2nd try: Add OnSyncEnabled/OnWriteEnabled notification handling to operation runner

Original review: https://chromiumcodereview.appspot.com/11238054
1st submit (http://crrev.com/163584) got reverted because it broke win_aura/win7 bots.

* Slightly changing ownership relationship: now LocalFileSyncContext owns both SyncableFileOperationRunner and LocalFileSyncStatus (previously SyncContext used to own OperationRunner that owns SyncStatus)
* Adding OnSyncEnabled, OnWriteEnabled notifications to the LocalFileSyncStatus
* Let SyncableFileOperation listen to OnWriteEnabled notification so that it can automatically start write operations (as far as the # of operations is less than the given max)

BUG=148897, 154234
TEST=SyncableFileOperationRunnerTest.*

Review URL: https://codereview.chromium.org/11271003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@163812 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 120 insertions, 43 deletions

diff --git a/webkit/fileapi/syncable/local_file_sync_context.cc b/webkit/fileapi/syncable/local_file_sync_context.cc
index a325829..531e67f 100644
--- a/webkit/fileapi/syncable/local_file_sync_context.cc
+++ b/webkit/fileapi/syncable/local_file_sync_context.cc
@@ -17,6 +17,10 @@
 
 namespace fileapi {
 
+namespace {
+const int kMaxConcurrentSyncableOperation = 3;
+}  // namespace
+
 LocalFileSyncContext::LocalFileSyncContext(
     base::SingleThreadTaskRunner* ui_task_runner,
     base::SingleThreadTaskRunner* io_task_runner)
@@ -68,12 +72,18 @@ LocalFileSyncContext::operation_runner() const {
   return base::WeakPtr<SyncableFileOperationRunner>();
 }
 
+LocalFileSyncStatus* LocalFileSyncContext::sync_status() const {
+  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  return sync_status_.get();
+}
+
 LocalFileSyncContext::~LocalFileSyncContext() {
 }
 
 void LocalFileSyncContext::ShutdownOnIOThread() {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
   operation_runner_.reset();
+  sync_status_.reset();
 }
 
 void LocalFileSyncContext::InitializeFileSystemContextOnIOThread(
@@ -98,8 +108,13 @@ void LocalFileSyncContext::InitializeFileSystemContextOnIOThread(
                    make_scoped_refptr(file_system_context)));
     return;
   }
-  if (!operation_runner_.get())
-    operation_runner_.reset(new SyncableFileOperationRunner);
+  if (!operation_runner_.get()) {
+    DCHECK(!sync_status_.get());
+    sync_status_.reset(new LocalFileSyncStatus);
+    operation_runner_.reset(new SyncableFileOperationRunner(
+            kMaxConcurrentSyncableOperation,
+            sync_status_.get()));
+  }
   file_system_context->set_sync_context(this);
   DidInitialize(source_url, file_system_context, SYNC_STATUS_OK);
 }
diff --git a/webkit/fileapi/syncable/local_file_sync_context.h b/webkit/fileapi/syncable/local_file_sync_context.h
index c02d81a..74b4a90 100644
--- a/webkit/fileapi/syncable/local_file_sync_context.h
+++ b/webkit/fileapi/syncable/local_file_sync_context.h
@@ -18,6 +18,7 @@
 #include "base/memory/weak_ptr.h"
 #include "googleurl/src/gurl.h"
 #include "webkit/fileapi/syncable/file_change.h"
+#include "webkit/fileapi/syncable/local_file_sync_status.h"
 #include "webkit/fileapi/syncable/sync_status_code.h"
 #include "webkit/storage/webkit_storage_export.h"
 
@@ -27,6 +28,7 @@ class SingleThreadTaskRunner;
 
 namespace fileapi {
 
+class FileChange;
 class FileSystemContext;
 class LocalFileChangeTracker;
 class SyncableFileOperationRunner;
@@ -59,12 +61,15 @@ class WEBKIT_STORAGE_EXPORT LocalFileSyncContext
   // OperationRunner is accessible only on IO thread.
   base::WeakPtr<SyncableFileOperationRunner> operation_runner() const;
 
+  // SyncContext is accessible only on IO thread.
+  LocalFileSyncStatus* sync_status() const;
+
  private:
   typedef std::deque<StatusCallback> StatusCallbackQueue;
   friend class base::RefCountedThreadSafe<LocalFileSyncContext>;
   friend class CannedSyncableFileSystem;
 
-  ~LocalFileSyncContext();
+  virtual ~LocalFileSyncContext();
 
   void ShutdownOnIOThread();
 
@@ -91,16 +96,22 @@ class WEBKIT_STORAGE_EXPORT LocalFileSyncContext
   // OperationRunner. This must be accessed only on IO thread.
   scoped_ptr<SyncableFileOperationRunner> operation_runner_;
 
+  // Keeps track of writing/syncing status.
+  // This must be accessed only on IO thread.
+  scoped_ptr<LocalFileSyncStatus> sync_status_;
+
   // Pointers to file system contexts that have been initialized for
   // synchronization (i.e. that own this instance).
   // This must be accessed only on UI thread.
   std::set<FileSystemContext*> file_system_contexts_;
 
+  // Accessed only on UI thread.
   std::map<FileSystemContext*, StatusCallbackQueue>
       pending_initialize_callbacks_;
 
   // Origin to context map. (Assuming that as far as we're in the same
   // profile single origin wouldn't belong to multiple FileSystemContexts.)
+  // Accessed only on UI thread.
   std::map<GURL, FileSystemContext*> origin_to_contexts_;
 
   DISALLOW_COPY_AND_ASSIGN(LocalFileSyncContext);
diff --git a/webkit/fileapi/syncable/local_file_sync_status.cc b/webkit/fileapi/syncable/local_file_sync_status.cc
index 3be00e3..e543306 100644
--- a/webkit/fileapi/syncable/local_file_sync_status.cc
+++ b/webkit/fileapi/syncable/local_file_sync_status.cc
@@ -21,8 +21,10 @@ void LocalFileSyncStatus::StartWriting(const FileSystemURL& url) {
 void LocalFileSyncStatus::EndWriting(const FileSystemURL& url) {
   DCHECK(CalledOnValidThread());
   int count = --writing_[url];
-  if (count == 0)
+  if (count == 0) {
     writing_.erase(url);
+    FOR_EACH_OBSERVER(Observer, observer_list_, OnSyncEnabled(url));
+  }
 }
 
 void LocalFileSyncStatus::StartSyncing(const FileSystemURL& url) {
@@ -34,6 +36,7 @@ void LocalFileSyncStatus::StartSyncing(const FileSystemURL& url) {
 void LocalFileSyncStatus::EndSyncing(const FileSystemURL& url) {
   DCHECK(CalledOnValidThread());
   syncing_.erase(url);
+  FOR_EACH_OBSERVER(Observer, observer_list_, OnWriteEnabled(url));
 }
 
 bool LocalFileSyncStatus::IsWriting(const FileSystemURL& url) const {
@@ -46,6 +49,16 @@ bool LocalFileSyncStatus::IsWritable(const FileSystemURL& url) const {
   return !IsChildOrParentSyncing(url);
 }
 
+void LocalFileSyncStatus::AddObserver(Observer* observer) {
+  DCHECK(CalledOnValidThread());
+  observer_list_.AddObserver(observer);
+}
+
+void LocalFileSyncStatus::RemoveObserver(Observer* observer) {
+  DCHECK(CalledOnValidThread());
+  observer_list_.RemoveObserver(observer);
+}
+
 bool LocalFileSyncStatus::IsChildOrParentWriting(
     const FileSystemURL& url) const {
   DCHECK(CalledOnValidThread());
diff --git a/webkit/fileapi/syncable/local_file_sync_status.h b/webkit/fileapi/syncable/local_file_sync_status.h
index f99b84e..f0ce581 100644
--- a/webkit/fileapi/syncable/local_file_sync_status.h
+++ b/webkit/fileapi/syncable/local_file_sync_status.h
@@ -10,6 +10,7 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/observer_list.h"
 #include "base/threading/non_thread_safe.h"
 #include "webkit/fileapi/file_system_url.h"
 
@@ -27,6 +28,15 @@ namespace fileapi {
 // while the target url is in syncing must fail and vice versa.
 class WEBKIT_STORAGE_EXPORT LocalFileSyncStatus : public base::NonThreadSafe {
  public:
+  class WEBKIT_STORAGE_EXPORT Observer {
+   public:
+    Observer() {}
+    virtual void OnSyncEnabled(const FileSystemURL& url) = 0;
+    virtual void OnWriteEnabled(const FileSystemURL& url) = 0;
+   private:
+    DISALLOW_COPY_AND_ASSIGN(Observer);
+  };
+
   LocalFileSyncStatus();
   ~LocalFileSyncStatus();
 
@@ -50,6 +60,9 @@ class WEBKIT_STORAGE_EXPORT LocalFileSyncStatus : public base::NonThreadSafe {
   // Returns true if the |url| is enabled for writing (i.e. not in syncing).
   bool IsWritable(const FileSystemURL& url) const;
 
+  void AddObserver(Observer* observer);
+  void RemoveObserver(Observer* observer);
+
  private:
   typedef std::map<FileSystemURL, int64, FileSystemURL::Comparator> URLCountMap;
   typedef std::set<FileSystemURL, FileSystemURL::Comparator> URLSet;
@@ -63,6 +76,8 @@ class WEBKIT_STORAGE_EXPORT LocalFileSyncStatus : public base::NonThreadSafe {
   // If this flag is set sync process is running on the file.
   URLSet syncing_;
 
+  ObserverList<Observer> observer_list_;
+
   DISALLOW_COPY_AND_ASSIGN(LocalFileSyncStatus);
 };
 
diff --git a/webkit/fileapi/syncable/syncable_file_operation_runner.cc b/webkit/fileapi/syncable/syncable_file_operation_runner.cc
index ef9c899..4389a4d 100644
--- a/webkit/fileapi/syncable/syncable_file_operation_runner.cc
+++ b/webkit/fileapi/syncable/syncable_file_operation_runner.cc
@@ -41,38 +41,56 @@ void SyncableFileOperationRunner::Task::Start(LocalFileSyncStatus* status) {
 
 // SyncableFileOperationRunner -------------------------------------------------
 
-SyncableFileOperationRunner::SyncableFileOperationRunner()
-    : sync_status_(new LocalFileSyncStatus) {
+SyncableFileOperationRunner::SyncableFileOperationRunner(
+    int64 max_inflight_tasks,
+    LocalFileSyncStatus* sync_status)
+    : sync_status_(sync_status),
+      max_inflight_tasks_(max_inflight_tasks),
+      num_inflight_tasks_(0) {
   DCHECK(CalledOnValidThread());
+  sync_status_->AddObserver(this);
 }
 
 SyncableFileOperationRunner::~SyncableFileOperationRunner() {
   DCHECK(CalledOnValidThread());
-  for_each(pending_operations_.begin(), pending_operations_.end(),
+  for_each(pending_tasks_.begin(), pending_tasks_.end(),
            SyncableFileOperationRunner::Task::CancelAndDelete);
 }
 
+void SyncableFileOperationRunner::OnSyncEnabled(const FileSystemURL& url) {
+}
+
+void SyncableFileOperationRunner::OnWriteEnabled(const FileSystemURL& url) {
+  DCHECK(CalledOnValidThread());
+  RunNextRunnableTask();
+}
+
 void SyncableFileOperationRunner::PostOperationTask(scoped_ptr<Task> task) {
   DCHECK(CalledOnValidThread());
-  pending_operations_.push_back(task.release());
+  pending_tasks_.push_back(task.release());
   RunNextRunnableTask();
 }
 
 void SyncableFileOperationRunner::RunNextRunnableTask() {
   DCHECK(CalledOnValidThread());
-  for (std::list<Task*>::iterator iter = pending_operations_.begin();
-       iter != pending_operations_.end(); ++iter) {
+  for (std::list<Task*>::iterator iter = pending_tasks_.begin();
+       iter != pending_tasks_.end() && ShouldStartMoreTasks();) {
     if ((*iter)->IsRunnable(sync_status())) {
+      ++num_inflight_tasks_;
+      DCHECK_GE(num_inflight_tasks_, 1);
       scoped_ptr<Task> task(*iter);
-      pending_operations_.erase(iter);
+      pending_tasks_.erase(iter++);
       task->Start(sync_status());
-      return;
+      continue;
     }
+    ++iter;
   }
 }
 
 void SyncableFileOperationRunner::OnOperationCompleted(
     const std::vector<FileSystemURL>& target_paths) {
+  --num_inflight_tasks_;
+  DCHECK_GE(num_inflight_tasks_, 0);
   for (size_t i = 0; i < target_paths.size(); ++i) {
     DCHECK(sync_status()->IsWriting(target_paths[i]));
     sync_status()->EndWriting(target_paths[i]);
@@ -80,4 +98,8 @@ void SyncableFileOperationRunner::OnOperationCompleted(
   RunNextRunnableTask();
 }
 
+bool SyncableFileOperationRunner::ShouldStartMoreTasks() const {
+  return num_inflight_tasks_ < max_inflight_tasks_;
+}
+
 }  // namespace fileapi
diff --git a/webkit/fileapi/syncable/syncable_file_operation_runner.h b/webkit/fileapi/syncable/syncable_file_operation_runner.h
index 95cf5c6..cd63138 100644
--- a/webkit/fileapi/syncable/syncable_file_operation_runner.h
+++ b/webkit/fileapi/syncable/syncable_file_operation_runner.h
@@ -14,18 +14,19 @@
 #include "base/memory/weak_ptr.h"
 #include "base/threading/non_thread_safe.h"
 #include "webkit/fileapi/file_system_url.h"
+#include "webkit/fileapi/syncable/local_file_sync_status.h"
 #include "webkit/storage/webkit_storage_export.h"
 
 namespace fileapi {
 
 class FileSystemURL;
-class LocalFileSyncStatus;
 
 // This class must run only on IO thread.
 // Owned by LocalFileSyncContext.
 class WEBKIT_STORAGE_EXPORT SyncableFileOperationRunner
     : public base::NonThreadSafe,
-      public base::SupportsWeakPtr<SyncableFileOperationRunner> {
+      public base::SupportsWeakPtr<SyncableFileOperationRunner>,
+      public LocalFileSyncStatus::Observer {
  public:
   // Represents an operation task (which usually wraps one FileSystemOperation).
   class Task {
@@ -50,15 +51,20 @@ class WEBKIT_STORAGE_EXPORT SyncableFileOperationRunner
     DISALLOW_COPY_AND_ASSIGN(Task);
   };
 
-  SyncableFileOperationRunner();
-  ~SyncableFileOperationRunner();
+  SyncableFileOperationRunner(int64 max_inflight_tasks,
+                              LocalFileSyncStatus* sync_status);
+  virtual ~SyncableFileOperationRunner();
+
+  // LocalFileSyncStatus::Observer overrides.
+  virtual void OnSyncEnabled(const FileSystemURL& url) OVERRIDE;
+  virtual void OnWriteEnabled(const FileSystemURL& url) OVERRIDE;
 
   // Runs the given |task| if no sync operation is running on any of
-  // its target_paths(). This also runs pending operations that have become
+  // its target_paths(). This also runs pending tasks that have become
   // runnable (before running the given operation).
-  // If there're ongoing sync operations on the target_paths this method
+  // If there're ongoing sync tasks on the target_paths this method
   // just queues up the |task|.
-  // Pending operations are cancelled when this class is destructed.
+  // Pending tasks are cancelled when this class is destructed.
   void PostOperationTask(scoped_ptr<Task> task);
 
   // Runs a next runnable task (if there's any).
@@ -68,14 +74,25 @@ class WEBKIT_STORAGE_EXPORT SyncableFileOperationRunner
   // writable and may start a next runnable task.
   void OnOperationCompleted(const std::vector<FileSystemURL>& target_paths);
 
-  // For syncable file systems.
-  LocalFileSyncStatus* sync_status() const { return sync_status_.get(); }
+  LocalFileSyncStatus* sync_status() const { return sync_status_; }
+
+  int64 num_pending_tasks() const {
+    return static_cast<int64>(pending_tasks_.size());
+  }
+
+  int64 num_inflight_tasks() const { return num_inflight_tasks_; }
 
  private:
-  // Keeps track of the writing/syncing status.
-  scoped_ptr<LocalFileSyncStatus> sync_status_;
+  // Returns true if we should start more tasks.
+  bool ShouldStartMoreTasks() const;
+
+  // Keeps track of the writing/syncing status. Not owned.
+  LocalFileSyncStatus* sync_status_;
+
+  std::list<Task*> pending_tasks_;
 
-  std::list<Task*> pending_operations_;
+  const int64 max_inflight_tasks_;
+  int64 num_inflight_tasks_;
 
   DISALLOW_COPY_AND_ASSIGN(SyncableFileOperationRunner);
 };
diff --git a/webkit/fileapi/syncable/syncable_file_operation_runner_unittest.cc b/webkit/fileapi/syncable/syncable_file_operation_runner_unittest.cc
index 2e7a160..f72cfbb 100644
--- a/webkit/fileapi/syncable/syncable_file_operation_runner_unittest.cc
+++ b/webkit/fileapi/syncable/syncable_file_operation_runner_unittest.cc
@@ -75,16 +75,8 @@ class SyncableFileOperationRunnerTest : public testing::Test {
     return file_system_.URL(path);
   }
 
-  LocalFileSyncContext* sync_context() {
-    return file_system_.file_system_context()->sync_context();
-  }
-
-  SyncableFileOperationRunner* operation_runner() {
-    return sync_context()->operation_runner().get();
-  }
-
   LocalFileSyncStatus* sync_status() {
-    return operation_runner()->sync_status();
+    return file_system_.file_system_context()->sync_context()->sync_status();
   }
 
   void ResetCallbackStatus() {
@@ -164,7 +156,6 @@ TEST_F(SyncableFileOperationRunnerTest, SimpleQueue) {
   ASSERT_TRUE(sync_status()->IsWritable(URL(kFile)));
 
   ResetCallbackStatus();
-  operation_runner()->RunNextRunnableTask();
   MessageLoop::current()->RunAllPending();
   EXPECT_EQ(2, callback_count_);
 
@@ -185,14 +176,11 @@ TEST_F(SyncableFileOperationRunnerTest, WriteToParentAndChild) {
   sync_status()->StartSyncing(URL(kDir));
   ASSERT_FALSE(sync_status()->IsWritable(URL(kDir)));
 
-  // Writes to kParent, kDir and kChild should be all queued up.
+  // Writes to kParent and kChild should be all queued up.
   ResetCallbackStatus();
   file_system_.NewOperation()->Truncate(
       URL(kChild), 1, ExpectStatus(FROM_HERE, base::PLATFORM_FILE_OK));
   file_system_.NewOperation()->Remove(
-      URL(kDir), true /* recursive */,
-      ExpectStatus(FROM_HERE, base::PLATFORM_FILE_OK));
-  file_system_.NewOperation()->Remove(
       URL(kParent), true /* recursive */,
       ExpectStatus(FROM_HERE, base::PLATFORM_FILE_OK));
   MessageLoop::current()->RunAllPending();
@@ -217,9 +205,8 @@ TEST_F(SyncableFileOperationRunnerTest, WriteToParentAndChild) {
   ASSERT_TRUE(sync_status()->IsWritable(URL(kDir)));
 
   ResetCallbackStatus();
-  operation_runner()->RunNextRunnableTask();
   MessageLoop::current()->RunAllPending();
-  EXPECT_EQ(3, callback_count_);
+  EXPECT_EQ(2, callback_count_);
 }
 
 TEST_F(SyncableFileOperationRunnerTest, CopyAndMove) {
@@ -262,7 +249,6 @@ TEST_F(SyncableFileOperationRunnerTest, CopyAndMove) {
   // Finish syncing the "dest-copy2" directory to unlock Copy.
   sync_status()->EndSyncing(URL("dest-copy2"));
   ResetCallbackStatus();
-  operation_runner()->RunNextRunnableTask();
   MessageLoop::current()->RunAllPending();
   EXPECT_EQ(1, callback_count_);
 
@@ -273,7 +259,6 @@ TEST_F(SyncableFileOperationRunnerTest, CopyAndMove) {
   // Finish syncing the kParent to unlock Move.
   sync_status()->EndSyncing(URL(kParent));
   ResetCallbackStatus();
-  operation_runner()->RunNextRunnableTask();
   MessageLoop::current()->RunAllPending();
   EXPECT_EQ(1, callback_count_);
 
@@ -299,7 +284,6 @@ TEST_F(SyncableFileOperationRunnerTest, Write) {
 
   sync_status()->EndSyncing(URL(kFile));
   ResetCallbackStatus();
-  operation_runner()->RunNextRunnableTask();
 
   while (!write_complete_)
     MessageLoop::current()->RunAllPending();