Bug 932728: Run DBus on the I/O thread, r=qdot
authorThomas Zimmermann <tdz@users.sourceforge.net>
Tue, 29 Oct 2013 16:03:40 +0100
changeset 160194 0ddb34970ecb74ec84e58dfe8ca8e95bc70a9bf8
parent 160193 93089287545bf81c60670d4620a848636b33bb05
child 160195 5e7da9785fb6bd70b5a934177c3c2d21a1404d38
push id3953
push userkwierso@gmail.com
push dateFri, 13 Dec 2013 03:25:06 +0000
treeherderfx-team@1bc33fa19b24 [default view] [failures only]
perfherder[talos] [build metrics] [platform microbench] (compared to previous push)
reviewersqdot
bugs932728
milestone29.0a1
Bug 932728: Run DBus on the I/O thread, r=qdot This patch converts DBusWatcher to run on the I/O thread. When a caller starts DBus, StartDBus creates a new connection and adds it to the I/O thread's poll loop. DBusWatchers are created and removed automatically by the DBus libary. The I/O thread provides all features of the DBus thread. So most of the existing code has been removed or rewritten. The former includes the control socket and the DBus thread, the latter is in the DBusWatcher code.
ipc/dbus/DBusThread.cpp
ipc/dbus/DBusThread.h
ipc/dbus/RawDBusConnection.cpp
--- a/ipc/dbus/DBusThread.cpp
+++ b/ipc/dbus/DBusThread.cpp
@@ -61,619 +61,253 @@
 #if defined(MOZ_WIDGET_GONK)
 #include <android/log.h>
 #define LOG(args...)  __android_log_print(ANDROID_LOG_INFO, "GonkDBus", args);
 #else
 #define BTDEBUG true
 #define LOG(args...) if (BTDEBUG) printf(args);
 #endif
 
