Fix MessageReader not to invoke message callback from done callback.

In the current implementation MessageReader may read a message while processing
done callback for the previous message. The message handler may not expect that
which may lead to crashes. E.g. ChannelMultiplex was written such that
OnPacketReceived() is not reentrant.

BUG=148512


Review URL: https://chromiumcodereview.appspot.com/10919259

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

3 files changed, 78 insertions, 20 deletions

diff --git a/remoting/protocol/message_reader.cc b/remoting/protocol/message_reader.cc
index 309f87d..926b058 100644
--- a/remoting/protocol/message_reader.cc
+++ b/remoting/protocol/message_reader.cc
@@ -9,6 +9,7 @@
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/thread_task_runner_handle.h"
+#include "base/single_thread_task_runner.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/socket/socket.h"
@@ -88,23 +89,23 @@ void MessageReader::OnDataReceived(net::IOBuffer* data, int data_size) {
 
   // Get list of all new messages first, and then call the callback
   // for all of them.
-  std::vector<CompoundBuffer*> new_messages;
   while (true) {
     CompoundBuffer* buffer = message_decoder_.GetNextMessage();
     if (!buffer)
       break;
-    new_messages.push_back(buffer);
+    pending_messages_++;
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::Bind(&MessageReader::RunCallback,
+                   weak_factory_.GetWeakPtr(),
+                   base::Passed(scoped_ptr<CompoundBuffer>(buffer))));
   }
+}
 
-  pending_messages_ += new_messages.size();
-
-  for (std::vector<CompoundBuffer*>::iterator it = new_messages.begin();
-       it != new_messages.end(); ++it) {
-    message_received_callback_.Run(
-        scoped_ptr<CompoundBuffer>(*it),
-        base::Bind(&MessageReader::OnMessageDone,
-                   weak_factory_.GetWeakPtr()));
-  }
+void MessageReader::RunCallback(scoped_ptr<CompoundBuffer> message) {
+  message_received_callback_.Run(
+      message.Pass(), base::Bind(&MessageReader::OnMessageDone,
+                                 weak_factory_.GetWeakPtr()));
 }
 
 void MessageReader::OnMessageDone() {
@@ -112,8 +113,8 @@ void MessageReader::OnMessageDone() {
   pending_messages_--;
   DCHECK_GE(pending_messages_, 0);
 
-  if (!read_pending_)
-    DoRead();  // Start next read if necessary.
+  // Start next read if necessary.
+  DoRead();
 }
 
 }  // namespace protocol
