Fixed the racy code around the message-only window in base::MessagePumpForUI on Windows.

base/message_pump_win.cc

 // Copyright (c) 2012 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 "base/message_pump_win.h"
 
 #include <math.h>
 
 #include "base/debug/trace_event.h"
 #include "base/message_loop.h"
 #include "base/metrics/histogram.h"
-#include "base/process_util.h"
-#include "base/win/wrapped_window_proc.h"
 
 namespace {
 
 enum MessageLoopProblems {
   MESSAGE_POST_ERROR,
   COMPLETION_POST_ERROR,
   SET_TIMER_ERROR,
   MESSAGE_LOOP_PROBLEM_MAX,
 };
 
+// Possible states of the message pump:
+//   - kPumpIdle: the thread is sleeping, waiting for work to be posted.
+//   - kPumpHaveWork: a window message or completion packet has been posted.
+//   - kPumpDisabled: the pump is disabled, no more work can be posted.
+static const LONG kPumpIdle = 0;
+static const LONG kPumpHaveWork = 1;
+static const LONG kPumpDisabled = 2;
+
+// Used to wake up the message loop thread in the case if it is waiting for
+// another thread to exit MessagePumpForUI::ScheduleWork().
+VOID CALLBACK DummyApc(ULONG_PTR) {}
+
 }  // namespace
 
 namespace base {
 
-static const wchar_t kWndClass[] = L"Chrome_MessagePumpWindow";
-
 // Message sent to get an additional time slice for pumping (processing) another
 // task (a series of such messages creates a continuous task pump).
 static const int kMsgHaveWork = WM_USER + 1;
 
+// Used by MessagePumpUI to wake up the thread and check any pending timers.
+static const int kTimerId = 1;
+
 //-----------------------------------------------------------------------------
 // MessagePumpWin public:
 
+MessagePumpWin::MessagePumpWin() : pump_state_(kPumpIdle), state_(NULL) {}
+
 void MessagePumpWin::AddObserver(MessagePumpObserver* observer) {
   observers_.AddObserver(observer);
 }
 
 void MessagePumpWin::RemoveObserver(MessagePumpObserver* observer) {
   observers_.RemoveObserver(observer);
 }
 
 void MessagePumpWin::WillProcessMessage(const MSG& msg) {
   FOR_EACH_OBSERVER(MessagePumpObserver, observers_, WillProcessEvent(msg));
@@ skipping 42 matching lines @@
   if (delay < 0)
     delay = 0;
 
   return delay;
 }
 
 //-----------------------------------------------------------------------------
 // MessagePumpForUI public:
 
 MessagePumpForUI::MessagePumpForUI()
-    : instance_(NULL),
-      message_filter_(new MessageFilter) {
-  InitMessageWnd();
+    : message_filter_(new MessageFilter),
+      window_(new win::MessageWindow()) {
+  CHECK(DuplicateHandle(GetCurrentProcess(),
+                        GetCurrentThread(),
+                        GetCurrentProcess(),
+                        thread_.Receive(),
+                        THREAD_SET_CONTEXT,
+                        FALSE,    // no not inherit this handle
+                        0));
+  CHECK(window_->Create(this));
 }
 
 MessagePumpForUI::~MessagePumpForUI() {
-  DestroyWindow(message_hwnd_);
-  UnregisterClass(kWndClass, instance_);
 }
 
 void MessagePumpForUI::ScheduleWork() {
-  if (InterlockedExchange(&have_work_, 1))
-    return;  // Someone else continued the pumping.
+  if (InterlockedCompareExchange(&pump_state_, kPumpHaveWork,
+                                 kPumpIdle) != kPumpIdle) {
+    // Either someone else continued the pumping or the pump is disabled.
+    return;
+  }
 
   // Make sure the MessagePump does some work for us.
-  BOOL ret = PostMessage(message_hwnd_, kMsgHaveWork,
-                         reinterpret_cast<WPARAM>(this), 0);
-  if (ret)
-    return;  // There was room in the Window Message queue.
+  if (PostMessage(window_->hwnd(), kMsgHaveWork, 0, 0))
+    return;
 
   // We have failed to insert a have-work message, so there is a chance that we
   // will starve tasks/timers while sitting in a nested message loop.  Nested
   // loops only look at Windows Message queues, and don't look at *our* task
   // queues, etc., so we might not get a time slice in such. :-(
   // We could abort here, but the fear is that this failure mode is plausibly
   // common (queue is full, of about 2000 messages), so we'll do a near-graceful
   // recovery.  Nested loops are pretty transient (we think), so this will
   // probably be recoverable.
-  InterlockedExchange(&have_work_, 0);  // Clarify that we didn't really insert.
   UMA_HISTOGRAM_ENUMERATION("Chrome.MessageLoopProblem", MESSAGE_POST_ERROR,
                             MESSAGE_LOOP_PROBLEM_MAX);
+
+  // Clarify that we didn't really insert.
+  InterlockedExchange(&pump_state_, kPumpIdle);

[darin (slow to review), 2013/06/10 19:28:45]

nit: Shouldn't this use InterlockedCompareExchange(&pump_state_, kPumpIdle,
kPumpHaveWork) so that you don't mistakenly clobber kPumpDisabled?

Are you sure using a tri-state variable here is best?  Maybe it would be cleaner
to have two variables, one for indicating work and the other for indicating
shutdown?

[alexeypa (please no reviews), 2013/06/10 20:16:12]

No, at this point |pump_state_| is guaranteed to be |kPumpHaveWork| and the
current thread is the only thread that is allowed to modify |pump_state_|: 

- threads posting work and calling ScheduleWork() will bail out.
- the message loop thread will wait for |kPumpHaveWork| to be cleared.
Having two variables will increase the number of states to four and make
transitions between the states more complex and, therefore, - error prone.

+
+  // Try to wake up the message loop thread by posting an APC. This might not

[darin (slow to review), 2013/06/10 19:28:45]

I'm not sure I fully understand the conditions that lead to reaching this code. 
I also worry a little bit since sometimes other code runs nested message loops
on our thread not using our MessageLoop infrastructure.  For example, a library
function might implement its own nested message loop.  I'm not sure what it
means for our APC to run in that context.

[alexeypa (please no reviews), 2013/06/10 20:16:12]

I don't have any data on hand but two reasons that come to my mind are:

1. The message queue is full. The receiving thread is spinning, blocked on
something or became unresponsive because any other reason.
2. The receiving thread has been terminated.

There could be other reasons.
APC will not be invoked unless the thread is in alertable state. Which means
either _our_ message pump is pumping events or 3rd party code doing an alertable
wait. In either case this CL either improve or does not change behavior:

1. If it is our pump, it will be waken up via APC, while before we would hope
that someone will post a message to us.
2. If it is not our code, than there is no change in behavior.

+  // work while a nested loop still running (see the comment above) but this
+  // will unblock WillDestroyCurrentMessageLoop() if it is waiting for
+  // ScheduleWork() to complete.
+  //
+  // According to the UMA metrics posting an I/O completion packet has very low
+  // error rate. Queuing an APC hits roughly the same path in the kernel so
+  // the error rate should be low as well. Given that we do it only when
+  // PostMessage() fails it should be safe to CHECK() here.
+  CHECK(QueueUserAPC(&DummyApc, thread_, 0));
 }
 
 void MessagePumpForUI::ScheduleDelayedWork(const TimeTicks& delayed_work_time) {
+  DCHECK(window_->CalledOnValidThread());
+
   //
   // 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
   // mechanism because the application can enter modal windows loops where it
   // is not running our MessageLoop; the only way to have our timers fire in
   // these cases is to post messages there.
   //
   // To provide sub-10ms timers, we process timers directly from our run loop.
   // For the common case, timers will be processed there as the run loop does
   // its normal work.  However, we *also* set the system timer so that WM_TIMER
   // events fire.  This mops up the case of timers not being able to work in
   // modal message loops.  It is possible for the SetTimer to pop and have no
   // pending timers, because they could have already been processed by the
   // run loop itself.
   //
   // We use a single SetTimer corresponding to the timer that will expire
   // soonest.  As new timers are created and destroyed, we update SetTimer.
   // Getting a spurrious SetTimer event firing is benign, as we'll just be
   // processing an empty timer queue.
   //
   delayed_work_time_ = delayed_work_time;
 
   int delay_msec = GetCurrentDelay();
   DCHECK_GE(delay_msec, 0);
   if (delay_msec < USER_TIMER_MINIMUM)
     delay_msec = USER_TIMER_MINIMUM;
 
   // Create a WM_TIMER event that will wake us up to check for any pending
   // timers (in case we are running within a nested, external sub-pump).
-  BOOL ret = SetTimer(message_hwnd_, reinterpret_cast<UINT_PTR>(this),
-                      delay_msec, NULL);
+  BOOL ret = SetTimer(window_->hwnd(), kTimerId, delay_msec, NULL);
   if (ret)
     return;
   // If we can't set timers, we are in big trouble... but cross our fingers for
   // now.
   // TODO(jar): If we don't see this error, use a CHECK() here instead.
   UMA_HISTOGRAM_ENUMERATION("Chrome.MessageLoopProblem", SET_TIMER_ERROR,
                             MESSAGE_LOOP_PROBLEM_MAX);
 }
 
+void MessagePumpForUI::WillDestroyCurrentMessageLoop() {
+  DCHECK(window_->CalledOnValidThread());
+
+  // Disable the message pump. If |pump_state_ == kPumpHaveWork| then
+  // ScheduleWork() might be still running on a different thread. Wait until
+  // |kMsgHaveWork| is received or |pump_state_| is reset back to |kPumpIdle|.
+  while (InterlockedCompareExchange(&pump_state_, kPumpDisabled,
+                                    kPumpIdle) == kPumpHaveWork) {
+    MSG msg;
+    if (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE)) {
+      if (msg.message == kMsgHaveWork && msg.hwnd == window_->hwnd()) {
+        // Now that we received |kMsgHaveWork| the pump can be safely disabled.
+        InterlockedExchange(&pump_state_, kPumpDisabled);
+        break;
+      }
+    }
+
+    // Wait until |kMsgHaveWork| is posted or an APC is received.
+    WaitForWork();
+  }
+
+  // At this point the pump is disabled and other threads exited ScheduleWork().
+  window_.reset();
+}
+
 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;
   int peek_count;
   for (peek_count = 0; peek_count < kMaxPeekCount; ++peek_count) {
     MSG msg;
     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("Loop.PumpOutPendingPaintMessages Peeks", peek_count);
 }
 
 //-----------------------------------------------------------------------------
 // MessagePumpForUI private:
 
