[Sync] Split DirectoryChangeListener for thread-safety

chrome/browser/sync/syncable/syncable.cc

 // Copyright (c) 2011 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/syncable/syncable.h"
 
 #include "build/build_config.h"
 
 #include <sys/stat.h>
 #if defined(OS_POSIX)
@@ skipping 18 matching lines @@
 
 #include "base/hash_tables.h"
 #include "base/file_util.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/perftimer.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
 #include "base/stl_util-inl.h"
 #include "base/time.h"
+#include "base/tracked.h"
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
 #include "chrome/browser/sync/engine/syncer.h"
 #include "chrome/browser/sync/engine/syncer_util.h"
 #include "chrome/browser/sync/protocol/proto_value_conversions.h"
 #include "chrome/browser/sync/protocol/service_constants.h"
 #include "chrome/browser/sync/syncable/directory_backing_store.h"
-#include "chrome/browser/sync/syncable/directory_change_listener.h"
+#include "chrome/browser/sync/syncable/directory_change_delegate.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/syncable/syncable-inl.h"
 #include "chrome/browser/sync/syncable/syncable_changes_version.h"
 #include "chrome/browser/sync/syncable/syncable_columns.h"
 #include "chrome/browser/sync/syncable/syncable_enum_conversions.h"
+#include "chrome/browser/sync/syncable/transaction_observer.h"
+#include "chrome/browser/sync/util/logging.h"
 #include "chrome/common/deprecated/event_sys-inl.h"
 #include "net/base/escape.h"
 
 namespace {
 enum InvariantCheckLevel {
   OFF = 0,
   VERIFY_IN_MEMORY = 1,
   FULL_DB_VERIFICATION = 2
 };
 
@@ skipping 253 matching lines @@
   SetFieldValues(*this, kernel_info,
                  &GetBitTempString, &Value::CreateBooleanValue,
                  BIT_TEMPS_BEGIN, BIT_TEMPS_END - 1);
 
   return kernel_info;
 }
 
 ///////////////////////////////////////////////////////////////////////////
 // Directory
 
-void Directory::init_kernel(const std::string& name) {
+void Directory::InitKernel(const std::string& name,
+                           DirectoryChangeDelegate* delegate) {
   DCHECK(kernel_ == NULL);
-  kernel_ = new Kernel(FilePath(), name, KernelLoadInfo());
+  kernel_ = new Kernel(FilePath(), name, KernelLoadInfo(), delegate);
 }
 
 Directory::PersistedKernelInfo::PersistedKernelInfo()
     : next_id(0) {
   for (int i = FIRST_REAL_MODEL_TYPE; i < MODEL_TYPE_COUNT; ++i) {
     reset_download_progress(ModelTypeFromInt(i));
   }
   autofill_migration_state = NOT_DETERMINED;
   memset(&autofill_migration_debug_info, 0,
          sizeof(autofill_migration_debug_info));
@@ skipping 10 matching lines @@
 }
 
 Directory::SaveChangesSnapshot::SaveChangesSnapshot()
     : kernel_info_status(KERNEL_SHARE_INFO_INVALID) {
 }
 
 Directory::SaveChangesSnapshot::~SaveChangesSnapshot() {}
 
 Directory::Kernel::Kernel(const FilePath& db_path,
                           const string& name,
-                          const KernelLoadInfo& info)
+                          const KernelLoadInfo& info,
+                          DirectoryChangeDelegate* delegate)
     : db_path(db_path),
       refcount(1),
       name(name),
       metahandles_index(new Directory::MetahandlesIndex),
       ids_index(new Directory::IdsIndex),
       parent_id_child_index(new Directory::ParentIdChildIndex),
       client_tag_index(new Directory::ClientTagIndex),
       unapplied_update_metahandles(new MetahandleSet),
       unsynced_metahandles(new MetahandleSet),
       dirty_metahandles(new MetahandleSet),
       metahandles_to_purge(new MetahandleSet),
       channel(new Directory::Channel(syncable::DIRECTORY_DESTROYED)),
       info_status(Directory::KERNEL_SHARE_INFO_VALID),
       persisted_info(info.kernel_info),
       cache_guid(info.cache_guid),
