make new syncer thread the default.

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>
 #include <queue>
 #include <string>
 #include <vector>
 
 #include "base/base64.h"
 #include "base/command_line.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop.h"
 #include "base/observer_list.h"
 #include "base/sha1.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
 #include "base/synchronization/lock.h"
 #include "base/task.h"
+#include "base/time.h"
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
 #include "chrome/browser/sync/engine/all_status.h"
 #include "chrome/browser/sync/engine/change_reorder_buffer.h"
 #include "chrome/browser/sync/engine/model_safe_worker.h"
+#include "chrome/browser/sync/engine/nudge_source.h"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/net/syncapi_server_connection_manager.h"
 #include "chrome/browser/sync/engine/syncer.h"
 #include "chrome/browser/sync/engine/syncer_thread.h"
-#include "chrome/browser/sync/engine/syncer_thread2.h"
-#include "chrome/browser/sync/engine/syncer_thread_adapter.h"
 #include "chrome/browser/sync/engine/http_post_provider_factory.h"
 #include "chrome/browser/sync/js_arg_list.h"
 #include "chrome/browser/sync/js_backend.h"
 #include "chrome/browser/sync/js_event_router.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
 #include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 #include "chrome/browser/sync/protocol/app_specifics.pb.h"
 #include "chrome/browser/sync/protocol/autofill_specifics.pb.h"
 #include "chrome/browser/sync/protocol/bookmark_specifics.pb.h"
 #include "chrome/browser/sync/protocol/extension_specifics.pb.h"
 #include "chrome/browser/sync/protocol/nigori_specifics.pb.h"
 #include "chrome/browser/sync/protocol/preference_specifics.pb.h"
 #include "chrome/browser/sync/protocol/proto_value_conversions.h"
 #include "chrome/browser/sync/protocol/service_constants.h"
 #include "chrome/browser/sync/protocol/session_specifics.pb.h"
 #include "chrome/browser/sync/protocol/sync.pb.h"
 #include "chrome/browser/sync/protocol/theme_specifics.pb.h"
 #include "chrome/browser/sync/protocol/typed_url_specifics.pb.h"
 #include "chrome/browser/sync/sessions/sync_session.h"
 #include "chrome/browser/sync/sessions/sync_session_context.h"
 #include "chrome/browser/sync/syncable/autofill_migration.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/browser/sync/syncable/model_type_payload_map.h"
+#include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/syncable/nigori_util.h"
 #include "chrome/browser/sync/syncable/syncable.h"
 #include "chrome/browser/sync/util/crypto_helpers.h"
 #include "chrome/common/chrome_switches.h"
 #include "chrome/common/deprecated/event_sys.h"
 #include "chrome/common/net/gaia/gaia_authenticator.h"
 #include "content/browser/browser_thread.h"
 #include "net/base/network_change_notifier.h"
 
+using base::TimeDelta;
 using browser_sync::AllStatus;
 using browser_sync::Cryptographer;
 using browser_sync::KeyParams;
 using browser_sync::ModelSafeRoutingInfo;
 using browser_sync::ModelSafeWorker;
 using browser_sync::ModelSafeWorkerRegistrar;
 using browser_sync::ServerConnectionEvent;
 using browser_sync::ServerConnectionEvent2;
 using browser_sync::ServerConnectionEventListener;
 using browser_sync::SyncEngineEvent;
 using browser_sync::SyncEngineEventListener;
 using browser_sync::Syncer;
 using browser_sync::SyncerThread;
-using browser_sync::SyncerThreadAdapter;
 using browser_sync::kNigoriTag;
 using browser_sync::sessions::SyncSessionContext;
 using std::list;
 using std::hex;
 using std::string;
 using std::vector;
 using syncable::Directory;
 using syncable::DirectoryManager;
 using syncable::Entry;
+using syncable::ModelTypeBitSet;
 using syncable::SPECIFICS;
 using sync_pb::AutofillProfileSpecifics;
 
 typedef GoogleServiceAuthError AuthError;
 
 static const int kThreadExitTimeoutMsec = 60000;
 static const int kSSLPort = 443;
 static const int kSyncerThreadDelayMsec = 250;
 
 #if defined(OS_CHROMEOS)
