Sync: Convert AddressWatchThread from pthreads to chrome threads.

chrome/browser/sync/engine/syncapi.cc

 // Copyright (c) 2006-2009 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 "build/build_config.h"
 
 #if defined(OS_WIN)
 #include <windows.h>
 #include <iphlpapi.h>
 #endif
 
 #include <iomanip>
 #include <list>
 #include <string>
 #include <vector>
 
 #include "base/basictypes.h"
 #include "base/command_line.h"
 #include "base/platform_thread.h"
 #include "base/scoped_ptr.h"
 #include "base/string_util.h"
+#include "base/task.h"
 #include "chrome/browser/sync/engine/all_status.h"
 #include "chrome/browser/sync/engine/auth_watcher.h"
 #include "chrome/browser/sync/engine/change_reorder_buffer.h"
 #include "chrome/browser/sync/engine/client_command_channel.h"
 #include "chrome/browser/sync/engine/model_safe_worker.h"
 #include "chrome/browser/sync/engine/net/gaia_authenticator.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"
@@ skipping 31 matching lines @@
 using std::hex;
 using std::string;
 using std::vector;
 using syncable::Directory;
 using syncable::DirectoryManager;
 
 static const int kServerReachablePollingIntervalMsec = 60000 * 60;
 static const int kThreadExitTimeoutMsec = 60000;
 static const int kSSLPort = 443;
 