-// static
-LRESULT CALLBACK MessagePumpForUI::WndProcThunk(
-    HWND hwnd, UINT message, WPARAM wparam, LPARAM lparam) {
+bool MessagePumpForUI::HandleMessage(HWND hwnd,
+                                     UINT message,
+                                     WPARAM wparam,
+                                     LPARAM lparam,
+                                     LRESULT* result) {
   switch (message) {
     case kMsgHaveWork:
-      reinterpret_cast<MessagePumpForUI*>(wparam)->HandleWorkMessage();
+      HandleWorkMessage();
       break;
+
     case WM_TIMER:
-      reinterpret_cast<MessagePumpForUI*>(wparam)->HandleTimerMessage();
+      HandleTimerMessage();
       break;
   }
-  return DefWindowProc(hwnd, message, wparam, lparam);
+
+  // Do default processing for all messages.
+  return false;
 }
 
 void MessagePumpForUI::DoRunLoop() {
+  DCHECK(window_->CalledOnValidThread());
+
   // 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
@@ skipping 17 matching lines @@
     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));
+      KillTimer(window_->hwnd(), kTimerId);
     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() {
-  WNDCLASSEX wc = {0};
-  wc.cbSize = sizeof(wc);
-  wc.lpfnWndProc = base::win::WrappedWindowProc<WndProcThunk>;
-  wc.hInstance = base::GetModuleFromAddress(wc.lpfnWndProc);
-  wc.lpszClassName = kWndClass;
-  instance_ = wc.hInstance;
-  RegisterClassEx(&wc);
-
-  message_hwnd_ =
-      CreateWindow(kWndClass, 0, 0, 0, 0, 0, 0, HWND_MESSAGE, 0, instance_, 0);
-  DCHECK(message_hwnd_);
-}
-
 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);
