Worker changes to prepare for lock-free shutdown.

* Make worker observe the destruction of the thread where it does work. 
* Make work done signal a memeber of worker and set it when worker's working
  thread is destroyed so that syncer won't be blocked indefinitely on work
  that's not gonna run.
* Ref-count worker in session context so that worker remains valid when syncer
  is still iterating through workers but registrar has erased it from its worker map.
* Add RequestStop() and Stopped() to allow worker to return early when it's 
  stopped while waiting to run work on its working thread.

This cl by itself should have no real impact because the added functions are not used.

BUG=19757

Review URL: https://chromiumcodereview.appspot.com/14046031

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

7 files changed, 128 insertions, 16 deletions

diff --git a/sync/internal_api/public/engine/model_safe_worker.cc b/sync/internal_api/public/engine/model_safe_worker.cc
index d7bf906..3e2096a 100644
--- a/sync/internal_api/public/engine/model_safe_worker.cc
+++ b/sync/internal_api/public/engine/model_safe_worker.cc
@@ -80,6 +80,58 @@ std::string ModelSafeGroupToString(ModelSafeGroup group) {
   }
 }
 
+ModelSafeWorker::ModelSafeWorker(WorkerLoopDestructionObserver* observer)
+    : stopped_(false),
+      work_done_or_stopped_(false, false),
+      observer_(observer) {}
+
 ModelSafeWorker::~ModelSafeWorker() {}
 
+void ModelSafeWorker::RequestStop() {
+  base::AutoLock al(stopped_lock_);
+
+  // Set stop flag but don't signal work_done_or_stopped_ to unblock sync loop
+  // because the worker may be working and depending on sync command object
+  // living on sync thread. his prevents any *further* tasks from being posted
+  // to worker threads (see DoWorkAndWaitUntilDone below), but note that one
+  // may already be posted.
+  stopped_ = true;
+}
+
+SyncerError ModelSafeWorker::DoWorkAndWaitUntilDone(const WorkCallback& work) {
+  {
+    base::AutoLock al(stopped_lock_);
+    if (stopped_)
+      return CANNOT_DO_WORK;
+
+    CHECK(!work_done_or_stopped_.IsSignaled());
+  }
+
+  return DoWorkAndWaitUntilDoneImpl(work);
+}
+
+bool ModelSafeWorker::IsStopped() {
+  base::AutoLock al(stopped_lock_);
+  return stopped_;
+}
+
+void ModelSafeWorker::WillDestroyCurrentMessageLoop() {
+  {
+    base::AutoLock al(stopped_lock_);
+    stopped_ = true;
+
+    // Must signal to unblock syncer if it's waiting for a posted task to
+    // finish. At this point, all pending tasks posted to the loop have been
+    // destroyed (see MessageLoop::~MessageLoop). So syncer will be blocked
+    // indefinitely without signaling here.
+    work_done_or_stopped_.Signal();
+
+    DVLOG(1) << ModelSafeGroupToString(GetModelSafeGroup())
+        << " worker stops on destruction of its working thread.";
+  }
+
+  if (observer_)
+    observer_->OnWorkerLoopDestroyed(GetModelSafeGroup());
+}
+
 }  // namespace syncer
diff --git a/sync/internal_api/public/engine/model_safe_worker.h b/sync/internal_api/public/engine/model_safe_worker.h
index fbec6b1..94925df 100644
--- a/sync/internal_api/public/engine/model_safe_worker.h
+++ b/sync/internal_api/public/engine/model_safe_worker.h
@@ -11,6 +11,9 @@
 
 #include "base/callback.h"
 #include "base/memory/ref_counted.h"
