[NaCl SDK] nacl_io: Add flow control the JavaScript pipes.

native_client_sdk/src/libraries/nacl_io/devfs/jspipe_event_emitter.cc

+// Copyright 2013 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 "nacl_io/devfs/jspipe_event_emitter.h"
+
+#define __STDC_FORMAT_MACROS
+#include <errno.h>
+#include <inttypes.h>
+#include <string.h>
+
+#include <algorithm>
+
+#define TRACE(format, ...) \
+    LOG_TRACE("jspipe[%s]: " format, name_.c_str(), ##__VA_ARGS__)
+#define ERROR(format, ...) \
+    LOG_ERROR("jspipe[%s]: " format, name_.c_str(), ##__VA_ARGS__)
+
+#include "nacl_io/log.h"
+#include "nacl_io/pepper_interface.h"
+
+namespace {
+const size_t kMaxPostMessageSize = 64*1024;
+}
+
+namespace nacl_io {
+
+JSPipeEventEmitter::JSPipeEventEmitter(PepperInterface* ppapi, size_t size)
+  : input_fifo_(size),
+    post_message_buffer_size_(size),
+    bytes_sent_(0),
+    bytes_acked_(0),
+    bytes_read_(0),
+    ppapi_(ppapi) {
+  if (ppapi) {
+    messaging_iface_ = ppapi->GetMessagingInterface();

[binji, 2014/05/01 20:22:31]

initialize to NULL if ppapi is NULL

[Sam Clegg, 2014/05/01 22:16:55]

Done, and added a test.

+    var_iface_ = ppapi->GetVarInterface();
+    array_iface_ = ppapi->GetVarArrayInterface();
+    buffer_iface_ = ppapi->GetVarArrayBufferInterface();
+    dict_iface_ = ppapi->GetVarDictionaryInterface();
+  }
+  UpdateStatus_Locked();
+}
+
+void JSPipeEventEmitter::UpdateStatus_Locked() {
+  uint32_t status = 0;
+  if (!input_fifo_.IsEmpty())
+    status |= POLLIN;
+
+  if (GetOSpace() > 0)
+    status |= POLLOUT;
+
+  ClearEvents_Locked(~status);
+  RaiseEvents_Locked(status);
+}
+
+size_t JSPipeEventEmitter::Read_Locked(char* data, size_t len) {
+  size_t out_len = input_fifo_.Read(data, len);
+  if (out_len > 0) {
+    bytes_read_ += out_len;
+    SendAckMessage(bytes_read_);

[binji, 2014/05/01 20:22:31]

Why ack total count instead of the amount read?

[binji, 2014/05/01 20:22:31]

check error

[Sam Clegg, 2014/05/01 22:16:55]

The ack messages always contain the running total.  Makes debugging much easier.

[Sam Clegg, 2014/05/01 22:16:55]

Done.

+  }
+
+  UpdateStatus_Locked();
+  return out_len;
+}
+
+Error JSPipeEventEmitter::SendWriteMessage(const void* buf, size_t count) {
+  TRACE("SendWriteMessage [%zd] total=%" PRIi64, count, bytes_sent_);
+  if (!var_iface_ || !buffer_iface_)
+    return EIO;
+
+  // Copy payload data in a new ArrayBuffer
+  PP_Var buffer = buffer_iface_->Create(count);
+  memcpy(buffer_iface_->Map(buffer), buf, count);
+  buffer_iface_->Unmap(buffer);
+
+  SendMessageToJS("write", buffer);

[binji, 2014/05/01 20:22:31]

check error

[Sam Clegg, 2014/05/01 22:16:55]

Done.

+
+  var_iface_->Release(buffer);
+  return 0;
+}
+
+Error JSPipeEventEmitter::SetName(const char* name) {
+  // name can only be set once
+  if (!name_.empty())
+    return EIO;
+
+  // new name must not be empty
+  if (!name || strlen(name) == 0)
+    return EIO;
+
+  TRACE("set name: %s", name);
+  name_ = name;
+  return 0;
+}
+
+Error JSPipeEventEmitter::SendMessageToJS(const char* message_type,
+                                          PP_Var payload) {
+  if (!messaging_iface_ || !var_iface_ || !dict_iface_ || !array_iface_)
+    return EIO;
+
+  // Construct two element array containing message type and the payload
+  PP_Var array = array_iface_->Create();
+  PP_Var message_type_var = var_iface_->VarFromUtf8(message_type,

[binji, 2014/05/01 20:22:31]

are you expecting arbitrary message types? If not, maybe just an int/enum. write
= 0, ack = 1, etc.

Even if you don't go that route, you can cache "write" and "ack" vars.

+                                                    strlen(message_type));
+  array_iface_->Set(array, 0, message_type_var);

[binji, 2014/05/01 20:22:31]

check errors on these pepper functions

+  array_iface_->Set(array, 1, payload);
+  var_iface_->Release(message_type_var);
+
+  // Create dict object with pipe name as key and array as value.
+  PP_Var dict = dict_iface_->Create();
+  PP_Var pipe_name = var_iface_->VarFromUtf8(name_.c_str(), name_.size());

[binji, 2014/05/01 20:22:31]

cache this in SetName because it never changes?

+  dict_iface_->Set(dict, pipe_name, array);
+  var_iface_->Release(pipe_name);
+  var_iface_->Release(array);
+
+  // Send the dict via PostMessage
+  messaging_iface_->PostMessage(ppapi_->GetInstance(), dict);
+
+  // Release the dict
+  var_iface_->Release(dict);
+  return 0;
+}
+
+Error JSPipeEventEmitter::SendAckMessage(int64_t byte_count) {
+  TRACE("SendAckMessage %"PRIi64, byte_count);
+  PP_Var payload;
+  payload.type = PP_VARTYPE_INT32;
+  payload.value.as_int = (int32_t)byte_count;
+
+  return SendMessageToJS("ack", payload);
+}
+
+size_t JSPipeEventEmitter::HandleJSWrite(const char* data, size_t len) {
+  size_t out_len = input_fifo_.Write(data, len);
+  UpdateStatus_Locked();
+  return out_len;
+}
+
+void JSPipeEventEmitter::HandleJSAck(int64_t byte_count) {
+  if (byte_count > bytes_sent_ || byte_count < 0) {
+    ERROR("HandleAck unexpected byte count: %"PRIi64, byte_count);
+    return;
+  }
+
+  bytes_acked_ = byte_count;
+  TRACE("HandleAck: %" PRIi64 "/%" PRIi64, bytes_acked_, bytes_sent_);
+  UpdateStatus_Locked();
+}
+
+size_t JSPipeEventEmitter::Write_Locked(const char* data, size_t len) {
+  if (GetOSpace() == 0)
+    return -1;

[binji, 2014/05/01 20:22:31]

should this return ssize_t then?

[Sam Clegg, 2014/05/01 22:16:55]

Done.

+
+  if (len > GetOSpace())
+    len = GetOSpace();
+
+  // Limit the size of the data we send with PostMessage to kMaxPostMessageSize
+  if (len > kMaxPostMessageSize)
+    len = kMaxPostMessageSize;
+
+  SendWriteMessage(data, len);

[binji, 2014/05/01 20:22:31]

check error

[Sam Clegg, 2014/05/01 22:16:55]

Done.

+  bytes_sent_ += len;
+
+  UpdateStatus_Locked();
+  return len;
+}
+
+}  // namespace nacl_io