-      next_metahandle(info.max_metahandle + 1) {
+      next_metahandle(info.max_metahandle + 1),
+      delegate(delegate),
+      observers(new ObserverListThreadSafe<TransactionObserver>()) {
+  DCHECK(delegate);
 }
 
 void Directory::Kernel::AddRef() {
   base::subtle::NoBarrier_AtomicIncrement(&refcount, 1);
 }
 
 void Directory::Kernel::Release() {
   if (!base::subtle::NoBarrier_AtomicIncrement(&refcount, -1))
     delete this;
 }
 
-void Directory::Kernel::AddChangeListener(
-    DirectoryChangeListener* listener) {
-  base::AutoLock lock(change_listeners_lock_);
-  change_listeners_.AddObserver(listener);
-}
-
-void Directory::Kernel::RemoveChangeListener(
-    DirectoryChangeListener* listener) {
-  base::AutoLock lock(change_listeners_lock_);
-  change_listeners_.RemoveObserver(listener);
-}
-
-// Note: it is possible that a listener will remove itself after we
-// have made a copy, but before the copy is consumed. This could
-// theoretically result in accessing a garbage pointer, but can only
-// occur when an about:sync window is closed in the middle of a
-// notification.  See crbug.com/85481.
-void Directory::Kernel::CopyChangeListeners(
-    ObserverList<DirectoryChangeListener>* change_listeners) {
-  DCHECK_EQ(0U, change_listeners->size());
-  base::AutoLock lock(change_listeners_lock_);
-  ObserverListBase<DirectoryChangeListener>::Iterator it(change_listeners_);
-  DirectoryChangeListener* obs;
-  while ((obs = it.GetNext()) != NULL)
-    change_listeners->AddObserver(obs);
-}
-
 Directory::Kernel::~Kernel() {
   CHECK_EQ(0, refcount);
   delete channel;
   delete unsynced_metahandles;
   delete unapplied_update_metahandles;
   delete dirty_metahandles;
   delete metahandles_to_purge;
   delete parent_id_child_index;
   delete client_tag_index;
   delete ids_index;
   STLDeleteElements(metahandles_index);
   delete metahandles_index;
 }
 
 Directory::Directory() : kernel_(NULL), store_(NULL) {
 }
 
 Directory::~Directory() {
   Close();
 }
 
-DirOpenResult Directory::Open(const FilePath& file_path, const string& name) {
-  const DirOpenResult result = OpenImpl(file_path, name);
+DirOpenResult Directory::Open(const FilePath& file_path, const string& name,
+                              DirectoryChangeDelegate* delegate) {
+  const DirOpenResult result = OpenImpl(file_path, name, delegate);
   if (OPENED != result)
     Close();
   return result;
 }
 
 void Directory::InitializeIndices() {
   MetahandlesIndex::iterator it = kernel_->metahandles_index->begin();
   for (; it != kernel_->metahandles_index->end(); ++it) {
     EntryKernel* entry = *it;
     InitializeIndexEntry<ParentIdAndHandleIndexer>(entry,
         kernel_->parent_id_child_index);
     InitializeIndexEntry<IdIndexer>(entry, kernel_->ids_index);
     InitializeIndexEntry<ClientTagIndexer>(entry, kernel_->client_tag_index);
     if (entry->ref(IS_UNSYNCED))
       kernel_->unsynced_metahandles->insert(entry->ref(META_HANDLE));
     if (entry->ref(IS_UNAPPLIED_UPDATE))
       kernel_->unapplied_update_metahandles->insert(entry->ref(META_HANDLE));
     DCHECK(!entry->is_dirty());
   }
 }
 
 DirectoryBackingStore* Directory::CreateBackingStore(
     const string& dir_name, const FilePath& backing_filepath) {
   return new DirectoryBackingStore(dir_name, backing_filepath);
 }
 
 DirOpenResult Directory::OpenImpl(const FilePath& file_path,
-                                  const string& name) {
+                                  const string& name,
+                                  DirectoryChangeDelegate* delegate) {
   DCHECK_EQ(static_cast<DirectoryBackingStore*>(NULL), store_);
   FilePath db_path(file_path);
   file_util::AbsolutePath(&db_path);
   store_ = CreateBackingStore(name, db_path);
 
   KernelLoadInfo info;
   // Temporary indices before kernel_ initialized in case Load fails. We 0(1)
   // swap these later.
   MetahandlesIndex metas_bucket;
   DirOpenResult result = store_->Load(&metas_bucket, &info);
   if (OPENED != result)
     return result;
 
-  kernel_ = new Kernel(db_path, name, info);
+  kernel_ = new Kernel(db_path, name, info, delegate);
   kernel_->metahandles_index->swap(metas_bucket);
   InitializeIndices();
   return OPENED;
 }
 
 void Directory::Close() {
   if (store_)
     delete store_;
   store_ = NULL;
   if (kernel_) {
@@ skipping 175 matching lines @@
     CHECK_EQ(kernel_->dirty_metahandles->count(handle), 0U);
     // TODO(tim): Bug 49278.
     CHECK(!kernel_->unsynced_metahandles->count(handle));
     CHECK(!kernel_->unapplied_update_metahandles->count(handle));
   }
 
   return safe;
 }
 
 void Directory::TakeSnapshotForSaveChanges(SaveChangesSnapshot* snapshot) {
-  ReadTransaction trans(this, __FILE__, __LINE__);
+  ReadTransaction trans(this, FROM_HERE);
   ScopedKernelLock lock(this);
   // Deep copy dirty entries from kernel_->metahandles_index into snapshot and
   // clear dirty flags.
 
   for (MetahandleSet::const_iterator i = kernel_->dirty_metahandles->begin();
        i != kernel_->dirty_metahandles->end(); ++i) {
     EntryKernel* entry = GetEntryByHandle(*i, &lock);
     if (!entry)
       continue;
     // Skip over false positives; it happens relatively infrequently.
@@ skipping 37 matching lines @@
   // Handle success or failure.
   if (success)
     VacuumAfterSaveChanges(snapshot);
   else
     HandleSaveChangesFailure(snapshot);
   return success;
 }
 
 void Directory::VacuumAfterSaveChanges(const SaveChangesSnapshot& snapshot) {
   // Need a write transaction as we are about to permanently purge entries.
-  WriteTransaction trans(this, VACUUM_AFTER_SAVE, __FILE__, __LINE__);
+  WriteTransaction trans(this, VACUUM_AFTER_SAVE, FROM_HERE);
   ScopedKernelLock lock(this);
   kernel_->flushed_metahandles.Push(0);  // Begin flush marker
   // Now drop everything we can out of memory.
   for (OriginalEntries::const_iterator i = snapshot.dirty_metas.begin();
        i != snapshot.dirty_metas.end(); ++i) {
     kernel_->needle.put(META_HANDLE, i->ref(META_HANDLE));
     MetahandlesIndex::iterator found =
         kernel_->metahandles_index->find(&kernel_->needle);
     EntryKernel* entry = (found == kernel_->metahandles_index->end() ?
                           NULL : *found);
@@ skipping 20 matching lines @@
 void Directory::PurgeEntriesWithTypeIn(const std::set<ModelType>& types) {
   if (types.count(UNSPECIFIED) != 0U || types.count(TOP_LEVEL_FOLDER) != 0U) {
     NOTREACHED() << "Don't support purging unspecified or top level entries.";
     return;
   }
 
   if (types.empty())
     return;
 
   {
-    WriteTransaction trans(this, PURGE_ENTRIES, __FILE__, __LINE__);
+    WriteTransaction trans(this, PURGE_ENTRIES, FROM_HERE);
     {
       ScopedKernelLock lock(this);
       MetahandlesIndex::iterator it = kernel_->metahandles_index->begin();
       while (it != kernel_->metahandles_index->end()) {
         const sync_pb::EntitySpecifics& local_specifics = (*it)->ref(SPECIFICS);
         const sync_pb::EntitySpecifics& server_specifics =
             (*it)->ref(SERVER_SPECIFICS);
         ModelType local_type = GetModelTypeFromSpecifics(local_specifics);
         ModelType server_type = GetModelTypeFromSpecifics(server_specifics);
 
@@ skipping 395 matching lines @@
     int64 elapsed_ms = check_timer.Elapsed().InMilliseconds();
     if (elapsed_ms > max_ms) {
       VLOG(1) << "Cutting Invariant check short after " << elapsed_ms
               << "ms. Processed " << entries_done << "/" << handles.size()
               << " entries";
       return;
     }
   }
 }
 
-void Directory::AddChangeListener(DirectoryChangeListener* listener) {
-  kernel_->AddChangeListener(listener);
+void Directory::AddTransactionObserver(TransactionObserver* observer) {
+  kernel_->observers->AddObserver(observer);
 }
 
-void Directory::RemoveChangeListener(DirectoryChangeListener* listener) {
-  kernel_->RemoveChangeListener(listener);
+void Directory::RemoveTransactionObserver(TransactionObserver* observer) {
+  kernel_->observers->RemoveObserver(observer);
 }
 
 ///////////////////////////////////////////////////////////////////////////////
 // ScopedKernelLock
 
 ScopedKernelLock::ScopedKernelLock(const Directory* dir)
   :  scoped_lock_(dir->kernel_->mutex), dir_(const_cast<Directory*>(dir)) {
 }
 
 ///////////////////////////////////////////////////////////////////////////
 // Transactions
 
-// Helper functions/macros to do logging with a source file/line.
-
-namespace {
-
-bool VlogIsOn(const char* source_file, int verbose_level) {
-  return (verbose_level <=
-          logging::GetVlogLevelHelper(source_file, ::strlen(source_file)));
-}
-
-}  // namespace
-
-#define VLOG_SRC_STREAM(source_file, line, verbose_level)               \
-  logging::LogMessage(source_file, line, -verbose_level).stream()
-
-#define VLOG_SRC(source_file, line, verbose_level)                      \
-  LAZY_STREAM(VLOG_SRC_STREAM(source_file, line, verbose_level),        \
-              VLOG_IS_ON(verbose_level) ||                              \
-              VlogIsOn(source_file, verbose_level))
-
 void BaseTransaction::Lock() {
   base::TimeTicks start_time = base::TimeTicks::Now();
 
   dirkernel_->transaction_mutex.Acquire();
 
   time_acquired_ = base::TimeTicks::Now();
   const base::TimeDelta elapsed = time_acquired_ - start_time;
-  VLOG_SRC(source_file_, line_, 1)
+  VLOG_LOC(from_here_, 1)
       << name_ << " transaction waited "
       << elapsed.InSecondsF() << " seconds.";
 }
 
-BaseTransaction::BaseTransaction(Directory* directory, const char* name,
-    const char* source_file, int line, WriterTag writer)
+BaseTransaction::BaseTransaction(
+    Directory* directory, const char* name,
+    const tracked_objects::Location& from_here, WriterTag writer)
   : directory_(directory), dirkernel_(directory->kernel_), name_(name),
-    source_file_(source_file), line_(line), writer_(writer) {
+    from_here_(from_here), writer_(writer) {
+  dirkernel_->observers->Notify(
+      &TransactionObserver::OnTransactionStart, from_here_, writer_);
   Lock();
 }
 
-BaseTransaction::BaseTransaction(Directory* directory)
-    : directory_(directory),
-      dirkernel_(NULL),
-      name_(NULL),
-      source_file_(NULL),
-      line_(0),
-      writer_(INVALID) {
+BaseTransaction::~BaseTransaction() {
+  dirkernel_->observers->Notify(
+      &TransactionObserver::OnTransactionEnd, from_here_, writer_);
 }
 
-BaseTransaction::~BaseTransaction() {}
-
 void BaseTransaction::UnlockAndLog(OriginalEntries* entries) {
-  // Work while trasnaction mutex is held
+  // Work while transaction mutex is held
   ModelTypeBitSet models_with_changes;
   if (!NotifyTransactionChangingAndEnding(entries, &models_with_changes))
     return;
 
   // Work after mutex is relased.
   NotifyTransactionComplete(models_with_changes);
 }
 
 bool BaseTransaction::NotifyTransactionChangingAndEnding(
     OriginalEntries* entries,
     ModelTypeBitSet* models_with_changes) {
   dirkernel_->transaction_mutex.AssertAcquired();
 
   scoped_ptr<OriginalEntries> originals(entries);
   const base::TimeDelta elapsed = base::TimeTicks::Now() - time_acquired_;
-  VLOG_SRC(source_file_, line_, 1)
+  VLOG_LOC(from_here_, 1)
         << name_ << " transaction completed in " << elapsed.InSecondsF()
         << " seconds.";
 
-  ObserverList<DirectoryChangeListener> change_listeners;
-  dirkernel_->CopyChangeListeners(&change_listeners);
-
-  if (NULL == originals.get() || originals->empty() ||
-      (change_listeners.size() == 0)) {
+  if (NULL == originals.get() || originals->empty()) {
     dirkernel_->transaction_mutex.Release();
     return false;
   }
 
+  DirectoryChangeDelegate* const delegate = dirkernel_->delegate;
   if (writer_ == syncable::SYNCAPI) {
-    FOR_EACH_OBSERVER(DirectoryChangeListener,
-                      change_listeners,
-                      HandleCalculateChangesChangeEventFromSyncApi(
-                          *originals.get(),
-                          writer_,
-                          this));
+    delegate->HandleCalculateChangesChangeEventFromSyncApi(*originals, this);
   } else {
-    FOR_EACH_OBSERVER(DirectoryChangeListener,
-                      change_listeners,
-                      HandleCalculateChangesChangeEventFromSyncer(
-                          *originals.get(),
-                          writer_,
-                          this));
+    delegate->HandleCalculateChangesChangeEventFromSyncer(*originals, this);
   }
 
-  // Set |*models_with_changes| to the union of the return values of
-  // the HandleTransactionEndingChangeEvent call to each
-  // DirectoryChangeListener.
-  {
-    ObserverList<DirectoryChangeListener>::Iterator it(change_listeners);
-    DirectoryChangeListener* obs = NULL;
-    while ((obs = it.GetNext()) != NULL) {
-      *models_with_changes |= obs->HandleTransactionEndingChangeEvent(this);
-    }
-  };
+  *models_with_changes = delegate->HandleTransactionEndingChangeEvent(this);
+
+  dirkernel_->observers->Notify(
+      &TransactionObserver::OnTransactionMutate,
+      from_here_, writer_, *originals, *models_with_changes);
 
   // Release the transaction. Note, once the transaction is released this thread
   // can be interrupted by another that was waiting for the transaction,
   // resulting in this code possibly being interrupted with another thread
-  // performing following the same code path. From this point foward, only
+  // performing following the same code path. From this point forward, only
   // local state can be touched.
   dirkernel_->transaction_mutex.Release();
   return true;
 }
 
 void BaseTransaction::NotifyTransactionComplete(
     ModelTypeBitSet models_with_changes) {
-  ObserverList<DirectoryChangeListener> change_listeners;
-  dirkernel_->CopyChangeListeners(&change_listeners);
-  FOR_EACH_OBSERVER(DirectoryChangeListener,
-                    change_listeners,
-                    HandleTransactionCompleteChangeEvent(
-                        models_with_changes));
+  dirkernel_->delegate->HandleTransactionCompleteChangeEvent(
+      models_with_changes);
 }
 
-#undef VLOG_SRC
+#undef VLOG_LOC
 
-#undef VLOG_SRC_STREAM
+#undef VLOG_LOC_STREAM
 
-ReadTransaction::ReadTransaction(Directory* directory, const char* file,
-                                 int line)
-  : BaseTransaction(directory, "Read", file, line, INVALID) {
+ReadTransaction::ReadTransaction(Directory* directory,
+                                 const tracked_objects::Location& location)
+  : BaseTransaction(directory, "Read", location, INVALID) {
 }
 
 ReadTransaction::ReadTransaction(const ScopedDirLookup& scoped_dir,
-                                 const char* file, int line)
-  : BaseTransaction(scoped_dir.operator -> (), "Read", file, line, INVALID) {
+                                 const tracked_objects::Location& location)
+  : BaseTransaction(scoped_dir.operator -> (), "Read", location, INVALID) {
 }
 
 ReadTransaction::~ReadTransaction() {
   UnlockAndLog(NULL);
 }
 
 WriteTransaction::WriteTransaction(Directory* directory, WriterTag writer,
-                                   const char* file, int line)
-  : BaseTransaction(directory, "Write", file, line, writer),
+                                   const tracked_objects::Location& location)
+  : BaseTransaction(directory, "Write", location, writer),
     originals_(new OriginalEntries) {
 }
 
 WriteTransaction::WriteTransaction(const ScopedDirLookup& scoped_dir,
-                                   WriterTag writer, const char* file, int line)
-  : BaseTransaction(scoped_dir.operator -> (), "Write", file, line, writer),
+                                   WriterTag writer,
+                                   const tracked_objects::Location& location)
+  : BaseTransaction(scoped_dir.operator -> (), "Write", location, writer),
     originals_(new OriginalEntries) {
 }
 
-WriteTransaction::WriteTransaction(Directory *directory)
-    : BaseTransaction(directory),
-      originals_(new OriginalEntries) {
-}
-
 void WriteTransaction::SaveOriginal(EntryKernel* entry) {
   if (NULL == entry)
     return;
   OriginalEntries::iterator i = originals_->lower_bound(*entry);
   if (i == originals_->end() ||
       i->ref(META_HANDLE) != entry->ref(META_HANDLE)) {
     originals_->insert(i, *entry);
   }
 }
 
@@ skipping 668 matching lines @@
   CHECK(result);
   for (iterator i = GetParentChildIndexLowerBound(lock, parent_id),
            end = GetParentChildIndexUpperBound(lock, parent_id);
        i != end; ++i) {
     DCHECK_EQ(parent_id, (*i)->ref(PARENT_ID));
     result->push_back((*i)->ref(META_HANDLE));
   }
 }
 
 }  // namespace syncable