-struct ThreadParams {
+struct AddressWatchTaskParams {
   browser_sync::ServerConnectionManager* conn_mgr;
 #if defined(OS_WIN)
   HANDLE exit_flag;
 #endif
 };
 
 // This thread calls CheckServerReachable() whenever a change occurs in the
 // table that maps IP addresses to interfaces, for example when the user
 // unplugs his network cable.
-void* AddressWatchThread(void* arg) {
-  PlatformThread::SetName("SyncEngine_AddressWatcher");
-  LOG(INFO) << "starting the address watch thread";
+class AddressWatchTask : public Task {
+ public:
+  explicit AddressWatchTask(AddressWatchTaskParams* params)
+      : params_(params) {}
+  virtual ~AddressWatchTask() {}
+
+  virtual void Run() {
+    LOG(INFO) << "starting the address watch thread";
 #if defined(OS_WIN)
-  const ThreadParams* const params = reinterpret_cast<const ThreadParams*>(arg);
-  OVERLAPPED overlapped = {0};
-  overlapped.hEvent = CreateEvent(NULL, FALSE, TRUE, NULL);
-  HANDLE file;
-  DWORD rc = WAIT_OBJECT_0;
-  while (true) {
-    // Only call NotifyAddrChange() after the IP address has changed or if this
-    // is the first time through the loop.
-    if (WAIT_OBJECT_0 == rc) {
-      ResetEvent(overlapped.hEvent);
-      DWORD notify_result = NotifyAddrChange(&file, &overlapped);
-      if (ERROR_IO_PENDING != notify_result) {
-        LOG(ERROR) << "NotifyAddrChange() returned unexpected result "
-            << hex << notify_result;
+    OVERLAPPED overlapped = {0};
+    overlapped.hEvent = CreateEvent(NULL, FALSE, TRUE, NULL);
+    HANDLE file;
+    DWORD rc = WAIT_OBJECT_0;
+    while (true) {
+      // Only call NotifyAddrChange() after the IP address has changed or if
+      // this is the first time through the loop.
+      if (WAIT_OBJECT_0 == rc) {
+        ResetEvent(overlapped.hEvent);
+        DWORD notify_result = NotifyAddrChange(&file, &overlapped);
+        if (ERROR_IO_PENDING != notify_result) {
+          LOG(ERROR) << "NotifyAddrChange() returned unexpected result "
+              << hex << notify_result;
+          break;
+        }
+      }
+      HANDLE events[] = { overlapped.hEvent, params_->exit_flag };
+      rc = WaitForMultipleObjects(ARRAYSIZE(events), events, FALSE,
+                                  kServerReachablePollingIntervalMsec);
+
+      // If the exit flag was signaled, the thread will exit.
+      if (WAIT_OBJECT_0 + 1 == rc)
         break;
-      }
+
+      params_->conn_mgr->CheckServerReachable();
     }
-    HANDLE events[] = { overlapped.hEvent, params->exit_flag };
-    rc = WaitForMultipleObjects(ARRAYSIZE(events), events, FALSE,
-                                kServerReachablePollingIntervalMsec);
-
-    // If the exit flag was signaled, the thread will exit.
-    if (WAIT_OBJECT_0 + 1 == rc)
-      break;
-
-    params->conn_mgr->CheckServerReachable();
-  }
-  CloseHandle(overlapped.hEvent);
+    CloseHandle(overlapped.hEvent);
 #else
   // TODO(zork): Add this functionality to Linux.
 #endif
-  LOG(INFO) << "The address watch thread has stopped";
-  return 0;
-}
+    LOG(INFO) << "The address watch thread has stopped";
+  }
+
+ private:
+  AddressWatchTaskParams* const params_;
+  DISALLOW_COPY_AND_ASSIGN(AddressWatchTask);
+};
 
 namespace sync_api {
 class ModelSafeWorkerBridge;
 
 static const PSTR_CHAR kBookmarkSyncUserSettingsDatabase[] =
     PSTR("BookmarkSyncSettings.sqlite3");
 static const PSTR_CHAR kDefaultNameForNewNodes[] = PSTR(" ");
 
 // The list of names which are reserved for use by the server.
 static const char16* kForbiddenServerNames[] =
@@ skipping 536 matching lines @@
  private:
   const std::string gaia_source_;
   scoped_ptr<HttpPostProviderFactory> post_factory_;
   DISALLOW_COPY_AND_ASSIGN(BridgedGaiaAuthenticator);
 };
 
 //////////////////////////////////////////////////////////////////////////
 // SyncManager's implementation: SyncManager::SyncInternal
 class SyncManager::SyncInternal {
  public:
-  typedef PThreadScopedLock<PThreadMutex> MutexLock;
   explicit SyncInternal(SyncManager* sync_manager)
       : observer_(NULL),
         command_channel_(0),
         auth_problem_(AUTH_PROBLEM_NONE),
         sync_manager_(sync_manager),
         notification_pending_(false),
+        address_watch_thread_("SyncEngine_AddressWatcher"),
         initialized_(false) {
   }
 
   ~SyncInternal() { }
 
   bool Init(const PathString& database_location,
             const std::string& sync_server_and_path,
             int port,
             const char* gaia_service_id,
             const char* gaia_source,
@@ skipping 79 matching lines @@
 
   // See SyncManager::SetupForTestMode for information.
   void SetupForTestMode(const sync_char16* test_username);
 
   // See SyncManager::Shutdown for information.
   void Shutdown();
 
   // Whether we're initialized to the point of being able to accept changes
   // (and hence allow transaction creation). See initialized_ for details.
   bool initialized() const {
-    MutexLock lock(&initialized_mutex_);
+    AutoLock lock(initialized_mutex_);
     return initialized_;
   }
  private:
   // Try to authenticate using persisted credentials from a previous successful
   // authentication. If no such credentials exist, calls OnAuthError on the
   // client to collect credentials. Otherwise, there exist local credentials
   // that were once used for a successful auth, so we'll try to re-use these.
   // Failure of that attempt will be communicated as normal using OnAuthError.
   // Since this entry point will bypass normal GAIA authentication and try to
   // authenticate directly with the sync service using a cached token,
@@ skipping 104 matching lines @@
 
   // Our cache of a recent authentication problem. If no authentication problem
   // occurred, or if the last problem encountered has been cleared (by a
   // subsequent AuthWatcherEvent), this is set to AUTH_PROBLEM_NONE.
   AuthProblem auth_problem_;
 
   // The sync dir_manager to which we belong.
   SyncManager* const sync_manager_;
 
   // Parameters for our thread listening to network status changes.
-  ThreadParams address_watch_params_;
-  thread_handle address_watch_thread_;
+  base::Thread address_watch_thread_;
+  AddressWatchTaskParams address_watch_params_;
 
   // True if the next SyncCycle should notify peers of an update.
   bool notification_pending_;
 
   // Set to true once Init has been called, and we know of an authenticated
   // valid) username either from a fresh authentication attempt (as in
   // first-use case) or from a previous attempt stored in our UserSettings
   // (as in the steady-state), and the syncable::Directory has been opened,
   // meaning we are ready to accept changes.  Protected by initialized_mutex_
   // as it can get read/set by both the SyncerThread and the AuthWatcherThread.
   bool initialized_;
-  mutable PThreadMutex initialized_mutex_;
+  mutable Lock initialized_mutex_;
 };
 
 SyncManager::SyncManager() {
   data_ = new SyncInternal(this);
 }
 
 bool SyncManager::Init(const sync_char16* database_location,
                        const char* sync_server_and_path,
                        int sync_server_port,
                        const char* gaia_service_id,
@@ skipping 62 matching lines @@
   share_.dir_manager.reset(new DirectoryManager(database_location));
 
   string client_id = user_settings_->GetClientId();
   connection_manager_.reset(new SyncAPIServerConnectionManager(
       sync_server_and_path, port, use_ssl, user_agent, client_id));
 
   // TODO(timsteele): This is temporary windows crap needed to listen for
   // network status changes. We should either pump this up to the embedder to
   // do (and call us in CheckServerReachable, for ex), or at least make this
   // platform independent in here.
-  // TODO(ncarter): When this gets cleaned up, the implementation of
-  // CreatePThread can also be removed.
 #if defined(OS_WIN)
   HANDLE exit_flag = CreateEvent(NULL, TRUE /*manual reset*/, FALSE, NULL);
   address_watch_params_.exit_flag = exit_flag;
 #endif
   address_watch_params_.conn_mgr = connection_manager();
-  address_watch_thread_ = CreatePThread(AddressWatchThread,
-                                        &address_watch_params_);
-#if defined(OS_WIN)
-  DCHECK(NULL != address_watch_thread_);
-#endif
+
+  bool address_watch_started = address_watch_thread_.Start();
+  DCHECK(address_watch_started);
+  address_watch_thread_.message_loop()->PostTask(FROM_HERE,
+      new AddressWatchTask(&address_watch_params_));
 
   // Hand over the bridged POST factory to be owned by the connection
   // dir_manager.
   connection_manager()->SetHttpPostProviderFactory(post_factory);
 
   // Watch various objects for aggregated status.
   allstatus()->WatchConnectionManager(connection_manager());
 
   std::string gaia_url = browser_sync::kGaiaUrl;
   const char* service_id = gaia_service_id ?
@@ skipping 43 matching lines @@
   return true;
 }
 
 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.
   {
-    MutexLock lock(&initialized_mutex_);
+    AutoLock lock(initialized_mutex_);
     if (initialized_)
       return;
     initialized_ = true;
   }
 
   // Notify that initialization is complete.
   if (observer_)
     observer_->OnInitializationComplete();
 }
 
@@ skipping 107 matching lines @@
 
   // We don't want to process any more events.
   dir_change_hookup_.reset();
   syncer_event_.reset();
   authwatcher_hookup_.reset();
 
 #if defined(OS_WIN)
   // Stop the address watch thread by signaling the exit flag.
   // TODO(timsteele): Same as todo in Init().
   SetEvent(address_watch_params_.exit_flag);
-  const DWORD wait_result = WaitForSingleObject(address_watch_thread_,
-                                                kThreadExitTimeoutMsec);
-  LOG_IF(ERROR, WAIT_FAILED == wait_result) << "Waiting for addr change thread "
-      "to exit failed. GetLastError(): " << hex << GetLastError();
-  LOG_IF(ERROR, WAIT_TIMEOUT == wait_result) << "Thread exit timeout expired";
+#endif
+
+  address_watch_thread_.Stop();
+
+#if defined(OS_WIN)
   CloseHandle(address_watch_params_.exit_flag);
 #endif
 }
 
 // Listen to model changes, filter out ones initiated by the sync API, and
 // saves the rest (hopefully just backend Syncer changes resulting from
 // ApplyUpdates) to data_->changelist.
 void SyncManager::SyncInternal::HandleChangeEvent(
     const syncable::DirectoryChangeEvent& event) {
   if (event.todo == syncable::DirectoryChangeEvent::TRANSACTION_COMPLETE) {
@@ skipping 327 matching lines @@
 BaseTransaction::~BaseTransaction() {
   delete lookup_;
 }
 
 UserShare* SyncManager::GetUserShare() const {
   DCHECK(data_->initialized()) << "GetUserShare requires initialization!";
   return data_->GetUserShare();
 }
 
 }  // namespace sync_api