Make sure that the UI window created by base::MessagePumpForUI is destoyed on the same thread (Wind…

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/stringprintf.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 kWndClassFormat[] = L"Chrome_MessagePumpWindow_%p";
 
 // 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;
 
 //-----------------------------------------------------------------------------
 // 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 44 matching lines @@
 
   return delay;
 }
 
 //-----------------------------------------------------------------------------
 // MessagePumpForUI public:
 
 MessagePumpForUI::MessagePumpForUI()
     : atom_(0),
       message_filter_(new MessageFilter) {
+  CHECK(DuplicateHandle(GetCurrentProcess(),
+                        GetCurrentThread(),
+                        GetCurrentProcess(),
+                        thread_.Receive(),
+                        THREAD_SET_CONTEXT,
+                        FALSE,    // no not inherit this handle
+                        0));
   InitMessageWnd();
 }
 
 MessagePumpForUI::~MessagePumpForUI() {
-  DestroyWindow(message_hwnd_);
-  UnregisterClass(MAKEINTATOM(atom_),
-                  base::GetModuleFromAddress(&WndProcThunk));
+  DCHECK(!atom_);
+  DCHECK(!message_hwnd_);
 }
 
 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(message_hwnd_, kMsgHaveWork, reinterpret_cast<WPARAM>(this),
+                  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);
+
+  // Try to wake up the message loop thread by posting an APC. This might not
+  // 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) {
   //
   // 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.
   //
@@ skipping 23 matching lines @@
                       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() {
+  // 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,

[rvargas (doing something else), 2013/06/12 00:22:23]

To be honest I prefer the version with a lock here.

ScheduleWork can use Try instead of Acquire (either someone else is starting the
pump or the loop is going away) so a regular critical section should be very
close to an interlocked operation.

We should only roll our own locks (like this implementation) after finding that
a regular lock is hurting performance.

[alexeypa (please no reviews), 2013/06/12 18:05:22]

Well, Darin's reaction was exactly the opposite. He was concerned about
performance penalty the lock adds. See https://codereview.chromium.org/16020005.
Try vs Acquire should not make much difference. The blocking can occur only when
MessageLoop is being destroyed. So when ScheduleWork() enters the critical
section it is always free meaning EnterCriticalSection() never enters the kernel
nor it spins.
Check out my comments at https://codereview.chromium.org/16020005/#msg9.

Another reason to keep the interlocked logic that that it wasn't introduced in
this CL. It existed before, I just added the3rd state to it.

[rvargas (doing something else), 2013/06/12 18:24:24]

Yes, I read all the comments on the original CL before sending my comments
out...

I agree that Try vs Acquire should be close to the same (you _can_ have multiple
threads calling ScheduleWork at the same time so there is a small difference).
But the fact that we don't care about possible contention during the normal
lifetime of the pump is lost on your original version with a lock.

... and that is part of my argument against writing our own lock. The current
version has a very simple "lock" (a boolean value accessed with an interlocked
operation). This cl is extending that lock to a multi-state interlocked variable
aided with posting a task and and APC. All that logic is a complex
synchronization "primitive" and I don't see the justification for it.

Yes, the new behavior requires more synchronization to avoid the shutdown race
of using a closed handle. But a standard lock provides basically the same
performance of the current implementation (one interlocked access), and it is
way simpler.

[alexeypa (please no reviews), 2013/06/12 18:57:25]

OK, fair enough. I resurrected the version with the lock. PTAL.

+                                    kPumpIdle) == kPumpHaveWork) {
+    MSG msg;
+    if (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE)) {
+      if (msg.message == kMsgHaveWork && msg.hwnd == message_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().
+  DestroyWindow(message_hwnd_);
+  UnregisterClass(MAKEINTATOM(atom_),
+                  base::GetModuleFromAddress(&WndProcThunk));
+
+  // Let the destructor know that the resources have been freed.
+  message_hwnd_ = NULL;
+  atom_ = 0;
+}
+
 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.
@@ skipping 104 matching lines @@
 
 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);
 
   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.
@@ skipping 67 matching lines @@
   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_);
 
   // Since we discarded a kMsgHaveWork message, we must update the flag.
-  int old_have_work = InterlockedExchange(&have_work_, 0);
+  int old_have_work = InterlockedExchange(&pump_state_, kPumpIdle);
   DCHECK(old_have_work);
 
   // 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