libchromeos: MessageLoop implementation using base::MessageLoopForIO.

The new chromeos::BaseMessageLoop implementation uses the existing
base::MessageLoopForIO implementation, that in the Linux and ChromeOS
context uses base::MessagePumpLibevent, a libevent based implementation.

This chromeos::MessageLoop allows to mix calls to base::MessageLoopForIO
and chromeos::MessageLoop with few limitations. This is useful when using
other code in libchrome that already interacts with libchrome's
MessageLoopForIO.

Among the limitations, the support for canceling a task will not free
the memory associated with the delayed event until it actually fires.
Also, base::MessagePumpLibevent uses epoll(7) which doesn't support
watching for file descriptors that would not naturally block (such as
regular files). Attempting to watch for those file descriptors will
return kTaskIdNull in this implementation, but would work on
chromeos::GlibMessageLoop.

This patch moves most of chromeos::GlibMessageLoop existing tests to a
common chromeos::MessageLoop test that runs for both implementations.

BUG=chromium:506052
TEST=Added unittests.

Change-Id: I4553d149511806ed599fa2847af4372985456106
Reviewed-on: https://chromium-review.googlesource.com/285217
Reviewed-by: Alex Deymo <deymo@chromium.org>
Commit-Queue: Alex Deymo <deymo@chromium.org>
Trybot-Ready: Alex Deymo <deymo@chromium.org>
Tested-by: Alex Deymo <deymo@chromium.org>

8 files changed, 709 insertions, 177 deletions

