Implementation of MessagePump for Mojo.

I've refactored a bunch of the common MessagePump code into
message_pump_test. This allows us to share tests for custom
messagePumps and should allow moving MessagePumpForUI out later
on. Sadly it results in a ginormous macro. On the positive side it's
easy to use.

BUG=none
TEST=none
R=darin@chromium.org

Review URL: https://codereview.chromium.org/66193007

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@234540 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 336 insertions, 0 deletions

diff --git a/mojo/common/message_pump_mojo.cc b/mojo/common/message_pump_mojo.cc
new file mode 100644
index 0000000..ce60bb2
--- /dev/null
+++ b/mojo/common/message_pump_mojo.cc
@@ -0,0 +1,198 @@
+// Copyright 2013 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 "mojo/common/message_pump_mojo.h"
+
+#include <algorithm>
+#include <vector>
+
+#include "base/logging.h"
+#include "mojo/common/message_pump_mojo_handler.h"
+#include "mojo/common/scoped_message_pipe.h"
+
+namespace mojo {
+namespace common {
+
+// State needed for one iteration of MojoWaitMany. The first handle and flags
+// corresponds to that of the control pipe.
+struct MessagePumpMojo::WaitState {
+  std::vector<MojoHandle> handles;
+  std::vector<MojoWaitFlags> wait_flags;
+};
+
+struct MessagePumpMojo::RunState {
+ public:
+  RunState() : should_quit(false) {}
+
+  MojoHandle read_handle() const { return control_pipe.handle_0(); }
+  MojoHandle write_handle() const { return control_pipe.handle_1(); }
+
+  base::TimeTicks delayed_work_time;
+
+  // Used to wake up WaitForWork().
+  ScopedMessagePipe control_pipe;
+
+  bool should_quit;
+};
+
+MessagePumpMojo::MessagePumpMojo() : run_state_(NULL) {
+}
+
+MessagePumpMojo::~MessagePumpMojo() {
+}
+
+void MessagePumpMojo::AddHandler(MessagePumpMojoHandler* handler,
+                                 MojoHandle handle,
+                                 MojoWaitFlags wait_flags) {
+  DCHECK(handler);
+  DCHECK_NE(MOJO_HANDLE_INVALID, handle);
+  handlers_[handle].handler = handler;
+  handlers_[handle].wait_flags = wait_flags;
+
+  SignalControlPipe();
+}
+
+void MessagePumpMojo::RemoveHandler(MojoHandle handle) {
+  handlers_.erase(handle);
+  SignalControlPipe();
+}
+
+void MessagePumpMojo::Run(Delegate* delegate) {
+  RunState* old_state = run_state_;
+  RunState run_state;
+  // TODO: better deal with error handling.
+  CHECK_NE(run_state.control_pipe.handle_0(), MOJO_HANDLE_INVALID);
+  CHECK_NE(run_state.control_pipe.handle_1(), MOJO_HANDLE_INVALID);
+  run_state_ = &run_state;
+  bool more_work_is_plausible = true;
+  for (;;) {
+    const bool block = !more_work_is_plausible;
+    DoInternalWork(block);
+
+    // There isn't a good way to know if there are more handles ready, we assume
+    // not.
+    more_work_is_plausible = false;
+
+    if (run_state.should_quit)
+      break;
+
+    more_work_is_plausible |= delegate->DoWork();
+    if (run_state.should_quit)
+      break;
+
+    more_work_is_plausible |= delegate->DoDelayedWork(
+        &run_state.delayed_work_time);
+    if (run_state.should_quit)
+      break;
+
+    if (more_work_is_plausible)
+      continue;
+
+    more_work_is_plausible = delegate->DoIdleWork();
+    if (run_state.should_quit)
+      break;
+  }
+  run_state_ = old_state;
+}
+
+void MessagePumpMojo::Quit() {
+  if (run_state_)
+    run_state_->should_quit = true;
+}
+
+void MessagePumpMojo::ScheduleWork() {
+  SignalControlPipe();
+}
+
+void MessagePumpMojo::ScheduleDelayedWork(
+    const base::TimeTicks& delayed_work_time) {
+  if (!run_state_)
+    return;
+  run_state_->delayed_work_time = delayed_work_time;
+  SignalControlPipe();
+}
+
+void MessagePumpMojo::DoInternalWork(bool block) {
+  MojoDeadline deadline;
+  if (block && !run_state_->delayed_work_time.is_null()) {
+    const base::TimeDelta delta = run_state_->delayed_work_time -
+        base::TimeTicks::Now();
+    deadline = std::max(static_cast<MojoDeadline>(0),
+                        static_cast<MojoDeadline>(delta.InMicroseconds()));
+  } else {
+    deadline = 0;
+  }
+  const WaitState wait_state = GetWaitState();
+  const MojoResult result = MojoWaitMany(
+      &wait_state.handles.front(),
+      &wait_state.wait_flags.front(),
+      static_cast<uint32_t>(wait_state.handles.size()),
+      deadline);
+  if (result == 0) {
+    // Control pipe was written to.
+    uint32_t num_bytes = 0;
+    MojoReadMessage(run_state_->read_handle(), NULL, &num_bytes, NULL, 0,
+                    MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
+  } else if (result > 0) {
+    const size_t index = static_cast<size_t>(result);
+    DCHECK(handlers_.find(wait_state.handles[index]) != handlers_.end());
+    handlers_[wait_state.handles[index]].handler->OnHandleReady(
+        wait_state.handles[index]);
+  } else {
+    switch (result) {
+      case MOJO_RESULT_INVALID_ARGUMENT:
+      case MOJO_RESULT_FAILED_PRECONDITION:
+        RemoveFirstInvalidHandle(wait_state);
+        break;
+      case MOJO_RESULT_DEADLINE_EXCEEDED:
+        break;
+      default:
+        NOTREACHED();
+    }
+  }
+}
+
+void MessagePumpMojo::RemoveFirstInvalidHandle(const WaitState& wait_state) {
+  // TODO(sky): deal with control pipe going bad.
+  for (size_t i = 1; i < wait_state.handles.size(); ++i) {
+    const MojoResult result =
+        MojoWait(wait_state.handles[i], wait_state.wait_flags[i], 0);
+    if (result == MOJO_RESULT_INVALID_ARGUMENT ||
+        result == MOJO_RESULT_FAILED_PRECONDITION) {
+      DCHECK(handlers_.find(wait_state.handles[i]) != handlers_.end());
+      MessagePumpMojoHandler* handler =
+          handlers_[wait_state.handles[i]].handler;
+      handlers_.erase(wait_state.handles[i]);
+      handler->OnHandleError(wait_state.handles[i], result);
+      return;
+    } else {
+      DCHECK_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, result);
+    }
+  }
+}
+
+void MessagePumpMojo::SignalControlPipe() {
+  if (!run_state_)
+    return;
+
+  // TODO(sky): deal with error?
+  MojoWriteMessage(run_state_->write_handle(), NULL, 0, NULL, 0,
+                   MOJO_WRITE_MESSAGE_FLAG_NONE);
+}
+
+MessagePumpMojo::WaitState MessagePumpMojo::GetWaitState() const {
+  WaitState wait_state;
+  wait_state.handles.push_back(run_state_->write_handle());
+  wait_state.wait_flags.push_back(MOJO_WAIT_FLAG_READABLE);
+
+  for (HandleToHandler::const_iterator i = handlers_.begin();
+       i != handlers_.end(); ++i) {
+    wait_state.handles.push_back(i->first);
+    wait_state.wait_flags.push_back(i->second.wait_flags);
+  }
+  return wait_state;
+}
+
+}  // namespace common
+}  // namespace mojo
diff --git a/mojo/common/message_pump_mojo.h b/mojo/common/message_pump_mojo.h
new file mode 100644
index 0000000..4eba660
--- /dev/null
+++ b/mojo/common/message_pump_mojo.h
@@ -0,0 +1,82 @@
+// Copyright 2013 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.
+
+#ifndef MOJO_COMMON_MESSAGE_PUMP_MOJO_H_
+#define MOJO_COMMON_MESSAGE_PUMP_MOJO_H_
+
+#include <map>
+
+#include "base/message_loop/message_pump.h"
+#include "mojo/common/mojo_common_export.h"
+#include "mojo/public/system/core.h"
+
+namespace mojo {
+namespace common {
+
+class MessagePumpMojoHandler;
+
+// Mojo implementation of MessagePump.
+class MOJO_COMMON_EXPORT MessagePumpMojo : public base::MessagePump {
+ public:
+  MessagePumpMojo();
+  virtual ~MessagePumpMojo();
+
+  // Registers a MessagePumpMojoHandler for the specified handle. Only one
+  // handler can be registered for a specified handle. If there is an existing
+  // handler registered it is clobbered and silently removed.
+  // The handler is notified either when the handle is ready, or when it becomes
+  // invalid. If the handle becomes invalid the handler is removed and notified.
+  void AddHandler(MessagePumpMojoHandler* handler,
+                  MojoHandle handle,
+                  MojoWaitFlags wait_flags);
+
+  void RemoveHandler(MojoHandle handle);
+
+  // MessagePump:
+  virtual void Run(Delegate* delegate) OVERRIDE;
+  virtual void Quit() OVERRIDE;
+  virtual void ScheduleWork() OVERRIDE;
+  virtual void ScheduleDelayedWork(
+      const base::TimeTicks& delayed_work_time) OVERRIDE;
+
+ private:
+  struct RunState;
+  struct WaitState;
+
+  // Creates a MessagePumpMojoHandler and the set of MojoWaitFlags it was
+  // registered with.
+  struct Handler {
+    Handler() : handler(NULL), wait_flags(MOJO_WAIT_FLAG_NONE) {}
+
+    MessagePumpMojoHandler* handler;
+    MojoWaitFlags wait_flags;
+  };
+
+  typedef std::map<MojoHandle, Handler> HandleToHandler;
+
+  // Services the set of handles ready. If |block| is true this waits for a
+  // handle to become ready, otherwise this does not block.
+  void DoInternalWork(bool block);
+
+  // Removes the first invalid handle. This is called if MojoWaitMany finds an
+  // invalid handle.
+  void RemoveFirstInvalidHandle(const WaitState& wait_state);
+
+  void SignalControlPipe();
+
+  WaitState GetWaitState() const;
+
+  // If non-NULL we're running (inside Run()). Member is reference to value on
+  // stack.
+  RunState* run_state_;
+
+  HandleToHandler handlers_;
+
+  DISALLOW_COPY_AND_ASSIGN(MessagePumpMojo);
+};
+
+}  // namespace common
+}  // namespace mojo
+
+#endif  // MOJO_COMMON_MESSAGE_PUMP_MOJO_H_
diff --git a/mojo/common/message_pump_mojo_handler.h b/mojo/common/message_pump_mojo_handler.h
new file mode 100644
index 0000000..f623d5a
--- /dev/null
+++ b/mojo/common/message_pump_mojo_handler.h
@@ -0,0 +1,29 @@
+// Copyright 2013 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.
+
+#ifndef MOJO_COMMON_MESSAGE_PUMP_MOJO_HANDLER_H_
+#define MOJO_COMMON_MESSAGE_PUMP_MOJO_HANDLER_H_
+
+#include "mojo/common/mojo_common_export.h"
+#include "mojo/public/system/core.h"
+
+namespace mojo {
+namespace common {
+
+// Used by MessagePumpMojo to notify when a handle is either ready or has become
+// invalid.
+class MOJO_COMMON_EXPORT MessagePumpMojoHandler {
+ public:
+  virtual void OnHandleReady(MojoHandle handle) = 0;
+
+  virtual void OnHandleError(MojoHandle handle, MojoResult result) = 0;
+
+ protected:
+  virtual ~MessagePumpMojoHandler() {}
+};
+
+}  // namespace common
+}  // namespace mojo
+
+#endif  // MOJO_COMMON_MESSAGE_PUMP_MOJO_HANDLER_H_
diff --git a/mojo/common/message_pump_mojo_unittest.cc b/mojo/common/message_pump_mojo_unittest.cc
new file mode 100644
index 0000000..4276a0d
--- /dev/null
+++ b/mojo/common/message_pump_mojo_unittest.cc
@@ -0,0 +1,22 @@
+// Copyright 2013 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 "mojo/common/message_pump_mojo.h"
+
+#include "base/message_loop/message_loop_test.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace mojo {
+namespace common {
+namespace test {
+
+base::MessagePump* CreateMojoMessagePump() {
+  return new MessagePumpMojo();
+}
+
+RUN_MESSAGE_LOOP_TESTS(Mojo, &CreateMojoMessagePump);
+
+}  // namespace test
+}  // namespace common
+}  // namespace mojo
diff --git a/mojo/mojo.gyp b/mojo/mojo.gyp
index e95229f..f038fbc 100644
--- a/mojo/mojo.gyp
+++ b/mojo/mojo.gyp
@@ -73,6 +73,9 @@
         'common/bindings_support_impl.h',
         'common/handle_watcher.cc',
         'common/handle_watcher.h',
+        'common/message_pump_mojo.cc',
+        'common/message_pump_mojo.h',
+        'common/message_pump_mojo_handler.h',
         'common/scoped_message_pipe.cc',
         'common/scoped_message_pipe.h',
       ],
@@ -90,6 +93,7 @@
       'type': 'executable',
       'dependencies': [
         '../base/base.gyp:base',
+        '../base/base.gyp:base_message_loop_tests',
         '../testing/gtest.gyp:gtest',
         'mojo_common_lib',
         'mojo_run_all_unittests',
@@ -97,6 +101,7 @@
       ],
       'sources': [
         'common/handle_watcher_unittest.cc',
+        'common/message_pump_mojo_unittest.cc',
       ],
       'conditions': [
         ['OS == "win"', {