Change data pipe wrappers used by SerialConnection to use message pipe.

device/serial/data_receiver.cc

 // Copyright 2014 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 "device/serial/data_receiver.h"
 
 #include <limits>
 
 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
-#include "device/serial/async_waiter.h"
 
 namespace device {
 
 // Represents a receive that is not yet fulfilled.
 class DataReceiver::PendingReceive {
  public:
   PendingReceive(DataReceiver* receiver,
                  const ReceiveDataCallback& callback,
                  const ReceiveErrorCallback& error_callback,
                  int32_t fatal_error_value);
 
   // Dispatches |data| to |receive_callback_|.
-  void DispatchData(const void* data, uint32_t num_bytes);
+  void DispatchData(DataReceiver::PendingData* data);
 
   // Reports |error| to |receive_error_callback_| if it is an appropriate time.
   // Returns whether it dispatched |error|.
-  bool DispatchError(DataReceiver::PendingError* error,
-                     uint32_t bytes_received);
+  bool DispatchError(DataReceiver::PendingError* error);
 
   // Reports |fatal_error_value_| to |receive_error_callback_|.
   void DispatchFatalError();
 
+  bool dispatched() { return buffer_in_use_; }

[raymes, 2014/10/17 01:55:42]

why not just call this function "buffer_in_use()"

[Sam McNally, 2014/10/20 05:12:58]

Done.

+
  private:
   class Buffer;
 
   // Invoked when the user is finished with the ReadOnlyBuffer provided to
   // |receive_callback_|.
   void Done(uint32_t num_bytes);
 
   // The DataReceiver that owns this.
   DataReceiver* receiver_;
 
@@ skipping 31 matching lines @@
   // The DataReceiver whose data pipe we are providing a view.
   scoped_refptr<DataReceiver> receiver_;
 
   // The PendingReceive to which this buffer has been created in response.
   PendingReceive* pending_receive_;
 
   const char* buffer_;
   uint32_t buffer_size_;
 };
 
+// A buffer of data received from the DataSource.
+struct DataReceiver::PendingData {
+  explicit PendingData(mojo::Array<uint8_t> data)
+      : data(data.Pass()), offset(0) {}
+
+  mojo::Array<uint8_t> data;
+
+  // The offset within |data| at which the next read should begin.
+  uint32_t offset;
+};
+
 // Represents an error received from the DataSource.
 struct DataReceiver::PendingError {

[raymes, 2014/10/17 01:55:42]

Could we combine PendingData and PendingError? It would simplify ordering logic
a bit. I think it seems natural have the error in the queue of received data.
Having a couple of unused fields seems ok. We could call it a DataFrame or
something? This seems like it would simplify DataReceiver::ReceiveInternal quite
a bit as well.

[Sam McNally, 2014/10/20 05:12:58]

Done.

-  PendingError(uint32_t offset, int32_t error)
-      : offset(offset), error(error), dispatched(false) {}
+  PendingError(int32_t error, size_t queue_position)
+      : queue_position(queue_position), error(error), dispatched(false) {}
 
-  // The location within the data stream where the error occurred.
-  const uint32_t offset;
+  // The number of pending data buffers to be dispatched before this error.
+  size_t queue_position;
 
   // The value of the error that occurred.
   const int32_t error;
 
   // Whether the error has been dispatched to the user.
   bool dispatched;
 };
 
 DataReceiver::DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
                            uint32_t buffer_size,
                            int32_t fatal_error_value)
     : source_(source.Pass()),
       fatal_error_value_(fatal_error_value),
-      bytes_received_(0),
       shut_down_(false),
       weak_factory_(this) {
-  MojoCreateDataPipeOptions options = {
-      sizeof(options), MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, 1, buffer_size,
-  };
-  mojo::ScopedDataPipeProducerHandle remote_handle;
-  MojoResult result = mojo::CreateDataPipe(&options, &remote_handle, &handle_);
-  DCHECK_EQ(MOJO_RESULT_OK, result);
-  source_->Init(remote_handle.Pass());
   source_.set_client(this);
+  source_.set_error_handler(this);
+  source_->Init(buffer_size);
 }
 
 bool DataReceiver::Receive(const ReceiveDataCallback& callback,
                            const ReceiveErrorCallback& error_callback) {
   DCHECK(!callback.is_null() && !error_callback.is_null());
   if (pending_receive_ || shut_down_)
     return false;
   // When the DataSource encounters an error, it pauses transmission. When the
   // user starts a new receive following notification of the error (via
   // |error_callback| of the previous Receive call) of the error we can tell the
   // DataSource to resume transmission of data.
   if (pending_error_ && pending_error_->dispatched) {
     source_->Resume();
     pending_error_.reset();
   }
 
   pending_receive_.reset(
       new PendingReceive(this, callback, error_callback, fatal_error_value_));
-  base::MessageLoop::current()->PostTask(
-      FROM_HERE,
-      base::Bind(&DataReceiver::ReceiveInternal, weak_factory_.GetWeakPtr()));
+  ReceiveInternal();
   return true;
 }
 
 DataReceiver::~DataReceiver() {
   ShutDown();
 }
 
