worker: emit worker exit notifications on BroadcastChannel

doc/api/worker_threads.md

@@ -987,6 +987,25 @@ added: v15.4.0
 * Type: {Function} Invoked with a received message cannot be
   deserialized.
 
+### `broadcastChannel.onworkerexited`
+
+* 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,32 @@ void MessagePortData::AddToIncomingQueue(std::shared_ptr<Message> message) {
   }
 }
 
+void MessagePortData::AddWorkerExitNotification(uint64_t thread_id,
+                                                ExitCode exit_code) {
+  Mutex::ScopedLock lock(mutex_);
+  worker_exit_notifications_.emplace_back(WorkerExitNotification{
+      thread_id,
+      exit_code,
+  });
+
+  if (owner_ != nullptr) {
+    Debug(owner_, "Adding worker-exit notification");
+    owner_->TriggerAsync();
+  }
+}
+
+bool MessagePortData::GetWorkerExitNotification(
+    WorkerExitNotification* notification) {
+  Mutex::ScopedLock lock(mutex_);
+
+  if (worker_exit_notifications_.empty()) return false;
+
+  *notification = worker_exit_notifications_.front();
+  worker_exit_notifications_.pop_front();
+
+  return true;
+}
+
 void MessagePortData::Entangle(MessagePortData* a, MessagePortData* b) {
   auto group = std::make_shared<SiblingGroup>();
   group->Entangle({a, b});
@@ -823,6 +849,7 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
   HandleScope handle_scope(env()->isolate());
   Local<Context> context =
       object(env()->isolate())->GetCreationContextChecked();
+  Local<Function> emit_message = PersistentToLocal::Strong(emit_message_fn_);
 
   size_t processing_limit;
   if (mode == MessageProcessingMode::kNormalOperation) {
@@ -850,9 +877,43 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
       return;
     }
 
+    MessagePortData::WorkerExitNotification worker_exit;
+
+    if (data_->GetWorkerExitNotification(&worker_exit)) {
+      Debug(this,
+            "Worker exited: thread_id=%" PRIu64 ", exit_code=%d",
+            worker_exit.thread_id,
+            static_cast<int>(worker_exit.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(), worker_exit.thread_id))
+          .Check();
+
+      exit_info
+          ->Set(context,
+                FIXED_ONE_BYTE_STRING(env()->isolate(), "exitCode"),
+                v8::Integer::New(env()->isolate(),
+                                 static_cast<int>(worker_exit.exit_code)))
+          .Check();
+
+      Local<Value> argv[3];
+      argv[0] = exit_info;
+      argv[1] = Undefined(env()->isolate());
+      argv[2] = FIXED_ONE_BYTE_STRING(env()->isolate(), "workerexited");
+
+      if (MakeCallback(emit_message, arraysize(argv), argv).IsEmpty()) {
+        if (data_) TriggerAsync();
+        return;
+      }
+      continue;
+    }
+
     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());
@@ -901,6 +962,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=%" PRIu64 ", exit_code=%d",
+            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 +1662,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

@@ -149,6 +149,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 +189,9 @@ 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 +220,15 @@ 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_;
+  struct WorkerExitNotification {
+    uint64_t thread_id;
+    ExitCode exit_code;
+  };
+
+  bool GetWorkerExitNotification(WorkerExitNotification* notification);
+
+  std::deque<WorkerExitNotification> worker_exit_notifications_;
+
   MessagePort* owner_ = nullptr;
   std::shared_ptr<SiblingGroup> group_;
   friend class MessagePort;

test/parallel/test-worker-broadcastchannel.js

@@ -183,3 +183,30 @@
     "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);
+  }));
+}