Changed MessageReader so that it doesn't read from the socket if there are

remoting/protocol/message_reader.cc

 // Copyright (c) 2010 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 "remoting/protocol/message_reader.h"
 
 #include "base/message_loop.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/socket/socket.h"
 #include "remoting/base/compound_buffer.h"
 #include "remoting/proto/internal.pb.h"
 
 namespace remoting {
 namespace protocol {
 
 static const int kReadBufferSize = 4096;
 
+class MessageReader::DoneTaskHandler
+    : public base::RefCountedThreadSafe<MessageReader::DoneTaskHandler> {
+ public:
+  DoneTaskHandler(MessageReader* message_reader)
+      : message_loop_(MessageLoop::current()),
+        message_reader_(message_reader) {
+  }
+
+  ~DoneTaskHandler() { }
+
+  void ReleaseReader() {
+    DCHECK_EQ(message_loop_, MessageLoop::current());
+    AutoLock auto_lock(lock_);
+    message_reader_ = NULL;
+  }
+
+  void OnDone(CompoundBuffer* buffer) {

[awong, 2011/01/20 20:06:38]

2 thoughts:
  1) I wonder if we shouldn't delete the buffer back on the jingle_thread. 
Feels slightly cleaner, and we can collapse OnDone with ProcessDoneEvent.

  2)  Can't we replace this class with a WeakPtr & funciton?

Example:

  static void MessageDone(MessageLoop* run_on_me, WeakPtr<MessageReader>
message_reader, CompoundBuffer *b) {
   if (MessageLoop::current() != run_on_me) {
       // post back to self..
   }

   delete b;
   if (message_reader) {
     message_reader->OnMessageDone();
   }
}

?

[Sergey Ulanov, 2011/01/20 21:55:57]

It would be great if we could use WeakPtr here, but weak pointers are not thread
safe.

On 2011/01/20 20:06:38, awong wrote:
message_reader may be destroyed on another thread between the line above and the
line below.

[awong, 2011/01/20 22:08:02]

Good point.  Though I think if you do the post back to ensure you don't
reference message_reader_ except from the thread which controls the lifetime of
the underlying object anyways, you can use the weak_ptr here.

Then you could remove both the lock, and the ReleaseReader, no?
I thought message_reader was created/destroyed on one thread.  The run_on_me
repost should enforce serialization no?

[Sergey Ulanov, 2011/01/21 03:45:05]

Destructor may be called on a different thread, and my understanding is that
weak_ptr doesn't support that (correct me if I'm wrong here).
We don't have any guarantees about the thread on which the object is destroyed.
It is destroyed by scoped_ptr when ConnectionToHost/ConnectionToClient is
destroyed (which usually happens on a thread other than the network), and I
don't think that enforcing a specific thread for destructor is the right
solution.

[awong, 2011/01/21 19:19:33]

Something about this still does not feel right.  In particular, I agree with
alpha that calling OnMessageDone() under a lock is feels error prone.

So the race is that the Connection object can be deleted while a read is
outstanding?  Why can that happen?  Shouldn't we be guarnateeing that all
objects have stopped scheduling tasks before we start the teardown?

+    delete buffer;
+    ProcessDoneEvent();
+  }
+
+ private:
+  void ProcessDoneEvent() {
+    if (MessageLoop::current() != message_loop_) {
+      message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
+          this, &DoneTaskHandler::ProcessDoneEvent));
+      return;
+    }
+
+    AutoLock auto_lock(lock_);
+    if (message_reader_) {
+      message_reader_->OnMessageDone();
+    }
+  }
+
+  // Network message loop on which this object is created and which is
+  // used to call MessageReader::OnMessageDone().
+  MessageLoop* message_loop_;
+
+  // MessageReader that owns this object.
+  MessageReader* message_reader_;
+
+  // Must be locked whenever |message_reader_| is accessed
+  // (ReleaseReader() may be called on any thread).
+  Lock lock_;
+};
+
 MessageReader::MessageReader()
     : socket_(NULL),
