worker: implement SharedArrayBuffer/MessagePort transferring

node.gyp

@@ -224,6 +224,7 @@
         'src/node_i18n.cc',
         'src/pipe_wrap.cc',
         'src/process_wrap.cc',
+        'src/sharedarraybuffer-metadata.cc',
         'src/signal_wrap.cc',
         'src/spawn_sync.cc',
         'src/string_bytes.cc',
@@ -279,6 +280,7 @@
         'src/udp_wrap.h',
         'src/req-wrap.h',
         'src/req-wrap-inl.h',
+        'src/sharedarraybuffer-metadata.h',
         'src/string_bytes.h',
         'src/stream_base.h',
         'src/stream_base-inl.h',

src/env.h

@@ -92,6 +92,7 @@ class ModuleWrap;
   V(decorated_private_symbol, "node:decorated")                               \
   V(npn_buffer_private_symbol, "node:npnBuffer")                              \
   V(processed_private_symbol, "node:processed")                               \
+  V(sab_lifetimepartner_symbol, "node:sharedArrayBufferLifetimePartner")      \
   V(selected_npn_buffer_private_symbol, "node:selectedNpnBuffer")             \
   V(domain_private_symbol, "node:domain")                                     \
 
@@ -324,6 +325,7 @@ class ModuleWrap;
   V(promise_wrap_template, v8::ObjectTemplate)                                \
   V(push_values_to_array_function, v8::Function)                              \
   V(randombytes_constructor_template, v8::ObjectTemplate)                     \
+  V(sab_lifetimepartner_constructor_template, v8::FunctionTemplate)           \
   V(script_context_constructor_template, v8::FunctionTemplate)                \
   V(script_data_constructor_function, v8::Function)                           \
   V(secure_context_constructor_template, v8::FunctionTemplate)                \

src/node_messaging.cc

@@ -25,6 +25,7 @@ using v8::Maybe;
 using v8::MaybeLocal;
 using v8::Nothing;
 using v8::Object;
+using v8::SharedArrayBuffer;
 using v8::String;
 using v8::Value;
 using v8::ValueDeserializer;
@@ -55,6 +56,8 @@ Message& Message::operator=(Message&& other) {
   main_message_buf_ = other.main_message_buf_;
   other.main_message_buf_ = uv_buf_init(nullptr, 0);
   array_buffer_contents_ = std::move(other.array_buffer_contents_);
+  shared_array_buffers_ = std::move(other.shared_array_buffers_);
+  message_ports_ = std::move(other.message_ports_);
   return *this;
 }
 
@@ -65,13 +68,25 @@ namespace {
 class DeserializerDelegate : public ValueDeserializer::Delegate {
  public:
   DeserializerDelegate(Message* m,
-                       Environment* env)
-    : env_(env), msg_(m) {}
+                       Environment* env,
+                       const std::vector<MessagePort*>& message_ports)
+    : env_(env), msg_(m), message_ports_(message_ports) {}
+
+  MaybeLocal<Object> ReadHostObject(Isolate* isolate) override {
+    // Currently, only MessagePort hosts objects are supported, so identifying
+    // by the index in the message's MessagePort array is sufficient.
+    uint32_t id;
+    if (!deserializer->ReadUint32(&id))
+      return MaybeLocal<Object>();
+    CHECK_LE(id, message_ports_.size());
+    return message_ports_[id]->object();
+  };
 
   ValueDeserializer* deserializer = nullptr;
  private:
   Environment* env_;
   Message* msg_;
+  const std::vector<MessagePort*>& message_ports_;
 };
 
 }  // anonymous namespace
@@ -85,13 +100,27 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
     // This is for messages generated in C++ with the expectation that they
     // are handled in JS, e.g. serialized error messages from workers.
     CHECK(array_buffer_contents_.empty());
+    CHECK(shared_array_buffers_.empty());
+    CHECK(message_ports_.empty());
     char* buf = main_message_buf_.base;
     main_message_buf_.base = nullptr;
     return handle_scope.Escape(
         Buffer::New(env, buf, main_message_buf_.len).FromMaybe(Local<Value>()));
   }
 
-  DeserializerDelegate delegate(this, env);
+  // Create all necessary MessagePort handles.
+  std::vector<MessagePort*> ports(message_ports_.size());
+  for (uint32_t i = 0; i < message_ports_.size(); ++i) {
+    ports[i] = MessagePort::New(env,
+                                context,
+                                nullptr,
+                                std::move(message_ports_[i]));
+    if (ports[i] == nullptr)
+      return MaybeLocal<Value>();
+  }
+  message_ports_.clear();
+
+  DeserializerDelegate delegate(this, env, ports);
   ValueDeserializer deserializer(
       env->isolate(),
       reinterpret_cast<const uint8_t*>(main_message_buf_.base),
@@ -111,12 +140,30 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
   }
   array_buffer_contents_.clear();
 
+  for (uint32_t i = 0; i < shared_array_buffers_.size(); ++i) {
+    Local<SharedArrayBuffer> sab;
+    if (!shared_array_buffers_[i]->GetSharedArrayBuffer(env, context)
+            .ToLocal(&sab))
+      return MaybeLocal<Value>();
+    deserializer.TransferSharedArrayBuffer(i, sab);
+  }
+  shared_array_buffers_.clear();
+
   if (deserializer.ReadHeader(context).IsNothing())
     return MaybeLocal<Value>();
   return handle_scope.Escape(
       deserializer.ReadValue(context).FromMaybe(Local<Value>()));
 }
 
