worker: add Notification API

Author: ShogunPanda

doc/api/worker_threads.md

@@ -654,6 +654,34 @@ this matches its values.
 
 If this is used in the main thread, its value is an empty object.
 
+## `worker.getNotifications()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Returns: {Array<bigint>} A list of notifications registered in the current thread.
+
+Return the list of all notifications registered in the current thread via
+[`worker.registerNotification(callback)`](#workerregisternotificationcallback).
+
+## `worker.registerNotification(callback)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `callback` {Function} A function to execute when the notification is received.
+* Returns: {bigint} The notification ID
+
+Register a callback associated to a notification.
+
+A notification is a number which can be safely transferred and used in other threads to trigger
+the associated callback via [`sendNotification(id`)](#workersendnotificationid).
+
+The notification must be unregistered using [`unregisterNotification(id)`](#workerunregisternotificationid)
+or [`unregisterNotifications()`](#workerunregisternotifications).
+
 ## `worker.SHARE_ENV`
 
 <!-- YAML
@@ -685,6 +713,17 @@ new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
   });
 ```
 
+## `worker.sendNotification(id)`
+
+* `id` {bigint}: The notification ID
+
+Triggers the notification previously registered via [`worker.registerNotification(callback)`](#workerregisternotificationcallback).
+
+The callback in receiving thread will invoked via [`setImmediate`](./timers.md#setimmediatecallback-args).
+
+It will throw an error if the notification is not found.  It is safe to invoke a notification
+registered in the same thread.
+
 ## `worker.setEnvironmentData(key[, value])`
 
 <!-- YAML
@@ -732,6 +771,27 @@ added: v24.6.0
 A string identifier for the current thread or null if the thread is not running.
 On the corresponding worker object (if there is any), it is available as [`worker.threadName`][].
 
+## `worker.unregisterNotification(id)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `id` {number}
+
+Unregister a notification previously registered with [`worker.registerNotification(callback)`](#workerregisternotificationcallback).
+
+It will throw an error if the notification is not found. The notification must be unregistered from the same
+worker thread that registered it.
+
+## `worker.unregisterNotifications()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+Unregister all the notifications previously registered with [`worker.registerNotification(callback)`](#workerregisternotificationcallback).
+
 ## `worker.workerData`
 
 <!-- YAML

lib/internal/worker.js

@@ -5,6 +5,7 @@ const {
   ArrayPrototypeMap,
   ArrayPrototypePush,
   AtomicsAdd,
+  BigInt,
   Float64Array,
   FunctionPrototypeBind,
   MathMax,
@@ -43,6 +44,7 @@ const {
   ERR_INVALID_ARG_TYPE,
   ERR_INVALID_ARG_VALUE,
   ERR_OPERATION_FAILED,
+  ERR_OUT_OF_RANGE,
 } = errorCodes;
 
 const workerIo = require('internal/worker/io');
@@ -63,7 +65,13 @@ const { createMainThreadPort, destroyMainThreadPort } = require('internal/worker
 const { deserializeError } = require('internal/error_serdes');
 const { fileURLToPath, isURL, pathToFileURL } = require('internal/url');
 const { kEmptyObject } = require('internal/util');
-const { validateArray, validateString, validateObject, validateNumber } = require('internal/validators');
+const {
+  validateArray,
+  validateString,
+  validateObject,
+  validateNumber,
+  validateFunction,
+} = require('internal/validators');
 const {
   throwIfBuildingSnapshot,
 } = require('internal/v8/startup_snapshot');
@@ -80,6 +88,11 @@ const {
   kCodeRangeSizeMb,
   kStackSizeMb,
   kTotalResourceLimitCount,
+  getNotifications,
+  registerNotification: registerNotificationInternal,
+  sendNotification: sendNotificationInternal,
+  unregisterNotification: unregisterNotificationInternal,
+  unregisterNotifications,
 } = internalBinding('worker');
 
 const kHandle = Symbol('kHandle');
@@ -620,6 +633,37 @@ function eventLoopUtilization(util1, util2) {
   );
 }
 
+
+function registerNotification(listener) {
+  validateFunction(listener, 'listener');
+  return registerNotificationInternal(listener);
+}
+
+function unregisterNotification(id) {
+  if (typeof id !== 'number' && typeof id !== 'bigint') {
+    throw new ERR_INVALID_ARG_TYPE('id', ['number', 'bigint'], id);
+  }
+
+  if (id < 0) {
+    throw new ERR_OUT_OF_RANGE('id', `>= 1`, id);
+  }
+
+  unregisterNotificationInternal(BigInt(id));
+}
+
+function sendNotification(id) {
+  if (typeof id !== 'number' && typeof id !== 'bigint') {
+    throw new ERR_INVALID_ARG_TYPE('id', ['number', 'bigint'], id);
+  }
+
+  if (id < 0) {
+    throw new ERR_OUT_OF_RANGE('id', `>= 1`, id);
+  }
+
+  sendNotificationInternal(BigInt(id));
+}
+
+
 module.exports = {
   ownsProcessState,
   kIsOnline,
@@ -635,4 +679,9 @@ module.exports = {
   threadName,
   InternalWorker,
   Worker,
+  getNotifications,
+  registerNotification,
+  sendNotification,
+  unregisterNotification,
+  unregisterNotifications,
 };

lib/worker_threads.js

@@ -10,6 +10,11 @@ const {
   threadId,
   threadName,
   Worker,
+  getNotifications,
+  registerNotification,
+  sendNotification,
+  unregisterNotification,
+  unregisterNotifications,
 } = require('internal/worker');
 
 const {
@@ -54,4 +59,9 @@ module.exports = {
   setEnvironmentData,
   getEnvironmentData,
   locks,
+  getNotifications,
+  registerNotification,
+  sendNotification,
+  unregisterNotification,
+  unregisterNotifications,
 };

src/env.h

@@ -199,6 +199,7 @@ class NODE_EXTERN_PRIVATE IsolateData : public MemoryRetainer {
 
   size_t max_young_gen_size = 1;
   std::unordered_map<const char*, v8::Eternal<v8::String>> static_str_map;
+  std::set<uint64_t> worker_notifications_;
 
   inline v8::Isolate* isolate() const;
   IsolateData(const IsolateData&) = delete;

src/node_errors.h

@@ -99,6 +99,7 @@ void OOMErrorHandler(const char* location, const v8::OOMDetails& details);
   V(ERR_INVALID_PACKAGE_CONFIG, Error)                                         \
   V(ERR_INVALID_OBJECT_DEFINE_PROPERTY, TypeError)                             \
   V(ERR_INVALID_MODULE, Error)                                                 \
+  V(ERR_INVALID_NOTIFICATION, Error)                                           \
   V(ERR_INVALID_STATE, Error)                                                  \
   V(ERR_INVALID_THIS, TypeError)                                               \
   V(ERR_INVALID_URL, TypeError)                                                \
@@ -218,6 +219,7 @@ ERRORS_WITH_CODE(V)
   V(ERR_INVALID_ADDRESS, "Invalid socket address")                             \
   V(ERR_INVALID_INVOCATION, "Invalid invocation")                              \
   V(ERR_INVALID_MODULE, "No such module")                                      \
+  V(ERR_INVALID_NOTIFICATION, "Invalid notification")                          \
   V(ERR_INVALID_STATE, "Invalid state")                                        \
   V(ERR_INVALID_THIS, "Value of \"this\" is the wrong type")                   \
   V(ERR_INVALID_URL_SCHEME, "The URL must be of scheme file:")                 \

src/node_worker.cc

@@ -34,6 +34,7 @@ using v8::HandleScope;
 using v8::Integer;
 using v8::Isolate;
 using v8::Local;
+using v8::LocalVector;
 using v8::Locker;
 using v8::Maybe;
 using v8::Name;
@@ -221,6 +222,8 @@ class WorkerThreadData {
   }
 
   ~WorkerThreadData() {
+    Worker::UnregisterAllNotifications(isolate_data_.get());
+
     Debug(w_, "Worker %llu dispose isolate", w_->thread_id_.id);
     Isolate* isolate;
     {
@@ -1281,6 +1284,127 @@ void Worker::LoopStartTime(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(loop_start_time / 1e6);
 }
 
+void Worker::GetNotifications(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  Isolate* isolate = env->isolate();
+  LocalVector<Value> notifications(isolate);
+
+  for (auto notification_id : env->isolate_data()->worker_notifications_) {
+    notifications.push_back(v8::BigInt::New(isolate, notification_id));
+  }
+
+  args.GetReturnValue().Set(
+      Array::New(isolate, notifications.data(), notifications.size()));
+}
+
+void Worker::RegisterNotification(const FunctionCallbackInfo<Value>& args) {
+  CHECK(args[0]->IsFunction());
+
+  Environment* env = Environment::GetCurrent(args);
+  Isolate* isolate = env->isolate();
+  NotificationData* notification =
+      new NotificationData(env->thread_id(), env, args[0].As<v8::Function>());
+
+  notification->Register(env->isolate_data());
+  args.GetReturnValue().Set(v8::BigInt::New(isolate, notification->id_));
+}
+
+void Worker::SendNotification(const FunctionCallbackInfo<Value>& args) {
+  CHECK(args[0]->IsBigInt());
+
+  uint64_t notification_id = args[0].As<v8::BigInt>()->Uint64Value();
+  NotificationData* notification = workers_notifications_[notification_id];
+
+  if (notification == nullptr) {
+    return THROW_ERR_INVALID_NOTIFICATION(
+        args.GetIsolate(), "Invalid notification %llun", notification_id);
+  }
+
+  uv_async_send(&notification->async_);
+}
+
+void Worker::UnregisterNotification(const FunctionCallbackInfo<Value>& args) {
+  CHECK(args[0]->IsBigInt());
+
+  Environment* env = Environment::GetCurrent(args);
+  uint64_t notification_id = args[0].As<v8::BigInt>()->Uint64Value();
+  NotificationData* notification = workers_notifications_[notification_id];
+
+  if (notification == nullptr || notification->thread_id_ != env->thread_id()) {
+    return THROW_ERR_INVALID_NOTIFICATION(
+        args.GetIsolate(), "Invalid notification %un", notification_id);
+  }
+
+  notification->Unregister(env->isolate_data());
+}
+
+void Worker::UnregisterNotifications(const FunctionCallbackInfo<Value>& args) {
+  UnregisterAllNotifications(Environment::GetCurrent(args)->isolate_data());
+}
+
+void Worker::UnregisterAllNotifications(IsolateData* isolate_data) {
+  for (auto notification_id : isolate_data->worker_notifications_) {
+    workers_notifications_[notification_id]->Unregister(isolate_data);
+  }
+
+  // Wait for the uv_close callbacks to trigger
+  if (!isolate_data->worker_notifications_.empty()) {
+    uv_run(isolate_data->event_loop(), UV_RUN_ONCE);
+  }
+
+  isolate_data->worker_notifications_.clear();
+}
+
+NotificationData::NotificationData(uint64_t thread_id,
+                                   Environment* env,
+                                   v8::Local<v8::Function> callback)
+    : id_(++notifications_counter_), thread_id_(thread_id), env_(env) {
+  this->async_.data = this;
+  this->callback_.Reset(env->isolate(), callback);
+}
+
+NotificationData::~NotificationData() {
+  workers_notifications_.erase(this->id_);
+
+  if (!this->callback_.IsEmpty()) {
+    this->callback_.Reset();
+  }
+}
+
+void NotificationData::Register(IsolateData* isolate_data) {
+  isolate_data->worker_notifications_.insert(this->id_);
+  workers_notifications_.emplace(this->id_, this);
+
+  uv_async_init(env_->event_loop(), &this->async_, [](uv_async_t* handle) {
+    reinterpret_cast<NotificationData*>(handle->data)->Execute();
+  });
+}
+
+void NotificationData::Unregister(IsolateData* isolate_data) {
+  uv_close(reinterpret_cast<uv_handle_t*>(&this->async_),
+           [](uv_handle_t* handle) {
+             NotificationData* notification =
+                 reinterpret_cast<NotificationData*>(handle->data);
+             delete notification;
+           });
+
+  // Wait for the uv_close callbacks to trigger
+  uv_run(isolate_data->event_loop(), UV_RUN_ONCE);
+
+  isolate_data->worker_notifications_.erase(this->id_);
+  workers_notifications_.erase(this->id_);
+}
+
+void NotificationData::Execute() {
+  this->env_->SetImmediate(
+      [this](Environment* env) {
+        v8::Isolate* isolate = this->env_->isolate();
+        this->callback_.Get(isolate)->Call(
+            isolate->GetCurrentContext(), Null(isolate), 0, nullptr);
+      },
+      CallbackFlags::kUnrefed);
+}
+
 namespace {
 
 // Return the MessagePort that is global for this Environment and communicates
@@ -1380,6 +1504,19 @@ void CreateWorkerPerIsolateProperties(IsolateData* isolate_data,
   }
 
   SetMethod(isolate, target, "getEnvMessagePort", GetEnvMessagePort);
+
+  SetMethod(isolate, target, "getNotifications", Worker::GetNotifications);
+  SetMethod(
+      isolate, target, "registerNotification", Worker::RegisterNotification);
+  SetMethod(isolate, target, "sendNotification", Worker::SendNotification);
+  SetMethod(isolate,
+            target,
+            "unregisterNotification",
+            Worker::UnregisterNotification);
+  SetMethod(isolate,
+            target,
+            "unregisterNotifications",
+            Worker::UnregisterNotifications);
 }
 
 void CreateWorkerPerContextProperties(Local<Object> target,
@@ -1458,6 +1595,11 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(Worker::CpuUsage);
   registry->Register(Worker::StartCpuProfile);
   registry->Register(Worker::StopCpuProfile);
+  registry->Register(Worker::GetNotifications);
+  registry->Register(Worker::RegisterNotification);
+  registry->Register(Worker::SendNotification);
+  registry->Register(Worker::UnregisterNotification);
+  registry->Register(Worker::UnregisterNotifications);
 }
 
 }  // anonymous namespace