[Sync] Ensure all accesses to SyncInternal's observers are thread safe.

chrome/browser/sync/engine/syncapi.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/engine/syncapi.h"
 
 #include <algorithm>
 #include <bitset>
 #include <iomanip>
 #include <list>
@@ skipping 1219 matching lines @@
 
   // SyncNotifierObserver implementation.
   virtual void OnNotificationStateChange(
       bool notifications_enabled);
 
   virtual void OnIncomingNotification(
       const syncable::ModelTypePayloadMap& type_payloads);
 
   virtual void StoreState(const std::string& cookie);
 
+  bool HaveObservers();

[akalin, 2011/06/08 22:39:38]

make const

[Nicolas Zea, 2011/06/08 23:10:23]

Done.

+
   void AddObserver(SyncManager::Observer* observer);
 
   void RemoveObserver(SyncManager::Observer* observer);
 
   // Accessors for the private members.
   DirectoryManager* dir_manager() { return share_.dir_manager.get(); }
   SyncAPIServerConnectionManager* connection_manager() {
     return connection_manager_.get();
   }
   SyncerThread* syncer_thread() { return syncer_thread_.get(); }
@@ skipping 270 matching lines @@
   browser_sync::JsArgList FindNodesContainingString(
       const browser_sync::JsArgList& args);
 
   // We couple the DirectoryManager and username together in a UserShare member
   // so we can return a handle to share_ to clients of the API for use when
   // constructing any transaction type.
   UserShare share_;
 
   MessageLoop* core_message_loop_;
 
+  // We have to lock around every observers_ access because it can get accessed
+  // from any thread and added to/removed from on the core thread.
+  base::Lock observers_lock_;

[akalin, 2011/06/08 22:39:38]

make mutable

[Nicolas Zea, 2011/06/08 23:10:23]

Done.

   ObserverList<SyncManager::Observer> observers_;
 
   browser_sync::JsEventRouter* parent_router_;
 
   // The ServerConnectionManager used to abstract communication between the
   // client (the Syncer) and the sync server.
   scoped_ptr<SyncAPIServerConnectionManager> connection_manager_;
 
   // The thread that runs the Syncer. Needs to be explicitly Start()ed.
   scoped_ptr<SyncerThread> syncer_thread_;
@@ skipping 41 matching lines @@
   // Map used to store the notification info to be displayed in about:sync page.
   NotificationInfoMap notification_info_map_;
 
   browser_sync::JsDirectoryChangeListener js_directory_change_listener_;
 
   JsMessageHandlerMap js_message_handlers_;
 };
 const int SyncManager::SyncInternal::kDefaultNudgeDelayMilliseconds = 200;
 const int SyncManager::SyncInternal::kPreferencesNudgeDelayMilliseconds = 2000;
 
+// We have to read the current observer list while holding a lock because it
+// can be added/removed from another thread. We then notify the actual observers
+// while the lock is released.

[akalin, 2011/06/08 22:39:38]

Add a TODO explaining the problem with notifying garbage pointers, why it
doesn't happen often (only happens when you close last about:sync window in the
middle of a notify), but that we should fix it.

Also file a bug and link it, for good measure.  (Feel free to assign to me.)

[Nicolas Zea, 2011/06/08 23:10:23]

Done.

+#define NOTIFY_SYNCMANAGER_OBSERVERS(function)                          \

[akalin, 2011/06/08 22:39:38]

I think it would be cleaner to define a utility function
CopyObserverList(ObserverList* observers) that would copy observers_ into
observers under lock and use that.  That way, callers would just add:

ObserverList<> observers;
CopyObserverList(&observers);
FOR_EACH(..., observers, ...);

Cleaner than having a big macro.

[Nicolas Zea, 2011/06/08 23:10:23]

Done.