+void Message::AddSharedArrayBuffer(
+    SharedArrayBufferMetadataReference reference) {
+  shared_array_buffers_.push_back(reference);
+}
+
+void Message::AddMessagePort(std::unique_ptr<MessagePortData>&& data) {
+  message_ports_.emplace_back(std::move(data));
+}
+
 namespace {
 
 class SerializerDelegate : public ValueSerializer::Delegate {
@@ -128,12 +175,62 @@ class SerializerDelegate : public ValueSerializer::Delegate {
     env_->isolate()->ThrowException(Exception::Error(message));
   }
 
+  Maybe<bool> WriteHostObject(Isolate* isolate, Local<Object> object) override {
+    if (env_->message_port_constructor_template()->HasInstance(object)) {
+      return WriteMessagePort(Unwrap<MessagePort>(object));
+    }
+
+    env_->ThrowError("Cannot serialize unknown type of host object");
+    return Nothing<bool>();
+  }
+
+  Maybe<uint32_t> GetSharedArrayBufferId(
+      Isolate* isolate, Local<SharedArrayBuffer> shared_array_buffer) override {
+    uint32_t i;
+    for (i = 0; i < seen_shared_array_buffers_.size(); ++i) {
+      if (seen_shared_array_buffers_[i] == shared_array_buffer)
+        return Just(i);
+    }
+
+    SharedArrayBufferMetadataReference reference(
+        SharedArrayBufferMetadata::ForIncomingSharedArrayBuffer(env_,
+                                                          context_,
+                                                          shared_array_buffer));
+    if (!reference) {
+      return Nothing<uint32_t>();
+    }
+    seen_shared_array_buffers_.push_back(shared_array_buffer);
+    msg_->AddSharedArrayBuffer(reference);
+    return Just(i);
+  }
+
+  void Finish() {
+    for (MessagePort* port : ports_) {
+      port->Close();
+      msg_->AddMessagePort(port->Detach());
+    }
+  }
+
   ValueSerializer* serializer = nullptr;
 
  private:
+  Maybe<bool> WriteMessagePort(MessagePort* port) {
+    for (uint32_t i = 0; i < ports_.size(); i++) {
+      if (ports_[i] == port) {
+        serializer->WriteUint32(i);
+        return Just(true);
+      }
+    }
+
+    env_->ThrowError("MessagePort was not listed in transferList");
+    return Nothing<bool>();
+  }
+
   Environment* env_;
   Local<Context> context_;
   Message* msg_;
+  std::vector<Local<SharedArrayBuffer>> seen_shared_array_buffers_;
+  std::vector<MessagePort*> ports_;
 
   friend class worker::Message;
 };
@@ -159,7 +256,7 @@ Maybe<bool> Message::Serialize(Environment* env,
       Local<Value> entry;
       if (!transfer_list->Get(context, i).ToLocal(&entry))
         return Nothing<bool>();
-      // Currently, we support ArrayBuffers.
+      // Currently, we support ArrayBuffers and MessagePorts.
       if (entry->IsArrayBuffer()) {
         Local<ArrayBuffer> ab = entry.As<ArrayBuffer>();
         if (!ab->IsNeuterable() || ab->IsExternal())
@@ -168,6 +265,12 @@ Maybe<bool> Message::Serialize(Environment* env,
         array_buffers.push_back(ab);
         serializer.TransferArrayBuffer(id, ab);
         continue;
+      } else if (env->message_port_constructor_template()
+                    ->HasInstance(entry)) {
+        MessagePort* port = Unwrap<MessagePort>(entry.As<Object>());
+        CHECK_NE(port, nullptr);
+        delegate.ports_.push_back(port);
+        continue;
       }
 
       env->ThrowError("Found invalid object in transferList");
@@ -188,6 +291,7 @@ Maybe<bool> Message::Serialize(Environment* env,
                     contents.ByteLength()));
   }
 
+  delegate.Finish();
   std::pair<uint8_t*, size_t> data = serializer.Release();
   main_message_buf_.base = reinterpret_cast<char*>(data.first);
   main_message_buf_.len = data.second;

src/node_messaging.h

@@ -5,8 +5,8 @@
 
 #include "env.h"
 #include "node_mutex.h"
+#include "sharedarraybuffer-metadata.h"
 #include <list>
-#include <memory>
 
 namespace node {
 namespace worker {
@@ -27,7 +27,6 @@ class MessagePort;
 
 // Any further flagged message codes are defined by the modules that use them.
 
-
 // Represents a single communication message. The only non-standard extension
 // here is passing of a separate flag that the Workers implementation uses
 // for internal cross-thread information passing.
@@ -54,10 +53,19 @@ class Message {
                             v8::Local<v8::Value> input,
                             v8::Local<v8::Value> transfer_list);
 
+  // Internal method of Message that is called when a new SharedArrayBuffer
+  // object is encountered in the incoming value's structure.
+  void AddSharedArrayBuffer(SharedArrayBufferMetadataReference ref);
+  // Internal method of Message that is called once serialization finishes
+  // and that transfers ownership of `data` to this message.
+  void AddMessagePort(std::unique_ptr<MessagePortData>&& data);
+
  private:
   int32_t flag_ = MESSAGE_FLAG_NONE;
   uv_buf_t main_message_buf_;
   std::vector<uv_buf_t> array_buffer_contents_;
+  std::vector<SharedArrayBufferMetadataReference> shared_array_buffers_;
+  std::vector<std::unique_ptr<MessagePortData>> message_ports_;
 
   friend class MessagePort;
 };