+                                       MWMO_ALERTABLE | MWMO_INPUTAVAILABLE);

[darin (slow to review), 2013/06/10 19:28:45]

I'm a little worried that making this call alertable might result in wakeups
caused by other sources.  What does it mean if WaitForWork() returns early in
those cases (i.e., when we are not shutting down)?  It is probably safe to do
another pass through DoRunLoop().

[alexeypa (please no reviews), 2013/06/10 20:16:12]

It means that it will be possible to post APCs to the thread. Code in our code
base does not do it because it simply does not work currently. 3rd party code
that is doing that is currently broken anyway.

 
   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 may appear if the child window has
     // capture).
     // The subsequent PeekMessages call may fail 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
     // some time 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;
+  } else if (WAIT_IO_COMPLETION == result) {
+    // The wait was ended by one or more APCs. It could be cause
+    // MessagePumpUI::ScheduleWork() is trying to wake up the message loop
+    // thread.
+    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.
   if (!state_) {
     // Since we handled a kMsgHaveWork message, we must still update this flag.
-    InterlockedExchange(&have_work_, 0);
+    InterlockedExchange(&pump_state_, kPumpIdle);
     return;
   }
 
   // Let whatever would have run had we not been putting messages in the queue
   // run now.  This is an attempt to make our dummy message not starve other
   // messages that may be in the Windows message queue.
   ProcessPumpReplacementMessage();
 
   // Now give the delegate a chance to do some work.  He'll let us know if he
   // needs to do more work.
   if (state_->delegate->DoWork())
     ScheduleWork();
 }
 
 void MessagePumpForUI::HandleTimerMessage() {
-  KillTimer(message_hwnd_, reinterpret_cast<UINT_PTR>(this));
+  KillTimer(window_->hwnd(), kTimerId);
 
   // If we are being called outside of the context of Run, then don't do
   // anything.  This could correspond to a MessageBox call or something of
   // that sort.
   if (!state_)
     return;
 
   state_->delegate->DoDelayedWork(&delayed_work_time_);
   if (!delayed_work_time_.is_null()) {
     // A bit gratuitous to set delayed_work_time_ again, but oh well.
@@ skipping 23 matching lines @@
                "message", msg.message);
   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_)