+  do {                                                                  \
+    ObserverList<SyncManager::Observer> temp_observers;                 \
+    {                                                                   \
+      base::AutoLock lock(observers_lock_);                             \
+      ObserverListBase<SyncManager::Observer>::Iterator it(observers_); \
+      SyncManager::Observer* obs;                                       \
+      while ((obs = it.GetNext()) != NULL)                              \
+        temp_observers.AddObserver(obs);                                \
+    }                                                                   \
+    FOR_EACH_OBSERVER(SyncManager::Observer, temp_observers, function); \
+  } while (0)
+
 SyncManager::Observer::~Observer() {}
 
 SyncManager::SyncManager() {
   data_ = new SyncInternal(this);
 }
 
 bool SyncManager::Init(const FilePath& database_location,
                        const char* sync_server_and_path,
                        int sync_server_port,
                        bool use_ssl,
@@ skipping 200 matching lines @@
 
     ReadNode node(&trans);
     if (!node.InitByTagLookup(kNigoriTag)) {
       NOTREACHED();
       return;
     }
 
     nigori.CopyFrom(node.GetNigoriSpecifics());
     Cryptographer::UpdateResult result = cryptographer->Update(nigori);
     if (result == Cryptographer::NEEDS_PASSPHRASE) {
-       FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                         OnPassphraseRequired(sync_api::REASON_DECRYPTION));
+       NOTIFY_SYNCMANAGER_OBSERVERS(
+          OnPassphraseRequired(sync_api::REASON_DECRYPTION));
     }
 
     // Refresh list of encrypted datatypes.
     encrypted_types = GetEncryptedTypes(&trans);
   }
 
 
 
   // Ensure any datatypes that need encryption are encrypted.
   EncryptDataTypes(encrypted_types);
 }
 
 void SyncManager::SyncInternal::StartSyncingNormally() {
-   // Start the syncer thread. This won't actually
-   // result in any syncing until at least the
-   // DirectoryManager broadcasts the OPENED event,
-   // and a valid server connection is detected.
+  // Start the syncer thread. This won't actually
+  // result in any syncing until at least the
+  // DirectoryManager broadcasts the OPENED event,
+  // and a valid server connection is detected.
   if (syncer_thread())  // NULL during certain unittests.
     syncer_thread()->Start(SyncerThread::NORMAL_MODE, NULL);
 }
 
 void SyncManager::SyncInternal::MarkAndNotifyInitializationComplete() {
   // There is only one real time we need this mutex.  If we get an auth
   // success, and before the initial sync ends we get an auth failure.  In this
   // case we'll be listening to both the AuthWatcher and Syncer, and it's a race
   // between their respective threads to call MarkAndNotify.  We need to make
   // sure the observer is notified once and only once.
   {
     base::AutoLock lock(initialized_mutex_);
     if (initialized_)
       return;
     initialized_ = true;
   }
 
   // Notify that initialization is complete.
-  FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                    OnInitializationComplete());
+  NOTIFY_SYNCMANAGER_OBSERVERS(OnInitializationComplete());
 }
 
 void SyncManager::SyncInternal::SendNotification() {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
   if (!sync_notifier_) {
     VLOG(1) << "Not sending notification: sync_notifier_ is NULL";
     return;
   }
   sync_notifier_->SendNotification();
 }
 
 bool SyncManager::SyncInternal::OpenDirectory() {
   DCHECK(!initialized()) << "Should only happen once";
 
   bool share_opened = dir_manager()->Open(username_for_share());
   DCHECK(share_opened);
   if (!share_opened) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnStopSyncingPermanently());
