worker: emit worker exit notifications on BroadcastChannel

doc/api/worker_threads.md

@@ -987,6 +987,29 @@ added: v15.4.0
 * Type: {Function} Invoked with a received message cannot be
   deserialized.
 
+### `broadcastChannel.onworkerexited`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {Function} Invoked when worker associated with the
+  `BroadcastChannel` terminates.
+
+The callback receives an object with the following properties:
+
+* `threadId` {number} The ID of the worker thread that terminated.
+* `exitCode` {number} The exit code with which the worker terminated.
+
+The `exitCode` is the value passed to `process.exit()` when the worker
+explicitly exits. If the worker terminates without explicitly specifying
+an exit code, the corresponding exit code is reported.
+
+The `workerexited` event is emitted only when the worker's execution
+environment is stopping. Closing a `BroadcastChannel` or its underlying
+`MessagePort` does not by itself indicate that a worker has exited and
+does not emit this event.
+
 ### `broadcastChannel.postMessage(message)`
 
 <!-- YAML

lib/internal/worker/io.js

@@ -74,6 +74,7 @@ const kIncrementsPortRef = Symbol('kIncrementsPortRef');
 const kName = Symbol('kName');
 const kOnMessage = Symbol('kOnMessage');
 const kOnMessageError = Symbol('kOnMessageError');
+const kOnWorkerExited = Symbol('kOnWorkerExited');
 const kPort = Symbol('kPort');
 const kWaitingStreams = Symbol('kWaitingStreams');
 const kWritableCallback = Symbol('kWritableCallback');
