Index: mojo/services/network/public/cpp/web_socket_write_queue.cc |
diff --git a/mojo/services/network/public/cpp/web_socket_write_queue.cc b/mojo/services/network/public/cpp/web_socket_write_queue.cc |
index 9f2d13237ef0da16533ba320cb87ea1db1dd0f67..fb55da45b300251c40e74bfdcc1d58031f78c215 100644 |
--- a/mojo/services/network/public/cpp/web_socket_write_queue.cc |
+++ b/mojo/services/network/public/cpp/web_socket_write_queue.cc |
@@ -5,6 +5,7 @@ |
#include "network/public/cpp/web_socket_write_queue.h" |
#include "base/bind.h" |
+#include "third_party/mojo/src/mojo/public/cpp/environment/logging.h" |
namespace mojo { |
@@ -19,10 +20,12 @@ struct WebSocketWriteQueue::Operation { |
}; |
WebSocketWriteQueue::WebSocketWriteQueue(DataPipeProducerHandle handle) |
- : handle_(handle), is_waiting_(false) { |
+ : handle_(handle), is_busy_(false), destructed_(nullptr) { |
} |
WebSocketWriteQueue::~WebSocketWriteQueue() { |
+ if (destructed_) |
+ *destructed_ = true; |
} |
void WebSocketWriteQueue::Write(const char* data, |
@@ -34,41 +37,61 @@ void WebSocketWriteQueue::Write(const char* data, |
op->data_ = data; |
queue_.push_back(op); |
- MojoResult result = MOJO_RESULT_SHOULD_WAIT; |
- if (!is_waiting_) |
- result = TryToWrite(); |
+ if (!is_busy_) { |
+ is_busy_ = true; |
+ // This call may reset |is_busy_| to false. |
+ TryToWrite(); |
+ } |
- // If we have to wait, make a local copy of the data so we know it will |
- // live until we need it. |
- if (result == MOJO_RESULT_SHOULD_WAIT) { |
+ if (is_busy_) { |
+ // If we have to wait, make a local copy of the data so we know it will |
+ // live until we need it. |
op->data_copy_.resize(num_bytes); |
memcpy(&op->data_copy_[0], data, num_bytes); |
op->data_ = &op->data_copy_[0]; |
} |
} |
-MojoResult WebSocketWriteQueue::TryToWrite() { |
- Operation* op = queue_[0]; |
- uint32_t bytes_written = op->num_bytes_; |
- MojoResult result = WriteDataRaw( |
- handle_, op->data_, &bytes_written, MOJO_WRITE_DATA_FLAG_ALL_OR_NONE); |
- if (result == MOJO_RESULT_SHOULD_WAIT) { |
- Wait(); |
- return result; |
- } |
- |
- // 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 result; |
- |
- op->callback_.Run(op->data_); // may delete |this| |
- return result; |
+void WebSocketWriteQueue::TryToWrite() { |
+ MOJO_DCHECK(is_busy_); |
+ MOJO_DCHECK(!queue_.empty()); |
+ do { |
+ Operation* op = queue_[0]; |
+ uint32_t bytes_written = op->num_bytes_; |
+ MojoResult result = WriteDataRaw( |
+ handle_, op->data_, &bytes_written, MOJO_WRITE_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()); |
+ |
+ // http://crbug.com/490193 This should run callback as well. May need to |
+ // change the callback signature. |
+ if (result != MOJO_RESULT_OK) |
+ return; |
+ |
+ bool destructed = false; |
+ destructed_ = &destructed; |
+ |
+ // May delete |this|. In that case, |destructed| will be set to true. |
+ // Because |is_busy_| is true during the whole process, even if Write() is |
+ // called by the callback, TryToWrite() won't be re-entered. So we don't |
+ // have to worry about |destructed_| being re-written by nested calls. |
+ op->callback_.Run(op->data_); |
+ |
+ destructed_ = nullptr; |
+ if (destructed) |
+ return; |
+ } while (!queue_.empty()); |
+ is_busy_ = false; |
} |
void WebSocketWriteQueue::Wait() { |
- is_waiting_ = true; |
+ MOJO_DCHECK(is_busy_); |
handle_watcher_.Start(handle_, |
MOJO_HANDLE_SIGNAL_WRITABLE, |
MOJO_DEADLINE_INDEFINITE, |
@@ -77,7 +100,7 @@ void WebSocketWriteQueue::Wait() { |
} |
void WebSocketWriteQueue::OnHandleReady(MojoResult result) { |
- is_waiting_ = false; |
+ MOJO_DCHECK(is_busy_); |
TryToWrite(); |
} |