+    NOTIFY_SYNCMANAGER_OBSERVERS(OnStopSyncingPermanently());
 
     LOG(ERROR) << "Could not open share for:" << username_for_share();
     return false;
   }
 
   // Database has to be initialized for the guid to be available.
   syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
   if (!lookup.good()) {
     NOTREACHED();
     return false;
@@ skipping 63 matching lines @@
   registrar_->GetModelSafeRoutingInfo(&routes);
   syncable::ModelTypeSet enabled_types;
   for (ModelSafeRoutingInfo::const_iterator it = routes.begin();
        it != routes.end(); ++it) {
     enabled_types.insert(it->first);
   }
   sync_notifier_->UpdateEnabledTypes(enabled_types);
 }
 
 void SyncManager::SyncInternal::RaiseAuthNeededEvent() {
-  FOR_EACH_OBSERVER(
-      SyncManager::Observer, observers_,
+  NOTIFY_SYNCMANAGER_OBSERVERS(
       OnAuthError(AuthError(AuthError::INVALID_GAIA_CREDENTIALS)));
 }
 
 void SyncManager::SyncInternal::SetUsingExplicitPassphrasePrefForMigration(
     WriteTransaction* const trans) {
   WriteNode node(trans);
   if (!node.InitByTagLookup(kNigoriTag)) {
     // TODO(albertb): Plumb an UnrecoverableError all the way back to the PSS.
     NOTREACHED();
     return;
   }
   sync_pb::NigoriSpecifics specifics(node.GetNigoriSpecifics());
   specifics.set_using_explicit_passphrase(true);
   node.SetNigoriSpecifics(specifics);
 }
 
 void SyncManager::SyncInternal::SetPassphrase(
     const std::string& passphrase, bool is_explicit) {
   // We do not accept empty passphrases.
   if (passphrase.empty()) {
     VLOG(1) << "Rejecting empty passphrase.";
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
+    NOTIFY_SYNCMANAGER_OBSERVERS(
         OnPassphraseRequired(sync_api::REASON_SET_PASSPHRASE_FAILED));
     return;
   }
 
   // All accesses to the cryptographer are protected by a transaction.
   WriteTransaction trans(GetUserShare());
   Cryptographer* cryptographer = trans.GetCryptographer();
   KeyParams params = {"localhost", "dummy", passphrase};
 
   if (cryptographer->has_pending_keys()) {
     if (!cryptographer->DecryptPendingKeys(params)) {
       VLOG(1) << "Passphrase failed to decrypt pending keys.";
-      FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
+      NOTIFY_SYNCMANAGER_OBSERVERS(
           OnPassphraseRequired(sync_api::REASON_SET_PASSPHRASE_FAILED));
       return;
     }
 
     // TODO(tim): If this is the first time the user has entered a passphrase
     // since the protocol changed to store passphrase preferences in the cloud,
     // make sure we update this preference. See bug 62103.
     // TODO(jhawkins): Verify that this logic may be removed now that the
     // migration is no longer supported.
     if (is_explicit)
@@ skipping 25 matching lines @@
     sync_pb::NigoriSpecifics specifics(node.GetNigoriSpecifics());
     specifics.clear_encrypted();
     cryptographer->GetKeys(specifics.mutable_encrypted());
     specifics.set_using_explicit_passphrase(is_explicit);
     node.SetNigoriSpecifics(specifics);
     ReEncryptEverything(&trans);
   }
 
   std::string bootstrap_token;
   cryptographer->GetBootstrapToken(&bootstrap_token);
-  FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                    OnPassphraseAccepted(bootstrap_token));
+  NOTIFY_SYNCMANAGER_OBSERVERS(OnPassphraseAccepted(bootstrap_token));
 }
 
 bool SyncManager::SyncInternal::IsUsingExplicitPassphrase() {
   ReadTransaction trans(&share_);
   ReadNode node(&trans);
   if (!node.InitByTagLookup(kNigoriTag)) {
     // TODO(albertb): Plumb an UnrecoverableError all the way back to the PSS.
     NOTREACHED();
     return false;
   }
@@ skipping 97 matching lines @@
       WriteNode child(trans);
       if (!child.InitByIdLookup(child_id)) {
         NOTREACHED();
         return;
       }
       child.SetPasswordSpecifics(child.GetPasswordSpecifics());
       child_id = child.GetSuccessorId();
     }
   }
 
-  FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                    OnEncryptionComplete(encrypted_types));
+  NOTIFY_SYNCMANAGER_OBSERVERS(OnEncryptionComplete(encrypted_types));
 }
 
 SyncManager::~SyncManager() {
   delete data_;
 }
 
 void SyncManager::AddObserver(Observer* observer) {
   data_->AddObserver(observer);
 }
 
@@ skipping 72 matching lines @@
   // TODO(akalin): CheckServerReachable() can block, which may cause
   // jank if we try to shut down sync.  Fix this.
   connection_manager()->CheckServerReachable();
 }
 
 void SyncManager::SyncInternal::OnServerConnectionEvent(
     const ServerConnectionEvent& event) {
   allstatus_.HandleServerConnectionEvent(event);
   if (event.connection_code ==
       browser_sync::HttpResponse::SERVER_CONNECTION_OK) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnAuthError(AuthError::None()));