@@ -367,8 +368,10 @@ class BroadcastChannel extends EventTarget {
     this[kOnMessage] = FunctionPrototypeBind(onMessageEvent, this, 'message');
     this[kOnMessageError] =
       FunctionPrototypeBind(onMessageEvent, this, 'messageerror');
+    this[kOnWorkerExited] = FunctionPrototypeBind(onMessageEvent, this, 'workerexited');
     this[kHandle].on('message', this[kOnMessage]);
     this[kHandle].on('messageerror', this[kOnMessageError]);
+    this[kHandle].on('workerexited', this[kOnWorkerExited]);
   }
 
   [inspect.custom](depth, options) {
@@ -407,8 +410,10 @@ class BroadcastChannel extends EventTarget {
       return;
     this[kHandle].off('message', this[kOnMessage]);
     this[kHandle].off('messageerror', this[kOnMessageError]);
+    this[kHandle].off('workerexited', this[kOnWorkerExited]);
     this[kOnMessage] = undefined;
     this[kOnMessageError] = undefined;
+    this[kOnWorkerExited] = undefined;
     this[kHandle].close();
     this[kHandle] = undefined;
   }
@@ -468,6 +473,7 @@ ObjectDefineProperties(BroadcastChannel.prototype, {
 
 defineEventHandler(BroadcastChannel.prototype, 'message');
 defineEventHandler(BroadcastChannel.prototype, 'messageerror');
+defineEventHandler(BroadcastChannel.prototype, 'workerexited');
 
 function markAsUncloneable(obj) {
   if ((typeof obj !== 'object' && typeof obj !== 'function') || obj === null) {

src/node_messaging.cc

@@ -648,6 +648,19 @@ void MessagePortData::AddToIncomingQueue(std::shared_ptr<Message> message) {
   }
 }
 
+void MessagePortData::AddWorkerExitNotification(uint64_t thread_id,
+                                                ExitCode exit_code) {
+  Mutex::ScopedLock lock(mutex_);
+  auto message =
+      std::make_shared<Message>(WorkerExitNotification{thread_id, exit_code});
+  AddToIncomingQueue(std::move(message));
+
+  if (owner_ != nullptr) {
+    Debug(owner_, "Adding worker-exit notification");
+    owner_->TriggerAsync();
+  }
+}
+
 void MessagePortData::Entangle(MessagePortData* a, MessagePortData* b) {
   auto group = std::make_shared<SiblingGroup>();
   group->Entangle({a, b});
@@ -774,9 +787,11 @@ MessagePort* MessagePort::New(
   return port;
 }
 
-MaybeLocal<Value> MessagePort::ReceiveMessage(Local<Context> context,
-                                              MessageProcessingMode mode,
-                                              Local<Value>* port_list) {
+MaybeLocal<Value> MessagePort::ReceiveMessage(
+    Local<Context> context,
+    MessageProcessingMode mode,
+    Local<Value>* port_list,
+    std::optional<WorkerExitNotification>* worker_exit) {
   std::shared_ptr<Message> received;
   {
     // Get the head of the message queue.
@@ -801,6 +816,13 @@ MaybeLocal<Value> MessagePort::ReceiveMessage(Local<Context> context,
     data_->incoming_messages_.pop_front();
   }
 
+  if (received->IsWorkerExitMessage()) {
+    if (worker_exit != nullptr) {
+      *worker_exit = received->worker_exit_notification();
+      return env()->no_message_symbol();
+    }
+  }
+
   if (received->IsCloseMessage()) {
     Close();
     return env()->no_message_symbol();
@@ -819,11 +841,9 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
   // context, it will call the constructor and trigger the async handle empty.
   // Because all data was sent from the previous context.
   if (IsDetached()) return;
-
   HandleScope handle_scope(env()->isolate());
   Local<Context> context =
       object(env()->isolate())->GetCreationContextChecked();
-
   size_t processing_limit;
   if (mode == MessageProcessingMode::kNormalOperation) {
     Mutex::ScopedLock lock(data_->mutex_);
@@ -850,36 +870,65 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
       return;
     }
 
-    HandleScope handle_scope(env()->isolate());
     Context::Scope context_scope(context);
     Local<Function> emit_message = PersistentToLocal::Strong(emit_message_fn_);
 
     Local<Value> payload;
     Local<Value> port_list = Undefined(env()->isolate());
     Local<Value> message_error;
     Local<Value> argv[3];
+    std::optional<WorkerExitNotification> worker_exit;
 
     {
       // Catch any exceptions from parsing the message itself (not from
       // emitting it) as 'messageeror' events.
       TryCatchScope try_catch(env());
-      if (!ReceiveMessage(context, mode, &port_list).ToLocal(&payload)) {
+      if (!ReceiveMessage(context, mode, &port_list, &worker_exit)
+               .ToLocal(&payload)) {
         if (try_catch.HasCaught() && !try_catch.HasTerminated())
           message_error = try_catch.Exception();
         goto reschedule;
       }
     }
-    if (payload == env()->no_message_symbol()) break;
+    if (payload == env()->no_message_symbol() && !worker_exit.has_value())
+      break;
 
     if (!env()->can_call_into_js()) {
       Debug(this, "MessagePort drains queue because !can_call_into_js()");
       // In this case there is nothing to do but to drain the current queue.
       continue;
     }
 
-    argv[0] = payload;
-    argv[1] = port_list;
-    argv[2] = env()->message_string();
+    if (worker_exit.has_value()) {
+      const WorkerExitNotification& notification = *worker_exit;
+
+      Debug(this,
+            "Worker exited: thread_id=%d, exit_code=%d",
+            static_cast<int>(notification.thread_id),
+            static_cast<int>(notification.exit_code));
+      Local<Object> exit_info = Object::New(env()->isolate());
+
+      exit_info
+          ->Set(context,
+                FIXED_ONE_BYTE_STRING(env()->isolate(), "threadId"),
+                v8::Number::New(env()->isolate(),
+                                static_cast<double>(notification.thread_id)))
+          .Check();
+
+      exit_info
+          ->Set(context,
+                FIXED_ONE_BYTE_STRING(env()->isolate(), "exitCode"),
+                v8::Integer::New(env()->isolate(),
+                                 static_cast<int>(notification.exit_code)))
+          .Check();
+      argv[0] = exit_info;
+      argv[1] = Undefined(env()->isolate());
+      argv[2] = FIXED_ONE_BYTE_STRING(env()->isolate(), "workerexited");
+    } else {
+      argv[0] = payload;
+      argv[1] = port_list;
+      argv[2] = env()->message_string();
+    }
 
     if (MakeCallback(emit_message, arraysize(argv), argv).IsEmpty()) {
     reschedule:
@@ -901,6 +950,20 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
 void MessagePort::OnClose() {
   Debug(this, "MessagePort::OnClose()");
   if (data_) {
+    Environment* environment = env();
+    if (environment->is_stopping()) {
+      const uint64_t thread_id = environment->thread_id();
+      const ExitCode exit_code = environment->exit_code(ExitCode::kNoFailure);
+
+      Debug(this,
+            "Worker exiting: thread_id=%d, exit_code=%d",
+            static_cast<int>(thread_id),
+            static_cast<int>(exit_code));
+
+      if (data_->group_) {
+        data_->group_->NotifyWorkerExit(data_.get(), thread_id, exit_code);
+      }
+    }
     // Detach() returns move(data_).
     Detach()->Disentangle();
   }
@@ -1587,6 +1650,18 @@ void SiblingGroup::Disentangle(MessagePortData* data) {
     (*(ports_.begin()))->AddToIncomingQueue(std::make_shared<Message>());
 }
 
+void SiblingGroup::NotifyWorkerExit(MessagePortData* exiting_port,
+                                    uint64_t thread_id,
+                                    ExitCode exit_code) {
+  RwLock::ScopedReadLock lock(group_mutex_);
+
+  for (MessagePortData* port : ports_) {
+    if (port == exiting_port) continue;
+
+    port->AddWorkerExitNotification(thread_id, exit_code);
+  }
+}
+
 SiblingGroup::Map SiblingGroup::groups_;
 Mutex SiblingGroup::groups_mutex_;
 

src/node_messaging.h

@@ -17,6 +17,11 @@ namespace worker {
 class MessagePortData;
 class MessagePort;
 
+struct WorkerExitNotification {
+  uint64_t thread_id;
+  ExitCode exit_code;
+};
+
 typedef MaybeStackBuffer<v8::Value, 8> TransferList;
 
 // Used to represent the in-flight structure of an object that is being
@@ -49,6 +54,10 @@ class Message : public MemoryRetainer {
   // V8 ValueSerializer API. If `payload` is empty, this message indicates
   // that the receiving message port should close itself.
   explicit Message(MallocedBuffer<char>&& payload = MallocedBuffer<char>());
+
+  explicit Message(WorkerExitNotification notification)
+      : worker_exit_notification_(std::move(notification)) {}
+
   ~Message() = default;
 
   Message(Message&& other) = default;
@@ -60,6 +69,14 @@ class Message : public MemoryRetainer {
   // This is the last message to be received by a MessagePort.
   bool IsCloseMessage() const;
 
+  bool IsWorkerExitMessage() const {
+    return worker_exit_notification_.has_value();
+  }
+
+  const WorkerExitNotification& worker_exit_notification() const {
+    return worker_exit_notification_.value();
+  }
+
   // Deserialize the contained JS value. May only be called once, and only
   // after Serialize() has been called (e.g. by another thread).
   v8::MaybeLocal<v8::Value> Deserialize(
@@ -118,6 +135,7 @@ class Message : public MemoryRetainer {
   std::vector<std::unique_ptr<TransferData>> transferables_;
   std::vector<v8::CompiledWasmModule> wasm_modules_;
   std::optional<v8::SharedValueConveyor> shared_value_conveyor_;
+  std::optional<WorkerExitNotification> worker_exit_notification_;
 
   friend class MessagePort;
 };
@@ -149,6 +167,10 @@ class SiblingGroup final : public std::enable_shared_from_this<SiblingGroup> {
   void Entangle(std::initializer_list<MessagePortData*> data);
   void Disentangle(MessagePortData* data);
 
+  void NotifyWorkerExit(MessagePortData* exiting_port,
+                        uint64_t thread_id,
+                        ExitCode exit_code);
+
   const std::string& name() const { return name_; }
 
   size_t size() const { return ports_.size(); }
@@ -185,6 +207,8 @@ class MessagePortData : public TransferData {
   v8::Maybe<bool> Dispatch(
       std::shared_ptr<Message> message,
       std::string* error = nullptr);
+  // Internal worker-exit notification.
+  void AddWorkerExitNotification(uint64_t thread_id, ExitCode exit_code);
 
   // Turns `a` and `b` into siblings, i.e. connects the sending side of one
   // to the receiving side of the other. This is not thread-safe.
@@ -213,6 +237,8 @@ class MessagePortData : public TransferData {
   // once that is available with C++17, because std::shared_ptr comes with
   // overhead that is only necessary for BroadcastChannel.
   std::deque<std::shared_ptr<Message>> incoming_messages_;
+  bool GetWorkerExitNotification(WorkerExitNotification* notification);
+  std::deque<WorkerExitNotification> worker_exit_notifications_;
   MessagePort* owner_ = nullptr;
   std::shared_ptr<SiblingGroup> group_;
   friend class MessagePort;
@@ -308,7 +334,8 @@ class MessagePort : public HandleWrap {
   v8::MaybeLocal<v8::Value> ReceiveMessage(
       v8::Local<v8::Context> context,
       MessageProcessingMode mode,
-      v8::Local<v8::Value>* port_list = nullptr);
+      v8::Local<v8::Value>* port_list = nullptr,
+      std::optional<WorkerExitNotification>* worker_exit = nullptr);
 
   std::unique_ptr<MessagePortData> data_ = nullptr;
   bool receiving_messages_ = false;

test/parallel/test-worker-broadcastchannel.js

@@ -183,3 +183,30 @@ assert.throws(() => new BroadcastChannel(), {
     "BroadcastChannel { name: 'channel5', active: false }"
   );
 }
+
+{
+  const bc = new BroadcastChannel('channel6');
+
+  const worker = new Worker(`
+    const { BroadcastChannel } = require('worker_threads');
+
+    const bc = new BroadcastChannel('channel6');
+
+    // Keep the BroadcastChannel alive long enough for the exit
+    // notification to be observed by the parent.
+    setImmediate(() => {
+      process.exit(42);
+    });
+  `, { eval: true });
+
+  bc.onworkerexited = common.mustCall((event) => {
+    assert.strictEqual(event.data.threadId, worker.threadId);
+    assert.strictEqual(event.data.exitCode, 42);
+
+    bc.close();
+  });
+
+  worker.on('exit', common.mustCall((exitCode) => {
+    assert.strictEqual(exitCode, 42);
+  }));
+}