-#define DEFAULT_INITIAL_POLLFD_COUNT 8
-
 namespace mozilla {
 namespace ipc {
 
-class DBusWatcher
+class DBusWatcher : public MessageLoopForIO::Watcher
 {
 public:
-  DBusWatcher()
-  : mConnection(nullptr)
-  { }
-
-  ~DBusWatcher()
+  DBusWatcher(RawDBusConnection* aConnection, DBusWatch* aWatch)
+  : mConnection(aConnection),
+    mWatch(aWatch)
   {
-    // Connection has been released
-    MOZ_ASSERT(!mConnection);
+    MOZ_ASSERT(mConnection);
+    MOZ_ASSERT(mWatch);
   }
 
-  bool Initialize();
-  void CleanUp();
-
-  void WakeUp();
-  bool Stop();
+  ~DBusWatcher()
+  { }
 
-  bool Poll();
+  void StartWatching();
+  void StopWatching();
 
-  bool AddWatch(DBusWatch* aWatch);
-  void RemoveWatch(DBusWatch* aWatch);
-
-  void HandleWatchAdd();
-  void HandleWatchRemove();
+  static void        FreeFunction(void* aData);
+  static dbus_bool_t AddWatchFunction(DBusWatch* aWatch, void* aData);
+  static void        RemoveWatchFunction(DBusWatch* aWatch, void* aData);
+  static void        ToggleWatchFunction(DBusWatch* aWatch, void* aData);
 
   RawDBusConnection* GetConnection();
 
 private:
-  struct PollFdComparator {
-    bool Equals(const pollfd& a, const pollfd& b) const {
-      return ((a.fd == b.fd) && (a.events == b.events));
-    }
-    bool LessThan(const pollfd& a, const pollfd&b) const {
-      return false;
-    }
-  };
+  void OnFileCanReadWithoutBlocking(int aFd);
+  void OnFileCanWriteWithoutBlocking(int aFd);
 
-  enum DBusEventTypes {
-    DBUS_EVENT_LOOP_EXIT = 1,
-    DBUS_EVENT_LOOP_ADD = 2,
-    DBUS_EVENT_LOOP_REMOVE = 3,
-    DBUS_EVENT_LOOP_WAKEUP = 4
-  };
-
-  static unsigned int UnixEventsToDBusFlags(short events);
-  static short        DBusFlagsToUnixEvents(unsigned int flags);
+  // Read watcher for libevent. Only to be accessed on IO Thread.
+  MessageLoopForIO::FileDescriptorWatcher mReadWatcher;
 
-  static dbus_bool_t AddWatchFunction(DBusWatch* aWatch, void* aData);
-  static void        RemoveWatchFunction(DBusWatch* aWatch, void* aData);
-  static void        ToggleWatchFunction(DBusWatch* aWatch, void* aData);
-  static void        DBusWakeupFunction(void* aData);
-
-  bool SetUp();
+  // Write watcher for libevent. Only to be accessed on IO Thread.
+  MessageLoopForIO::FileDescriptorWatcher mWriteWatcher;
 
-  // Information about the sockets we're polling. Socket counts
-  // increase/decrease depending on how many add/remove watch signals
-  // we're received via the control sockets.
-  nsTArray<pollfd> mPollData;
-  nsTArray<DBusWatch*> mWatchData;
-
-  // Sockets for receiving dbus control information (watch
-  // add/removes, loop shutdown, etc...)
-  ScopedClose mControlFdR;
-  ScopedClose mControlFdW;
-
+  // DBus structures
   RawDBusConnection* mConnection;
+  DBusWatch* mWatch;
 };
 
 RawDBusConnection*
 DBusWatcher::GetConnection()
 {
   return mConnection;
 }
 
-bool
-DBusWatcher::Initialize()
+void DBusWatcher::StartWatching()
 {
-  if (!SetUp()) {
-    CleanUp();
-    return false;
-  }
+  MOZ_ASSERT(!NS_IsMainThread());
+  MOZ_ASSERT(mWatch);
+
+  int fd = dbus_watch_get_unix_fd(mWatch);
 
-  return true;
+  MessageLoopForIO* ioLoop = MessageLoopForIO::current();
+  ioLoop->WatchFileDescriptor(fd, true, MessageLoopForIO::WATCH_READ,
+                              &mReadWatcher, this);
+  ioLoop->WatchFileDescriptor(fd, true, MessageLoopForIO::WATCH_WRITE,
+                              &mWriteWatcher, this);
 }
 
-void
-DBusWatcher::CleanUp()
+void DBusWatcher::StopWatching()
 {
   MOZ_ASSERT(!NS_IsMainThread());
 
-  dbus_connection_set_wakeup_main_function(mConnection->GetConnection(),
-                                           nullptr, nullptr, nullptr);
-  dbus_bool_t success =
-    dbus_connection_set_watch_functions(mConnection->GetConnection(),
-                                        nullptr, nullptr, nullptr,
-                                        nullptr, nullptr);
-  if (success != TRUE) {
-    NS_WARNING("dbus_connection_set_watch_functions failed");
-  }
-
-#ifdef DEBUG
-  LOG("Removing DBus Sockets\n");
-#endif
-  if (mControlFdW.get()) {
-    mControlFdW.dispose();
-  }
-  if (mControlFdR.get()) {
-    mControlFdR.dispose();
-  }
-  mPollData.Clear();
-
-  // DBusWatch pointers are maintained by DBus, so we won't leak by
-  // clearing.
-  mWatchData.Clear();
-
-  delete mConnection;
-  mConnection = nullptr;
-}
-
-void
-DBusWatcher::WakeUp()
-{
-  static const char control = DBUS_EVENT_LOOP_WAKEUP;
-
-  struct pollfd fds = {
-    mControlFdW.get(),
-    POLLOUT,
-    0
-  };
-
-  int nfds = TEMP_FAILURE_RETRY(poll(&fds, 1, 0));
-  NS_ENSURE_TRUE_VOID(nfds == 1);
-  NS_ENSURE_TRUE_VOID(fds.revents == POLLOUT);
-
-  ssize_t res = TEMP_FAILURE_RETRY(
-    write(mControlFdW.get(), &control, sizeof(control)));
-  if (res < 0) {
-    NS_WARNING("Cannot write wakeup bit to DBus controller!");
-  }
-}
-
-bool
-DBusWatcher::Stop()
-{
-  static const char data = DBUS_EVENT_LOOP_EXIT;
-
-  ssize_t res =
-    TEMP_FAILURE_RETRY(write(mControlFdW.get(), &data, sizeof(data)));
-  NS_ENSURE_TRUE(res == 1, false);
-
-  return true;
-}
-
-bool
-DBusWatcher::Poll()
-{
-  int res = TEMP_FAILURE_RETRY(poll(mPollData.Elements(),
-                                    mPollData.Length(), -1));
-  NS_ENSURE_TRUE(res > 0, false);
-
-  bool continueThread = true;
-
-  nsTArray<pollfd>::size_type i = 0;
-
-  while (i < mPollData.Length()) {
-    if (mPollData[i].revents == POLLIN) {
-      if (mPollData[i].fd == mControlFdR.get()) {
-        char data;
-        res = TEMP_FAILURE_RETRY(read(mControlFdR.get(), &data, sizeof(data)));
-        NS_ENSURE_TRUE(res > 0, false);
-
-        switch (data) {
-          case DBUS_EVENT_LOOP_EXIT:
-            continueThread = false;
-            break;
-          case DBUS_EVENT_LOOP_ADD:
-            HandleWatchAdd();
-            break;
-          case DBUS_EVENT_LOOP_REMOVE:
-            HandleWatchRemove();
-            // don't increment i, or we'll skip one element
-            continue;
-          case DBUS_EVENT_LOOP_WAKEUP:
-            NS_ProcessPendingEvents(NS_GetCurrentThread(),
-                                    PR_INTERVAL_NO_TIMEOUT);
-            break;
-          default:
-#if DEBUG
-            nsCString warning("unknown command ");
-            warning.AppendInt(data);
-            NS_WARNING(warning.get());
-#endif
-            break;
-        }
-      } else {
-        short events = mPollData[i].revents;
-        mPollData[i].revents = 0;
-
-        dbus_watch_handle(mWatchData[i], UnixEventsToDBusFlags(events));
-
-        DBusDispatchStatus dbusDispatchStatus;
-        do {
-          dbusDispatchStatus =
-            dbus_connection_dispatch(mConnection->GetConnection());
-        } while (dbusDispatchStatus == DBUS_DISPATCH_DATA_REMAINS);
-
-        // Break at this point since we don't know if the operation
-        // was destructive
-        break;
-      }
-    }
-
-    ++i;
-  }
-
-  return continueThread;
-}
-
-bool
-DBusWatcher::AddWatch(DBusWatch* aWatch)
-{
-  static const char control = DBUS_EVENT_LOOP_ADD;
-
-  if (dbus_watch_get_enabled(aWatch) == FALSE) {
-    return true;
-  }
-
-  // note that we can't just send the watch and inspect it later
-  // because we may get a removeWatch call before this data is reacted
-  // to by our eventloop and remove this watch..  reading the add first
-  // and then inspecting the recently deceased watch would be bad.
-  ssize_t res =
-    TEMP_FAILURE_RETRY(write(mControlFdW.get(),&control, sizeof(control)));
-  if (res < 0) {
-    LOG("Cannot write DBus add watch control data to socket!\n");
-    return false;
-  }
-
-  int fd = dbus_watch_get_unix_fd(aWatch);
-  res = TEMP_FAILURE_RETRY(write(mControlFdW.get(), &fd, sizeof(fd)));
-  if (res < 0) {
-    LOG("Cannot write DBus add watch descriptor data to socket!\n");
-    return false;
-  }
-
-  unsigned int flags = dbus_watch_get_flags(aWatch);
-  res = TEMP_FAILURE_RETRY(write(mControlFdW.get(), &flags, sizeof(flags)));
-  if (res < 0) {
-    LOG("Cannot write DBus add watch flag data to socket!\n");
-    return false;
-  }
-
-  res = TEMP_FAILURE_RETRY(write(mControlFdW.get(), &aWatch, sizeof(aWatch)));
-  if (res < 0) {
-    LOG("Cannot write DBus add watch struct data to socket!\n");
-    return false;
-  }
-
-  return true;
-}
-
-void
-DBusWatcher::RemoveWatch(DBusWatch* aWatch)
-{
-  static const char control = DBUS_EVENT_LOOP_REMOVE;
-
-  ssize_t res =
-    TEMP_FAILURE_RETRY(write(mControlFdW.get(), &control, sizeof(control)));
-  if (res < 0) {
-    LOG("Cannot write DBus remove watch control data to socket!\n");
-    return;
-  }
-
-  int fd = dbus_watch_get_unix_fd(aWatch);
-  res = TEMP_FAILURE_RETRY(write(mControlFdW.get(), &fd, sizeof(fd)));
-  if (res < 0) {
-    LOG("Cannot write DBus remove watch descriptor data to socket!\n");
-    return;
-  }
-
-  unsigned int flags = dbus_watch_get_flags(aWatch);
-  res = TEMP_FAILURE_RETRY(write(mControlFdW.get(), &flags, sizeof(flags)));
-  if (res < 0) {
-    LOG("Cannot write DBus remove watch flag data to socket!\n");
-    return;
-  }
-}
-
-void
-DBusWatcher::HandleWatchAdd()
-{
-  int fd;
-  ssize_t res = TEMP_FAILURE_RETRY(read(mControlFdR.get(), &fd, sizeof(fd)));
-  if (res < 0) {
-    LOG("Cannot read DBus watch add descriptor data from socket!\n");
-    return;
-  }
-
-  unsigned int flags;
-  res = TEMP_FAILURE_RETRY(read(mControlFdR.get(), &flags, sizeof(flags)));
-  if (res < 0) {
-    LOG("Cannot read DBus watch add flag data from socket!\n");
-    return;
-  }
-
-  DBusWatch* watch;
-  res = TEMP_FAILURE_RETRY(read(mControlFdR.get(), &watch, sizeof(watch)));
-  if (res < 0) {
-    LOG("Cannot read DBus watch add watch data from socket!\n");
-    return;
-  }
-
-  struct pollfd p = {
-    fd, // .fd
-    DBusFlagsToUnixEvents(flags), // .events
-    0 // .revents
-  };
-  if (mPollData.Contains(p, PollFdComparator())) {
-    return;
-  }
-  mPollData.AppendElement(p);
-  mWatchData.AppendElement(watch);
-}
-
-void
-DBusWatcher::HandleWatchRemove()
-{
-  int fd;
-  ssize_t res = TEMP_FAILURE_RETRY(read(mControlFdR.get(), &fd, sizeof(fd)));
-  if (res < 0) {
-    LOG("Cannot read DBus watch remove descriptor data from socket!\n");
-    return;
-  }
-
-  unsigned int flags;
-  res = TEMP_FAILURE_RETRY(read(mControlFdR.get(), &flags, sizeof(flags)));
-  if (res < 0) {
-    LOG("Cannot read DBus watch remove flag data from socket!\n");
-    return;
-  }
-
-  struct pollfd p = {
-    fd, // .fd
-    DBusFlagsToUnixEvents(flags), // .events
-    0 // .revents
-  };
-  int index = mPollData.IndexOf(p, 0, PollFdComparator());
-  // There are times where removes can be requested for watches that
-  // haven't been added (for example, whenever gecko comes up after
-  // adapters have already been enabled), so check to make sure we're
-  // using the watch in the first place
-  if (index < 0) {
-    LOG("DBus requested watch removal of non-existant socket, ignoring...");
-    return;
-  }
-  mPollData.RemoveElementAt(index);
-
-  // DBusWatch pointers are maintained by DBus, so we won't leak by
-  // removing.
-  mWatchData.RemoveElementAt(index);
-}
-
-// Flag conversion
-
-unsigned int
-DBusWatcher::UnixEventsToDBusFlags(short events)
-{
-  return (events & DBUS_WATCH_READABLE ? POLLIN : 0) |
-         (events & DBUS_WATCH_WRITABLE ? POLLOUT : 0) |
-         (events & DBUS_WATCH_ERROR ? POLLERR : 0) |
-         (events & DBUS_WATCH_HANGUP ? POLLHUP : 0);
-}
-
-short
-DBusWatcher::DBusFlagsToUnixEvents(unsigned int flags)
-{
-  return (flags & POLLIN ? DBUS_WATCH_READABLE : 0) |
-         (flags & POLLOUT ? DBUS_WATCH_WRITABLE : 0) |
-         (flags & POLLERR ? DBUS_WATCH_ERROR : 0) |
-         (flags & POLLHUP ? DBUS_WATCH_HANGUP : 0);
+  mReadWatcher.StopWatchingFileDescriptor();
+  mWriteWatcher.StopWatchingFileDescriptor();
 }
 
 // DBus utility functions, used as function pointers in DBus setup
 
+void
+DBusWatcher::FreeFunction(void* aData)
+{
+  delete static_cast<DBusWatcher*>(aData);
+}
+
 dbus_bool_t
 DBusWatcher::AddWatchFunction(DBusWatch* aWatch, void* aData)
 {
-  MOZ_ASSERT(aData);
-  DBusWatcher* dbusWatcher = static_cast<DBusWatcher*>(aData);
-  return dbusWatcher->AddWatch(aWatch);
+  MOZ_ASSERT(!NS_IsMainThread());
+
+  RawDBusConnection* connection = static_cast<RawDBusConnection*>(aData);
+
+  DBusWatcher* dbusWatcher = new DBusWatcher(connection, aWatch);
+  dbus_watch_set_data(aWatch, dbusWatcher, DBusWatcher::FreeFunction);
+
+  if (dbus_watch_get_enabled(aWatch)) {
+    dbusWatcher->StartWatching();
+  }
+
+  return TRUE;
 }
 
 void
 DBusWatcher::RemoveWatchFunction(DBusWatch* aWatch, void* aData)
 {
-  MOZ_ASSERT(aData);
-  DBusWatcher* dbusWatcher = static_cast<DBusWatcher*>(aData);
-  dbusWatcher->RemoveWatch(aWatch);
+  MOZ_ASSERT(!NS_IsMainThread());
+
+  DBusWatcher* dbusWatcher =
+    static_cast<DBusWatcher*>(dbus_watch_get_data(aWatch));
+  dbusWatcher->StopWatching();
 }
 
 void
 DBusWatcher::ToggleWatchFunction(DBusWatch* aWatch, void* aData)
 {
-  MOZ_ASSERT(aData);
-  DBusWatcher* dbusWatcher = static_cast<DBusWatcher*>(aData);
+  MOZ_ASSERT(!NS_IsMainThread());
+
+  DBusWatcher* dbusWatcher =
+    static_cast<DBusWatcher*>(dbus_watch_get_data(aWatch));
 
   if (dbus_watch_get_enabled(aWatch)) {
-    dbusWatcher->AddWatch(aWatch);
+    dbusWatcher->StartWatching();
   } else {
-    dbusWatcher->RemoveWatch(aWatch);
+    dbusWatcher->StopWatching();
   }
 }
 
 void
-DBusWatcher::DBusWakeupFunction(void* aData)
+DBusWatcher::OnFileCanReadWithoutBlocking(int aFd)
 {
-  MOZ_ASSERT(aData);
-  DBusWatcher* dbusWatcher = static_cast<DBusWatcher*>(aData);
-  dbusWatcher->WakeUp();
+  MOZ_ASSERT(!NS_IsMainThread());
+
+  dbus_watch_handle(mWatch, DBUS_WATCH_READABLE);
+
+  DBusDispatchStatus dbusDispatchStatus;
+  do {
+    dbusDispatchStatus =
+      dbus_connection_dispatch(mConnection->GetConnection());
+  } while (dbusDispatchStatus == DBUS_DISPATCH_DATA_REMAINS);
 }
 
-bool
-DBusWatcher::SetUp()
+void
+DBusWatcher::OnFileCanWriteWithoutBlocking(int aFd)
 {
   MOZ_ASSERT(!NS_IsMainThread());
 
-  // If we already have a connection, exit
-  if (mConnection) {
-    return false;
-  }
+  dbus_watch_handle(mWatch, DBUS_WATCH_WRITABLE);
+}
 
-  // socketpair opens two sockets for the process to communicate on.
-  // This is how android's implementation of the dbus event loop
-  // communicates with itself in relation to IPC signals. These
-  // sockets are contained sequentially in the same struct in the
-  // android code, but we break them out into class members here.
-  // Therefore we read into a local array and then copy.
-
-  int sockets[2];
-  if (socketpair(AF_LOCAL, SOCK_STREAM, 0, sockets) < 0) {
-    return false;
+class WatchDBusConnectionTask : public Task
+{
+public:
+  WatchDBusConnectionTask(RawDBusConnection* aConnection)
+  : mConnection(aConnection)
+  {
+    MOZ_ASSERT(mConnection);
   }
 
-  mControlFdR.rwget() = sockets[0];
-  mControlFdW.rwget() = sockets[1];
-
-  pollfd* p = mPollData.AppendElement();
-
-  p->fd = mControlFdR.get();
-  p->events = POLLIN;
-  p->revents = 0;
-
-  // Due to the fact that mPollData and mWatchData have to match, we
-  // push a null to the front of mWatchData since it has the control
-  // fd in the first slot of mPollData.
-
-  mWatchData.AppendElement(static_cast<DBusWatch*>(nullptr));
-
-  RawDBusConnection* connection = new RawDBusConnection();
-
-  // If we can't establish a connection to dbus, nothing else will work
-  nsresult rv = connection->EstablishDBusConnection();
-  if (NS_FAILED(rv)) {
-    NS_WARNING("Cannot create DBus Connection for DBus Thread!");
-    return false;
-  }
-
-  dbus_bool_t success =
-    dbus_connection_set_watch_functions(connection->GetConnection(),
-                                        AddWatchFunction, RemoveWatchFunction,
-                                        ToggleWatchFunction, this, nullptr);
-  NS_ENSURE_TRUE(success == TRUE, false);
-
-  dbus_connection_set_wakeup_main_function(connection->GetConnection(),
-                                           DBusWakeupFunction,
-                                           this, nullptr);
-
-  mConnection = connection;
-
-  return true;
-}
-
-// Main task for polling the DBus system
-
-class DBusPollTask : public nsRunnable
-{
-public:
-  DBusPollTask(DBusWatcher* aDBusWatcher)
-  : mDBusWatcher(aDBusWatcher)
-  { }
-
-  NS_IMETHOD Run()
+  void Run()
   {
     MOZ_ASSERT(!NS_IsMainThread());
 
-    bool continueThread;
-
-    do {
-      continueThread = mDBusWatcher->Poll();
-    } while (continueThread);
-
-    mDBusWatcher->CleanUp();
-
-    nsIThread* thread;
-    nsresult rv = NS_GetCurrentThread(&thread);
-    NS_ENSURE_SUCCESS(rv, rv);
-
-    nsRefPtr<nsIRunnable> runnable =
-      NS_NewRunnableMethod(thread, &nsIThread::Shutdown);
-    rv = NS_DispatchToMainThread(runnable);
-    NS_ENSURE_SUCCESS(rv, rv);
-
-    return NS_OK;
+    dbus_bool_t success =
+      dbus_connection_set_watch_functions(mConnection->GetConnection(),
+                                          DBusWatcher::AddWatchFunction,
+                                          DBusWatcher::RemoveWatchFunction,
+                                          DBusWatcher::ToggleWatchFunction,
+                                          mConnection, nullptr);
+    NS_ENSURE_TRUE_VOID(success == TRUE);
   }
 
 private:
-  nsAutoPtr<DBusWatcher> mDBusWatcher;
+  RawDBusConnection* mConnection;
 };
 
-static DBusWatcher*            gDBusWatcher;
-static StaticRefPtr<nsIThread> gDBusServiceThread;
+class DeleteDBusConnectionTask : public Task
+{
+public:
+  DeleteDBusConnectionTask(RawDBusConnection* aConnection)
+  : mConnection(aConnection)
+  {
+    MOZ_ASSERT(mConnection);
+  }
+
+  void Run()
+  {
+    MOZ_ASSERT(!NS_IsMainThread());
 
-// Startup/Shutdown utility functions
+    // This command closes the DBus connection and all instances of
+    // DBusWatch will be removed and free'd.
+    delete mConnection;
+  }
+
+private:
+  RawDBusConnection* mConnection;
+};
+
+ // Startup/Shutdown utility functions
+
+static RawDBusConnection* gDBusConnection;
 
 bool
 StartDBus()
 {
   MOZ_ASSERT(!NS_IsMainThread());
-  NS_ENSURE_TRUE(!gDBusWatcher, true);
-
-  nsAutoPtr<DBusWatcher> dbusWatcher(new DBusWatcher());
-
-  bool eventLoopStarted = dbusWatcher->Initialize();
-  NS_ENSURE_TRUE(eventLoopStarted, false);
-
-  nsresult rv;
+  NS_ENSURE_TRUE(!gDBusConnection, true);
 
-  if (!gDBusServiceThread) {
-    nsIThread* dbusServiceThread;
-    rv = NS_NewNamedThread("DBus Thread", &dbusServiceThread);
-    NS_ENSURE_SUCCESS(rv, false);
-    gDBusServiceThread = dbusServiceThread;
-  }
-
-#ifdef DEBUG
-  LOG("DBus Thread Starting\n");
-#endif
-
-  nsRefPtr<nsIRunnable> pollTask(new DBusPollTask(dbusWatcher));
-  NS_ENSURE_TRUE(pollTask, false);
-
-  rv = gDBusServiceThread->Dispatch(pollTask, NS_DISPATCH_NORMAL);
+  RawDBusConnection* connection = new RawDBusConnection();
+  nsresult rv = connection->EstablishDBusConnection();
   NS_ENSURE_SUCCESS(rv, false);
 
-  gDBusWatcher = dbusWatcher.forget();
+  XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
+    new WatchDBusConnectionTask(connection));
+
+  gDBusConnection = connection;
 
   return true;
 }
 
 bool
 StopDBus()
 {
   MOZ_ASSERT(!NS_IsMainThread());
-  NS_ENSURE_TRUE(gDBusServiceThread, true);
-
-  DBusWatcher* dbusWatcher = gDBusWatcher;
-  gDBusWatcher = nullptr;
+  NS_ENSURE_TRUE(gDBusConnection, true);
 
-  if (dbusWatcher && !dbusWatcher->Stop()) {
-    return false;
-  }
+  RawDBusConnection* connection = gDBusConnection;
+  gDBusConnection = nullptr;
 
-  gDBusServiceThread = nullptr;
+  XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
+    new DeleteDBusConnectionTask(connection));
 
   return true;
 }
 
 nsresult
-DispatchToDBusThread(nsIRunnable* event)
+DispatchToDBusThread(Task* task)
 {
-  nsRefPtr<nsIThread> dbusServiceThread(gDBusServiceThread);
-  DBusWatcher* dbusWatcher = gDBusWatcher;
-
-  NS_ENSURE_TRUE(dbusServiceThread.get() && dbusWatcher,
-                 NS_ERROR_NOT_INITIALIZED);
-
-  nsresult rv = dbusServiceThread->Dispatch(event, NS_DISPATCH_NORMAL);
-  NS_ENSURE_SUCCESS(rv, rv);
-
-  dbusWatcher->WakeUp();
+  XRE_GetIOMessageLoop()->PostTask(FROM_HERE, task);
 
   return NS_OK;
 }
 
 RawDBusConnection*
 GetDBusConnection()
 {
-  NS_ENSURE_TRUE(gDBusWatcher, nullptr);
+  NS_ENSURE_TRUE(gDBusConnection, nullptr);
 
-  return gDBusWatcher->GetConnection();
+  return gDBusConnection;
 }
 
 }
 }