+  if (msg.message == kMsgHaveWork && msg.hwnd == window_->hwnd())
     return ProcessPumpReplacementMessage();
 
   if (CallMsgFilter(const_cast<MSG*>(&msg), kMessageFilterCode))
     return true;
 
   WillProcessMessage(msg);
 
   if (!message_filter_->ProcessMessage(msg)) {
     if (state_->dispatcher) {
       if (!state_->dispatcher->Dispatch(msg))
         state_->should_quit = true;
     } else {
       TranslateMessage(&msg);
       DispatchMessage(&msg);
     }
   }
 
   DidProcessMessage(msg);
   return true;
 }
 
 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
   // 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
+  // then resets the pump_state_ 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!!
 
   bool have_message = false;
   MSG msg;
   // We should not process all window messages if we are in the context of an
   // OS modal loop, i.e. in the context of a windows API call like MessageBox.
   // This is to ensure that these messages are peeked out by the OS modal loop.
   if (MessageLoop::current()->os_modal_loop()) {
     // We only peek out WM_PAINT and WM_TIMER here for reasons mentioned above.
     have_message = PeekMessage(&msg, NULL, WM_PAINT, WM_PAINT, PM_REMOVE) ||
                    PeekMessage(&msg, NULL, WM_TIMER, WM_TIMER, PM_REMOVE);
   } else {
     have_message = !!message_filter_->DoPeekMessage(&msg, NULL, 0, 0,
                                                     PM_REMOVE);
   }
 
   DCHECK(!have_message || kMsgHaveWork != msg.message ||
-         msg.hwnd != message_hwnd_);
+         msg.hwnd != window_->hwnd());
 
   // Since we discarded a kMsgHaveWork message, we must update the flag.
