As an intermediate step towards having a message pump handling IO through completion ports, this CL creates two types of message pumps for windows: one to handle UI threads and another one to handle IO threads.

I'm basically moving stuff around, except by the fact that now UI threads will not process APCs or wait on objects (those things only happen on message loops of TYPE_IO)

Review URL: http://codereview.chromium.org/3094

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

4 files changed, 267 insertions, 146 deletions

diff --git a/base/message_loop.cc b/base/message_loop.cc
index 91c40d3..60a8ce5 100644
--- a/base/message_loop.cc
+++ b/base/message_loop.cc
@@ -78,12 +78,15 @@ MessageLoop::MessageLoop(Type type)
   DCHECK(!current()) << "should only have one message loop per thread";
   lazy_tls_ptr.Pointer()->Set(this);
 
-  // TODO(darin): Choose the pump based on the requested type.
 #if defined(OS_WIN)
+  // TODO(rvargas): Get rid of the OS guards.
   if (type_ == TYPE_DEFAULT) {
     pump_ = new base::MessagePumpDefault();
+  } else if (type_ == TYPE_IO) {
+    pump_ = new base::MessagePumpForIO();
   } else {
-    pump_ = new base::MessagePumpWin();
+    DCHECK(type_ == TYPE_UI);
+    pump_ = new base::MessagePumpForUI();
   }
 #elif defined(OS_POSIX)
 #if defined(OS_MACOSX)