--- a/ipc/dbus/DBusThread.h
+++ b/ipc/dbus/DBusThread.h
@@ -4,17 +4,17 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
  * You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 #ifndef mozilla_ipc_dbus_gonk_dbusthread_h__
 #define mozilla_ipc_dbus_gonk_dbusthread_h__
 
 #include "nscore.h"
 
-class nsIRunnable;
+class Task;
 
 namespace mozilla {
 namespace ipc {
 
 class RawDBusConnection;
 
 /**
  * Starts the DBus thread, which handles returning signals to objects
@@ -29,23 +29,28 @@ bool StartDBus();
  * Stop the DBus thread, assuming it's currently running. Should be
  * called from main thread.
  *
  * @return True on thread stopping correctly, false otherwise
  */
 bool StopDBus();
 
 /**
- * Dispatch an event to the DBus thread
+ * Dispatch a task to the DBus I/O thread
  *
- * @param event An nsIRunnable to run in the DBus thread
+ * @param task A task to run on the DBus I/O thread
  * @return NS_OK on success, or an error code otherwise
  */
 nsresult
-DispatchToDBusThread(nsIRunnable* event);
+DispatchToDBusThread(Task* task);
 
+/**
+ * Returns the connection to the DBus server
+ *
+ * @return The DBus connection on success, or nullptr otherwise
+ */
 RawDBusConnection*
 GetDBusConnection(void);
 
 }
 }
 
 #endif