-  int old_have_work = InterlockedExchange(&have_work_, 0);
-  DCHECK(old_have_work);
+  int old_pump_state = InterlockedExchange(&pump_state_, kPumpIdle);
+  DCHECK(old_pump_state);
 
   // We don't need a special time slice if we didn't have_message to process.
   if (!have_message)
     return false;
 
   // Guarantee we'll get another time slice in the case where we go into native
   // windows code.   This ScheduleWork() may hurt performance a tiny bit when
   // tasks appear very infrequently, but when the event queue is busy, the
   // kMsgHaveWork events get (percentage wise) rarer and rarer.
   ScheduleWork();
   return ProcessMessageHelper(msg);
 }
 
 void MessagePumpForUI::SetMessageFilter(
     scoped_ptr<MessageFilter> message_filter) {
   message_filter_ = message_filter.Pass();
 }
 
 //-----------------------------------------------------------------------------
 // MessagePumpForIO public:
 
 MessagePumpForIO::MessagePumpForIO() {
   port_.Set(CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1));
   DCHECK(port_.IsValid());
 }
 
 void MessagePumpForIO::ScheduleWork() {
-  if (InterlockedExchange(&have_work_, 1))
+  if (InterlockedExchange(&pump_state_, kPumpHaveWork))
     return;  // Someone else continued the pumping.
 
   // Make sure the MessagePump does some work for us.
   BOOL ret = PostQueuedCompletionStatus(port_, 0,
                                         reinterpret_cast<ULONG_PTR>(this),
                                         reinterpret_cast<OVERLAPPED*>(this));
   if (ret)
     return;  // Post worked perfectly.
 
   // See comment in MessagePumpForUI::ScheduleWork() for this error recovery.
-  InterlockedExchange(&have_work_, 0);  // Clarify that we didn't succeed.
+  InterlockedExchange(&pump_state_, kPumpIdle);
   UMA_HISTOGRAM_ENUMERATION("Chrome.MessageLoopProblem", COMPLETION_POST_ERROR,
                             MESSAGE_LOOP_PROBLEM_MAX);
 }
 
 void MessagePumpForIO::ScheduleDelayedWork(const TimeTicks& 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;
 }
@@ skipping 124 matching lines @@
   item->handler = KeyToHandler(key, &item->has_valid_io_context);
   item->context = reinterpret_cast<IOContext*>(overlapped);
   return true;
 }
 
 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);
+    InterlockedExchange(&pump_state_, kPumpIdle);
     return true;
   }
   return false;
 }
 
 // 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) {
@@ skipping 40 matching lines @@
 
 // static
 MessagePumpForIO::IOHandler* MessagePumpForIO::KeyToHandler(
     ULONG_PTR key,
     bool* has_valid_io_context) {
   *has_valid_io_context = ((key & 1) == 0);
   return reinterpret_cast<IOHandler*>(key & ~static_cast<ULONG_PTR>(1));
 }
 
 }  // namespace base