@@ -576,7 +579,7 @@ void MessageLoopForUI::PumpOutPendingPaintMessages() {
 #if defined(OS_WIN)
 
 void MessageLoopForIO::WatchObject(HANDLE object, Watcher* watcher) {
-  pump_win()->WatchObject(object, watcher);
+  pump_io()->WatchObject(object, watcher);
 }
 
 #elif defined(OS_POSIX)
diff --git a/base/message_loop.h b/base/message_loop.h
index 25b3cfa..cc67587 100644
--- a/base/message_loop.h
+++ b/base/message_loop.h
@@ -423,6 +423,12 @@ class MessageLoopForUI : public MessageLoop {
   void WillProcessMessage(const MSG& message);
   void DidProcessMessage(const MSG& message);
   void PumpOutPendingPaintMessages();
+
+ protected:
+  // TODO(rvargas): Make this platform independent.
+  base::MessagePumpWin* pump_ui() {
+    return static_cast<base::MessagePumpForUI*>(pump_.get());
+  }
 #endif  // defined(OS_WIN)
 };
 
@@ -452,11 +458,17 @@ class MessageLoopForIO : public MessageLoop {
   }
 
 #if defined(OS_WIN)
-  typedef base::MessagePumpWin::Watcher Watcher;
+  typedef base::MessagePumpForIO::Watcher Watcher;
 
   // Please see MessagePumpWin for definitions of these methods.
   void WatchObject(HANDLE object, Watcher* watcher);
 
+ protected:
+  // TODO(rvargas): Make this platform independent.
+  base::MessagePumpForIO* pump_io() {
+    return static_cast<base::MessagePumpForIO*>(pump_.get());
+  }
+
 #elif defined(OS_POSIX)
   typedef base::MessagePumpLibevent::Watcher Watcher;
 
@@ -464,7 +476,7 @@ class MessageLoopForIO : public MessageLoop {
   void WatchSocket(int socket, short interest_mask, 
                    struct event* e, Watcher* watcher);
   void UnwatchSocket(struct event* e);
-#endif  // defined(OS_WIN)
+#endif  // defined(OS_POSIX)
 };
 
 // Do not add any member variables to MessageLoopForIO!  This is important b/c
diff --git a/base/message_pump_win.cc b/base/message_pump_win.cc
index 0b52e7b..9c96475 100644
--- a/base/message_pump_win.cc
+++ b/base/message_pump_win.cc
@@ -35,38 +35,6 @@ MessagePumpWin::~MessagePumpWin() {
   DestroyWindow(message_hwnd_);
 }
 
-void MessagePumpWin::WatchObject(HANDLE object, Watcher* watcher) {
-  DCHECK(object);
-  DCHECK_NE(object, INVALID_HANDLE_VALUE);
-
-  std::vector<HANDLE>::iterator it =
-      find(objects_.begin(), objects_.end(), object);
-  if (watcher) {
-    if (it == objects_.end()) {
-     static size_t warning_multiple = 1;
-     if (objects_.size() >= warning_multiple * MAXIMUM_WAIT_OBJECTS / 2) {
-       LOG(INFO) << "More than " << warning_multiple * MAXIMUM_WAIT_OBJECTS / 2
-           << " objects being watched";
-       // This DCHECK() is an artificial limitation, meant to warn us if we
-       // start creating too many objects.  It can safely be raised to a higher
-       // level, and the program is designed to handle much larger values.
-       // Before raising this limit, make sure that there is a very good reason
-       // (in your debug testing) to be watching this many objects.
-       DCHECK(2 <= warning_multiple);
-       ++warning_multiple;
-      }
-      objects_.push_back(object);
-      watchers_.push_back(watcher);
-    } else {
-      watchers_[it - objects_.begin()] = watcher;
-    }
-  } else if (it != objects_.end()) {
-    std::vector<HANDLE>::difference_type index = it - objects_.begin();
-    objects_.erase(it);
-    watchers_.erase(watchers_.begin() + index);
-  }
-}
-
 void MessagePumpWin::AddObserver(Observer* observer) {
   observers_.AddObserver(observer);
 }
@@ -176,7 +144,7 @@ void MessagePumpWin::ScheduleDelayedWork(const Time& delayed_work_time) {
 }
 
 //-----------------------------------------------------------------------------
-// MessagePumpWin private:
+// MessagePumpWin protected:
 
 // static
 LRESULT CALLBACK MessagePumpWin::WndProcThunk(
@@ -244,8 +212,199 @@ void MessagePumpWin::HandleTimerMessage() {
   }
 }
 
-void MessagePumpWin::DoRunLoop() {
-  // IF this was just a simple PeekMessage() loop (servicing all passible work
+bool MessagePumpWin::ProcessNextWindowsMessage() {
+  MSG msg;
+  if (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE))
+    return ProcessMessageHelper(msg);
+  return false;
+}
+
+bool MessagePumpWin::ProcessMessageHelper(const MSG& msg) {
+  if (WM_QUIT == msg.message) {
+    // Repost the QUIT message so that it will be retrieved by the primary
+    // GetMessage() loop.
+    state_->should_quit = true;
+    PostQuitMessage(static_cast<int>(msg.wParam));
+    return false;
+  }
+
+  // While running our main message pump, we discard kMsgHaveWork messages.
+  if (msg.message == kMsgHaveWork && msg.hwnd == message_hwnd_)
+    return ProcessPumpReplacementMessage();
+
+  WillProcessMessage(msg);
+
+  if (state_->dispatcher) {
+    if (!state_->dispatcher->Dispatch(msg))
+      state_->should_quit = true;
+  } else {
+    TranslateMessage(&msg);
+    DispatchMessage(&msg);
+  }
+
+  DidProcessMessage(msg);
+  return true;
+}
+
+bool MessagePumpWin::ProcessPumpReplacementMessage() {
+  // When we encounter a kMsgHaveWork message, this method is called to peek
+  // and process a replacement message, such as a WM_PAINT or WM_TIMER.  The
+  // goal is to make the kMsgHaveWork as non-intrusive as possible, even though
+  // a continuous stream of such messages are posted.  This method carefully
+  // peeks a message while there is no chance for a kMsgHaveWork to be pending,
+  // then resets the have_work_ flag (allowing a replacement kMsgHaveWork to
+  // possibly be posted), and finally dispatches that peeked replacement.  Note
+  // that the re-post of kMsgHaveWork may be asynchronous to this thread!!
+
+  MSG msg;
+  bool have_message = (0 != PeekMessage(&msg, NULL, 0, 0, PM_REMOVE));
+  DCHECK(!have_message || kMsgHaveWork != msg.message ||
+         msg.hwnd != message_hwnd_);
+  
+  // Since we discarded a kMsgHaveWork message, we must update the flag.
+  InterlockedExchange(&have_work_, 0);
+
+  // TODO(darin,jar): There is risk of being lost in a sub-pump within the call
+  // to ProcessMessageHelper, which could result in no longer getting a
+  // kMsgHaveWork message until the next out-of-band call to ScheduleWork.
+
+  return have_message && ProcessMessageHelper(msg);
+}
+
+int MessagePumpWin::GetCurrentDelay() const {
+  if (delayed_work_time_.is_null())
+    return -1;
+
+  // Be careful here.  TimeDelta has a precision of microseconds, but we want a
+  // value in milliseconds.  If there are 5.5ms left, should the delay be 5 or
+  // 6?  It should be 6 to avoid executing delayed work too early.
+  double timeout = ceil((delayed_work_time_ - Time::Now()).InMillisecondsF());
+
+  // If this value is negative, then we need to run delayed work soon.
+  int delay = static_cast<int>(timeout);
+  if (delay < 0)
+    delay = 0;
+
+  return delay;
+}
+
+//-----------------------------------------------------------------------------
+// MessagePumpForUI private:
+
+void MessagePumpForUI::DoRunLoop() {
+  // IF this was just a simple PeekMessage() loop (servicing all possible work
+  // queues), then Windows would try to achieve the following order according
+  // to MSDN documentation about PeekMessage with no filter):
+  //    * Sent messages
+  //    * Posted messages
+  //    * Sent messages (again)
+  //    * WM_PAINT messages
+  //    * WM_TIMER messages
+  //
+  // Summary: none of the above classes is starved, and sent messages has twice
+  // the chance of being processed (i.e., reduced service time).
+
+  for (;;) {
+    // If we do any work, we may create more messages etc., and more work may
+    // possibly be waiting in another task group.  When we (for example)
+    // ProcessNextWindowsMessage(), there is a good chance there are still more
+    // messages waiting.  On the other hand, when any of these methods return
+    // having done no work, then it is pretty unlikely that calling them again
+    // quickly will find any work to do.  Finally, if they all say they had no
+    // work, then it is a good time to consider sleeping (waiting) for more
+    // work.
+
+    bool more_work_is_plausible = ProcessNextWindowsMessage();
+    if (state_->should_quit)
+      break;
+
+    more_work_is_plausible |= state_->delegate->DoWork();
+    if (state_->should_quit)
+      break;
+
+    more_work_is_plausible |=
+        state_->delegate->DoDelayedWork(&delayed_work_time_);
+    // If we did not process any delayed work, then we can assume that our
+    // existing WM_TIMER if any will fire when delayed work should run.  We
+    // don't want to disturb that timer if it is already in flight.  However,
+    // if we did do all remaining delayed work, then lets kill the WM_TIMER.
+    if (more_work_is_plausible && delayed_work_time_.is_null())
+      KillTimer(message_hwnd_, reinterpret_cast<UINT_PTR>(this));
+    if (state_->should_quit)
+      break;
+
+    if (more_work_is_plausible)
+      continue;
+
+    more_work_is_plausible = state_->delegate->DoIdleWork();
+    if (state_->should_quit)
+      break;
+
+    if (more_work_is_plausible)
+      continue;
+
+    WaitForWork();  // Wait (sleep) until we have work to do again.
+  }
+}
+
+void MessagePumpForUI::WaitForWork() {
+  // Wait until a message is available, up to the time needed by the timer
+  // manager to fire the next set of timers.
+  int delay = GetCurrentDelay();
+  if (delay < 0)  // Negative value means no timers waiting.
+    delay = INFINITE;
+
+  DWORD result;
+  result = MsgWaitForMultipleObjectsEx(0, NULL, delay, QS_ALLINPUT,
+                                       MWMO_INPUTAVAILABLE);
+
+  if (WAIT_OBJECT_0 == result)
+    return;  // A WM_* message is available.
+
+  DCHECK_NE(WAIT_FAILED, result) << GetLastError();
+}
+
+//-----------------------------------------------------------------------------
+// MessagePumpForIO public:
+
+void MessagePumpForIO::WatchObject(HANDLE object, Watcher* watcher) {
+  DCHECK(object);
+  DCHECK_NE(object, INVALID_HANDLE_VALUE);
+
+  std::vector<HANDLE>::iterator it =
+      find(objects_.begin(), objects_.end(), object);
+  if (watcher) {
+    if (it == objects_.end()) {
+     static size_t warning_multiple = 1;
+     if (objects_.size() >= warning_multiple * MAXIMUM_WAIT_OBJECTS / 2) {
+       LOG(INFO) << "More than " << warning_multiple * MAXIMUM_WAIT_OBJECTS / 2
+           << " objects being watched";
+       // This DCHECK() is an artificial limitation, meant to warn us if we
+       // start creating too many objects.  It can safely be raised to a higher
+       // level, and the program is designed to handle much larger values.
+       // Before raising this limit, make sure that there is a very good reason
+       // (in your debug testing) to be watching this many objects.
+       DCHECK(2 <= warning_multiple);
+       ++warning_multiple;
+      }
+      objects_.push_back(object);
+      watchers_.push_back(watcher);
+    } else {
+      watchers_[it - objects_.begin()] = watcher;
+    }
+  } else if (it != objects_.end()) {
+    std::vector<HANDLE>::difference_type index = it - objects_.begin();
+    objects_.erase(it);
+    watchers_.erase(watchers_.begin() + index);
+  }
+}
+
+
+//-----------------------------------------------------------------------------
+// MessagePumpForIO private:
+
+void MessagePumpForIO::DoRunLoop() {
+  // IF this was just a simple PeekMessage() loop (servicing all possible work
   // queues), then Windows would try to achieve the following order according
   // to MSDN documentation about PeekMessage with no filter):
   //    * Sent messages
@@ -316,7 +475,7 @@ void MessagePumpWin::DoRunLoop() {
 // open, etc.)
 static const int kMultipleWaitPollingInterval = 20;
 
-void MessagePumpWin::WaitForWork() {
+void MessagePumpForIO::WaitForWork() {
   // Wait until either an object is signaled or a message is available.  Handle
   // (without returning) any APCs (only the IO thread currently has APCs.)
 
@@ -386,68 +545,7 @@ void MessagePumpWin::WaitForWork() {
   }
 }
 
-bool MessagePumpWin::ProcessNextWindowsMessage() {
-  MSG msg;
-  if (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE))
-    return ProcessMessageHelper(msg);
-  return false;
-}
-
-bool MessagePumpWin::ProcessMessageHelper(const MSG& msg) {
-  if (WM_QUIT == msg.message) {
-    // Repost the QUIT message so that it will be retrieved by the primary
-    // GetMessage() loop.
-    state_->should_quit = true;
-    PostQuitMessage(static_cast<int>(msg.wParam));
-    return false;
-  }
-
-  // While running our main message pump, we discard kMsgHaveWork messages.
-  if (msg.message == kMsgHaveWork && msg.hwnd == message_hwnd_)
-    return ProcessPumpReplacementMessage();
-
-  WillProcessMessage(msg);
-
-  if (state_->dispatcher) {
-    if (!state_->dispatcher->Dispatch(msg))
-      state_->should_quit = true;
-  } else {
-    TranslateMessage(&msg);
-    DispatchMessage(&msg);
-  }
-
-  DidProcessMessage(msg);
-  return true;
-}
-
-bool MessagePumpWin::ProcessPumpReplacementMessage() {
-  // When we encounter a kMsgHaveWork message, this method is called to peek
-  // and process a replacement message, such as a WM_PAINT or WM_TIMER.  The
-  // goal is to make the kMsgHaveWork as non-intrusive as possible, even though
-  // a continuous stream of such messages are posted.  This method carefully
-  // peeks a message while there is no chance for a kMsgHaveWork to be pending,
-  // then resets the have_work_ flag (allowing a replacement kMsgHaveWork to
-  // possibly be posted), and finally dispatches that peeked replacement.  Note
-  // that the re-post of kMsgHaveWork may be asynchronous to this thread!!
-
-  MSG msg;
-  bool have_message = (0 != PeekMessage(&msg, NULL, 0, 0, PM_REMOVE));
-  DCHECK(!have_message || kMsgHaveWork != msg.message ||
-         msg.hwnd != message_hwnd_);
-  
-  // Since we discarded a kMsgHaveWork message, we must update the flag.
-  InterlockedExchange(&have_work_, 0);
-
-  // TODO(darin,jar): There is risk of being lost in a sub-pump within the call
-  // to ProcessMessageHelper, which could result in no longer getting a
-  // kMsgHaveWork message until the next out-of-band call to ScheduleWork.
-
-  return have_message && ProcessMessageHelper(msg);
-}
-
-// Note: MsgWaitMultipleObjects() can't take a nil list, and that is why I had
-// to use SleepEx() to handle APCs when there were no objects.
-bool MessagePumpWin::ProcessNextObject() {
+bool MessagePumpForIO::ProcessNextObject() {
   size_t total_objs = objects_.size();
   if (!total_objs) {
     return false;
@@ -481,7 +579,7 @@ bool MessagePumpWin::ProcessNextObject() {
   return false;  // We serviced nothing.
 }
 
-bool MessagePumpWin::SignalWatcher(size_t object_index) {
+bool MessagePumpForIO::SignalWatcher(size_t object_index) {
   // Signal the watcher corresponding to the given index.
 
   DCHECK(objects_.size() > object_index);
@@ -499,22 +597,4 @@ bool MessagePumpWin::SignalWatcher(size_t object_index) {
   return true;
 }
 
-int MessagePumpWin::GetCurrentDelay() const {
-  if (delayed_work_time_.is_null())
-    return -1;
-
-  // Be careful here.  TimeDelta has a precision of microseconds, but we want a
-  // value in milliseconds.  If there are 5.5ms left, should the delay be 5 or
-  // 6?  It should be 6 to avoid executing delayed work too early.
-  double timeout = ceil((delayed_work_time_ - Time::Now()).InMillisecondsF());
-
-  // If this value is negative, then we need to run delayed work soon.
-  int delay = static_cast<int>(timeout);
-  if (delay < 0)
-    delay = 0;
-
-  return delay;
-}
-
 }  // namespace base
-
diff --git a/base/message_pump_win.h b/base/message_pump_win.h
index 0de3b89..0192e3c 100644
--- a/base/message_pump_win.h
+++ b/base/message_pump_win.h
@@ -62,15 +62,6 @@ namespace base {
 //
 class MessagePumpWin : public MessagePump {
  public:
-  // Used with WatchObject to asynchronously monitor the signaled state of a
-  // HANDLE object.
-  class Watcher {
-   public:
-    virtual ~Watcher() {}
-    // Called from MessageLoop::Run when a signalled object is detected.
-    virtual void OnObjectSignaled(HANDLE object) = 0;
-  };
-
   // An Observer is an object that receives global notifications from the
   // MessageLoop.
   //
@@ -106,11 +97,7 @@ class MessagePumpWin : public MessagePump {
   };
 
   MessagePumpWin();
-  ~MessagePumpWin();
-
-  // Have the current thread's message loop watch for a signaled object.
-  // Pass a null watcher to stop watching the object.
-  void WatchObject(HANDLE, Watcher*);
+  virtual ~MessagePumpWin();
 
   // Add an Observer, which will start receiving notifications immediately.
   void AddObserver(Observer* observer);
@@ -138,7 +125,7 @@ class MessagePumpWin : public MessagePump {
   virtual void ScheduleWork();
   virtual void ScheduleDelayedWork(const Time& delayed_work_time);
 
- private:
+ protected:
   struct RunState {
     Delegate* delegate;
     Dispatcher* dispatcher;
@@ -152,26 +139,18 @@ class MessagePumpWin : public MessagePump {
 
   static LRESULT CALLBACK WndProcThunk(
       HWND hwnd, UINT message, WPARAM wparam, LPARAM lparam);
+  virtual void DoRunLoop() = 0;
   void InitMessageWnd();
   void HandleWorkMessage();
   void HandleTimerMessage();
-  void DoRunLoop();
-  void WaitForWork();
   bool ProcessNextWindowsMessage();
   bool ProcessMessageHelper(const MSG& msg);
   bool ProcessPumpReplacementMessage();
-  bool ProcessNextObject();
-  bool SignalWatcher(size_t object_index);
   int GetCurrentDelay() const;
 
   // A hidden message-only window.
   HWND message_hwnd_;
 
-  // A vector of objects (and corresponding watchers) that are routinely
-  // serviced by this message pump.
-  std::vector<HANDLE> objects_;
-  std::vector<Watcher*> watchers_;
-
   ObserverList<Observer> observers_;
 
   // The time at which delayed work should run.
@@ -186,6 +165,53 @@ class MessagePumpWin : public MessagePump {
   RunState* state_;
 };
 
+//-----------------------------------------------------------------------------
+// MessagePumpForUI extends MessagePumpWin with methods that are particular to a
+// MessageLoop instantiated with TYPE_UI.
+//
+class MessagePumpForUI : public MessagePumpWin {
+ public:
+  MessagePumpForUI() {}
+  virtual ~MessagePumpForUI() {}
+ private:
+  virtual void DoRunLoop();
+  void WaitForWork();
+};
+
+//-----------------------------------------------------------------------------
+// MessagePumpForIO extends MessagePumpWin with methods that are particular to a
+// MessageLoop instantiated with TYPE_IO.
+//
+class MessagePumpForIO : public MessagePumpWin {
+ public:
+  // Used with WatchObject to asynchronously monitor the signaled state of a
+  // HANDLE object.
+  class Watcher {
+   public:
+    virtual ~Watcher() {}
+    // Called from MessageLoop::Run when a signalled object is detected.
+    virtual void OnObjectSignaled(HANDLE object) = 0;
+  };
+
+  MessagePumpForIO() {}
+  virtual ~MessagePumpForIO() {}
+
+  // Have the current thread's message loop watch for a signaled object.
+  // Pass a null watcher to stop watching the object.
+  void WatchObject(HANDLE, Watcher*);
+
+ private:
+  virtual void DoRunLoop();
+  void WaitForWork();
+  bool ProcessNextObject();
+  bool SignalWatcher(size_t object_index);
+
+  // A vector of objects (and corresponding watchers) that are routinely
+  // serviced by this message pump.
+  std::vector<HANDLE> objects_;
+  std::vector<Watcher*> watchers_;
+};
+
 }  // namespace base
 
 #endif  // BASE_MESSAGE_PUMP_WIN_H_