-void DataReceiver::OnError(uint32_t offset, int32_t error) {
+void DataReceiver::OnError(int32_t error) {
   if (shut_down_)
     return;
 
   if (pending_error_) {
     // When OnError is called by the DataSource, transmission of data is
     // suspended. Thus we shouldn't receive another call to OnError until we
     // have fully dealt with the error and called Resume to resume transmission
     // (see Receive()). Under normal operation we should never get here, but if
     // we do (e.g. in the case of a hijacked service process) just shut down.
     ShutDown();
     return;
   }
-  pending_error_.reset(new PendingError(offset, error));
-  if (pending_receive_ &&
-      pending_receive_->DispatchError(pending_error_.get(), bytes_received_)) {
-    pending_receive_.reset();
-    waiter_.reset();
-  }
+  pending_error_.reset(new PendingError(error, pending_data_buffers_.size()));
+  if (pending_receive_)
+    ReceiveInternal();
+}
+
+void DataReceiver::OnData(mojo::Array<uint8_t> data) {
+  pending_data_buffers_.push(
+      linked_ptr<PendingData>(new PendingData(data.Pass())));
+  if (pending_receive_)
+    ReceiveInternal();
 }
 
 void DataReceiver::OnConnectionError() {
   ShutDown();
 }
 
 void DataReceiver::Done(uint32_t bytes_consumed) {
   if (shut_down_)
     return;
 
   DCHECK(pending_receive_);
-  MojoResult result = mojo::EndReadDataRaw(handle_.get(), bytes_consumed);
-  DCHECK_EQ(MOJO_RESULT_OK, result);
+  PendingData& pending_data = *pending_data_buffers_.front();
+  pending_data.offset += bytes_consumed;
+  if (pending_data.offset == pending_data.data.size()) {

[raymes, 2014/10/17 01:55:42]

Maybe we should at least DCHECK here that pending_data.offset <=
pending_data.data.size(). 

[Sam McNally, 2014/10/20 05:12:58]

Done.

+    if (pending_error_)
+      pending_error_->queue_position--;
+    pending_data_buffers_.pop();
+  }
   pending_receive_.reset();
-  bytes_received_ += bytes_consumed;
-}
-
-void DataReceiver::OnDoneWaiting(MojoResult result) {
-  DCHECK(pending_receive_ && !shut_down_ && waiter_);
-  waiter_.reset();
-  if (result != MOJO_RESULT_OK) {
-    ShutDown();
-    return;
-  }
-  ReceiveInternal();
 }
 
 void DataReceiver::ReceiveInternal() {
   if (shut_down_)
     return;
   DCHECK(pending_receive_);
-  if (pending_error_ &&
-      pending_receive_->DispatchError(pending_error_.get(), bytes_received_)) {
+  if (pending_receive_->dispatched())
+    return;
+
+  if (pending_error_ && pending_receive_->DispatchError(pending_error_.get())) {
     pending_receive_.reset();
-    waiter_.reset();
     return;
   }
-
-  const void* data;
-  uint32_t num_bytes = std::numeric_limits<uint32_t>::max();
-  MojoResult result = mojo::BeginReadDataRaw(
-      handle_.get(), &data, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
-  if (result == MOJO_RESULT_OK) {
-    if (!CheckErrorNotInReadRange(num_bytes)) {
-      ShutDown();
-      return;
-    }
-
-    pending_receive_->DispatchData(data, num_bytes);
-    return;
-  }
-  if (result == MOJO_RESULT_SHOULD_WAIT) {
-    waiter_.reset(new AsyncWaiter(
-        handle_.get(),
-        MOJO_HANDLE_SIGNAL_READABLE,
-        base::Bind(&DataReceiver::OnDoneWaiting, weak_factory_.GetWeakPtr())));
-    return;
-  }
-  ShutDown();
-}
-
-bool DataReceiver::CheckErrorNotInReadRange(uint32_t num_bytes) {
-  DCHECK(pending_receive_);
-  if (!pending_error_)
-    return true;
-
-  DCHECK_NE(bytes_received_, pending_error_->offset);
-  DCHECK_NE(num_bytes, 0u);
-  uint32_t potential_bytes_received = bytes_received_ + num_bytes;
-  // bytes_received_ can overflow so we must consider two cases:
-  //   1. Both |bytes_received_| and |pending_error_->offset| have overflowed an
-  //      equal number of times. In this case, |potential_bytes_received| must
-  //      be in the range (|bytes_received|, |pending_error_->offset|]. Below
-  //      this range can only occur if |bytes_received_| overflows before
-  //      |pending_error_->offset|. Above can only occur if |bytes_received_|
-  //      overtakes |pending_error_->offset|.
-  //   2. |pending_error_->offset| has overflowed once more than
-  //      |bytes_received_|. In this case, |potential_bytes_received| must not
-  //      be in the range (|pending_error_->offset|, |bytes_received_|].
-  if ((bytes_received_ < pending_error_->offset &&
-       (potential_bytes_received > pending_error_->offset ||
-        potential_bytes_received <= bytes_received_)) ||
-      (bytes_received_ > pending_error_->offset &&
-       potential_bytes_received > pending_error_->offset &&
-       potential_bytes_received <= bytes_received_)) {
-    return false;
-  }
-  return true;
+  if (!pending_data_buffers_.empty())
+    pending_receive_->DispatchData(pending_data_buffers_.front().get());
 }
 
 void DataReceiver::ShutDown() {
   shut_down_ = true;
   if (pending_receive_)
     pending_receive_->DispatchFatalError();
   pending_error_.reset();
-  waiter_.reset();
 }
 
 DataReceiver::PendingReceive::PendingReceive(
     DataReceiver* receiver,
     const ReceiveDataCallback& callback,
     const ReceiveErrorCallback& error_callback,
     int32_t fatal_error_value)
     : receiver_(receiver),
       receive_callback_(callback),
       receive_error_callback_(error_callback),
       fatal_error_value_(fatal_error_value),
       buffer_in_use_(false) {
 }
 
-void DataReceiver::PendingReceive::DispatchData(const void* data,
-                                                uint32_t num_bytes) {
+void DataReceiver::PendingReceive::DispatchData(
+    DataReceiver::PendingData* data) {
   DCHECK(!buffer_in_use_);
   buffer_in_use_ = true;
-  receive_callback_.Run(scoped_ptr<ReadOnlyBuffer>(
-      new Buffer(receiver_, this, static_cast<const char*>(data), num_bytes)));
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          receive_callback_,
+          base::Passed(scoped_ptr<ReadOnlyBuffer>(new Buffer(
+              receiver_,
+              this,
+              reinterpret_cast<char*>(&data->data[0]) + data->offset,
+              static_cast<uint32_t>(data->data.size() - data->offset))))));
 }
 
-bool DataReceiver::PendingReceive::DispatchError(PendingError* error,
-                                                 uint32_t bytes_received) {
+bool DataReceiver::PendingReceive::DispatchError(PendingError* error) {
   DCHECK(!error->dispatched);
-  if (buffer_in_use_ || bytes_received != error->offset)
+  if (buffer_in_use_ || error->queue_position != 0)
     return false;
 
   error->dispatched = true;
-  receive_error_callback_.Run(error->error);
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE, base::Bind(receive_error_callback_, error->error));
   return true;
 }
 
 void DataReceiver::PendingReceive::DispatchFatalError() {
   receive_error_callback_.Run(fatal_error_value_);
 }
 
 void DataReceiver::PendingReceive::Done(uint32_t bytes_consumed) {
   DCHECK(buffer_in_use_);
   buffer_in_use_ = false;
@@ skipping 32 matching lines @@
   buffer_size_ = 0;
 }
 
 void DataReceiver::PendingReceive::Buffer::DoneWithError(
     uint32_t bytes_consumed,
     int32_t error) {
   Done(bytes_consumed);
 }
 
 }  // namespace device