+#include "base/message_loop.h"
+#include "base/synchronization/lock.h"
+#include "base/synchronization/waitable_event.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/base/model_type_invalidation_map.h"
@@ -44,28 +47,74 @@ enum ModelSafeGroup {
 
 SYNC_EXPORT std::string ModelSafeGroupToString(ModelSafeGroup group);
 
+// WorkerLoopDestructionObserver is notified when the thread where it works
+// is going to be destroyed.
+class WorkerLoopDestructionObserver {
+ public:
+  virtual void OnWorkerLoopDestroyed(ModelSafeGroup group) = 0;
+};
+
 // The Syncer uses a ModelSafeWorker for all tasks that could potentially
 // modify syncable entries (e.g under a WriteTransaction). The ModelSafeWorker
 // only knows how to do one thing, and that is take some work (in a fully
 // pre-bound callback) and have it performed (as in Run()) from a thread which
 // is guaranteed to be "model-safe", where "safe" refers to not allowing us to
 // cause an embedding application model to fall out of sync with the
-// syncable::Directory due to a race.
+// syncable::Directory due to a race. Each ModelSafeWorker is affiliated with
+// a thread and does actual work on that thread. On the destruction of that
+// thread, the affiliated worker is effectively disabled to do more
+// work and will notify its observer.
 class SYNC_EXPORT ModelSafeWorker
-    : public base::RefCountedThreadSafe<ModelSafeWorker> {
+    : public base::RefCountedThreadSafe<ModelSafeWorker>,
+      public MessageLoop::DestructionObserver {
  public:
-  // Any time the Syncer performs model modifications (e.g employing a
-  // WriteTransaction), it should be done by this method to ensure it is done
-  // from a model-safe thread.
-  virtual SyncerError DoWorkAndWaitUntilDone(const WorkCallback& work) = 0;
+  // Subclass should implement to observe destruction of the loop where
+  // it actually does work.
+  virtual void RegisterForLoopDestruction() = 0;
+
+  // If not stopped, call DoWorkAndWaitUntilDoneImpl() to do work. Otherwise
+  // return CANNOT_DO_WORK.
+  SyncerError DoWorkAndWaitUntilDone(const WorkCallback& work);
+
+  // Soft stop worker by setting stopped_ flag. Called when sync is disabled
+  // or browser is shutting down.
+  void RequestStop();
 
   virtual ModelSafeGroup GetModelSafeGroup() = 0;
 
+  // MessageLoop::DestructionObserver implementation.
+  virtual void WillDestroyCurrentMessageLoop() OVERRIDE;
+
  protected:
+  friend class base::RefCountedThreadSafe<ModelSafeWorker>;
+
+  explicit ModelSafeWorker(WorkerLoopDestructionObserver* observer);
   virtual ~ModelSafeWorker();
 
+  // Any time the Syncer performs model modifications (e.g employing a
+  // WriteTransaction), it should be done by this method to ensure it is done
+  // from a model-safe thread.
+  virtual SyncerError DoWorkAndWaitUntilDoneImpl(const WorkCallback& work) = 0;
+
+  base::WaitableEvent* work_done_or_stopped() {
+    return &work_done_or_stopped_;
+  }
+
+  // Return true if the worker was stopped. Thread safe.
+  bool IsStopped();
+
  private:
-  friend class base::RefCountedThreadSafe<ModelSafeWorker>;
+  // Whether the worker should/can do more work. Set when sync is disabled or
+  // when the worker's working thread is to be destroyed.
+  base::Lock stopped_lock_;
+  bool stopped_;
+
+  // Signal set when work on native thread is finished or when native thread
+  // is to be destroyed so no more work can be done.
+  base::WaitableEvent work_done_or_stopped_;
+
+  // Notified when working thread of the worker is to be destroyed.
+  WorkerLoopDestructionObserver* observer_;
 };
 
 // A map that details which ModelSafeGroup each ModelType
diff --git a/sync/internal_api/public/engine/passive_model_worker.cc b/sync/internal_api/public/engine/passive_model_worker.cc
index 50d00cf..8b1a4e3 100644
--- a/sync/internal_api/public/engine/passive_model_worker.cc
+++ b/sync/internal_api/public/engine/passive_model_worker.cc
@@ -8,13 +8,20 @@
 
 namespace syncer {
 
-PassiveModelWorker::PassiveModelWorker(const base::MessageLoop* sync_loop)
-    : sync_loop_(sync_loop) {}
+PassiveModelWorker::PassiveModelWorker(const base::MessageLoop* sync_loop,
+                                       WorkerLoopDestructionObserver* observer)
+    : ModelSafeWorker(observer),
+      sync_loop_(sync_loop) {
+}
 
 PassiveModelWorker::~PassiveModelWorker() {
 }
 
-SyncerError PassiveModelWorker::DoWorkAndWaitUntilDone(
+void PassiveModelWorker::RegisterForLoopDestruction() {
+  NOTREACHED();
+}
+
+SyncerError PassiveModelWorker::DoWorkAndWaitUntilDoneImpl(
     const WorkCallback& work) {
   DCHECK_EQ(base::MessageLoop::current(), sync_loop_);
   // Simply do the work on the current thread.
diff --git a/sync/internal_api/public/engine/passive_model_worker.h b/sync/internal_api/public/engine/passive_model_worker.h
index a6ea011..f834c83 100644
--- a/sync/internal_api/public/engine/passive_model_worker.h
+++ b/sync/internal_api/public/engine/passive_model_worker.h
@@ -22,13 +22,17 @@ namespace syncer {
 // thread).
 class SYNC_EXPORT PassiveModelWorker : public ModelSafeWorker {
  public:
-  explicit PassiveModelWorker(const base::MessageLoop* sync_loop);
+  explicit PassiveModelWorker(const base::MessageLoop* sync_loop,
+                              WorkerLoopDestructionObserver* observer);
 
   // ModelSafeWorker implementation. Called on the sync thread.
-  virtual SyncerError DoWorkAndWaitUntilDone(
-      const WorkCallback& work) OVERRIDE;
+  virtual void RegisterForLoopDestruction() OVERRIDE;
   virtual ModelSafeGroup GetModelSafeGroup() OVERRIDE;
 
+ protected:
+  virtual SyncerError DoWorkAndWaitUntilDoneImpl(
+      const WorkCallback& work) OVERRIDE;
+
  private:
   virtual ~PassiveModelWorker();
 
diff --git a/sync/internal_api/public/internal_components_factory.h b/sync/internal_api/public/internal_components_factory.h
index f49f88a..1d5b122 100644
--- a/sync/internal_api/public/internal_components_factory.h
+++ b/sync/internal_api/public/internal_components_factory.h
@@ -71,7 +71,7 @@ class SYNC_EXPORT InternalComponentsFactory {
   virtual scoped_ptr<sessions::SyncSessionContext> BuildContext(
       ServerConnectionManager* connection_manager,
       syncable::Directory* directory,
-      const std::vector<ModelSafeWorker*> workers,
+      const std::vector<ModelSafeWorker*>& workers,
       ExtensionsActivityMonitor* monitor,
       ThrottledDataTypeTracker* throttled_data_type_tracker,
       const std::vector<SyncEngineEventListener*>& listeners,
diff --git a/sync/internal_api/public/internal_components_factory_impl.h b/sync/internal_api/public/internal_components_factory_impl.h
index 197ce46..646da5f 100644
--- a/sync/internal_api/public/internal_components_factory_impl.h
+++ b/sync/internal_api/public/internal_components_factory_impl.h
@@ -26,7 +26,7 @@ class SYNC_EXPORT InternalComponentsFactoryImpl
   virtual scoped_ptr<sessions::SyncSessionContext> BuildContext(
       ServerConnectionManager* connection_manager,
       syncable::Directory* directory,
-      const std::vector<ModelSafeWorker*> workers,
+      const std::vector<ModelSafeWorker*>& workers,
       ExtensionsActivityMonitor* monitor,
       ThrottledDataTypeTracker* throttled_data_type_tracker,
       const std::vector<SyncEngineEventListener*>& listeners,
diff --git a/sync/internal_api/public/test/test_internal_components_factory.h b/sync/internal_api/public/test/test_internal_components_factory.h
index 9db26e2..665999d 100644
--- a/sync/internal_api/public/test/test_internal_components_factory.h
+++ b/sync/internal_api/public/test/test_internal_components_factory.h
@@ -32,7 +32,7 @@ class TestInternalComponentsFactory : public InternalComponentsFactory {
   virtual scoped_ptr<sessions::SyncSessionContext> BuildContext(
       ServerConnectionManager* connection_manager,
       syncable::Directory* directory,
-      const std::vector<ModelSafeWorker*> workers,
+      const std::vector<ModelSafeWorker*>& workers,
       ExtensionsActivityMonitor* monitor,
       ThrottledDataTypeTracker* throttled_data_type_tracker,
       const std::vector<SyncEngineEventListener*>& listeners,