@@ skipping 1160 matching lines @@
 
   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();
   }
-  SyncerThreadAdapter* syncer_thread() { return syncer_thread_.get(); }
+  SyncerThread* syncer_thread() { return syncer_thread_.get(); }
   UserShare* GetUserShare() { return &share_; }
 
   // Return the currently active (validated) username for use with syncable
   // types.
   const std::string& username_for_share() const {
     return share_.name;
   }
 
   Status GetStatus();
 
@@ skipping 226 matching lines @@
 
   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<SyncerThreadAdapter> syncer_thread_;
+  scoped_ptr<SyncerThread> syncer_thread_;
 
   // The SyncNotifier which notifies us when updates need to be downloaded.
   sync_notifier::SyncNotifier* sync_notifier_;
 
   // A multi-purpose status watch object that aggregates stats from various
   // sync components.
   AllStatus allstatus_;
 
   // Each element of this array is a store of change records produced by
   // HandleChangeEvent during the CALCULATE_CHANGES step.  The changes are
@@ skipping 127 matching lines @@
 
 void SyncManager::EncryptDataTypes(
     const syncable::ModelTypeSet& encrypted_types) {
   data_->EncryptDataTypes(encrypted_types);
 }
 
 bool SyncManager::IsUsingExplicitPassphrase() {
   return data_ && data_->IsUsingExplicitPassphrase();
 }
 
