feat: support lazy initializing ports

Signed-off-by: Daeyeon Jeong <[email protected]>

Author: daeyeon

deps/message-port/CMakeLists.txt

@@ -9,6 +9,3 @@ add_library(nd-message-port async-uv.cc message-port.cc debug-mem-trace.cc)
 target_link_libraries(nd-message-port PUBLIC libuv::libuv nd-utils)
 target_include_directories(nd-message-port PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
 target_compile_definitions(nd-message-port PUBLIC MSG_PORTS_DEFS)
-if(BUILD_SHARED_LIBS)
-  target_compile_options(nd-message-port PRIVATE -fvisibility=hidden)
-endif()

deps/message-port/VERSION

@@ -45,22 +45,22 @@ AsyncUV::~AsyncUV() {
   });
 }
 
-void AsyncUV::Send(uv_loop_t* loop, Task task) {
+bool AsyncUV::Send(uv_loop_t* loop, Task task) {
   if (loop == nullptr) {
-    EnqueueTask(task);
-    return;
+    return false;
   }
-  (new AsyncUV(loop, task))->Send();
+  return (new AsyncUV(loop, task))->Send();
 }
 
-void AsyncUV::EnqueueTask(Task task) {
+size_t AsyncUV::EnqueueTask(Task task) {
   std::lock_guard<std::mutex> lock(queue_mutex_);
   queue_.push(task);
+  return queue_.size();
 }
 
 bool AsyncUV::DrainPendingTasks(uv_loop_t* loop) {
   std::lock_guard<std::mutex> lock(queue_mutex_);
-  TRACE(MSGPORT, "drain pending queue", queue_.size());
+  TRACE(MSGPORT, "drain pending tasks", queue_.size());
 
   if (loop == nullptr) {
     return false;
@@ -73,6 +73,20 @@ bool AsyncUV::DrainPendingTasks(uv_loop_t* loop) {
   return true;
 }
 
+void AsyncUV::DeletePendingTasks() {
+  std::lock_guard<std::mutex> lock(queue_mutex_);
+  TRACE(MSGPORT, "delete pending tasks", queue_.size());
+  if (!queue_.empty()) {
+    std::queue<Task> empty;
+    std::swap(queue_, empty);
+  }
+}
+
+bool AsyncUV::IsPendingTasksEmpty() {
+  std::lock_guard<std::mutex> lock(queue_mutex_);
+  return queue_.empty();
+}
+
 void AsyncUV::Init(uv_loop_t* loop, Task task) {
   task_ = task;
 
@@ -88,9 +102,10 @@ void AsyncUV::Init(uv_loop_t* loop, Task task) {
   TRACE_ADD(ASYNC_UV, uv_h_);
 }
 
-void AsyncUV::Send() {
+bool AsyncUV::Send() {
   if (!uv_h_) {
-    return;
+    return false;
   }
   uv_async_send(uv_h_);
+  return true;
 }

deps/message-port/async-uv.cc

@@ -33,15 +33,17 @@ class EXPORT_API AsyncUV {
 
   // If loop is nullptr, the task is enqueued to the pending queue.
   // When a valid loop is provided later, all pending tasks will be executed.
-  static void Send(uv_loop_t* loop, Task task);
+  static bool Send(uv_loop_t* loop, Task task);
   static bool DrainPendingTasks(uv_loop_t* loop);
-  static void EnqueueTask(Task task);
+  static size_t EnqueueTask(Task task);
+  static void DeletePendingTasks();
+  static bool IsPendingTasksEmpty();
 
   AsyncUV(uv_loop_t* loop = nullptr, Task task = nullptr);
   ~AsyncUV();
 
   void Init(uv_loop_t* loop, Task task);
-  void Send();
+  bool Send();
 
  private:
   uv_async_t* uv_h_;

deps/message-port/async-uv.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <future>
+#include <memory>
 
 #ifndef EXPORT_API
 #define EXPORT_API __attribute__((visibility("default")))
@@ -36,6 +37,8 @@ struct EXPORT_API Channel {
   static Channel New(std::shared_future<uv_loop_t*> loop,
                      const char* origin = nullptr);
   static void DrainPendingMessages(uv_loop_t* loop);
+  // This deletes messages if any exists in the pending queue.
+  static void DeletePendingMessages();
 
   void Reset();
 };

deps/message-port/channel.h

@@ -102,7 +102,8 @@ Port::Result Port::PostMessage(std::shared_ptr<MessageEvent> event) {
   }
 
   // Check if the sink has a receiver.
-  if (sink->internal_->callback == nullptr) {
+  if (internal_->loop != nullptr && sink->internal_->callback == nullptr) {
+    // TODO: Enqueue events if the number of the events in queues is acceptable.
     TRACE(MSGPORT, "sink has no callback.");
     return Result::NoOnMessage;
   }
@@ -134,21 +135,24 @@ Port::Result Port::PostMessage(std::shared_ptr<MessageEvent> event) {
     }
   }
 
-  // Sends a task immediately if the loop handle is valid. If the loop handle is
-  // nullptr, the task is enqueued to the pending queue and will be executed
-  // later when a valid loop handle is provided.
-  AsyncUV::Send(internal_->loop,
-                [event = event, sink_weak = internal_->sink](uv_async_t*) {
-                  // event: shared_ptr, sink_weak: weak_ptr
-                  auto sink = sink_weak.lock();
-                  if (sink && sink->internal_->callback) {
-                    // Since sink is locked, event->target() is always valid
-                    // inside the callback.
-                    sink->internal_->callback(event.get());
-                  } else {
-                    TRACE(MSGPORT, "sink port released, or no callback");
-                  }
-                });
+  auto task = [event = event, sink_weak = internal_->sink](uv_async_t*) {
+    // event: shared_ptr, sink_weak: weak_ptr
+    auto sink = sink_weak.lock();
+    if (sink && sink->internal_->callback) {
+      // Since sink is locked, event->target() is always valid
+      // inside the callback.
+      sink->internal_->callback(event.get());
+    } else {
+      TRACE(MSGPORT, "sink port released, or no callback");
+    }
+  };
+
+  if (internal_->loop == nullptr) {
+    AsyncUV::EnqueueTask(std::move(task));
+    return Result::MessageEventQueued;
+  }
+
+  AsyncUV::Send(internal_->loop, std::move(task));
   return Result::NoError;
 }
 
@@ -191,6 +195,10 @@ void Channel::DrainPendingMessages(uv_loop_t* loop) {
   AsyncUV::DrainPendingTasks(loop);
 }
 
+void Channel::DeletePendingMessages() {
+  AsyncUV::DeletePendingTasks();
+}
+
 void Channel::Reset() {
   port1.reset();
   port2.reset();

deps/message-port/message-port.cc

@@ -52,6 +52,7 @@ class EXPORT_API Port {
   using OnMessageCallback = std::function<void(const MessageEvent*)>;
   enum Result {
     NoError = 0,
+    MessageEventQueued,
     NoSink,
     NoOnMessage,
     InvalidMessageEvent,

deps/message-port/message-port.h

@@ -26,7 +26,7 @@
 
 #ifdef LWNODE
 #include "lwnode.h"
-#include <nd-vm-message-channel.h>
+#include <nd-vm-main-message-port.h>
 #include <uv-loop-holder.h>
 #endif
 
@@ -359,10 +359,12 @@ Environment::Environment(IsolateData* isolate_data,
   }
 
 #if defined(LWNODE)
-  message_channel_ = new MessageChannel();
+  channel_ = Channel::New(uv_promise_.get_future(), "embedder");
+  main_message_port_ =
+      new MainMessagePort(channel_.port2, std::move(uv_promise_));
   loop_holder_ = new LoopHolderUV(isolate_data->event_loop());
-  LWNode::InitMessageChannel(
-      context, message_channel_, loop_holder_, isolate_data->event_loop());
+  LWNode::InitMainMessagePort(
+      context, main_message_port_, loop_holder_, isolate_data->event_loop());
 #endif
 
 #if HAVE_INSPECTOR
@@ -495,8 +497,8 @@ Environment::~Environment() {
 
 #if defined(LWNODE)
   // @lwnode
-  delete message_channel_;
-  message_channel_ = nullptr;
+  delete main_message_port_;
+  main_message_port_ = nullptr;
 
   delete loop_holder_;
   loop_holder_ = nullptr;
@@ -1103,11 +1105,11 @@ void Environment::RunWeakRefCleanup() {
 
 #if defined(LWNODE)
 std::shared_ptr<Port> Environment::GetPort() {
-  return message_channel_->port1();
+  return channel_.port1;
 }
 
-MessageChannel* Environment::message_channel() {
-  return message_channel_;
+MainMessagePort* Environment::main_message_port() {
+  return main_message_port_;
 }
 #endif
 

deps/node/src/env.cc

@@ -51,9 +51,12 @@
 #include <vector>
 
 // @lwnode
-class MessageChannel;
+#include <future>
+#include <channel.h>
+class MainMessagePort;
 class Port;
 class LoopHolderUV;
+struct Channel;
 // @lwnode
 
 namespace node {
@@ -1406,11 +1409,13 @@ class Environment : public MemoryRetainer {
   // Port
  public:
   std::shared_ptr<Port> GetPort();
-  MessageChannel* message_channel();
+  MainMessagePort* main_message_port();
 
  private:
-  MessageChannel* message_channel_;
-  LoopHolderUV* loop_holder_;
+  std::promise<uv_loop_t*> uv_promise_;
+  Channel channel_;
+  MainMessagePort* main_message_port_ = nullptr;
+  LoopHolderUV* loop_holder_ = nullptr;
 #endif
 };
 

deps/node/src/env.h

@@ -77,7 +77,7 @@
         'src/lwnode/lwnode-gc-strategy.cc',
         'src/lwnode/nd-mod-base.cc',
         'src/lwnode/nd-mod-message-port.cc',
-        'src/lwnode/nd-vm-message-channel.cc',
+        'src/lwnode/nd-vm-main-message-port.cc',
       ],
       'defines': ['V8_PROMISE_INTERNAL_FIELD_COUNT=1',
         'LWNODE_ENABLE_EXPERIMENTAL_SERIALIZATION=1',

escargotshim.gyp

@@ -28,7 +28,7 @@ class ContextRef;
 class ValueRef;
 }  // namespace Escargot
 
-class MessageChannel;
+class MainMessagePort;
 class Port;
 class LoopHolderUV;
 
@@ -38,17 +38,17 @@ namespace LWNode {
 
 enum ContextEmbedderIndex {
   // Others are listed in deps/node/src/node_context_data.h.
-  kMessageChannel = 90,
+  kMainMessagePort = 90,
   kLoopHolder = 91,
 };
 
 void InitializeProcessMethods(v8::Local<v8::Object> target,
                               v8::Local<v8::Context> context);
 
-void InitMessageChannel(v8::Local<v8::Context> context,
-                        MessageChannel* channel,
-                        LoopHolderUV* loop_holder,
-                        uv_loop_t* loop);
+void InitMainMessagePort(v8::Local<v8::Context> context,
+                         MainMessagePort* port,
+                         LoopHolderUV* loop_holder,
+                         uv_loop_t* loop);
 
 void IdleGC(v8::Isolate* isolate = nullptr);
 void initDebugger();

include/lwnode/lwnode.h

@@ -52,6 +52,7 @@ class EXPORT_API Port {
   using OnMessageCallback = std::function<void(const MessageEvent*)>;
   enum Result {
     NoError = 0,
+    MessageEventQueued,
     NoSink,
     NoOnMessage,
     InvalidMessageEvent,

include/lwnode/message-port.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <future>
 #include <memory>
 
 namespace Escargot {
@@ -26,25 +27,22 @@ class FunctionObjectRef;
 class Port;
 using uv_loop_t = struct uv_loop_s;
 
-class MessageChannel {
+class MainMessagePort {
  public:
-  MessageChannel();
-  ~MessageChannel();
+  MainMessagePort(std::shared_ptr<Port> port,
+                  std::promise<uv_loop_t*>&& promise);
+  ~MainMessagePort();
 
   void Init(Escargot::ContextRef* context, uv_loop_t* loop);
-  void Start();
 
-  std::shared_ptr<Port> port1() { return port1_; }
-  std::shared_ptr<Port> port2() { return port2_; }
+  std::shared_ptr<Port> port() { return port_; }
   Escargot::ContextRef* context() { return context_; }
 
   void SetMessageEventClass(Escargot::FunctionObjectRef* klass);
   Escargot::FunctionObjectRef* MessageEventClass();
 
  private:
-  // In runtime perspective, port1 is a sink.
-  std::shared_ptr<Port> port1_;
-  std::shared_ptr<Port> port2_;
+  std::shared_ptr<Port> port_;
   Escargot::ContextRef* context_;
   uv_loop_t* uv_loop_;
 

include/lwnode/nd-vm-message-channel.h renamed to include/lwnode/nd-vm-main-message-port.h

@@ -17,7 +17,7 @@
 #include "lwnode/lwnode.h"
 #include <EscargotPublic.h>
 #include <malloc.h>  // for malloc_trim
-#include <nd-vm-message-channel.h>
+#include <nd-vm-main-message-port.h>
 #include <uv-loop-holder.h>
 #include <codecvt>
 #include <fstream>
@@ -240,15 +240,15 @@ static ValueRef* Unref(ExecutionStateRef* state,
   return ValueRef::create(loop_holder->ref_count());
 }
 
-void InitMessageChannel(Local<Context> context,
-                        MessageChannel* channel,
+void InitMainMessagePort(Local<Context> context,
+                        MainMessagePort* main_port,
                         LoopHolderUV* loop_holder,
                         uv_loop_t* loop) {
   auto lwContext = CVAL(*context)->context();
   auto esContext = lwContext->get();
-  channel->Init(esContext, loop);
+  main_port->Init(esContext, loop);
 
-  lwContext->SetAlignedPointerInEmbedderData(kMessageChannel, channel);
+  lwContext->SetAlignedPointerInEmbedderData(kMainMessagePort, main_port);
   lwContext->SetAlignedPointerInEmbedderData(kLoopHolder, loop_holder);
 }