diff --git a/remoting/protocol/message_reader.h b/remoting/protocol/message_reader.h
index efba5e1..fb6c92e 100644
--- a/remoting/protocol/message_reader.h
+++ b/remoting/protocol/message_reader.h
@@ -50,6 +50,7 @@ class MessageReader : public base::NonThreadSafe {
   void OnRead(int result);
   void HandleReadResult(int result);
   void OnDataReceived(net::IOBuffer* data, int data_size);
+  void RunCallback(scoped_ptr<CompoundBuffer> message);
   void OnMessageDone();
 
   net::Socket* socket_;
diff --git a/remoting/protocol/message_reader_unittest.cc b/remoting/protocol/message_reader_unittest.cc
index c789b96..b3e6289 100644
--- a/remoting/protocol/message_reader_unittest.cc
+++ b/remoting/protocol/message_reader_unittest.cc
@@ -43,7 +43,26 @@ class MockMessageReceivedCallback {
 class MessageReaderTest : public testing::Test {
  public:
   MessageReaderTest()
-      : run_task_finished_(false, false) {
+      : in_callback_(false) {
+  }
+
+  // Following two methods are used by the ReadFromCallback test.
+  void AddSecondMessage(const base::Closure& task) {
+    AddMessage(kTestMessage2);
+    in_callback_ = true;
+    task.Run();
+    in_callback_ = false;
+  }
+
+  void OnSecondMessage(const base::Closure& task) {
+    EXPECT_FALSE(in_callback_);
+    task.Run();
+  }
+
+  // Used by the DeleteFromCallback() test.
+  void DeleteReader(const base::Closure& task) {
+    reader_.reset();
+    task.Run();
   }
 
  protected:
@@ -73,11 +92,6 @@ class MessageReaderTest : public testing::Test {
     return result == expected;
   }
 
-  void RunClosure(const base::Closure& task) {
-    task.Run();
-    run_task_finished_.Signal();
-  }
-
   void OnMessage(scoped_ptr<CompoundBuffer> buffer,
                  const base::Closure& done_callback) {
     messages_.push_back(buffer.release());
@@ -85,11 +99,11 @@ class MessageReaderTest : public testing::Test {
   }
 
   MessageLoop message_loop_;
-  base::WaitableEvent run_task_finished_;
   scoped_ptr<MessageReader> reader_;
   FakeSocket socket_;
   MockMessageReceivedCallback callback_;
   std::vector<CompoundBuffer*> messages_;
+  bool in_callback_;
 };
 
 // Receive one message and process it with delay
@@ -103,6 +117,7 @@ TEST_F(MessageReaderTest, OneMessage_Delay) {
       .WillOnce(SaveArg<0>(&done_task));
 
   InitReader();
+  message_loop_.RunAllPending();
 
   Mock::VerifyAndClearExpectations(&callback_);
   Mock::VerifyAndClearExpectations(&socket_);
@@ -127,6 +142,7 @@ TEST_F(MessageReaderTest, OneMessage_Instant) {
       .WillOnce(CallDoneTask());
 
   InitReader();
+  message_loop_.RunAllPending();
 
   EXPECT_TRUE(socket_.read_pending());
   EXPECT_EQ(1U, messages_.size());
@@ -146,6 +162,7 @@ TEST_F(MessageReaderTest, TwoMessages_Together) {
       .WillOnce(SaveArg<0>(&done_task2));
 
   InitReader();
+  message_loop_.RunAllPending();
 
   Mock::VerifyAndClearExpectations(&callback_);
   Mock::VerifyAndClearExpectations(&socket_);
@@ -158,10 +175,12 @@ TEST_F(MessageReaderTest, TwoMessages_Together) {
   EXPECT_FALSE(socket_.read_pending());
 
   done_task1.Run();
+  message_loop_.RunAllPending();
 
   EXPECT_FALSE(socket_.read_pending());
 
   done_task2.Run();
+  message_loop_.RunAllPending();
 
   EXPECT_TRUE(socket_.read_pending());
 }
@@ -180,6 +199,7 @@ TEST_F(MessageReaderTest, TwoMessages_Instant) {
       .WillOnce(SaveArg<0>(&done_task2));
 
   InitReader();
+  message_loop_.RunAllPending();
 
   Mock::VerifyAndClearExpectations(&callback_);
   Mock::VerifyAndClearExpectations(&socket_);
@@ -207,6 +227,7 @@ TEST_F(MessageReaderTest, TwoMessages_Instant2) {
       .WillOnce(CallDoneTask());
 
   InitReader();
+  message_loop_.RunAllPending();
 
   EXPECT_TRUE(socket_.read_pending());
 }
@@ -222,6 +243,7 @@ TEST_F(MessageReaderTest, TwoMessages_Separately) {
       .WillOnce(SaveArg<0>(&done_task));
 
   InitReader();
+  message_loop_.RunAllPending();
 
   Mock::VerifyAndClearExpectations(&callback_);
   Mock::VerifyAndClearExpectations(&socket_);
@@ -233,6 +255,7 @@ TEST_F(MessageReaderTest, TwoMessages_Separately) {
   EXPECT_FALSE(socket_.read_pending());
 
   done_task.Run();
+  message_loop_.RunAllPending();
 
   EXPECT_TRUE(socket_.read_pending());
 
@@ -241,6 +264,7 @@ TEST_F(MessageReaderTest, TwoMessages_Separately) {
       .Times(1)
       .WillOnce(SaveArg<0>(&done_task));
   AddMessage(kTestMessage2);
+  message_loop_.RunAllPending();
 
   EXPECT_TRUE(CompareResult(messages_[1], kTestMessage2));
 
@@ -266,5 +290,37 @@ TEST_F(MessageReaderTest, ReadError) {
   InitReader();
 }
 
+// Verify that we the OnMessage callback is not reentered.
+TEST_F(MessageReaderTest, ReadFromCallback) {
+  AddMessage(kTestMessage1);
+
+  EXPECT_CALL(callback_, OnMessage(_))
+      .Times(2)
+      .WillOnce(Invoke(this, &MessageReaderTest::AddSecondMessage))
+      .WillOnce(Invoke(this, &MessageReaderTest::OnSecondMessage));
+
+  InitReader();
+  message_loop_.RunAllPending();
+
+  EXPECT_TRUE(socket_.read_pending());
+}
+
+// Verify that we stop getting callbacks after deleting MessageReader.
+TEST_F(MessageReaderTest, DeleteFromCallback) {
+  base::Closure done_task1;
+  base::Closure done_task2;
+
+  AddMessage(kTestMessage1);
+  AddMessage(kTestMessage2);
+
+  // OnMessage() should never be called for the second message.
+  EXPECT_CALL(callback_, OnMessage(_))
+      .Times(1)
+      .WillOnce(Invoke(this, &MessageReaderTest::DeleteReader));
+
+  InitReader();
+  message_loop_.RunAllPending();
+}
+
 }  // namespace protocol
 }  // namespace remoting