Switch MessagePumpForIO to use completion ports on Windows.

Cleanup the separation between MessagePumpForUI and 
MessagePumpForIO, and convert the latter to use Completion
Ports instead of MsgWaitForMultipleobjects to sleep
when idle.

Remove all traces of Windows messages from MessagePumpForIO,
remove the transitional API of completion port notifications
and remove WatchObject API.

Modify all callers of RegisterIOHandler so that they are no
longer using RegisterIOContext, and also handle properly
the new semantics of completion ports (notifications even when
the IO completes immediately).

Add a new interface to allow proper cleanup of disk cache (to
replace code that was waiting for pending APCs from the destructor).

Add a way for the message pump to perform cleanup of abandoned IO.


BUG=B/1344358, 3497, 3630
TESt=unit tests
R=darin


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

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

1 files changed, 259 insertions, 407 deletions

diff --git a/base/message_pump_win.cc b/base/message_pump_win.cc
index 542fcf1..7ada5eb7 100644
--- a/base/message_pump_win.cc
+++ b/base/message_pump_win.cc
@@ -11,41 +11,6 @@
 
 using base::Time;
 
-namespace {
-
-class HandlerData : public base::MessagePumpForIO::Watcher {
- public:
-  typedef base::MessagePumpForIO::IOHandler IOHandler;
-  HandlerData(OVERLAPPED* context, IOHandler* handler)
-      : context_(context), handler_(handler) {}
-  ~HandlerData() {}
-
-  virtual void OnObjectSignaled(HANDLE object);
-
- private:
-  OVERLAPPED* context_;
-  IOHandler* handler_;
-  
-  DISALLOW_COPY_AND_ASSIGN(HandlerData);
-};
-
-void HandlerData::OnObjectSignaled(HANDLE object) {
-  DCHECK(object == context_->hEvent);
-  DWORD transfered;
-  DWORD error = ERROR_SUCCESS;
-  BOOL ret = GetOverlappedResult(NULL, context_, &transfered, FALSE);
-  if (!ret) {
-    error = GetLastError();
-    DCHECK(ERROR_HANDLE_EOF == error || ERROR_BROKEN_PIPE == error);
-    transfered = 0;
-  }
-
-  ResetEvent(context_->hEvent);
-  handler_->OnIOCompleted(context_, transfered, error);
-}
-
-}  // namespace
-
 namespace base {
 
 static const wchar_t kWndClass[] = L"Chrome_MessagePumpWindow";
@@ -54,24 +19,9 @@ static const wchar_t kWndClass[] = L"Chrome_MessagePumpWindow";
 // task (a series of such messages creates a continuous task pump).
 static const int kMsgHaveWork = WM_USER + 1;
 
-#ifndef NDEBUG
-// Force exercise of polling model.
-static const int kMaxWaitObjects = 8;
-#else
-static const int kMaxWaitObjects = MAXIMUM_WAIT_OBJECTS;
-#endif
-
 //-----------------------------------------------------------------------------
 // MessagePumpWin public:
 
-MessagePumpWin::MessagePumpWin() : have_work_(0), state_(NULL) {
-  InitMessageWnd();
-}
-
-MessagePumpWin::~MessagePumpWin() {
-  DestroyWindow(message_hwnd_);
-}
-
 void MessagePumpWin::AddObserver(Observer* observer) {
   observers_.AddObserver(observer);
 }
@@ -88,36 +38,6 @@ void MessagePumpWin::DidProcessMessage(const MSG& msg) {
   FOR_EACH_OBSERVER(Observer, observers_, DidProcessMessage(msg));
 }
 
-void MessagePumpWin::PumpOutPendingPaintMessages() {
-  // If we are being called outside of the context of Run, then don't try to do
-  // any work.
-  if (!state_)
-    return;
-
-  // Create a mini-message-pump to force immediate processing of only Windows
-  // WM_PAINT messages.  Don't provide an infinite loop, but do enough peeking
-  // to get the job done.  Actual common max is 4 peeks, but we'll be a little
-  // safe here.
-  const int kMaxPeekCount = 20;
-  bool win2k = win_util::GetWinVersion() <= win_util::WINVERSION_2000;
-  int peek_count;
-  for (peek_count = 0; peek_count < kMaxPeekCount; ++peek_count) {
-    MSG msg;
-    if (win2k) {
-      if (!PeekMessage(&msg, NULL, WM_PAINT, WM_PAINT, PM_REMOVE))
-        break;
-    } else {
-      if (!PeekMessage(&msg, NULL, 0, 0, PM_REMOVE | PM_QS_PAINT))
-        break;
-    }
-    ProcessMessageHelper(msg);
-    if (state_->should_quit)  // Handle WM_QUIT.
-      break;
-  }
-  // Histogram what was really being used, to help to adjust kMaxPeekCount.
-  DHISTOGRAM_COUNTS(L"Loop.PumpOutPendingPaintMessages Peeks", peek_count);
-}
-
 void MessagePumpWin::RunWithDispatcher(
     Delegate* delegate, Dispatcher* dispatcher) {
   RunState s;
@@ -139,7 +59,38 @@ void MessagePumpWin::Quit() {
   state_->should_quit = true;
 }
 
-void MessagePumpWin::ScheduleWork() {
+//-----------------------------------------------------------------------------
+// MessagePumpWin protected:
+
+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 public:
+
+MessagePumpForUI::MessagePumpForUI() {
+  InitMessageWnd();
+}
+
+MessagePumpForUI::~MessagePumpForUI() {
+  DestroyWindow(message_hwnd_);
+}
+
+void MessagePumpForUI::ScheduleWork() {
   if (InterlockedExchange(&have_work_, 1))
     return;  // Someone else continued the pumping.
 
@@ -147,7 +98,7 @@ void MessagePumpWin::ScheduleWork() {
   PostMessage(message_hwnd_, kMsgHaveWork, reinterpret_cast<WPARAM>(this), 0);
 }
 
-void MessagePumpWin::ScheduleDelayedWork(const Time& delayed_work_time) {
+void MessagePumpForUI::ScheduleDelayedWork(const Time& delayed_work_time) {
   //
   // We would *like* to provide high resolution timers.  Windows timers using
   // SetTimer() have a 10ms granularity.  We have to use WM_TIMER as a wakeup
@@ -180,24 +131,110 @@ void MessagePumpWin::ScheduleDelayedWork(const Time& delayed_work_time) {
   SetTimer(message_hwnd_, reinterpret_cast<UINT_PTR>(this), delay_msec, NULL);
 }
 
+void MessagePumpForUI::PumpOutPendingPaintMessages() {
+  // If we are being called outside of the context of Run, then don't try to do
+  // any work.
+  if (!state_)
+    return;
+
+  // Create a mini-message-pump to force immediate processing of only Windows
+  // WM_PAINT messages.  Don't provide an infinite loop, but do enough peeking
+  // to get the job done.  Actual common max is 4 peeks, but we'll be a little
+  // safe here.
+  const int kMaxPeekCount = 20;
+  bool win2k = win_util::GetWinVersion() <= win_util::WINVERSION_2000;
+  int peek_count;
+  for (peek_count = 0; peek_count < kMaxPeekCount; ++peek_count) {
+    MSG msg;
+    if (win2k) {
+      if (!PeekMessage(&msg, NULL, WM_PAINT, WM_PAINT, PM_REMOVE))
+        break;
+    } else {
+      if (!PeekMessage(&msg, NULL, 0, 0, PM_REMOVE | PM_QS_PAINT))
+        break;
+    }
+    ProcessMessageHelper(msg);
+    if (state_->should_quit)  // Handle WM_QUIT.
+      break;
+  }
+  // Histogram what was really being used, to help to adjust kMaxPeekCount.
+  DHISTOGRAM_COUNTS(L"Loop.PumpOutPendingPaintMessages Peeks", peek_count);
+}
+
 //-----------------------------------------------------------------------------
-// MessagePumpWin protected:
+// MessagePumpForUI private:
 
 // static
-LRESULT CALLBACK MessagePumpWin::WndProcThunk(
+LRESULT CALLBACK MessagePumpForUI::WndProcThunk(
     HWND hwnd, UINT message, WPARAM wparam, LPARAM lparam) {
   switch (message) {
     case kMsgHaveWork:
-      reinterpret_cast<MessagePumpWin*>(wparam)->HandleWorkMessage();
+      reinterpret_cast<MessagePumpForUI*>(wparam)->HandleWorkMessage();
       break;
     case WM_TIMER:
-      reinterpret_cast<MessagePumpWin*>(wparam)->HandleTimerMessage();
+      reinterpret_cast<MessagePumpForUI*>(wparam)->HandleTimerMessage();
       break;
   }
   return DefWindowProc(hwnd, message, wparam, lparam);
 }
 
-void MessagePumpWin::InitMessageWnd() {
+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::InitMessageWnd() {
   HINSTANCE hinst = GetModuleHandle(NULL);
 
   WNDCLASSEX wc = {0};
@@ -212,7 +249,41 @@ void MessagePumpWin::InitMessageWnd() {
   DCHECK(message_hwnd_);
 }
 
-void MessagePumpWin::HandleWorkMessage() {
+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) {
+    // A WM_* message is available.
+    // If a parent child relationship exists between windows across threads
+    // then their thread inputs are implicitly attached.
+    // This causes the MsgWaitForMultipleObjectsEx API to return indicating
+    // that messages are ready for processing (specifically mouse messages
+    // intended for the child window. Occurs if the child window has capture)
+    // The subsequent PeekMessages call fails to return any messages thus
+    // causing us to enter a tight loop at times.
+    // The WaitMessage call below is a workaround to give the child window
+    // sometime to process its input messages.
+    MSG msg = {0};
+    DWORD queue_status = GetQueueStatus(QS_MOUSE);
+    if (HIWORD(queue_status) & QS_MOUSE &&
+       !PeekMessage(&msg, NULL, WM_MOUSEFIRST, WM_MOUSELAST, PM_NOREMOVE)) {
+      WaitMessage();
+    }
+    return;
+  }
+
+  DCHECK_NE(WAIT_FAILED, result) << GetLastError();
+}
+
+void MessagePumpForUI::HandleWorkMessage() {
   // If we are being called outside of the context of Run, then don't try to do
   // any work.  This could correspond to a MessageBox call or something of that
   // sort.
@@ -233,7 +304,7 @@ void MessagePumpWin::HandleWorkMessage() {
     ScheduleWork();
 }
 
-void MessagePumpWin::HandleTimerMessage() {
+void MessagePumpForUI::HandleTimerMessage() {
   KillTimer(message_hwnd_, reinterpret_cast<UINT_PTR>(this));
 
   // If we are being called outside of the context of Run, then don't do
@@ -249,7 +320,7 @@ void MessagePumpWin::HandleTimerMessage() {
   }
 }
 
-bool MessagePumpWin::ProcessNextWindowsMessage() {
+bool MessagePumpForUI::ProcessNextWindowsMessage() {
   // If there are sent messages in the queue then PeekMessage internally
   // dispatches the message and returns false. We return true in this
   // case to ensure that the message loop peeks again instead of calling
@@ -266,7 +337,7 @@ bool MessagePumpWin::ProcessNextWindowsMessage() {
   return sent_messages_in_queue;
 }
 
-bool MessagePumpWin::ProcessMessageHelper(const MSG& msg) {
+bool MessagePumpForUI::ProcessMessageHelper(const MSG& msg) {
   if (WM_QUIT == msg.message) {
     // Repost the QUIT message so that it will be retrieved by the primary
     // GetMessage() loop.
@@ -293,7 +364,7 @@ bool MessagePumpWin::ProcessMessageHelper(const MSG& msg) {
   return true;
 }
 
-bool MessagePumpWin::ProcessPumpReplacementMessage() {
+bool MessagePumpForUI::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
@@ -307,7 +378,7 @@ bool MessagePumpWin::ProcessPumpReplacementMessage() {
   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);
 
@@ -318,233 +389,63 @@ bool MessagePumpWin::ProcessPumpReplacementMessage() {
   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;
+// MessagePumpForIO public:
 
-    WaitForWork();  // Wait (sleep) until we have work to do again.
-  }
+MessagePumpForIO::MessagePumpForIO() {
+  port_.Set(CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1));
+  DCHECK(port_.IsValid());
 }
 
-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) {
-    // A WM_* message is available.
-    // If a parent child relationship exists between windows across threads
-    // then their thread inputs are implicitly attached.
-    // This causes the MsgWaitForMultipleObjectsEx API to return indicating
-    // that messages are ready for processing (specifically mouse messages
-    // intended for the child window. Occurs if the child window has capture)
-    // The subsequent PeekMessages call fails to return any messages thus
-    // causing us to enter a tight loop at times.
-    // The WaitMessage call below is a workaround to give the child window
-    // sometime to process its input messages.
-    MSG msg = {0};
-    DWORD queue_status = GetQueueStatus(QS_MOUSE);
-    if (HIWORD(queue_status) & QS_MOUSE && 
-       !PeekMessage(&msg, NULL, WM_MOUSEFIRST, WM_MOUSELAST, PM_NOREMOVE)) {
-      WaitMessage();
-    }
-    return;  
-  }
+void MessagePumpForIO::ScheduleWork() {
+  if (InterlockedExchange(&have_work_, 1))
+    return;  // Someone else continued the pumping.
 
-  DCHECK_NE(WAIT_FAILED, result) << GetLastError();
+  // Make sure the MessagePump does some work for us.
+  BOOL ret = PostQueuedCompletionStatus(port_, 0,
+                                        reinterpret_cast<ULONG_PTR>(this),
+                                        reinterpret_cast<OVERLAPPED*>(this));
+  DCHECK(ret);
 }
 
-//-----------------------------------------------------------------------------
-// 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);
-  }
+void MessagePumpForIO::ScheduleDelayedWork(const Time& delayed_work_time) {
+  // We know that we can't be blocked right now since this method can only be
+  // called on the same thread as Run, so we only need to update our record of
+  // how long to sleep when we do sleep.
+  delayed_work_time_ = delayed_work_time;
 }
 
 void MessagePumpForIO::RegisterIOHandler(HANDLE file_handle,
                                          IOHandler* handler) {
-#if 0
-  // TODO(rvargas): This is just to give an idea of what this code will look
-  // like when we actually move to completion ports. Of course, we cannot
-  // do this without calling GetQueuedCompletionStatus().
   ULONG_PTR key = reinterpret_cast<ULONG_PTR>(handler);
   HANDLE port = CreateIoCompletionPort(file_handle, port_, key, 1);
-  if (!port_.IsValid())
-    port_.Set(port);
-#endif
-}
-
-void MessagePumpForIO::RegisterIOContext(OVERLAPPED* context,
-                                         IOHandler* handler) {
-  DCHECK(context->hEvent);
-  if (handler) {
-    HandlerData* watcher = new HandlerData(context, handler);
-    WatchObject(context->hEvent, watcher);
-  } else {
-    std::vector<HANDLE>::iterator it =
-      find(objects_.begin(), objects_.end(), context->hEvent);
-
-    if (it == objects_.end()) {
-      NOTREACHED();
-      return;
-    }
-
-    std::vector<HANDLE>::difference_type index = it - objects_.begin();
-    objects_.erase(it);
-    delete watchers_[index];
-    watchers_.erase(watchers_.begin() + index);
-  }
+  DCHECK(port == port_.Get());
 }
 
 //-----------------------------------------------------------------------------
 // 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
-  //    * 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 (same thing for ProcessNextObject(), which responds to
-    // only one signaled object; etc.).  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;
+    // WaitForIOCompletion(), 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.
 
-    more_work_is_plausible |= state_->delegate->DoWork();
+    bool more_work_is_plausible = state_->delegate->DoWork();
     if (state_->should_quit)
       break;
 
-    more_work_is_plausible |= ProcessNextObject();
+    more_work_is_plausible |= WaitForIOCompletion(0, NULL);
     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;
 
@@ -558,141 +459,92 @@ void MessagePumpForIO::DoRunLoop() {
     if (more_work_is_plausible)
       continue;
 
-    // We service APCs in WaitForWork, without returning.
     WaitForWork();  // Wait (sleep) until we have work to do again.
   }
 }
 
-// If we handle more than the OS limit on the number of objects that can be
-// waited for, we'll need to poll (sequencing through subsets of the objects
-// that can be passed in a single OS wait call).  The following is the polling
-// interval used in that (unusual) case. (I don't have a lot of justifcation
-// for the specific value, but it needed to be short enough that it would not
-// add a lot of latency, and long enough that we wouldn't thrash the CPU for no
-// reason... especially considering the silly user probably has a million tabs
-// open, etc.)
-static const int kMultipleWaitPollingInterval = 20;
-
+// Wait until IO completes, up to the time needed by the timer manager to fire
+// the next set of timers.
 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.)
-
-  // We do not support nested message loops when we have watched objects.  This
-  // is to avoid messy recursion problems.
-  DCHECK(objects_.empty() || state_->run_depth == 1) <<
-      "Cannot nest a message loop when there are watched objects!";
+  // We do not support nested IO message loops. This is to avoid messy
+  // recursion problems.
+  DCHECK(state_->run_depth == 1) << "Cannot nest an IO message loop!";
 
-  int wait_flags = MWMO_ALERTABLE | MWMO_INPUTAVAILABLE;
+  int timeout = GetCurrentDelay();
+  if (timeout < 0)  // Negative value means no timers waiting.
+    timeout = INFINITE;
 
-  bool use_polling = false;  // Poll if too many objects for one OS Wait call.
-  for (;;) {
-    // Do initialization here, in case APC modifies object list.
-    size_t total_objs = objects_.size();
-
-    int delay;
-    size_t polling_index = 0;  // The first unprocessed object index.
-    do {
-      size_t objs_len =
-          (polling_index < total_objs) ? total_objs - polling_index : 0;
-      if (objs_len >= MAXIMUM_WAIT_OBJECTS) {
-        objs_len = MAXIMUM_WAIT_OBJECTS - 1;
-        use_polling = true;
-      }
-      HANDLE* objs = objs_len ? polling_index + &objects_.front() : NULL;
-
-      // Only wait up to the time needed by the timer manager to fire the next
-      // set of timers.
-      delay = GetCurrentDelay();
-      if (use_polling && delay > kMultipleWaitPollingInterval)
-        delay = kMultipleWaitPollingInterval;
-      if (delay < 0)  // Negative value means no timers waiting.
-        delay = INFINITE;
-
-      DWORD result;
-      result = MsgWaitForMultipleObjectsEx(static_cast<DWORD>(objs_len), objs,
-                                           delay, QS_ALLINPUT, wait_flags);
-
-      if (WAIT_IO_COMPLETION == result) {
-        // We'll loop here when we service an APC.  At it currently stands,
-        // *ONLY* the IO thread uses *any* APCs, so this should have no impact
-        // on the UI thread.
-        break;  // Break to outer loop, and waitforwork() again.
-      }
-
-      // Use unsigned type to simplify range detection;
-      size_t signaled_index = result - WAIT_OBJECT_0;
-      if (signaled_index < objs_len) {
-        SignalWatcher(polling_index + signaled_index);
-        return;  // We serviced a signaled object.
-      }
-
-      if (objs_len == signaled_index)
-        return;  // A WM_* message is available.
-
-      DCHECK_NE(WAIT_FAILED, result) << GetLastError();
-
-      DCHECK(!objs || result == WAIT_TIMEOUT);
-      if (!use_polling)
-        return;
-      polling_index += objs_len;
-    } while (polling_index < total_objs);
-    // For compatibility, we didn't return sooner.  This made us do *some* wait
-    // call(s) before returning. This will probably change in next rev.
-    if (!delay || !GetCurrentDelay())
-      return;  // No work done, but timer is ready to fire.
-  }
+  WaitForIOCompletion(timeout, NULL);
 }
 
-bool MessagePumpForIO::ProcessNextObject() {
-  size_t total_objs = objects_.size();
-  if (!total_objs) {
-    return false;
+bool MessagePumpForIO::WaitForIOCompletion(DWORD timeout, IOHandler* filter) {
+  IOItem item;
+  if (completed_io_.empty() || !MatchCompletedIOItem(filter, &item)) {
+    // We have to ask the system for another IO completion.
+    if (!GetIOItem(timeout, &item))
+      return false;
+
+    if (ProcessInternalIOItem(item))
+      return true;
   }
 
-  size_t polling_index = 0;  // The first unprocessed object index.
-  do {
-    DCHECK(polling_index < total_objs);
-    size_t objs_len = total_objs - polling_index;
-    if (objs_len >= kMaxWaitObjects)
-      objs_len = kMaxWaitObjects - 1;
-    HANDLE* objs = polling_index + &objects_.front();
-
-    // Identify 1 pending object, or allow an IO APC to be completed.
-    DWORD result = WaitForMultipleObjectsEx(static_cast<DWORD>(objs_len), objs,
-                                            FALSE,    // 1 signal is sufficient.
-                                            0,        // Wait 0ms.
-                                            false);  // Not alertable (no APC).
-
-    // Use unsigned type to simplify range detection;
-    size_t signaled_index = result - WAIT_OBJECT_0;
-    if (signaled_index < objs_len) {
-      SignalWatcher(polling_index + signaled_index);
-      return true;  // We serviced a signaled object.
+  if (item.context->handler) {
+    if (filter && item.handler != filter) {
+      // Save this item for later
+      completed_io_.push_back(item);
+    } else {
+      DCHECK(item.context->handler == item.handler);
+      item.handler->OnIOCompleted(item.context, item.bytes_transfered,
+                                  item.error);
     }
-
-    // If an handle is invalid, it will be WAIT_FAILED.
-    DCHECK_EQ(WAIT_TIMEOUT, result) << GetLastError();
-    polling_index += objs_len;
-  } while (polling_index < total_objs);
-  return false;  // We serviced nothing.
+  } else {
+    // The handler must be gone by now, just cleanup the mess.
+    delete item.context;
+  }
+  return true;
 }
 
-bool MessagePumpForIO::SignalWatcher(size_t object_index) {
-  // Signal the watcher corresponding to the given index.
-
-  DCHECK(objects_.size() > object_index);
+// Asks the OS for another IO completion result.
+bool MessagePumpForIO::GetIOItem(DWORD timeout, IOItem* item) {
+  memset(item, 0, sizeof(*item));
+  ULONG_PTR key = NULL;
+  OVERLAPPED* overlapped = NULL;
+  if (!GetQueuedCompletionStatus(port_.Get(), &item->bytes_transfered, &key,
+                                 &overlapped, timeout)) {
+    if (!overlapped)
+      return false;  // Nothing in the queue.
+    item->error = GetLastError();
+    item->bytes_transfered = 0;
+  }
 
-  // On reception of OnObjectSignaled() to a Watcher object, it may call
-  // WatchObject(). watchers_ and objects_ will be modified. This is expected,
-  // so don't be afraid if, while tracing a OnObjectSignaled() function, the
-  // corresponding watchers_[result] is non-existant.
-  watchers_[object_index]->OnObjectSignaled(objects_[object_index]);
+  item->handler = reinterpret_cast<IOHandler*>(key);
+  item->context = reinterpret_cast<IOContext*>(overlapped);
+  return true;
+}
 
-  // Signaled objects tend to be removed from the watch list, and then added
-  // back (appended).  As a result, they move to the end of the objects_ array,
-  // and this should make their service "fair" (no HANDLEs should be starved).
+bool MessagePumpForIO::ProcessInternalIOItem(const IOItem& item) {
+  if (this == reinterpret_cast<MessagePumpForIO*>(item.context) &&
+      this == reinterpret_cast<MessagePumpForIO*>(item.handler)) {
+    // This is our internal completion.
+    DCHECK(!item.bytes_transfered);
+    InterlockedExchange(&have_work_, 0);
+    return true;
+  }
+  return false;
+}
 
-  return true;
+// Returns a completion item that was previously received.
+bool MessagePumpForIO::MatchCompletedIOItem(IOHandler* filter, IOItem* item) {
+  DCHECK(!completed_io_.empty());
+  for (std::list<IOItem>::iterator it = completed_io_.begin();
+       it != completed_io_.end(); ++it) {
+    if (!filter || it->handler == filter) {
+      *item = *it;
+      completed_io_.erase(it);
+      return true;
+    }
+  }
+  return false;
 }
 
 }  // namespace base