Mojo: Add RawChannel tests for calling Shutdown() from OnReadMessage()/OnFatalError().

Also make RawChannel not call its delegate after Shutdown().

R=sky@chromium.org

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

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

4 files changed, 166 insertions, 21 deletions

diff --git a/mojo/system/raw_channel.cc b/mojo/system/raw_channel.cc
index 80f2176..d9febeb 100644
--- a/mojo/system/raw_channel.cc
+++ b/mojo/system/raw_channel.cc
@@ -92,8 +92,8 @@ size_t RawChannel::WriteBuffer::GetTotalBytesToWrite() const {
 }
 
 RawChannel::RawChannel()
-    : delegate_(NULL),
-      message_loop_for_io_(NULL),
+    : message_loop_for_io_(NULL),
+      delegate_(NULL),
       read_stopped_(false),
       write_stopped_(false),
       weak_ptr_factory_(this) {
@@ -140,10 +140,11 @@ void RawChannel::Shutdown() {
   LOG_IF(WARNING, !write_buffer_->message_queue_.empty())
       << "Shutting down RawChannel with write buffer nonempty";
 
-  weak_ptr_factory_.InvalidateWeakPtrs();
-
+  // Reset the delegate so that it won't receive further calls.
+  delegate_ = NULL;
   read_stopped_ = true;
   write_stopped_ = true;
+  weak_ptr_factory_.InvalidateWeakPtrs();
 
   OnShutdownNoLock(read_buffer_.Pass(), write_buffer_.Pass());
 }
@@ -254,6 +255,7 @@ void RawChannel::OnReadCompleted(bool result, size_t bytes_read) {
       DCHECK_EQ(message_view.total_size(), message_size);
 
       // Dispatch the message.
+      DCHECK(delegate_);
       delegate_->OnReadMessage(message_view);
       if (read_stopped_) {
         // |Shutdown()| was called in |OnReadMessage()|.
@@ -328,7 +330,8 @@ void RawChannel::OnWriteCompleted(bool result, size_t bytes_written) {
 void RawChannel::CallOnFatalError(Delegate::FatalError fatal_error) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
   // TODO(vtl): Add a "write_lock_.AssertNotAcquired()"?
-  delegate_->OnFatalError(fatal_error);
+  if (delegate_)
+    delegate_->OnFatalError(fatal_error);
 }
 
 bool RawChannel::OnWriteCompletedNoLock(bool result, size_t bytes_written) {
diff --git a/mojo/system/raw_channel.h b/mojo/system/raw_channel.h
index 4b60e36d..b4571f8 100644
--- a/mojo/system/raw_channel.h
+++ b/mojo/system/raw_channel.h
@@ -59,7 +59,7 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
 
     // Called when there's a fatal error, which leads to the channel no longer
     // being viable. This may call |Shutdown()| (on the |RawChannel()|), but
-    // must now destroy it.
+    // must not destroy it.
     //
     // For each raw channel, at most one |FATAL_ERROR_FAILED_READ| and at most
     // one |FATAL_ERROR_FAILED_WRITE| notification will be issued (both may be
@@ -78,7 +78,8 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
 
   // This must be called (on an I/O thread) before this object is used. Does
   // *not* take ownership of |delegate|. Both the I/O thread and |delegate| must
-  // remain alive for the lifetime of this object. Returns true on success. On
+  // remain alive until |Shutdown()| is called (unless this fails); |delegate|
+  // will no longer be used after |Shutdown()|. Returns true on success. On
   // failure, |Shutdown()| should *not* be called.
   bool Init(Delegate* delegate);
 
@@ -199,9 +200,9 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
 
   // Must be called on the I/O thread WITHOUT |write_lock_| held.
   virtual bool OnInit() = 0;
-  // On shutdown, passes the ownership of the buffers to subclasses, who may
-  // want to preserve them if there are pending read/write.
-  // Must be called on the I/O thread under |write_lock_|.
+  // On shutdown, passes the ownership of the buffers to subclasses, which may
+  // want to preserve them if there are pending read/write. Must be called on
+  // the I/O thread under |write_lock_|.
   virtual void OnShutdownNoLock(
       scoped_ptr<ReadBuffer> read_buffer,
       scoped_ptr<WriteBuffer> write_buffer) = 0;
@@ -225,10 +226,10 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
 
   // Set in |Init()| and never changed (hence usable on any thread without
   // locking):
-  Delegate* delegate_;
   base::MessageLoopForIO* message_loop_for_io_;
 
   // Only used on the I/O thread:
+  Delegate* delegate_;
   bool read_stopped_;
   scoped_ptr<ReadBuffer> read_buffer_;
 
diff --git a/mojo/system/raw_channel_posix.cc b/mojo/system/raw_channel_posix.cc
index 2b1a358..cebbc9d 100644
--- a/mojo/system/raw_channel_posix.cc
+++ b/mojo/system/raw_channel_posix.cc
@@ -111,8 +111,7 @@ RawChannel::IOResult RawChannelPosix::Read(size_t* bytes_read) {
 
   // |read_result == 0| means "end of file".
   if (read_result == 0 || (errno != EAGAIN && errno != EWOULDBLOCK)) {
-    if (read_result != 0)
-      PLOG(ERROR) << "read";
+    PLOG_IF(ERROR, read_result != 0) << "read";
 
     // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again.
     read_watcher_.reset();
@@ -174,8 +173,7 @@ RawChannel::IOResult RawChannelPosix::WriteNoLock(size_t* bytes_written) {
   }
 
   if (errno != EAGAIN && errno != EWOULDBLOCK) {
-    PLOG(ERROR) << "write of size "
-                << write_buffer_no_lock()->GetTotalBytesToWrite();
+    PLOG(ERROR) << "write";
     return IO_FAILED;
   }
 
diff --git a/mojo/system/raw_channel_unittest.cc b/mojo/system/raw_channel_unittest.cc
index 70aba1c..6c2b722 100644
--- a/mojo/system/raw_channel_unittest.cc
+++ b/mojo/system/raw_channel_unittest.cc
@@ -253,7 +253,7 @@ class ReadCheckerRawChannelDelegate : public RawChannel::Delegate {
     CHECK_EQ(fatal_error, FATAL_ERROR_FAILED_READ);
   }
 
-  // Wait for all the messages (of sizes |expected_sizes_|) to be seen.
+  // Waits for all the messages (of sizes |expected_sizes_|) to be seen.
   void Wait() {
     done_event_.Wait();
   }
@@ -362,7 +362,7 @@ class ReadCountdownRawChannelDelegate : public RawChannel::Delegate {
     CHECK_EQ(fatal_error, FATAL_ERROR_FAILED_READ);
   }
 
-  // Wait for all the messages to have been seen.
+  // Waits for all the messages to have been seen.
   void Wait() {
     done_event_.Wait();
   }
@@ -497,12 +497,11 @@ TEST_F(RawChannelTest, OnFatalError) {
 TEST_F(RawChannelTest, ReadUnaffectedByWriteFatalError) {
   const size_t kMessageCount = 5;
 
-  // Write into the other end a few messages.
+  // Write a few messages into the other end.
   uint32_t message_size = 1;
-  for (size_t count = 0; count < kMessageCount;
-       ++count, message_size += message_size / 2 + 1) {
+  for (size_t i = 0; i < kMessageCount;
+       i++, message_size += message_size / 2 + 1)
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), message_size));
-  }
 
   // Close the other end, which should make writing fail.
   handles[1].reset();
@@ -548,6 +547,150 @@ TEST_F(RawChannelTest, WriteMessageAfterShutdown) {
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 }
 
+// RawChannelTest.ShutdownOnReadMessage ----------------------------------------
+
+class ShutdownOnReadMessageRawChannelDelegate : public RawChannel::Delegate {
+ public:
+  explicit ShutdownOnReadMessageRawChannelDelegate(RawChannel* raw_channel)
+      : raw_channel_(raw_channel),
+        done_event_(false, false),
+        did_shutdown_(false) {}
+  virtual ~ShutdownOnReadMessageRawChannelDelegate() {}
+
+  // |RawChannel::Delegate| implementation (called on the I/O thread):
+  virtual void OnReadMessage(
+      const MessageInTransit::View& message_view) OVERRIDE {
+    EXPECT_FALSE(did_shutdown_);
+    EXPECT_TRUE(CheckMessageData(message_view.bytes(),
+                message_view.num_bytes()));
+    raw_channel_->Shutdown();
+    did_shutdown_ = true;
+    done_event_.Signal();
+  }
+  virtual void OnFatalError(FatalError /*fatal_error*/) OVERRIDE {
+    CHECK(false);  // Should not get called.
+  }
+
+  // Waits for shutdown.
+  void Wait() {
+    done_event_.Wait();
+    EXPECT_TRUE(did_shutdown_);
+  }
+
+ private:
+  RawChannel* const raw_channel_;
+  base::WaitableEvent done_event_;
+  bool did_shutdown_;
+
+  DISALLOW_COPY_AND_ASSIGN(ShutdownOnReadMessageRawChannelDelegate);
+};
+
+// TODO(vtl): crbug.com/366768
+#if defined(OS_WIN)
+#define MAYBE_ShutdownOnReadMessage DISABLED_ShutdownOnReadMessage
+#else
+#define MAYBE_ShutdownOnReadMessage ShutdownOnReadMessage
+#endif
+TEST_F(RawChannelTest, MAYBE_ShutdownOnReadMessage) {
+  // Write a few messages into the other end.
+  for (size_t count = 0; count < 5; count++)
+    EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), 10));
+
+  scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
+  ShutdownOnReadMessageRawChannelDelegate delegate(rc.get());
+  io_thread()->PostTaskAndWait(FROM_HERE,
+                               base::Bind(&InitOnIOThread, rc.get(),
+                                          base::Unretained(&delegate)));
+
+  // Wait for the delegate, which will shut the |RawChannel| down.
+  delegate.Wait();
+}
+
+// RawChannelTest.ShutdownOnFatalError{Read, Write} ----------------------------
+
+class ShutdownOnFatalErrorRawChannelDelegate : public RawChannel::Delegate {
+ public:
+  ShutdownOnFatalErrorRawChannelDelegate(RawChannel* raw_channel,
+                                         FatalError shutdown_on_error_type)
+      : raw_channel_(raw_channel),
+        shutdown_on_error_type_(shutdown_on_error_type),
+        done_event_(false, false),
+        did_shutdown_(false) {}
+  virtual ~ShutdownOnFatalErrorRawChannelDelegate() {}
+
+  // |RawChannel::Delegate| implementation (called on the I/O thread):
+  virtual void OnReadMessage(
+      const MessageInTransit::View& /*message_view*/) OVERRIDE {
+    CHECK(false);  // Should not get called.
+  }
+  virtual void OnFatalError(FatalError fatal_error) OVERRIDE {
+    EXPECT_FALSE(did_shutdown_);
+    if (fatal_error != shutdown_on_error_type_)
+      return;
+    raw_channel_->Shutdown();
+    did_shutdown_ = true;
+    done_event_.Signal();
+  }
+
+  // Waits for shutdown.
+  void Wait() {
+    done_event_.Wait();
+    EXPECT_TRUE(did_shutdown_);
+  }
+
+ private:
+  RawChannel* const raw_channel_;
+  const FatalError shutdown_on_error_type_;
+  base::WaitableEvent done_event_;
+  bool did_shutdown_;
+
+  DISALLOW_COPY_AND_ASSIGN(ShutdownOnFatalErrorRawChannelDelegate);
+};
+
+// TODO(vtl): crbug.com/366768
+#if defined(OS_WIN)
+#define MAYBE_ShutdownOnFatalErrorRead DISABLED_ShutdownOnFatalErrorRead
+#else
+#define MAYBE_ShutdownOnFatalErrorRead ShutdownOnFatalErrorRead
+#endif
+TEST_F(RawChannelTest, MAYBE_ShutdownOnFatalErrorRead) {
+  scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
+  ShutdownOnFatalErrorRawChannelDelegate delegate(
+      rc.get(), RawChannel::Delegate::FATAL_ERROR_FAILED_READ);
+  io_thread()->PostTaskAndWait(FROM_HERE,
+                               base::Bind(&InitOnIOThread, rc.get(),
+                                          base::Unretained(&delegate)));
+
+  // Close the handle of the other end, which should stuff fail.
+  handles[1].reset();
+
+  // Wait for the delegate, which will shut the |RawChannel| down.
+  delegate.Wait();
+}
+
+// TODO(vtl): crbug.com/366768
+#if defined(OS_WIN)
+#define MAYBE_ShutdownOnFatalErrorWrite DISABLED_ShutdownOnFatalErrorWrite
+#else
+#define MAYBE_ShutdownOnFatalErrorWrite ShutdownOnFatalErrorWrite
+#endif
+TEST_F(RawChannelTest, MAYBE_ShutdownOnFatalErrorWrite) {
+  scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
+  ShutdownOnFatalErrorRawChannelDelegate delegate(
+      rc.get(), RawChannel::Delegate::FATAL_ERROR_FAILED_WRITE);
+  io_thread()->PostTaskAndWait(FROM_HERE,
+                               base::Bind(&InitOnIOThread, rc.get(),
+                                          base::Unretained(&delegate)));
+
+  // Close the handle of the other end, which should stuff fail.
+  handles[1].reset();
+
+  EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
+
+  // Wait for the delegate, which will shut the |RawChannel| down.
+  delegate.Wait();
+}
+
 }  // namespace
 }  // namespace system
 }  // namespace mojo