+    NOTIFY_SYNCMANAGER_OBSERVERS(OnAuthError(AuthError::None()));
   }
 
   if (event.connection_code == browser_sync::HttpResponse::SYNC_AUTH_ERROR) {
-    FOR_EACH_OBSERVER(
-        SyncManager::Observer, observers_,
+    NOTIFY_SYNCMANAGER_OBSERVERS(
         OnAuthError(AuthError(AuthError::INVALID_GAIA_CREDENTIALS)));
   }
 }
 
 void SyncManager::SyncInternal::HandleTransactionCompleteChangeEvent(
     const syncable::ModelTypeBitSet& models_with_changes) {
   // This notification happens immediately after the transaction mutex is
   // released. This allows work to be performed without blocking other threads
   // from acquiring a transaction.
-  if (observers_.size() <= 0)
+  if (!HaveObservers())
     return;
 
   // Call commit.
   for (int i = 0; i < syncable::MODEL_TYPE_COUNT; ++i) {
     if (models_with_changes.test(i)) {
-      FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                        OnChangesComplete(syncable::ModelTypeFromInt(i)));
+      NOTIFY_SYNCMANAGER_OBSERVERS(
+          OnChangesComplete(syncable::ModelTypeFromInt(i)));
     }
   }
 }
 
 ModelTypeBitSet SyncManager::SyncInternal::HandleTransactionEndingChangeEvent(
     syncable::BaseTransaction* trans) {
   // This notification happens immediately before a syncable WriteTransaction
   // falls out of scope. It happens while the channel mutex is still held,
   // and while the transaction mutex is held, so it cannot be re-entrant.
-  if (observers_.size() <= 0 || ChangeBuffersAreEmpty())
+  if (!HaveObservers() || ChangeBuffersAreEmpty())
     return ModelTypeBitSet();
 
   // This will continue the WriteTransaction using a read only wrapper.
   // This is the last chance for read to occur in the WriteTransaction
   // that's closing. This special ReadTransaction will not close the
   // underlying transaction.
   ReadTransaction read_trans(GetUserShare(), trans);
 
   syncable::ModelTypeBitSet models_with_changes;
   for (int i = 0; i < syncable::MODEL_TYPE_COUNT; ++i) {
     if (change_buffers_[i].IsEmpty())
       continue;
 
     vector<ChangeRecord> ordered_changes;
     change_buffers_[i].GetAllChangesInTreeOrder(&read_trans, &ordered_changes);
     if (!ordered_changes.empty()) {
-      FOR_EACH_OBSERVER(
-          SyncManager::Observer, observers_,
+      NOTIFY_SYNCMANAGER_OBSERVERS(
           OnChangesApplied(syncable::ModelTypeFromInt(i), &read_trans,
                            &ordered_changes[0], ordered_changes.size()));
       models_with_changes.set(i, true);
     }
     change_buffers_[i].Clear();
   }
   return models_with_changes;
 }
 
 void SyncManager::SyncInternal::HandleCalculateChangesChangeEventFromSyncApi(
@@ skipping 128 matching lines @@
 void SyncManager::SyncInternal::RequestNudgeWithDataTypes(
     const TimeDelta& delay,
     browser_sync::NudgeSource source, const ModelTypeBitSet& types,
     const tracked_objects::Location& nudge_location) {
   if (syncer_thread())
      syncer_thread()->ScheduleNudge(delay, source, types, nudge_location);
 }
 
 void SyncManager::SyncInternal::OnSyncEngineEvent(
     const SyncEngineEvent& event) {
-  if (observers_.size() <= 0) {
+  if (!HaveObservers()) {
     VLOG(0) << "OnSyncEngineEvent returning because observers_.size() is zero";
     return;
   }
 
   // Only send an event if this is due to a cycle ending and this cycle
   // concludes a canonical "sync" process; that is, based on what is known
   // locally we are "all happy" and up-to-date.  There may be new changes on
   // the server, but we'll get them on a subsequent sync.
   //
   // Notifications are sent at the end of every sync cycle, regardless of
@@ skipping 27 matching lines @@
           if (!nigori.encrypted().blob().empty()) {
             DCHECK(!cryptographer->CanDecrypt(nigori.encrypted()));
             cryptographer->SetPendingKeys(nigori.encrypted());
           }
         }
 
         // If we've completed a sync cycle and the cryptographer isn't ready
         // yet, prompt the user for a passphrase.
         if (cryptographer->has_pending_keys()) {
           VLOG(1) << "OnPassPhraseRequired Sent";
-          FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                            OnPassphraseRequired(sync_api::REASON_DECRYPTION));
+          NOTIFY_SYNCMANAGER_OBSERVERS(
+              OnPassphraseRequired(sync_api::REASON_DECRYPTION));
         } else if (!cryptographer->is_ready()) {
           VLOG(1) << "OnPassphraseRequired sent because cryptographer is not "
                   << "ready";
-          FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                            OnPassphraseRequired(sync_api::REASON_ENCRYPTION));
+          NOTIFY_SYNCMANAGER_OBSERVERS(
+              OnPassphraseRequired(sync_api::REASON_ENCRYPTION));
         }
         // If everything is in order(we have the passphrase) then there is no
         // need to inform the listeners. They will just wait for sync
         // completion event and if no errors have been raised it means
         // encryption was succesful.
       }
     }
 
     if (!initialized()) {
       VLOG(0) << "OnSyncCycleCompleted not sent because sync api is not "
               << "initialized";
       return;
     }
 
     if (!event.snapshot->has_more_to_sync) {
       VLOG(1) << "OnSyncCycleCompleted sent";
-      FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                        OnSyncCycleCompleted(event.snapshot));
+      NOTIFY_SYNCMANAGER_OBSERVERS(OnSyncCycleCompleted(event.snapshot));
     }
 
     // This is here for tests, which are still using p2p notifications.
     // SendNotification does not do anything if we are using server based
     // notifications.
     // TODO(chron): Consider changing this back to track has_more_to_sync
     // only notify peers if a successful commit has occurred.
     bool is_notifiable_commit =
         (event.snapshot->syncer_status.num_successful_commits > 0);
     if (is_notifiable_commit) {
       allstatus_.IncrementNotifiableCommits();
       core_message_loop_->PostTask(
           FROM_HERE,
           NewRunnableMethod(
               this,
               &SyncManager::SyncInternal::SendNotification));
     }
   }
 
   if (event.what_happened == SyncEngineEvent::STOP_SYNCING_PERMANENTLY) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnStopSyncingPermanently());
