Fix WebSocket{Read,Write}Queue.

mojo/services/network/public/cpp/web_socket_read_queue.cc

Index: mojo/services/network/public/cpp/web_socket_read_queue.cc
diff --git a/mojo/services/network/public/cpp/web_socket_read_queue.cc b/mojo/services/network/public/cpp/web_socket_read_queue.cc
index 1bcf503fce17f476b0cab2ca2d3f935f6d85dc3e..e2c80c7f2bbf9dc111e1e031448c0381717962ad 100644
--- a/mojo/services/network/public/cpp/web_socket_read_queue.cc
+++ b/mojo/services/network/public/cpp/web_socket_read_queue.cc
@@ -5,6 +5,7 @@
 #include "network/public/cpp/web_socket_read_queue.h"
 
 #include "base/bind.h"
+#include "third_party/mojo/src/mojo/public/cpp/environment/logging.h"
 
 namespace mojo {
 
@@ -14,10 +15,12 @@ struct WebSocketReadQueue::Operation {
 };
 
 WebSocketReadQueue::WebSocketReadQueue(DataPipeConsumerHandle handle)
-    : handle_(handle), is_waiting_(false) {
+    : handle_(handle), is_busy_(false), destructed_(nullptr) {
 }
 
 WebSocketReadQueue::~WebSocketReadQueue() {
+  if (destructed_)
+    *destructed_ = true;
 }
 
 void WebSocketReadQueue::Read(uint32_t num_bytes,
@@ -27,34 +30,61 @@ void WebSocketReadQueue::Read(uint32_t num_bytes,
   op->callback_ = callback;
   queue_.push_back(op);
 
-  if (!is_waiting_)
-    TryToRead();
+  if (is_busy_)
+    return;
+
+  is_busy_ = true;
+  TryToRead();
 }
 
 void WebSocketReadQueue::TryToRead() {
-  Operation* op = queue_[0];
-  const void* buffer = NULL;
-  uint32_t bytes_read = op->num_bytes_;
-  MojoResult result = BeginReadDataRaw(
-      handle_, &buffer, &bytes_read, MOJO_READ_DATA_FLAG_ALL_OR_NONE);
-  if (result == MOJO_RESULT_SHOULD_WAIT) {
-    EndReadDataRaw(handle_, bytes_read);
-    Wait();
-    return;
-  }
+  MOJO_DCHECK(is_busy_);
+  MOJO_DCHECK(!queue_.empty());
+  do {
+    Operation* op = queue_[0];
+    const void* buffer = NULL;
+    uint32_t bytes_read = op->num_bytes_;
+    MojoResult result = BeginReadDataRaw(
+        handle_, &buffer, &bytes_read, MOJO_READ_DATA_FLAG_ALL_OR_NONE);
+    if (result == MOJO_RESULT_SHOULD_WAIT) {
+      Wait();
+      return;
+    }
 
-  // Ensure |op| is deleted, whether or not |this| goes away.
-  scoped_ptr<Operation> op_deleter(op);
-  queue_.weak_erase(queue_.begin());
-  if (result != MOJO_RESULT_OK)
-    return;
-  DataPipeConsumerHandle handle = handle_;
-  op->callback_.Run(static_cast<const char*>(buffer));  // may delete |this|
-  EndReadDataRaw(handle, bytes_read);
+    // Ensure |op| is deleted, whether or not |this| goes away.
+    scoped_ptr<Operation> op_deleter(op);
+    queue_.weak_erase(queue_.begin());
+
+    // http://crbug.com/490193 This should run callback as well. May need to
+    // change the callback signature.
+    if (result != MOJO_RESULT_OK)
+      return;
+
+    uint32_t num_bytes = op_deleter->num_bytes_;
+    MOJO_DCHECK(num_bytes <= bytes_read);
+    DataPipeConsumerHandle handle = handle_;
+
+    bool destructed = false;
+    destructed_ = &destructed;

[jam, 2015/05/20 18:23:13]

checkout this pattern:
https://code.google.com/p/chromium/codesearch#chromium/src/components/html_vi...

I had used booleans like you did in the past, but the weak factory approach is a
bit cleaner.

[yzshen1, 2015/05/20 19:23:32]

I agree that the pattern is cleaner.

One issue is that this folder is supposed to be consumed by WebSocket Mojo API
clients which may not depend on the chromium repo (base/ in particular). There
were discussions before about getting rid of existing base/ dependencies from
WebSocket{Read,Write}Queue.

There is presubmit check preventing me from adding new base/ includes.

If you think it is okay, I would like to leave it as it is. What do you think?
Thanks!

+
+    // May delete |this|. In that case, |destructed| will be set to true.
+    // Because |is_busy_| is true during the whole process, even if Read() is
+    // called by the callback, TryToRead() won't be re-entered. So we don't have
+    // to worry about |destructed_| being re-written by nested calls.
+    op->callback_.Run(static_cast<const char*>(buffer));
+
+    destructed_ = nullptr;
+
+    EndReadDataRaw(handle, num_bytes);
+
+    if (destructed)
+      return;
+  } while (!queue_.empty());
+  is_busy_ = false;
 }
 
 void WebSocketReadQueue::Wait() {
-  is_waiting_ = true;
+  MOJO_DCHECK(is_busy_);
   handle_watcher_.Start(
       handle_,
       MOJO_HANDLE_SIGNAL_READABLE,
@@ -63,7 +93,7 @@ void WebSocketReadQueue::Wait() {
 }
 
 void WebSocketReadQueue::OnHandleReady(MojoResult result) {
-  is_waiting_ = false;
+  MOJO_DCHECK(is_busy_);
   TryToRead();
 }