-bool SyncManager::RequestPause() {
-  if (data_->syncer_thread())
-    return data_->syncer_thread()->RequestPause();
-  return false;
-}
-
-bool SyncManager::RequestResume() {
-  if (data_->syncer_thread())
-    return data_->syncer_thread()->RequestResume();
-  return false;
-}
-
 void SyncManager::RequestNudge(const tracked_objects::Location& location) {
   if (data_->syncer_thread())
-    data_->syncer_thread()->NudgeSyncer(0, SyncerThread::kLocal, location);
+    data_->syncer_thread()->ScheduleNudge(
+        TimeDelta::FromMilliseconds(0), browser_sync::NUDGE_SOURCE_LOCAL,
+        ModelTypeBitSet(), location);
 }
 
 void SyncManager::RequestClearServerData() {
   if (data_->syncer_thread())
-    data_->syncer_thread()->NudgeSyncer(0, SyncerThread::kClearPrivateData,
-        FROM_HERE);
+    data_->syncer_thread()->ScheduleClearUserData();
 }
 
 void SyncManager::RequestConfig(const syncable::ModelTypeBitSet& types) {
   if (!data_->syncer_thread())
     return;
-  // It is an error for this to be called if new_impl is null.
   StartConfigurationMode(NULL);
-  data_->syncer_thread()->new_impl()->ScheduleConfig(types);
+  data_->syncer_thread()->ScheduleConfig(types);
 }
 
 void SyncManager::StartConfigurationMode(ModeChangeCallback* callback) {
   if (!data_->syncer_thread())
     return;
-  if (!data_->syncer_thread()->new_impl())
-    return;
-  data_->syncer_thread()->new_impl()->Start(
-      browser_sync::s3::SyncerThread::CONFIGURATION_MODE, callback);
+  data_->syncer_thread()->Start(
+      browser_sync::SyncerThread::CONFIGURATION_MODE, callback);
 }
 
 const std::string& SyncManager::GetAuthenticatedUsername() {
   DCHECK(data_);
   return data_->username_for_share();
 }
 
 bool SyncManager::SyncInternal::Init(
     const FilePath& database_location,
     const std::string& sync_server_and_path,
@@ skipping 17 matching lines @@
   sync_notifier_ = sync_notifier;
   sync_notifier_->AddObserver(this);
 
   share_.dir_manager.reset(new DirectoryManager(database_location));
 
   connection_manager_.reset(new SyncAPIServerConnectionManager(
       sync_server_and_path, port, use_ssl, user_agent, post_factory));
 
   net::NetworkChangeNotifier::AddIPAddressObserver(this);
 
-  bool new_syncer_thread = CommandLine::ForCurrentProcess()->HasSwitch(
-      switches::kNewSyncerThread);
-
-  if (new_syncer_thread) {
-    connection_manager()->AddListener(this);
-  } else {
-    connection_manager_hookup_.reset(
-        NewEventListenerHookup(connection_manager()->channel(), this,
-            &SyncManager::SyncInternal::HandleServerConnectionEvent));
-  }
+  connection_manager()->AddListener(this);
 
   // TODO(akalin): CheckServerReachable() can block, which may cause jank if we
   // try to shut down sync.  Fix this.
   core_message_loop_->PostTask(FROM_HERE,
       method_factory_.NewRunnableMethod(&SyncInternal::CheckServerReachable));
 
   // Test mode does not use a syncer context or syncer thread.
   if (!setup_for_test_mode_) {
     // Build a SyncSessionContext and store the worker in it.
     VLOG(1) << "Sync is bringing up SyncSessionContext.";
     std::vector<SyncEngineEventListener*> listeners;
     listeners.push_back(&allstatus_);
     listeners.push_back(this);
     SyncSessionContext* context = new SyncSessionContext(
         connection_manager_.get(),
         dir_manager(),
         model_safe_worker_registrar,
         listeners);
     context->set_account_name(credentials.email);
     // The SyncerThread takes ownership of |context|.
-    syncer_thread_.reset(new SyncerThreadAdapter(context,
-        new_syncer_thread));
+    syncer_thread_.reset(new SyncerThread(context, new Syncer()));
   }
 
   bool signed_in = SignIn(credentials);
 
   // Do this once the directory is opened.
   BootstrapEncryption(restored_key_for_bootstrapping);
   return signed_in;
 }
 
 void SyncManager::SyncInternal::BootstrapEncryption(
@@ skipping 34 matching lines @@
 
   // Refresh list of encrypted datatypes.
   syncable::ModelTypeSet encrypted_types =
       syncable::GetEncryptedDataTypesFromNigori(nigori);
 
   // Ensure any datatypes that need encryption are encrypted.
   EncryptDataTypes(encrypted_types);
 }
 
 void SyncManager::SyncInternal::StartSyncing() {
+   // 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();  // 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.
+    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_);
@@ skipping 32 matching lines @@
 
   // 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;
   }
 
   connection_manager()->set_client_id(lookup->cache_guid());
 
-  if (syncer_thread())
-    syncer_thread()->CreateSyncer(username_for_share());
-
   MarkAndNotifyInitializationComplete();
   dir_change_hookup_.reset(lookup->AddChangeObserver(this));
   return true;
 }
 
 bool SyncManager::SyncInternal::SignIn(const SyncCredentials& credentials) {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
   DCHECK(share_.name.empty());
   share_.name = credentials.email;
 
@@ skipping 305 matching lines @@
 }
 
 void SyncManager::Shutdown() {
   data_->Shutdown();
 }
 
 void SyncManager::SyncInternal::Shutdown() {
   method_factory_.RevokeAll();
 
   if (syncer_thread()) {
-    if (!syncer_thread()->Stop(kThreadExitTimeoutMsec)) {
-      LOG(FATAL) << "Unable to stop the syncer, it won't be happy...";
-    }
+    syncer_thread()->Stop();
     syncer_thread_.reset();
   }
 
   // We NULL out sync_notifer_ so that any pending tasks do not
   // trigger further notifications.
   // TODO(akalin): NULL the other member variables defensively, too.
   if (sync_notifier_) {
     sync_notifier_->RemoveObserver(this);
   }
 
@@ skipping 184 matching lines @@
       exists_unsynced_items = true;
 
       model_types[model_type] = true;
       if (model_type != syncable::PREFERENCES)
         only_preference_changes = false;
     }
   }
   if (exists_unsynced_items && syncer_thread()) {
     int nudge_delay = only_preference_changes ?
         kPreferencesNudgeDelayMilliseconds : kDefaultNudgeDelayMilliseconds;
-    syncer_thread()->NudgeSyncerWithDataTypes(
-        nudge_delay,
-        SyncerThread::kLocal,
+    syncer_thread()->ScheduleNudge(
+        TimeDelta::FromMilliseconds(nudge_delay),
+        browser_sync::NUDGE_SOURCE_LOCAL,
         model_types,
         FROM_HERE);
   }
 }
 
 void SyncManager::SyncInternal::SetExtraChangeRecordData(int64 id,
     syncable::ModelType type, ChangeReorderBuffer* buffer,
     Cryptographer* cryptographer, const syncable::EntryKernel& original,
     bool existed_before, bool exists_now) {
   // If this is a deletion and the datatype was encrypted, we need to decrypt it
@@ skipping 135 matching lines @@
         (event.snapshot->syncer_status.num_successful_commits > 0);
     if (new_notification) {
       core_message_loop_->PostTask(
           FROM_HERE,
           NewRunnableMethod(
               this,
               &SyncManager::SyncInternal::SendNotification));
     }
   }
 