+    NOTIFY_SYNCMANAGER_OBSERVERS(OnStopSyncingPermanently());
     return;
   }
 
   if (event.what_happened == SyncEngineEvent::CLEAR_SERVER_DATA_SUCCEEDED) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnClearServerDataSucceeded());
+    NOTIFY_SYNCMANAGER_OBSERVERS(OnClearServerDataSucceeded());
     return;
   }
 
   if (event.what_happened == SyncEngineEvent::CLEAR_SERVER_DATA_FAILED) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnClearServerDataFailed());
+    NOTIFY_SYNCMANAGER_OBSERVERS(OnClearServerDataFailed());
     return;
   }
 
   if (event.what_happened == SyncEngineEvent::UPDATED_TOKEN) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnUpdatedToken(event.updated_token));
+    NOTIFY_SYNCMANAGER_OBSERVERS(OnUpdatedToken(event.updated_token));
     return;
   }
 }
 
 void SyncManager::SyncInternal::SetParentJsEventRouter(
     browser_sync::JsEventRouter* router) {
   DCHECK(router);
   parent_router_ = router;
 
   // We might be called before OpenDirectory() or after shutdown.
@@ skipping 283 matching lines @@
   }
   if (VLOG_IS_ON(1)) {
     std::string encoded_state;
     base::Base64Encode(state, &encoded_state);
     VLOG(1) << "Writing notification state: " << encoded_state;
   }
   lookup->SetNotificationState(state);
   lookup->SaveChanges();
 }
 
+bool SyncManager::SyncInternal::HaveObservers() {
+  base::AutoLock lock(observers_lock_);
+  return observers_.size() > 0;
+}
+
 void SyncManager::SyncInternal::AddObserver(
     SyncManager::Observer* observer) {
+  base::AutoLock lock(observers_lock_);
   observers_.AddObserver(observer);
 }
 
 void SyncManager::SyncInternal::RemoveObserver(
     SyncManager::Observer* observer) {
+  base::AutoLock lock(observers_lock_);
   observers_.RemoveObserver(observer);
 }
 
 SyncManager::Status::Summary SyncManager::GetStatusSummary() const {
   return data_->GetStatus().summary;
 }
 
 SyncManager::Status SyncManager::GetDetailedStatus() const {
   return data_->GetStatus();
 }
@@ skipping 70 matching lines @@
 void SyncManager::TriggerOnIncomingNotificationForTest(
     const syncable::ModelTypeBitSet& model_types) {
   syncable::ModelTypePayloadMap model_types_with_payloads =
       syncable::ModelTypePayloadMapFromBitSet(model_types,
           std::string());
 
   data_->OnIncomingNotification(model_types_with_payloads);
 }
 
 }  // namespace sync_api