+      read_pending_(false),
+      pending_messages_(0),
       closed_(false),
       ALLOW_THIS_IN_INITIALIZER_LIST(
           read_callback_(this, &MessageReader::OnRead)) {
 }
 
 MessageReader::~MessageReader() {
+  done_task_handler_->ReleaseReader();
 }
 
 void MessageReader::Init(net::Socket* socket,
                          MessageReceivedCallback* callback) {
   message_received_callback_.reset(callback);
   DCHECK(socket);
   socket_ = socket;
+  done_task_handler_ = new DoneTaskHandler(this);
   DoRead();
 }
 
 void MessageReader::DoRead() {
-  while (!closed_) {
+  // Don't try to read again if there is another read pending or we
+  // have messages that we haven't finished processing yet.
+  while (!closed_ && !read_pending_ && pending_messages_ == 0) {

[Alpha Left Google, 2011/01/20 20:54:52]

The flag of |read_pending| is confusing to me. According to the code we never
issue two DoRead() at the same time. Also we should never read when there's
outstanding DoneTaskHandler running, which means when pending_messages > 0,
read_pending is never true. What I'm saying is that putting !read_pending_ in
the condition is hiding the problem (which i believe the current code has, see
my comments below..). Either |read_pending_| should be removed or used as a
DCHECK to make sure it's exclusive with pending_messages >0.

[Sergey Ulanov, 2011/01/20 21:55:57]

It is here mainly for the second iteration, after HandleReadResults() sets this
flag. I've added DCHECK.

     read_buffer_ = new net::IOBuffer(kReadBufferSize);
     int result = socket_->Read(
         read_buffer_, kReadBufferSize, &read_callback_);
     HandleReadResult(result);
-    if (result < 0)
-      break;
   }
 }
 
 void MessageReader::OnRead(int result) {
+  DCHECK(read_pending_);
+  read_pending_ = false;
+
   if (!closed_) {
     HandleReadResult(result);
     DoRead();
   }
 }
 
 void MessageReader::HandleReadResult(int result) {
   if (result > 0) {
     OnDataReceived(read_buffer_, result);
   } else {
     if (result == net::ERR_CONNECTION_CLOSED) {
       closed_ = true;
-    } else if (result != net::ERR_IO_PENDING) {
+    } else if (result == net::ERR_IO_PENDING) {
+      read_pending_ = true;
+    } else {
       LOG(ERROR) << "Read() returned error " << result;
     }
   }
 }
 
 void MessageReader::OnDataReceived(net::IOBuffer* data, int data_size) {
   message_decoder_.AddData(data, data_size);
 
   while (true) {
-    CompoundBuffer buffer;
-    if (!message_decoder_.GetNextMessage(&buffer))
+    CompoundBuffer* buffer = message_decoder_.GetNextMessage();
+    if (!buffer)
       break;
 
-    message_received_callback_->Run(&buffer);
+    pending_messages_++;

[Alpha Left Google, 2011/01/20 20:54:52]

I think there's a bug here. Incrementing here will not do what you intended to.
If the following Run() calls trigger OnMessageDone() then pending_messages will
immediately become 0. Either you always post a task in ProcessDoneEvent() or
change this to become:

vector<pair<CompoundBuffer*, Task*> > invocations;

while (true) {
  CompoundBuffer* buffer = message_decoder_.GetNextMessage();
  if (!buffer) break;
}

pending_messages_ += invocations.size();

while (invocations...) {
  message_received_callback_->Run(....);
}

[Sergey Ulanov, 2011/01/20 21:55:57]

Good catch! Will also add unittest for this case.

On 2011/01/20 20:54:52, Alpha wrote:

+    message_received_callback_->Run(buffer, NewRunnableMethod(
+        done_task_handler_.get(), &DoneTaskHandler::OnDone, buffer));
   }
 }
 
+void MessageReader::OnMessageDone() {
+  pending_messages_--;
+  DCHECK_GE(pending_messages_, 0);
+
+  DoRead(); // Start next read if neccessary.
+}
+
 }  // namespace protocol
 }  // namespace remoting