-  if (event.what_happened == SyncEngineEvent::SYNCER_THREAD_PAUSED) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnPaused());
-    return;
-  }
-
-  if (event.what_happened == SyncEngineEvent::SYNCER_THREAD_RESUMED) {
-    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
-                      OnResumed());
-    return;
-  }
-
   if (event.what_happened == SyncEngineEvent::STOP_SYNCING_PERMANENTLY) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                       OnStopSyncingPermanently());
     return;
   }
 
   if (event.what_happened == SyncEngineEvent::CLEAR_SERVER_DATA_SUCCEEDED) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                       OnClearServerDataSucceeded());
     return;
@@ skipping 136 matching lines @@
   return_args.Append(result);
   return browser_sync::JsArgList(return_args);
 }
 
 void SyncManager::SyncInternal::OnNotificationStateChange(
     bool notifications_enabled) {
   VLOG(1) << "P2P: Notifications enabled = "
           << (notifications_enabled ? "true" : "false");
   allstatus_.SetNotificationsEnabled(notifications_enabled);
   if (syncer_thread()) {
-    syncer_thread()->SetNotificationsEnabled(notifications_enabled);
+    syncer_thread()->set_notifications_enabled(notifications_enabled);
   }
   if (parent_router_) {
     ListValue args;
     args.Append(Value::CreateBooleanValue(notifications_enabled));
     // TODO(akalin): Tidy up grammar in event names.
     parent_router_->RouteJsEvent("onSyncNotificationStateChange",
                                  browser_sync::JsArgList(args), NULL);
   }
 }
 
 void SyncManager::SyncInternal::UpdateNotificationInfo(
     const syncable::ModelTypePayloadMap& type_payloads) {
   for (syncable::ModelTypePayloadMap::const_iterator it = type_payloads.begin();
        it != type_payloads.end(); ++it) {
     NotificationInfo* info = &notification_info_map_[it->first];
     info->total_count++;
     info->payload = it->second;
   }
 }
 
 void SyncManager::SyncInternal::OnIncomingNotification(
     const syncable::ModelTypePayloadMap& type_payloads) {
   if (!type_payloads.empty()) {
     if (syncer_thread()) {
-      syncer_thread()->NudgeSyncerWithPayloads(
-          kSyncerThreadDelayMsec,
-          SyncerThread::kNotification,
+      syncer_thread()->ScheduleNudgeWithPayloads(
+          TimeDelta::FromMilliseconds(kSyncerThreadDelayMsec),
+          browser_sync::NUDGE_SOURCE_NOTIFICATION,
           type_payloads, FROM_HERE);
     }
     allstatus_.IncrementNotificationsReceived();
     UpdateNotificationInfo(type_payloads);
   } else {
     LOG(WARNING) << "Sync received notification without any type information.";
   }
 
   if (parent_router_) {
     ListValue args;
@@ skipping 95 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