Mojo JS bindings: draining a DataPipe

mojo/bindings/js/drain_data.cc

Index: mojo/bindings/js/drain_data.cc
diff --git a/mojo/bindings/js/drain_data.cc b/mojo/bindings/js/drain_data.cc
new file mode 100644
index 0000000000000000000000000000000000000000..1a207d62c200f9a538597c59c3efc797b8e56006
--- /dev/null
+++ b/mojo/bindings/js/drain_data.cc
@@ -0,0 +1,125 @@
+// 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 "mojo/bindings/js/drain_data.h"
+
+#include "gin/array_buffer.h"
+#include "gin/converter.h"
+#include "gin/dictionary.h"
+#include "gin/per_context_data.h"
+#include "gin/per_isolate_data.h"
+#include "mojo/public/cpp/environment/environment.h"
+#include "mojo/public/cpp/system/core.h"
+
+namespace mojo {
+namespace js {
+
+DrainData::DrainData(v8::Isolate* isolate, mojo::Handle handle)
+    : isolate_(isolate),
+      handle_(DataPipeConsumerHandle(handle.value())),
+      wait_id_(0) {
+
+  v8::Handle<v8::Context> context(isolate_->GetCurrentContext());
+  runner_ = gin::PerContextData::From(context)->runner()->GetWeakPtr();
+
+  WaitForData();
+}
+
+v8::Handle<v8::Value> DrainData::GetPromise() {
+  CHECK(resolver_.IsEmpty());
+  v8::Handle<v8::Promise::Resolver> resolver(
+      v8::Promise::Resolver::New(isolate_));
+  resolver_.Reset(isolate_, resolver);
+  return resolver->GetPromise();
+}
+
+DrainData::~DrainData() {
+  if (wait_id_)
+    Environment::GetDefaultAsyncWaiter()->CancelWait(wait_id_);
+  resolver_.Reset();
+}
+
+void DrainData::WaitForData() {
+  wait_id_ = Environment::GetDefaultAsyncWaiter()->AsyncWait(
+      handle_.get().value(),
+      MOJO_HANDLE_SIGNAL_READABLE,
+      MOJO_DEADLINE_INDEFINITE,
+      &DrainData::WaitCompleted,
+      this);
+}
+
+void DrainData::DataReady(MojoResult result) {
+  wait_id_ = 0;
+  if (result != MOJO_RESULT_OK) {
+    DeliverData(result);
+    return;
+  }
+  while(result == MOJO_RESULT_OK) {

[Matt Perry, 2014/09/24 00:39:42]

space before (

[hansmuller, 2014/09/24 17:47:53]

Done.

+    result = ReadData();
+    if (result == MOJO_RESULT_SHOULD_WAIT)
+      WaitForData();
+    else if (result != MOJO_RESULT_OK)
+      DeliverData(result);
+  }
+}
+
+MojoResult DrainData::ReadData() {
+  const void* buffer;
+  uint32_t num_bytes = 0;
+  MojoResult result = BeginReadDataRaw(
+      handle_.get(), &buffer, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+  if (result != MOJO_RESULT_OK)
+    return result;
+  const char* p = static_cast<const char*>(buffer);
+  DataBuffer data_buffer(p, p + num_bytes);
+  data_buffers_.push_back(data_buffer);

[Matt Perry, 2014/09/24 00:39:41]

This will make an extra copy of the data. Alternative:

data_buffers_.push_back(DataBuffer());
data_buffers_.back() = DataBuffer(p, p + num_bytes);

I'm also not sure if the DataBuffers will be copied around inside data_buffer_
when it is resized. You may want to make data_buffer_ a vector of DataBuffer
pointers.

[hansmuller, 2014/09/24 17:47:53]

Good point, I've done that.

+  return EndReadDataRaw(handle_.get(), num_bytes);
+}
+
+void DrainData::DeliverData(MojoResult result) {
+  if (!runner_) {
+    delete this;
+    return;
+  }
+
+  uint32_t total_bytes = 0;
+  for (unsigned i = 0; i < data_buffers_.size(); i++)
+    total_bytes += data_buffers_[i].size();
+
+  // Create a total_bytes length ArrayBuffer return value.
+  gin::Runner::Scope scope(runner_.get());
+  v8::Handle<v8::ArrayBuffer> array_buffer =
+    v8::ArrayBuffer::New(isolate_, total_bytes);

[Matt Perry, 2014/09/24 00:39:42]

indent +2

[hansmuller, 2014/09/24 17:47:53]

Done.

+  gin::ArrayBuffer buffer;
+  ConvertFromV8(isolate_, array_buffer, &buffer);
+  CHECK_EQ(total_bytes, buffer.num_bytes());
+
+  // Copy the data_buffers into the ArrayBuffer.
+  char* array_buffer_ptr = static_cast<char*>(buffer.bytes());
+  unsigned offset = 0;
+  for (unsigned i = 0; i < data_buffers_.size(); i++) {
+    uint32_t num_bytes = data_buffers_[i].size();
+    if (!num_bytes)

[Matt Perry, 2014/09/24 00:39:41]

compare to 0 rather than check for logical false

[hansmuller, 2014/09/24 17:47:53]

Done.

+      continue;
+    const char* data_buffer_ptr = &(data_buffers_[i][0]);
+    memcpy(array_buffer_ptr + offset, data_buffer_ptr, num_bytes);

[Matt Perry, 2014/09/24 00:39:41]

It's a shame we have to copy the data again here. Is there no way to allocate a
v8::ArrayBuffer initially and write directly to that in ReadData(), skipping the
data_buffers_ step?

[hansmuller, 2014/09/24 17:47:53]

We don't know how much data is coming, so there isn't a simple way avoid
copying. We might change gin to enable dynamically changing an ArrayBuffer's
size, so that incoming data could be appended to the ArrayBuffer directly.
However doing a realloc for each incoming buffer of data might not be a net win.
There are other possibilities however it might be good to wait for some real
performance/usage data to see how much of a performance bottleneck this is.

+    offset += num_bytes;
+  }
+
+  v8::Handle<v8::Promise::Resolver> resolver(
+      v8::Local<v8::Promise::Resolver>::New(isolate_, resolver_));
+  if (result == MOJO_RESULT_FAILED_PRECONDITION) {

[Matt Perry, 2014/09/24 00:39:41]

So the promise never resolves if there's an error? Or does not calling Resolve()
on a Resolver result in it resolving with an error?

[hansmuller, 2014/09/24 17:47:53]

Sorry, that's my mistake! There's should be an else clause that calls Reject().

+    gin::Dictionary dictionary = gin::Dictionary::CreateEmpty(isolate_);
+    dictionary.Set("result", result);
+    dictionary.Set("buffer", array_buffer);
+    resolver->Resolve(ConvertToV8(isolate_, dictionary));
+  }
+
+  delete this;
+}
+
+
+}  // namespace js
+}  // namespace mojo
+