diff --git a/chromeos/message_loops/base_message_loop.cc b/chromeos/message_loops/base_message_loop.cc
new file mode 100644
index 0000000..142ae90
--- /dev/null
+++ b/chromeos/message_loops/base_message_loop.cc
@@ -0,0 +1,250 @@
+// Copyright 2015 The Chromium OS 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 <chromeos/message_loops/base_message_loop.h>
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <base/bind.h>
+
+#include <chromeos/location_logging.h>
+
+using base::Closure;
+
+namespace chromeos {
+
+BaseMessageLoop::BaseMessageLoop(base::MessageLoopForIO* base_loop)
+    : base_loop_(base_loop),
+      weak_ptr_factory_(this) {}
+
+BaseMessageLoop::~BaseMessageLoop() {
+  for (auto& io_task : io_tasks_) {
+    DVLOG_LOC(io_task.second.location(), 1)
+        << "Removing file descriptor watcher task_id " << io_task.first
+        << " leaked on BaseMessageLoop, scheduled from this location.";
+    io_task.second.fd_watcher()->StopWatchingFileDescriptor();
+  }
+
+  // Note all pending canceled delayed tasks when destroying the message loop.
+  size_t lazily_deleted_tasks = 0;
+  for (const auto& delayed_task : delayed_tasks_) {
+    if (delayed_task.second.closure.is_null()) {
+      lazily_deleted_tasks++;
+    } else {
+      DVLOG_LOC(delayed_task.second.location, 1)
+          << "Removing delayed task_id " << delayed_task.first
+          << " leaked on BaseMessageLoop, scheduled from this location.";
+    }
+  }
+  if (lazily_deleted_tasks) {
+    LOG(INFO) << "Leaking " << lazily_deleted_tasks << " canceled tasks.";
+  }
+}
+
+MessageLoop::TaskId BaseMessageLoop::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure &task,
+    base::TimeDelta delay) {
+  TaskId task_id =  NextTaskId();
+  bool base_scheduled = base_loop_->task_runner()->PostDelayedTask(
+      from_here,
+      base::Bind(&BaseMessageLoop::OnRanPostedTask,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 task_id),
+      delay);
+  DVLOG_LOC(from_here, 1) << "Scheduling delayed task_id " << task_id
+                          << " to run in " << delay << ".";
+  if (!base_scheduled)
+    return MessageLoop::kTaskIdNull;
+
+  delayed_tasks_.emplace(task_id,
+                         DelayedTask{from_here, task_id, std::move(task)});
+  return task_id;
+}
+
+MessageLoop::TaskId BaseMessageLoop::WatchFileDescriptor(
+    const tracked_objects::Location& from_here,
+    int fd,
+    WatchMode mode,
+    bool persistent,
+    const Closure &task) {
+  // base::MessageLoopForIO CHECKS that "fd >= 0", so we handle that case here.
+  if (fd < 0)
+    return MessageLoop::kTaskIdNull;
+
+  base::MessageLoopForIO::Mode base_mode = base::MessageLoopForIO::WATCH_READ;
+  switch (mode) {
+    case MessageLoop::kWatchRead:
+      base_mode = base::MessageLoopForIO::WATCH_READ;
+      break;
+    case MessageLoop::kWatchWrite:
+      base_mode = base::MessageLoopForIO::WATCH_WRITE;
+      break;
+    default:
+      return MessageLoop::kTaskIdNull;
+  }
+
+  TaskId task_id =  NextTaskId();
+  auto it_bool = io_tasks_.emplace(
+      std::piecewise_construct,
+      std::forward_as_tuple(task_id),
+      std::forward_as_tuple(from_here, this, task_id, persistent, task));
+  // This should always insert a new element.
+  DCHECK(it_bool.second);
+  IOTask* new_io_task = &it_bool.first->second;
+
+  bool scheduled = base_loop_->WatchFileDescriptor(
+      fd, persistent, base_mode, new_io_task->fd_watcher(), new_io_task);
+
+  DVLOG_LOC(from_here, 1)
+      << "Watching fd " << fd << " for "
+      << (mode == MessageLoop::kWatchRead ? "reading" : "writing")
+      << (persistent ? " persistently" : " just once")
+      << " as task_id " << task_id
+      << (scheduled ? " successfully" : " failed.");
+
+  if (!scheduled) {
+    io_tasks_.erase(task_id);
+    return MessageLoop::kTaskIdNull;
+  }
+  return task_id;
+}
+
+bool BaseMessageLoop::CancelTask(TaskId task_id) {
+  if (task_id == kTaskIdNull)
+    return false;
+  auto delayed_task_it = delayed_tasks_.find(task_id);
+  if (delayed_task_it == delayed_tasks_.end()) {
+    // This might be an IOTask then.
+    auto io_task_it = io_tasks_.find(task_id);
+    if (io_task_it == io_tasks_.end())
+      return false;
+
+    DVLOG_LOC(io_task_it->second.location(), 1)
+        << "Removing task_id " << task_id << " scheduled from this location.";
+    // Destroying the FileDescriptorWatcher implicitly stops watching the file
+    // descriptor.
+    io_tasks_.erase(io_task_it);
+    return true;
+  }
+  // A DelayedTask was found for this task_id at this point.
+
+  // Check if the callback was already canceled but we have the entry in
+  // delayed_tasks_ since it didn't fire yet in the message loop.
+  if (delayed_task_it->second.closure.is_null())
+    return false;
+
+  DVLOG_LOC(delayed_task_it->second.location, 1)
+      << "Removing task_id " << task_id << " scheduled from this location.";
+  // We reset to closure to a null Closure to release all the resources
+  // used by this closure at this point, but we don't remove the task_id from
+  // delayed_tasks_ since we can't tell base::MessageLoopForIO to not run it.
+  delayed_task_it->second.closure = Closure();
+
+  return true;
+}
+
+bool BaseMessageLoop::RunOnce(bool may_block) {
+  run_once_ = true;
+  if (!may_block)
+    base_loop_->RunUntilIdle();
+  else
+    base_loop_->Run();
+  // If the flag was reset to false, it means a closure was run.
+  if (!run_once_)
+    return true;
+
+  run_once_ = false;
+  return false;
+}
+
+void BaseMessageLoop::Run() {
+  base_loop_->Run();
+}
+
+void BaseMessageLoop::BreakLoop() {
+  base_loop_->QuitNow();
+}
+
+MessageLoop::TaskId BaseMessageLoop::NextTaskId() {
+  TaskId res;
+  do {
+    res = ++last_id_;
+    // We would run out of memory before we run out of task ids.
+  } while (!res ||
+           delayed_tasks_.find(res) != delayed_tasks_.end() ||
+           io_tasks_.find(res) != io_tasks_.end());
+  return res;
+}
+
+void BaseMessageLoop::OnRanPostedTask(MessageLoop::TaskId task_id) {
+  auto task_it = delayed_tasks_.find(task_id);
+  DCHECK(task_it != delayed_tasks_.end());
+  if (!task_it->second.closure.is_null()) {
+    DVLOG_LOC(task_it->second.location, 1)
+        << "Running delayed task_id " << task_id
+        << " scheduled from this location.";
+    // Mark the task as canceled while we are running it so CancelTask returns
+    // false.
+    Closure closure = std::move(task_it->second.closure);
+    task_it->second.closure = Closure();
+    closure.Run();
+
+    // If the |run_once_| flag is set, it is because we are instructed to run
+    // only once callback.
+    if (run_once_) {
+      run_once_ = false;
+      base_loop_->QuitNow();
+    }
+  }
+  delayed_tasks_.erase(task_it);
+}
+
+BaseMessageLoop::IOTask::IOTask(const tracked_objects::Location& location,
+                                BaseMessageLoop* loop,
+                                MessageLoop::TaskId task_id,
+                                bool persistent,
+                                const Closure& task)
+    : location_(location), loop_(loop), task_id_(task_id),
+      persistent_(persistent), closure_(task) {}
+
+void BaseMessageLoop::IOTask::OnFileCanReadWithoutBlocking(int fd) {
+  OnFileReady(fd);
+}
+
+void BaseMessageLoop::IOTask::OnFileCanWriteWithoutBlocking(int fd) {
+  OnFileReady(fd);
+}
+
+void BaseMessageLoop::IOTask::OnFileReady(int fd) {
+  // We can't access |this| after running the |closure_| since it could call
+  // CancelTask on its own task_id, so we copy the members we need now.
+  BaseMessageLoop* loop_ptr = loop_;
+
+  DVLOG_LOC(location_, 1)
+      << "Running task_id " << task_id_
+      << " for watching file descriptor " << fd
+      << ", scheduled from this location.";
+
+  if (persistent_) {
+    // In the persistent case we just run the callback. If this callback cancels
+    // the task id, we can't access |this| anymore.
+    closure_.Run();
+  } else {
+    // This will destroy |this|, the fd_watcher and therefore stop watching this
+    // file descriptor.
+    Closure closure_copy = std::move(closure_);
+    loop_->io_tasks_.erase(task_id_);
+    // Run the closure from the local copy we just made.
+    closure_copy.Run();
+  }
+
+  if (loop_ptr->run_once_) {
+    loop_ptr->run_once_ = false;
+    loop_ptr->base_loop_->QuitNow();
+  }
+}
+
+}  // namespace chromeos
diff --git a/chromeos/message_loops/base_message_loop.h b/chromeos/message_loops/base_message_loop.h
new file mode 100644
index 0000000..943f36d
--- /dev/null
+++ b/chromeos/message_loops/base_message_loop.h
@@ -0,0 +1,122 @@
+// Copyright 2015 The Chromium OS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef LIBCHROMEOS_CHROMEOS_MESSAGE_LOOPS_BASE_MESSAGE_LOOP_H_
+#define LIBCHROMEOS_CHROMEOS_MESSAGE_LOOPS_BASE_MESSAGE_LOOP_H_
+
+// BaseMessageLoop is a chromeos::MessageLoop implementation based on
+// base::MessageLoopForIO. This allows to mix new code using
+// chromeos::MessageLoop and legacy code using base::MessageLoopForIO in the
+// same thread and share a single main loop. This disadvantage of using this
+// class is a less efficient implementation of CancelTask() for delayed tasks
+// since base::MessageLoopForIO doesn't provide a way to remove the event.
+
+#include <map>
+#include <memory>
+
+#include <base/location.h>
+#include <base/memory/weak_ptr.h>
+#include <base/message_loop/message_loop.h>
+#include <base/time/time.h>
+
+#include <chromeos/chromeos_export.h>
+#include <chromeos/message_loops/message_loop.h>
+
+namespace chromeos {
+
+class CHROMEOS_EXPORT BaseMessageLoop : public MessageLoop {
+ public:
+  explicit BaseMessageLoop(base::MessageLoopForIO* base_loop);
+  ~BaseMessageLoop() override;
+
+  // MessageLoop overrides.
+  TaskId PostDelayedTask(const tracked_objects::Location& from_here,
+                         const base::Closure& task,
+                         base::TimeDelta delay) override;
+  using MessageLoop::PostDelayedTask;
+  TaskId WatchFileDescriptor(const tracked_objects::Location& from_here,
+                             int fd,
+                             WatchMode mode,
+                             bool persistent,
+                             const base::Closure& task) override;
+  using MessageLoop::WatchFileDescriptor;
+  bool CancelTask(TaskId task_id) override;
+  bool RunOnce(bool may_block) override;
+  void Run() override;
+  void BreakLoop() override;
+
+ private:
+  // Called by base::MessageLoopForIO when is time to call the callback
+  // scheduled with Post*Task() of id |task_id|, even if it was canceled.
+  void OnRanPostedTask(MessageLoop::TaskId task_id);
+
+  // Return a new unused task_id.
+  TaskId NextTaskId();
+
+  struct DelayedTask {
+    tracked_objects::Location location;
+
+    MessageLoop::TaskId task_id;
+    base::Closure closure;
+  };
+
+  std::map<MessageLoop::TaskId, DelayedTask> delayed_tasks_;
+
+  class IOTask : public base::MessageLoopForIO::Watcher {
+   public:
+    IOTask(const tracked_objects::Location& location,
+           BaseMessageLoop* loop,
+           MessageLoop::TaskId task_id,
+           bool persistent,
+           const base::Closure& task);
+
+    const tracked_objects::Location& location() const { return location_; }
+    base::MessageLoopForIO::FileDescriptorWatcher* fd_watcher() {
+      return &fd_watcher_;
+    }
+
+   private:
+    tracked_objects::Location location_;
+    BaseMessageLoop* loop_;
+
+    MessageLoop::TaskId task_id_;
+    bool persistent_;
+    base::Closure closure_;
+
+    base::MessageLoopForIO::FileDescriptorWatcher fd_watcher_;
+
+    // base::MessageLoopForIO::Watcher overrides:
+    void OnFileCanReadWithoutBlocking(int fd) override;
+    void OnFileCanWriteWithoutBlocking(int fd) override;
+
+    // Common implementation for both the read and write case.
+    void OnFileReady(int fd);
+
+    DISALLOW_COPY_AND_ASSIGN(IOTask);
+  };
+
+  std::map<MessageLoop::TaskId, IOTask> io_tasks_;
+
+  // Flag to mark that we should run the message loop only one iteration.
+  bool run_once_{false};
+
+  // The last used TaskId. While base::MessageLoopForIO doesn't allow to cancel
+  // delayed tasks, we handle that functionality by not running the callback
+  // if it fires at a later point.
+  MessageLoop::TaskId last_id_{kTaskIdNull};
+
+  // The pointer to the libchrome base::MessageLoopForIO we are wrapping with
+  // this interface.
+  base::MessageLoopForIO* base_loop_;
+
+  // We use a WeakPtrFactory to schedule tasks with the base::MessageLoopForIO
+  // since we can't cancel the callbacks we have scheduled there once this
+  // instance is destroyed.
+  base::WeakPtrFactory<BaseMessageLoop> weak_ptr_factory_;
+  DISALLOW_COPY_AND_ASSIGN(BaseMessageLoop);
+};
+
+}  // namespace chromeos
+
+#endif  // LIBCHROMEOS_CHROMEOS_MESSAGE_LOOPS_BASE_MESSAGE_LOOP_H_
diff --git a/chromeos/message_loops/fake_message_loop.cc b/chromeos/message_loops/fake_message_loop.cc
index 089591d..786cde2 100644
--- a/chromeos/message_loops/fake_message_loop.cc
+++ b/chromeos/message_loops/fake_message_loop.cc
@@ -15,7 +15,7 @@ FakeMessageLoop::FakeMessageLoop(base::SimpleTestClock* clock)
 
 MessageLoop::TaskId FakeMessageLoop::PostDelayedTask(
     const tracked_objects::Location& from_here,
-    const base::Closure &task,
+    const base::Closure& task,
     base::TimeDelta delay) {
   // If no SimpleTestClock was provided, we use the last time we fired a
   // callback. In this way, tasks scheduled from a Closure will have the right
@@ -38,7 +38,7 @@ MessageLoop::TaskId FakeMessageLoop::WatchFileDescriptor(
     int fd,
     WatchMode mode,
     bool persistent,
-    const base::Closure &task) {
+    const base::Closure& task) {
   MessageLoop::TaskId current_id = ++last_id_;
   // FakeMessageLoop is limited to only 2^64 tasks. That should be enough.
   CHECK(current_id);
diff --git a/chromeos/message_loops/fake_message_loop.h b/chromeos/message_loops/fake_message_loop.h
index 4d6e7ca..e9f0c51 100644
--- a/chromeos/message_loops/fake_message_loop.h
+++ b/chromeos/message_loops/fake_message_loop.h
@@ -37,14 +37,14 @@ class CHROMEOS_EXPORT FakeMessageLoop : public MessageLoop {
   ~FakeMessageLoop() override = default;
 
   TaskId PostDelayedTask(const tracked_objects::Location& from_here,
-                         const base::Closure &task,
+                         const base::Closure& task,
                          base::TimeDelta delay) override;
   using MessageLoop::PostDelayedTask;
   TaskId WatchFileDescriptor(const tracked_objects::Location& from_here,
                              int fd,
                              WatchMode mode,
                              bool persistent,
-                             const base::Closure &task) override;
+                             const base::Closure& task) override;
   using MessageLoop::WatchFileDescriptor;
   bool CancelTask(TaskId task_id) override;
   bool RunOnce(bool may_block) override;
diff --git a/chromeos/message_loops/glib_message_loop.h b/chromeos/message_loops/glib_message_loop.h
index 92f570b..12d5af9 100644
--- a/chromeos/message_loops/glib_message_loop.h
+++ b/chromeos/message_loops/glib_message_loop.h
@@ -24,14 +24,14 @@ class CHROMEOS_EXPORT GlibMessageLoop : public MessageLoop {
 
   // MessageLoop overrides.
   TaskId PostDelayedTask(const tracked_objects::Location& from_here,
-                         const base::Closure &task,
+                         const base::Closure& task,
                          base::TimeDelta delay) override;
   using MessageLoop::PostDelayedTask;
   TaskId WatchFileDescriptor(const tracked_objects::Location& from_here,
                              int fd,
                              WatchMode mode,
                              bool persistent,
-                             const base::Closure &task) override;
+                             const base::Closure& task) override;
   using MessageLoop::WatchFileDescriptor;
   bool CancelTask(TaskId task_id) override;
   bool RunOnce(bool may_block) override;
diff --git a/chromeos/message_loops/glib_message_loop_unittest.cc b/chromeos/message_loops/glib_message_loop_unittest.cc
index 13df023..0563bd5 100644
--- a/chromeos/message_loops/glib_message_loop_unittest.cc
+++ b/chromeos/message_loops/glib_message_loop_unittest.cc
@@ -19,33 +19,6 @@
 #include <chromeos/message_loops/message_loop_utils.h>
 
 using base::Bind;
-using base::TimeDelta;
-
-namespace {
-// Helper class to create and close a unidirectional pipe. Used to provide valid
-// file descriptors when testing watching for a file descriptor.
-class ScopedPipe {
- public:
-  ScopedPipe() {
-    int fds[2];
-    if (pipe(fds) != 0) {
-      PLOG(FATAL) << "Creating a pipe()";
-    }
-    reader = fds[0];
-    writer = fds[1];
-  }
-  ~ScopedPipe() {
-    if (reader != -1)
-      close(reader);
-    if (writer != -1)
-      close(writer);
-  }
-
-  // The reader and writer end of the pipe.
-  int reader{-1};
-  int writer{-1};
-};
-}  // namespace
 
 namespace chromeos {
 
@@ -61,80 +34,6 @@ class GlibMessageLoopTest : public ::testing::Test {
   std::unique_ptr<GlibMessageLoop> loop_;
 };
 
-TEST_F(GlibMessageLoopTest, CancelTaskInvalidValuesTest) {
-  EXPECT_FALSE(loop_->CancelTask(MessageLoop::kTaskIdNull));
-  EXPECT_FALSE(loop_->CancelTask(1234));
-}
-
-TEST_F(GlibMessageLoopTest, PostTaskTest) {
-  bool called = false;
-  TaskId task_id = loop_->PostTask(FROM_HERE,
-                                   Bind([&called]() { called = true; }));
-  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
-  MessageLoopRunMaxIterations(loop_.get(), 100);
-  EXPECT_TRUE(called);
-}
-
-// Tests that we can cancel tasks right after we schedule them.
-TEST_F(GlibMessageLoopTest, PostTaskCancelledTest) {
-  bool called = false;
-  TaskId task_id = loop_->PostTask(FROM_HERE,
-                                   Bind([&called]() { called = true; }));
-  EXPECT_TRUE(loop_->CancelTask(task_id));
-  MessageLoopRunMaxIterations(loop_.get(), 100);
-  EXPECT_FALSE(called);
-  // Can't remove a task you already removed.
-  EXPECT_FALSE(loop_->CancelTask(task_id));
-}
-
-TEST_F(GlibMessageLoopTest, PostDelayedTaskRunsEventuallyTest) {
-  bool called = false;
-  TaskId task_id = loop_->PostDelayedTask(FROM_HERE,
-                                          Bind([&called]() { called = true; }),
-                                          TimeDelta::FromMilliseconds(100));
-  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
-  MessageLoopRunUntil(loop_.get(),
-                      TimeDelta::FromSeconds(10),
-                      Bind([&called]() { return called; }));
-  // Check that the main loop finished before the 10 seconds timeout, so it
-  // finished due to the callback being called and not due to the timeout.
-  EXPECT_TRUE(called);
-}
-
-// Test that you can call the overloaded version of PostDelayedTask from
-// MessageLoop. This is important because only one of the two methods is
-// virtual, so you need to unhide the other when overriding the virtual one.
-TEST_F(GlibMessageLoopTest, PostDelayedTaskWithoutLocation) {
-  loop_->PostDelayedTask(Bind(&base::DoNothing), TimeDelta());
-  EXPECT_EQ(1, MessageLoopRunMaxIterations(loop_.get(), 100));
-}
-
-TEST_F(GlibMessageLoopTest, WatchForInvalidFD) {
-  bool called = false;
-  EXPECT_EQ(MessageLoop::kTaskIdNull, loop_->WatchFileDescriptor(
-      FROM_HERE, -1, MessageLoop::kWatchRead, true,
-      Bind([&called] { called = true; })));
-  EXPECT_EQ(MessageLoop::kTaskIdNull, loop_->WatchFileDescriptor(
-      FROM_HERE, -1, MessageLoop::kWatchWrite, true,
-      Bind([&called] { called = true; })));
-  EXPECT_EQ(0, MessageLoopRunMaxIterations(loop_.get(), 100));
-  EXPECT_FALSE(called);
-}
-
-TEST_F(GlibMessageLoopTest, CancelWatchedFileDescriptor) {
-  ScopedPipe pipe;
-  bool called = false;
-  TaskId task_id = loop_->WatchFileDescriptor(
-      FROM_HERE, pipe.reader, MessageLoop::kWatchRead, true,
-      Bind([&called] { called = true; }));
-  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
-  // The reader end is blocked because we didn't write anything to the writer
-  // end.
-  EXPECT_EQ(0, MessageLoopRunMaxIterations(loop_.get(), 100));
-  EXPECT_FALSE(called);
-  EXPECT_TRUE(loop_->CancelTask(task_id));
-}
-
 // When you watch a file descriptor for reading, the guaranties are that a
 // blocking call to read() on that file descriptor will not block. This should
 // include the case when the other end of a pipe is closed or the file is empty.
@@ -150,22 +49,6 @@ TEST_F(GlibMessageLoopTest, WatchFileDescriptorTriggersWhenEmpty) {
   EXPECT_TRUE(loop_->CancelTask(task_id));
 }
 
-TEST_F(GlibMessageLoopTest, WatchFileDescriptorTriggersWhenPipeClosed) {
-  ScopedPipe pipe;
-  bool called = false;
-  EXPECT_EQ(0, HANDLE_EINTR(close(pipe.writer)));
-  pipe.writer = -1;
-  TaskId task_id = loop_->WatchFileDescriptor(
-      FROM_HERE, pipe.reader, MessageLoop::kWatchRead, true,
-      Bind([&called] { called = true; }));
-  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
-  // The reader end is not blocked because we closed the writer end so a read on
-  // the reader end would return 0 bytes read.
-  EXPECT_NE(0, MessageLoopRunMaxIterations(loop_.get(), 10));
-  EXPECT_TRUE(called);
-  EXPECT_TRUE(loop_->CancelTask(task_id));
-}
-
 // Test that an invalid file descriptor triggers the callback.
 TEST_F(GlibMessageLoopTest, WatchFileDescriptorTriggersWhenInvalid) {
   int fd = HANDLE_EINTR(open("/dev/zero", O_RDONLY));
@@ -183,52 +66,4 @@ TEST_F(GlibMessageLoopTest, WatchFileDescriptorTriggersWhenInvalid) {
   EXPECT_TRUE(loop_->CancelTask(task_id));
 }
 
-// When a WatchFileDescriptor task is scheduled with |persistent| = true, we
-// should keep getting a call whenever the file descriptor is ready.
-TEST_F(GlibMessageLoopTest, WatchFileDescriptorPersistently) {
-  int fd = HANDLE_EINTR(open("/dev/zero", O_RDONLY));
-  int called = 0;
-  TaskId task_id = loop_->WatchFileDescriptor(
-      FROM_HERE, fd, MessageLoop::kWatchRead, true,
-      Bind([&called] { called++; }));
-  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
-  // We let the main loop run for 20 iterations to give it enough iterations to
-  // verify that our callback was called more than one. We only check that our
-  // callback is called more than once.
-  EXPECT_EQ(20, MessageLoopRunMaxIterations(loop_.get(), 20));
-  EXPECT_LT(1, called);
-  EXPECT_TRUE(loop_->CancelTask(task_id));
-  EXPECT_EQ(0, IGNORE_EINTR(close(fd)));
-}
-
-TEST_F(GlibMessageLoopTest, WatchFileDescriptorNonPersistent) {
-  int fd = HANDLE_EINTR(open("/dev/zero", O_RDONLY));
-  int called = 0;
-  TaskId task_id = loop_->WatchFileDescriptor(
-      fd, MessageLoop::kWatchRead, false, Bind([&called] { called++; }));
-  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
-  // We let the main loop run for 20 iterations but we just expect it to run
-  // at least once. The callback should be called exactly once since we
-  // scheduled it non-persistently. After it ran, we shouldn't be able to cancel
-  // this task.
-  EXPECT_LT(0, MessageLoopRunMaxIterations(loop_.get(), 20));
-  EXPECT_EQ(1, called);
-  EXPECT_FALSE(loop_->CancelTask(task_id));
-  EXPECT_EQ(0, IGNORE_EINTR(close(fd)));
-}
-
-// Test that we can cancel the task we are running, and should just fail.
-TEST_F(GlibMessageLoopTest, DeleteTaskFromSelf) {
-  bool cancel_result = true;  // We would expect this to be false.
-  GlibMessageLoop* loop_ptr = loop_.get();
-  TaskId task_id;
-  task_id = loop_->PostTask(
-      FROM_HERE,
-      Bind([&cancel_result, loop_ptr, &task_id]() {
-        cancel_result = loop_ptr->CancelTask(task_id);
-      }));
-  EXPECT_EQ(1, MessageLoopRunMaxIterations(loop_.get(), 100));
-  EXPECT_FALSE(cancel_result);
-}
-
 }  // namespace chromeos
diff --git a/chromeos/message_loops/message_loop.h b/chromeos/message_loops/message_loop.h
index 58fe5ea..700b066 100644
--- a/chromeos/message_loops/message_loop.h
+++ b/chromeos/message_loops/message_loop.h
@@ -44,21 +44,21 @@ class CHROMEOS_EXPORT MessageLoop {
   // at a later point.
   // This methond can only be called from the same thread running the main loop.
   virtual TaskId PostDelayedTask(const tracked_objects::Location& from_here,
-                                 const base::Closure &task,
+                                 const base::Closure& task,
                                  base::TimeDelta delay) = 0;
   // Variant without the Location for easier usage.
-  TaskId PostDelayedTask(const base::Closure &task,
+  TaskId PostDelayedTask(const base::Closure& task,
                          base::TimeDelta delay) {
     return PostDelayedTask(tracked_objects::Location(), task, delay);
   }
 
   // A convenience method to schedule a call with no delay.
   // This methond can only be called from the same thread running the main loop.
-  TaskId PostTask(const base::Closure &task) {
+  TaskId PostTask(const base::Closure& task) {
     return PostDelayedTask(task, base::TimeDelta());
   }
   TaskId PostTask(const tracked_objects::Location& from_here,
-                  const base::Closure &task) {
+                  const base::Closure& task) {
     return PostDelayedTask(from_here, task, base::TimeDelta());
   }
 
@@ -79,13 +79,13 @@ class CHROMEOS_EXPORT MessageLoop {
                                      int fd,
                                      WatchMode mode,
                                      bool persistent,
-                                     const base::Closure &task) = 0;
+                                     const base::Closure& task) = 0;
 
   // Convenience function to call WatchFileDescriptor() without a location.
   TaskId WatchFileDescriptor(int fd,
                              WatchMode mode,
                              bool persistent,
-                             const base::Closure &task) {
+                             const base::Closure& task) {
     return WatchFileDescriptor(
         tracked_objects::Location(), fd, mode, persistent, task);
   }
diff --git a/chromeos/message_loops/message_loop_unittest.cc b/chromeos/message_loops/message_loop_unittest.cc
new file mode 100644
index 0000000..30f400f
--- /dev/null
+++ b/chromeos/message_loops/message_loop_unittest.cc
@@ -0,0 +1,325 @@
+// Copyright 2015 The Chromium OS 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 <chromeos/message_loops/message_loop.h>
+
+// These are the common tests for all the chromeos::MessageLoop implementations
+// that should conform to this interface's contracts. For extra
+// implementation-specific tests see the particular implementation unittests in
+// the *_unittest.cc files.
+
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <memory>
+
+#include <base/bind.h>
+#include <base/location.h>
+#include <base/posix/eintr_wrapper.h>
+#include <gtest/gtest.h>
+
+#include <chromeos/bind_lambda.h>
+#include <chromeos/message_loops/base_message_loop.h>
+#include <chromeos/message_loops/glib_message_loop.h>
+#include <chromeos/message_loops/message_loop_utils.h>
+
+using base::Bind;
+using base::TimeDelta;
+
+namespace {
+// Helper class to create and close a unidirectional pipe. Used to provide valid
+// file descriptors when testing watching for a file descriptor.
+class ScopedPipe {
+ public:
+  ScopedPipe() {
+    int fds[2];
+    if (pipe(fds) != 0) {
+      PLOG(FATAL) << "Creating a pipe()";
+    }
+    reader = fds[0];
+    writer = fds[1];
+  }
+  ~ScopedPipe() {
+    if (reader != -1)
+      close(reader);
+    if (writer != -1)
+      close(writer);
+  }
+
+  // The reader and writer end of the pipe.
+  int reader{-1};
+  int writer{-1};
+};
+
+class ScopedSocketPair {
+ public:
+  ScopedSocketPair() {
+    int fds[2];
+    if (socketpair(PF_LOCAL, SOCK_STREAM, 0, fds) != 0) {
+      PLOG(FATAL) << "Creating a socketpair()";
+    }
+    left = fds[0];
+    right = fds[1];
+  }
+  ~ScopedSocketPair() {
+    if (left != -1)
+      close(left);
+    if (right != -1)
+      close(right);
+  }
+
+  // The left and right sockets are bi-directional connected and
+  // indistinguishable file descriptor. We named them left/right for easier
+  // reading.
+  int left{-1};
+  int right{-1};
+};
+}  // namespace
+
+namespace chromeos {
+
+using TaskId = MessageLoop::TaskId;
+
+template <typename T>
+class MessageLoopTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    MessageLoopSetUp();
+    EXPECT_TRUE(this->loop_.get());
+  }
+
+  std::unique_ptr<base::MessageLoopForIO> base_loop_;
+
+  std::unique_ptr<MessageLoop> loop_;
+
+ private:
+  // These MessageLoopSetUp() methods are used to setup each MessageLoop
+  // according to its constructor requirements.
+  void MessageLoopSetUp();
+};
+
+template <>
+void MessageLoopTest<GlibMessageLoop>::MessageLoopSetUp() {
+  loop_.reset(new GlibMessageLoop());
+}
+
+template <>
+void MessageLoopTest<BaseMessageLoop>::MessageLoopSetUp() {
+  base_loop_.reset(new base::MessageLoopForIO());
+  loop_.reset(new BaseMessageLoop(base::MessageLoopForIO::current()));
+}
+
+// This setups gtest to run each one of the following TYPED_TEST test cases on
+// on each implementation.
+typedef ::testing::Types<
+  GlibMessageLoop,
+  BaseMessageLoop> MessageLoopTypes;
+TYPED_TEST_CASE(MessageLoopTest, MessageLoopTypes);
+
+
+TYPED_TEST(MessageLoopTest, CancelTaskInvalidValuesTest) {
+  EXPECT_FALSE(this->loop_->CancelTask(MessageLoop::kTaskIdNull));
+  EXPECT_FALSE(this->loop_->CancelTask(1234));
+}
+
+TYPED_TEST(MessageLoopTest, PostTaskTest) {
+  bool called = false;
+  TaskId task_id = this->loop_->PostTask(FROM_HERE,
+                                         Bind([&called]() { called = true; }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  MessageLoopRunMaxIterations(this->loop_.get(), 100);
+  EXPECT_TRUE(called);
+}
+
+// Tests that we can cancel tasks right after we schedule them.
+TYPED_TEST(MessageLoopTest, PostTaskCancelledTest) {
+  bool called = false;
+  TaskId task_id = this->loop_->PostTask(FROM_HERE,
+                                         Bind([&called]() { called = true; }));
+  EXPECT_TRUE(this->loop_->CancelTask(task_id));
+  MessageLoopRunMaxIterations(this->loop_.get(), 100);
+  EXPECT_FALSE(called);
+  // Can't remove a task you already removed.
+  EXPECT_FALSE(this->loop_->CancelTask(task_id));
+}
+
+TYPED_TEST(MessageLoopTest, PostDelayedTaskRunsEventuallyTest) {
+  bool called = false;
+  TaskId task_id = this->loop_->PostDelayedTask(
+      FROM_HERE,
+      Bind([&called]() { called = true; }),
+      TimeDelta::FromMilliseconds(50));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  MessageLoopRunUntil(this->loop_.get(),
+                      TimeDelta::FromSeconds(10),
+                      Bind([&called]() { return called; }));
+  // Check that the main loop finished before the 10 seconds timeout, so it
+  // finished due to the callback being called and not due to the timeout.
+  EXPECT_TRUE(called);
+}
+
+// Test that you can call the overloaded version of PostDelayedTask from
+// MessageLoop. This is important because only one of the two methods is
+// virtual, so you need to unhide the other when overriding the virtual one.
+TYPED_TEST(MessageLoopTest, PostDelayedTaskWithoutLocation) {
+  this->loop_->PostDelayedTask(Bind(&base::DoNothing), TimeDelta());
+  EXPECT_EQ(1, MessageLoopRunMaxIterations(this->loop_.get(), 100));
+}
+
+TYPED_TEST(MessageLoopTest, WatchForInvalidFD) {
+  bool called = false;
+  EXPECT_EQ(MessageLoop::kTaskIdNull, this->loop_->WatchFileDescriptor(
+      FROM_HERE, -1, MessageLoop::kWatchRead, true,
+      Bind([&called] { called = true; })));
+  EXPECT_EQ(MessageLoop::kTaskIdNull, this->loop_->WatchFileDescriptor(
+      FROM_HERE, -1, MessageLoop::kWatchWrite, true,
+      Bind([&called] { called = true; })));
+  EXPECT_EQ(0, MessageLoopRunMaxIterations(this->loop_.get(), 100));
+  EXPECT_FALSE(called);
+}
+
+TYPED_TEST(MessageLoopTest, CancelWatchedFileDescriptor) {
+  ScopedPipe pipe;
+  bool called = false;
+  TaskId task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, pipe.reader, MessageLoop::kWatchRead, true,
+      Bind([&called] { called = true; }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  // The reader end is blocked because we didn't write anything to the writer
+  // end.
+  EXPECT_EQ(0, MessageLoopRunMaxIterations(this->loop_.get(), 100));
+  EXPECT_FALSE(called);
+  EXPECT_TRUE(this->loop_->CancelTask(task_id));
+}
+
+// When you watch a file descriptor for reading, the guaranties are that a
+// blocking call to read() on that file descriptor will not block. This should
+// include the case when the other end of a pipe is closed or the file is empty.
+TYPED_TEST(MessageLoopTest, WatchFileDescriptorTriggersWhenPipeClosed) {
+  ScopedPipe pipe;
+  bool called = false;
+  EXPECT_EQ(0, HANDLE_EINTR(close(pipe.writer)));
+  pipe.writer = -1;
+  TaskId task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, pipe.reader, MessageLoop::kWatchRead, true,
+      Bind([&called] { called = true; }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  // The reader end is not blocked because we closed the writer end so a read on
+  // the reader end would return 0 bytes read.
+  EXPECT_NE(0, MessageLoopRunMaxIterations(this->loop_.get(), 10));
+  EXPECT_TRUE(called);
+  EXPECT_TRUE(this->loop_->CancelTask(task_id));
+}
+
+// When a WatchFileDescriptor task is scheduled with |persistent| = true, we
+// should keep getting a call whenever the file descriptor is ready.
+TYPED_TEST(MessageLoopTest, WatchFileDescriptorPersistently) {
+  ScopedPipe pipe;
+  EXPECT_EQ(1, HANDLE_EINTR(write(pipe.writer, "a", 1)));
+
+  int called = 0;
+  TaskId task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, pipe.reader, MessageLoop::kWatchRead, true,
+      Bind([&called] { called++; }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  // We let the main loop run for 20 iterations to give it enough iterations to
+  // verify that our callback was called more than one. We only check that our
+  // callback is called more than once.
+  EXPECT_EQ(20, MessageLoopRunMaxIterations(this->loop_.get(), 20));
+  EXPECT_LT(1, called);
+  EXPECT_TRUE(this->loop_->CancelTask(task_id));
+}
+
+TYPED_TEST(MessageLoopTest, WatchFileDescriptorNonPersistent) {
+  ScopedPipe pipe;
+  EXPECT_EQ(1, HANDLE_EINTR(write(pipe.writer, "a", 1)));
+
+  int called = 0;
+  TaskId task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, pipe.reader, MessageLoop::kWatchRead, false,
+      Bind([&called] { called++; }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  // We let the main loop run for 20 iterations but we just expect it to run
+  // at least once. The callback should be called exactly once since we
+  // scheduled it non-persistently. After it ran, we shouldn't be able to cancel
+  // this task.
+  EXPECT_LT(0, MessageLoopRunMaxIterations(this->loop_.get(), 20));
+  EXPECT_EQ(1, called);
+  EXPECT_FALSE(this->loop_->CancelTask(task_id));
+}
+
+TYPED_TEST(MessageLoopTest, WatchFileDescriptorForReadAndWriteSimultaneously) {
+  ScopedSocketPair socks;
+  EXPECT_EQ(1, HANDLE_EINTR(write(socks.right, "a", 1)));
+  // socks.left should be able to read this "a" and should also be able to write
+  // without blocking since the kernel has some buffering for it.
+
+  TaskId read_task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, socks.left, MessageLoop::kWatchRead, true,
+      Bind([this, &read_task_id] {
+        EXPECT_TRUE(this->loop_->CancelTask(read_task_id))
+            << "task_id" << read_task_id;
+      }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, read_task_id);
+
+  TaskId write_task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, socks.left, MessageLoop::kWatchWrite, true,
+      Bind([this, &write_task_id] {
+        EXPECT_TRUE(this->loop_->CancelTask(write_task_id));
+      }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, write_task_id);
+
+  EXPECT_LT(0, MessageLoopRunMaxIterations(this->loop_.get(), 20));
+
+  EXPECT_FALSE(this->loop_->CancelTask(read_task_id));
+  EXPECT_FALSE(this->loop_->CancelTask(write_task_id));
+}
+
+// Test that we can cancel the task we are running, and should just fail.
+TYPED_TEST(MessageLoopTest, DeleteTaskFromSelf) {
+  bool cancel_result = true;  // We would expect this to be false.
+  MessageLoop* loop_ptr = this->loop_.get();
+  TaskId task_id;
+  task_id = this->loop_->PostTask(
+      FROM_HERE,
+      Bind([&cancel_result, loop_ptr, &task_id]() {
+        cancel_result = loop_ptr->CancelTask(task_id);
+      }));
+  EXPECT_EQ(1, MessageLoopRunMaxIterations(this->loop_.get(), 100));
+  EXPECT_FALSE(cancel_result);
+}
+
+// Test that we can cancel a non-persistent file descriptor watching callback,
+// which should fail.
+TYPED_TEST(MessageLoopTest, DeleteNonPersistenIOTaskFromSelf) {
+  ScopedPipe pipe;
+  TaskId task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, pipe.writer, MessageLoop::kWatchWrite, false /* persistent */,
+      Bind([this, &task_id] {
+        EXPECT_FALSE(this->loop_->CancelTask(task_id));
+        task_id = MessageLoop::kTaskIdNull;
+      }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  EXPECT_EQ(1, MessageLoopRunMaxIterations(this->loop_.get(), 100));
+  EXPECT_EQ(MessageLoop::kTaskIdNull, task_id);
+}
+
+// Test that we can cancel a persistent file descriptor watching callback from
+// the same callback.
+TYPED_TEST(MessageLoopTest, DeletePersistenIOTaskFromSelf) {
+  ScopedPipe pipe;
+  TaskId task_id = this->loop_->WatchFileDescriptor(
+      FROM_HERE, pipe.writer, MessageLoop::kWatchWrite, true /* persistent */,
+      Bind([this, &task_id] {
+        EXPECT_TRUE(this->loop_->CancelTask(task_id));
+        task_id = MessageLoop::kTaskIdNull;
+      }));
+  EXPECT_NE(MessageLoop::kTaskIdNull, task_id);
+  EXPECT_EQ(1, MessageLoopRunMaxIterations(this->loop_.get(), 100));
+  EXPECT_EQ(MessageLoop::kTaskIdNull, task_id);
+}
+
+}  // namespace chromeos