[SyncFS] Refine SyncProcessRunner's throttling algorithm for parallel task support

chrome/browser/sync_file_system/sync_process_runner.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "chrome/browser/sync_file_system/sync_process_runner.h"
 
 #include "base/format_macros.h"
 #include "chrome/browser/sync_file_system/logger.h"
 
 namespace sync_file_system {
@@ skipping 51 matching lines @@
 SyncProcessRunner::SyncProcessRunner(
     const std::string& name,
     Client* client,
     scoped_ptr<TimerHelper> timer_helper,
     int max_parallel_task)
     : name_(name),
       client_(client),
       max_parallel_task_(max_parallel_task),
       running_tasks_(0),
       timer_helper_(timer_helper.Pass()),
-      current_delay_(0),
-      last_delay_(0),
       pending_changes_(0),
       factory_(this) {
   DCHECK_LE(1, max_parallel_task_);
 
   DCHECK_EQ(1, max_parallel_task_)
       << "Parellel task execution is not yet implemented.";
   if (!timer_helper_)
     timer_helper_.reset(new BaseTimerHelper);
 }
 
 SyncProcessRunner::~SyncProcessRunner() {}
 
 void SyncProcessRunner::Schedule() {
+  if (running_tasks_ >= max_parallel_task_)
+    return;
+
   int64 delay = kSyncDelayInMilliseconds;
   if (pending_changes_ == 0) {
     ScheduleInternal(kSyncDelayMaxInMilliseconds);
     return;
   }
   switch (GetServiceState()) {
     case SYNC_SERVICE_RUNNING:
+      ResetThrottling();
       if (pending_changes_ > kPendingChangeThresholdForFastSync)
-        delay = kSyncDelayFastInMilliseconds;
+        ScheduleInternal(kSyncDelayFastInMilliseconds);
       else
-        delay = kSyncDelayInMilliseconds;
-      break;
+        ScheduleInternal(kSyncDelayInMilliseconds);
+      return;
 
     case SYNC_SERVICE_TEMPORARY_UNAVAILABLE:
-      delay = kSyncDelaySlowInMilliseconds;
-      if (last_delay_ >= kSyncDelaySlowInMilliseconds)
-        delay = last_delay_ * 2;
-      if (delay >= kSyncDelayMaxInMilliseconds)
-        delay = kSyncDelayMaxInMilliseconds;
-      break;
+      ThrottleSync(kSyncDelaySlowInMilliseconds);
+      ScheduleInternal(kSyncDelaySlowInMilliseconds);
+      return;
 
     case SYNC_SERVICE_AUTHENTICATION_REQUIRED:
     case SYNC_SERVICE_DISABLED:
-      delay = kSyncDelayMaxInMilliseconds;
-      break;
+      ThrottleSync(kSyncDelaySlowInMilliseconds);
+      ScheduleInternal(kSyncDelayMaxInMilliseconds);
+      return;
   }
-  ScheduleInternal(delay);
+
+  NOTREACHED();
+  ScheduleInternal(kSyncDelayMaxInMilliseconds);
 }
 
-void SyncProcessRunner::ScheduleIfNotRunning() {
-  if (!timer_helper_->IsRunning())
-    Schedule();
+void SyncProcessRunner::ThrottleSync(int64 base_delay) {
+  base::TimeTicks now = timer_helper_->Now();
+  base::TimeDelta elapsed = std::min(now, throttle_until_) - throttle_from_;
+  DCHECK(base::TimeDelta() <= elapsed);
+
+  throttle_from_ = now;
+  // Extend throttling duration by twice the elapsed time.
+  // That is, if the backoff repeats in a short period, the throttling period
+  // doesn't grow exponentially.  If the backoff happens on the end of
+  // throttling period, it causes another throttling period that is twice as
+  // long as previous.
+  base::TimeDelta base_delay_delta =
+      base::TimeDelta::FromMilliseconds(base_delay);
+  const base::TimeDelta max_delay =
+      base::TimeDelta::FromMilliseconds(kSyncDelayMaxInMilliseconds);
+  throttle_until_ =
+      std::min(now + max_delay,
+               std::max(now + base_delay_delta, throttle_until_ + 2 * elapsed));
+}
+
+void SyncProcessRunner::ResetOldThrottling() {

[nhiroki, 2014/07/10 08:17:21]

This is not used in this CL.

[tzik, 2014/07/11 06:00:48]

I forgot to call this. Now it's used in Finished().

+  if (throttle_until_ < base::TimeTicks::Now())
+    ResetThrottling();
+}
+
+void SyncProcessRunner::ResetThrottling() {
+  throttle_from_ = base::TimeTicks();
+  throttle_until_ = base::TimeTicks();
 }
 
 void SyncProcessRunner::OnChangesUpdated(
     int64 pending_changes) {
   DCHECK_GE(pending_changes, 0);
   int64 old_pending_changes = pending_changes_;
   pending_changes_ = pending_changes;
   if (old_pending_changes != pending_changes) {
     if (pending_changes == 0)
       client_->OnSyncIdle();
@@ skipping 11 matching lines @@
 SyncServiceState SyncProcessRunner::GetServiceState() {
   return client_->GetSyncServiceState();
 }
 
 void SyncProcessRunner::Finished(const base::TimeTicks& start_time,
                                  SyncStatusCode status) {
   DCHECK_LT(0, running_tasks_);
   DCHECK_LE(running_tasks_, max_parallel_task_);
   --running_tasks_;
   util::Log(logging::LOG_VERBOSE, FROM_HERE,
-            "[%s] * Finished (elapsed: %" PRId64 " sec)",
-            name_.c_str(),
-            (timer_helper_->Now() - start_time).InSeconds());
+            "[%s] * Finished (elapsed: %" PRId64 " ms)", name_.c_str(),
+            (timer_helper_->Now() - start_time).InMilliseconds());
+
   if (status == SYNC_STATUS_NO_CHANGE_TO_SYNC ||
-      status == SYNC_STATUS_FILE_BUSY)
+      status == SYNC_STATUS_FILE_BUSY) {
     ScheduleInternal(kSyncDelayMaxInMilliseconds);
-  else if (!WasSuccessfulSync(status) &&
-           GetServiceState() == SYNC_SERVICE_RUNNING)
-    ScheduleInternal(kSyncDelayWithSyncError);
-  else
-    Schedule();
+    return;
+  }
+
+  if (!WasSuccessfulSync(status) &&
+      GetServiceState() == SYNC_SERVICE_RUNNING) {
+    ThrottleSync(kSyncDelayWithSyncError);
+  }
+
+  Schedule();
 }
 
 void SyncProcessRunner::Run() {
   if (running_tasks_ >= max_parallel_task_)
     return;
   ++running_tasks_;
   last_scheduled_ = timer_helper_->Now();
   last_delay_ = current_delay_;
 
   util::Log(logging::LOG_VERBOSE, FROM_HERE,
             "[%s] * Started", name_.c_str());
 
   StartSync(base::Bind(&SyncProcessRunner::Finished, factory_.GetWeakPtr(),
                        last_scheduled_));
+  Schedule();
 }
 
 void SyncProcessRunner::ScheduleInternal(int64 delay) {
   base::TimeDelta time_to_next = base::TimeDelta::FromMilliseconds(delay);
+  base::TimeTicks now = timer_helper_->Now();
 
   if (timer_helper_->IsRunning()) {
     if (current_delay_ == delay)
       return;
 
-    base::TimeDelta elapsed = timer_helper_->Now() - last_scheduled_;
+    base::TimeDelta elapsed = now - last_scheduled_;
     if (elapsed < time_to_next) {
       time_to_next = time_to_next - elapsed;
     } else {
       time_to_next = base::TimeDelta::FromMilliseconds(
           kSyncDelayFastInMilliseconds);
     }
   }
 
+  if (now + time_to_next < throttle_until_)
+    time_to_next = throttle_until_ - now;
+
   if (current_delay_ != delay) {
     util::Log(logging::LOG_VERBOSE, FROM_HERE,
               "[%s] Scheduling task in %" PRId64 " secs",
               name_.c_str(), time_to_next.InSeconds());
   }
   current_delay_ = delay;
 
   timer_helper_->Start(
       FROM_HERE, time_to_next,
       base::Bind(&SyncProcessRunner::Run, base::Unretained(this)));
 }
 
 }  // namespace sync_file_system