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

chrome/browser/sync/engine/syncapi.cc

Index: chrome/browser/sync/engine/syncapi.cc
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index 3546df4da92605e92423f708b6ed00cf3a873014..5d84638fb9ed148c307e41c470d512f8c9687c44 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -1237,6 +1237,8 @@ class SyncManager::SyncInternal
 
   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);
@@ -1527,6 +1529,9 @@ class SyncManager::SyncInternal
 
   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_;
@@ -1588,6 +1593,22 @@ class SyncManager::SyncInternal
 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() {
@@ -1808,8 +1829,8 @@ void SyncManager::SyncInternal::BootstrapEncryption(
     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.
@@ -1823,10 +1844,10 @@ void SyncManager::SyncInternal::BootstrapEncryption(
 }
 
 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);
 }
@@ -1845,8 +1866,7 @@ void SyncManager::SyncInternal::MarkAndNotifyInitializationComplete() {
   }
 
   // Notify that initialization is complete.
-  FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                    OnInitializationComplete());
+  NOTIFY_SYNCMANAGER_OBSERVERS(OnInitializationComplete());
 }
 
 void SyncManager::SyncInternal::SendNotification() {
@@ -1864,8 +1884,7 @@ bool SyncManager::SyncInternal::OpenDirectory() {
   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;
@@ -1949,8 +1968,7 @@ void SyncManager::SyncInternal::UpdateEnabledTypes() {
 }
 
 void SyncManager::SyncInternal::RaiseAuthNeededEvent() {
-  FOR_EACH_OBSERVER(
-      SyncManager::Observer, observers_,
+  NOTIFY_SYNCMANAGER_OBSERVERS(
       OnAuthError(AuthError(AuthError::INVALID_GAIA_CREDENTIALS)));
 }
 
@@ -1972,7 +1990,7 @@ void SyncManager::SyncInternal::SetPassphrase(
   // 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;
   }
@@ -1985,7 +2003,7 @@ void SyncManager::SyncInternal::SetPassphrase(
   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;
     }
@@ -2031,8 +2049,7 @@ void SyncManager::SyncInternal::SetPassphrase(
 
   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() {
@@ -2150,8 +2167,7 @@ void SyncManager::SyncInternal::ReEncryptEverything(WriteTransaction* trans) {
     }
   }
 
-  FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                    OnEncryptionComplete(encrypted_types));
+  NOTIFY_SYNCMANAGER_OBSERVERS(OnEncryptionComplete(encrypted_types));
 }
 
 SyncManager::~SyncManager() {
@@ -2244,13 +2260,11 @@ void SyncManager::SyncInternal::OnServerConnectionEvent(
   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)));
   }
 }
@@ -2260,14 +2274,14 @@ void SyncManager::SyncInternal::HandleTransactionCompleteChangeEvent(
   // 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)));
     }
   }
 }
@@ -2277,7 +2291,7 @@ ModelTypeBitSet SyncManager::SyncInternal::HandleTransactionEndingChangeEvent(
   // 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.
@@ -2294,8 +2308,7 @@ ModelTypeBitSet SyncManager::SyncInternal::HandleTransactionEndingChangeEvent(
     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);
@@ -2444,7 +2457,7 @@ void SyncManager::SyncInternal::RequestNudgeWithDataTypes(
 
 void SyncManager::SyncInternal::OnSyncEngineEvent(
     const SyncEngineEvent& event) {
-  if (observers_.size() <= 0) {
+  if (!HaveObservers()) {
     VLOG(0) << "OnSyncEngineEvent returning because observers_.size() is zero";
     return;
   }
@@ -2492,13 +2505,13 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
         // 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
@@ -2515,8 +2528,7 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
 
     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.
@@ -2537,26 +2549,22 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
   }
 
   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;
   }
 }
@@ -2860,13 +2868,20 @@ void SyncManager::SyncInternal::StoreState(
   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);
 }