--- a/ipc/dbus/RawDBusConnection.cpp
+++ b/ipc/dbus/RawDBusConnection.cpp
@@ -1,15 +1,16 @@
 /* -*- Mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; tab-width: 40 -*- */
 /* vim: set ts=2 et sw=2 tw=80: */
 /* This Source Code Form is subject to the terms of the Mozilla Public
  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
  * You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 #include <dbus/dbus.h>
+#include "base/message_loop.h"
 #include "mozilla/Monitor.h"
 #include "nsThreadUtils.h"
 #include "DBusThread.h"
 #include "DBusUtils.h"
 #include "RawDBusConnection.h"
 
 #ifdef LOG
 #undef LOG
@@ -29,68 +30,67 @@ using namespace mozilla::ipc;
 
 //
 // Runnables
 //
 
 namespace mozilla {
 namespace ipc {
 
-class DBusConnectionSendRunnableBase : public nsRunnable
+class DBusConnectionSendTaskBase : public Task
 {
+public:
+  virtual ~DBusConnectionSendTaskBase()
+  { }
+
 protected:
-  DBusConnectionSendRunnableBase(DBusConnection* aConnection,
-                                 DBusMessage* aMessage)
+  DBusConnectionSendTaskBase(DBusConnection* aConnection,
+                             DBusMessage* aMessage)
   : mConnection(aConnection),
     mMessage(aMessage)
   {
     MOZ_ASSERT(mConnection);
     MOZ_ASSERT(mMessage);
   }
 
-  virtual ~DBusConnectionSendRunnableBase()
-  { }
-
   DBusConnection*   mConnection;
   DBusMessageRefPtr mMessage;
 };
 
 //
 // Sends a message and returns the message's serial number to the
 // disaptching thread. Only run it in DBus thread.
 //
-class DBusConnectionSendRunnable : public DBusConnectionSendRunnableBase
+class DBusConnectionSendTask : public DBusConnectionSendTaskBase
 {
 public:
-  DBusConnectionSendRunnable(DBusConnection* aConnection,
-                             DBusMessage* aMessage)
-  : DBusConnectionSendRunnableBase(aConnection, aMessage)
+  DBusConnectionSendTask(DBusConnection* aConnection,
+                         DBusMessage* aMessage)
+  : DBusConnectionSendTaskBase(aConnection, aMessage)
   { }
 
-  NS_IMETHOD Run()
-  {
-    MOZ_ASSERT(!NS_IsMainThread());
-
-    dbus_bool_t success = dbus_connection_send(mConnection, mMessage, nullptr);
+  virtual ~DBusConnectionSendTask()
+  { }
 
-    NS_ENSURE_TRUE(success == TRUE, NS_ERROR_FAILURE);
+  void Run() MOZ_OVERRIDE
+  {
+    MOZ_ASSERT(MessageLoop::current());
 
-    return NS_OK;
+    dbus_bool_t success = dbus_connection_send(mConnection,
+                                               mMessage,
+                                               nullptr);
+    NS_ENSURE_TRUE_VOID(success == TRUE);
   }
-
-protected:
-  ~DBusConnectionSendRunnable()
-  { }
 };
 
 //
 // Sends a message and executes a callback function for the reply. Only
 // run it in DBus thread.
 //
-class DBusConnectionSendWithReplyRunnable : public DBusConnectionSendRunnableBase
+class DBusConnectionSendWithReplyTask : public DBusConnectionSendTaskBase
 {
 private:
   class NotifyData
   {
   public:
     NotifyData(DBusReplyCallback aCallback, void* aData)
     : mCallback(aCallback),
       mData(aData)
@@ -125,56 +125,53 @@ private:
       dbus_message_unref(reply);
     }
 
     dbus_pending_call_cancel(aCall);
     dbus_pending_call_unref(aCall);
   }
 
 public:
-  DBusConnectionSendWithReplyRunnable(DBusConnection* aConnection,
-                                      DBusMessage* aMessage,
-                                      int aTimeout,
-                                      DBusReplyCallback aCallback,
-                                      void* aData)
-  : DBusConnectionSendRunnableBase(aConnection, aMessage),
+  DBusConnectionSendWithReplyTask(DBusConnection* aConnection,
+                                  DBusMessage* aMessage,
+                                  int aTimeout,
+                                  DBusReplyCallback aCallback,
+                                  void* aData)
+  : DBusConnectionSendTaskBase(aConnection, aMessage),
     mCallback(aCallback),
     mData(aData),
     mTimeout(aTimeout)
   { }
 
-  NS_IMETHOD Run()
+  virtual ~DBusConnectionSendWithReplyTask()
+  { }
+
+  void Run() MOZ_OVERRIDE
   {
-    MOZ_ASSERT(!NS_IsMainThread());
+    MOZ_ASSERT(MessageLoop::current());
 
     // Freed at end of Notify
     nsAutoPtr<NotifyData> data(new NotifyData(mCallback, mData));
-    NS_ENSURE_TRUE(data, NS_ERROR_OUT_OF_MEMORY);
+    NS_ENSURE_TRUE_VOID(data);
 
     DBusPendingCall* call;
 
     dbus_bool_t success = dbus_connection_send_with_reply(mConnection,
                                                           mMessage,
                                                           &call,
                                                           mTimeout);
-    NS_ENSURE_TRUE(success == TRUE, NS_ERROR_FAILURE);
+    NS_ENSURE_TRUE_VOID(success == TRUE);
 
     success = dbus_pending_call_set_notify(call, Notify, data, nullptr);
-    NS_ENSURE_TRUE(success == TRUE, NS_ERROR_FAILURE);
+    NS_ENSURE_TRUE_VOID(success == TRUE);
 
     data.forget();
     dbus_message_unref(mMessage);
-
-    return NS_OK;
   };
 
-protected:
-  ~DBusConnectionSendWithReplyRunnable()
-  { }
-
 private:
   DBusReplyCallback mCallback;
   void*             mData;
   int               mTimeout;
 };
 
 }
 }
@@ -216,18 +213,18 @@ void RawDBusConnection::ScopedDBusConnec
   if (ptr) {
     dbus_connection_close(ptr);
     dbus_connection_unref(ptr);
   }
 }
 
 bool RawDBusConnection::Send(DBusMessage* aMessage)
 {
-  nsRefPtr<DBusConnectionSendRunnable> t(
-    new DBusConnectionSendRunnable(mConnection, aMessage));
+  DBusConnectionSendTask* t =
+    new DBusConnectionSendTask(mConnection, aMessage);
   MOZ_ASSERT(t);
 
   nsresult rv = DispatchToDBusThread(t);
 
   if (NS_FAILED(rv)) {
     if (aMessage) {
       dbus_message_unref(aMessage);
     }
@@ -237,19 +234,19 @@ bool RawDBusConnection::Send(DBusMessage
   return true;
 }
 
 bool RawDBusConnection::SendWithReply(DBusReplyCallback aCallback,
                                       void* aData,
                                       int aTimeout,
                                       DBusMessage* aMessage)
 {
-  nsRefPtr<nsIRunnable> t(
-    new DBusConnectionSendWithReplyRunnable(mConnection, aMessage,
-                                            aTimeout, aCallback, aData));
+  DBusConnectionSendWithReplyTask* t =
+    new DBusConnectionSendWithReplyTask(mConnection, aMessage, aTimeout,
+                                        aCallback, aData);
   MOZ_ASSERT(t);
 
   nsresult rv = DispatchToDBusThread(t);
 
   if (NS_FAILED(rv)) {
     if (aMessage) {
       dbus_message_unref(aMessage);
     }