Mojo: Fix MultiprocessMessagePipeTest.QueueMessages on Mac.

Actually, it may theoretically have been flaky on Linux as well.

R=darin@chromium.org
BUG=350575

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

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

5 files changed, 36 insertions, 9 deletions

diff --git a/mojo/system/channel.cc b/mojo/system/channel.cc
index d26d46d..d7a720e 100644
--- a/mojo/system/channel.cc
+++ b/mojo/system/channel.cc
@@ -149,6 +149,12 @@ bool Channel::WriteMessage(scoped_ptr<MessageInTransit> message) {
   return raw_channel_->WriteMessage(message.Pass());
 }
 
+bool Channel::IsWriteBufferEmpty() {
+  base::AutoLock locker(lock_);
+  DCHECK(raw_channel_.get());
+  return raw_channel_->IsWriteBufferEmpty();
+}
+
 void Channel::DetachMessagePipeEndpoint(MessageInTransit::EndpointId local_id) {
   DCHECK_NE(local_id, MessageInTransit::kInvalidEndpointId);
 
diff --git a/mojo/system/channel.h b/mojo/system/channel.h
index 8268a94..83be20a 100644
--- a/mojo/system/channel.h
+++ b/mojo/system/channel.h
@@ -92,6 +92,11 @@ class MOJO_SYSTEM_IMPL_EXPORT Channel
   // This forwards |message| verbatim to |raw_channel_|.
   bool WriteMessage(scoped_ptr<MessageInTransit> message);
 
+  // See |RawChannel::IsWriteBufferEmpty()|.
+  // TODO(vtl): Maybe we shouldn't expose this, and instead have a
+  // |FlushWriteBufferAndShutdown()| or something like that.
+  bool IsWriteBufferEmpty();
+
   // This removes the message pipe/port's endpoint (with the given local ID,
   // returned by |AttachMessagePipeEndpoint()| from this channel. After this is
   // called, |local_id| may be reused for another message pipe.
diff --git a/mojo/system/multiprocess_message_pipe_unittest.cc b/mojo/system/multiprocess_message_pipe_unittest.cc
index e0eab3b..94e619a 100644
--- a/mojo/system/multiprocess_message_pipe_unittest.cc
+++ b/mojo/system/multiprocess_message_pipe_unittest.cc
@@ -14,6 +14,7 @@
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
+#include "base/threading/platform_thread.h"  // For |Sleep()|.
 #include "mojo/common/test/multiprocess_test_helper.h"
 #include "mojo/embedder/scoped_platform_handle.h"
 #include "mojo/system/channel.h"
@@ -47,6 +48,12 @@ class ChannelThread {
 
   void Stop() {
     if (channel_) {
+      // Hack to flush write buffers before quitting.
+      // TODO(vtl): Remove this once |Channel| has a
+      // |FlushWriteBufferAndShutdown()| (or whatever).
+      while (!channel_->IsWriteBufferEmpty())
+        base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(20));
+
       test_io_thread_.PostTaskAndWait(
           FROM_HERE,
           base::Bind(&ChannelThread::ShutdownChannelOnIOThread,
@@ -215,13 +222,7 @@ TEST_F(MultiprocessMessagePipeTest, Basic) {
 
 // Sends a bunch of messages to the child. Expects them "repeated" back. Waits
 // for the child to close its end before quitting.
-// TODO(yzshen): The test hangs on Mac. http://crbug.com/350575
-#if defined(OS_MACOSX)
-#define MAYBE_QueueMessages DISABLED_QueueMessages
-#else
-#define MAYBE_QueueMessages QueueMessages
-#endif
-TEST_F(MultiprocessMessagePipeTest, MAYBE_QueueMessages) {
+TEST_F(MultiprocessMessagePipeTest, QueueMessages) {
   helper()->StartChild("EchoEcho");
 
   scoped_refptr<MessagePipe> mp(new MessagePipe(
diff --git a/mojo/system/raw_channel.cc b/mojo/system/raw_channel.cc
index 029e18a..cfdc131 100644
--- a/mojo/system/raw_channel.cc
+++ b/mojo/system/raw_channel.cc
@@ -130,6 +130,9 @@ void RawChannel::Shutdown() {
 
   base::AutoLock locker(write_lock_);
 
+  LOG_IF(WARNING, !write_buffer_->message_queue_.empty())
+      << "Shutting down RawChannel with write buffer nonempty";
+
   weak_ptr_factory_.InvalidateWeakPtrs();
 
   read_stopped_ = true;
@@ -172,6 +175,12 @@ bool RawChannel::WriteMessage(scoped_ptr<MessageInTransit> message) {
   return result;
 }
 
+// Reminder: This must be thread-safe.
+bool RawChannel::IsWriteBufferEmpty() {
+  base::AutoLock locker(write_lock_);
+  return write_buffer_->message_queue_.empty();
+}
+
 RawChannel::ReadBuffer* RawChannel::read_buffer() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
   return read_buffer_.get();
diff --git a/mojo/system/raw_channel.h b/mojo/system/raw_channel.h
index 2eb28f6..8f760d6 100644
--- a/mojo/system/raw_channel.h
+++ b/mojo/system/raw_channel.h
@@ -85,10 +85,16 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
   // This must be called (on the I/O thread) before this object is destroyed.
   void Shutdown();
 
-  // This is thread-safe. It takes ownership of |message| (always, even on
-  // failure). Returns true on success.
+  // Writes the given message (or schedules it to be written). This is
+  // thread-safe. Returns true on success.
   bool WriteMessage(scoped_ptr<MessageInTransit> message);
 
+  // Returns true if the write buffer is empty (i.e., all messages written using
+  // |WriteMessage()| have actually been sent.
+  // TODO(vtl): We should really also notify our delegate when the write buffer
+  // becomes empty (or something like that).
+  bool IsWriteBufferEmpty();
+
  protected:
   // Return values of |[Schedule]Read()| and |[Schedule]WriteNoLock()|.
   enum IOResult {