worker: prevent event loop starvation through MessagePorts

Limit the number of messages processed without interruption on a
given `MessagePort` to prevent event loop starvation, but still
make sure that all messages are emitted that were already in the
queue when emitting began.

This aligns the behaviour better with the web.

Refs: https://github.com/nodejs/node/pull/28030

PR-URL: https://github.com/nodejs/node/pull/29315
Reviewed-By: Gus Caplan <me@gus.host>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Jeremiah Senkpiel <fishrock123@rocketmail.com>

1 files changed, 19 insertions, 0 deletions

diff --git a/src/node_messaging.cc b/src/node_messaging.cc
index 5aec784f60..19065fdb7d 100644
--- a/src/node_messaging.cc
+++ b/src/node_messaging.cc
@@ -604,11 +604,30 @@ void MessagePort::OnMessage() {
   HandleScope handle_scope(env()->isolate());
   Local<Context> context = object(env()->isolate())->CreationContext();
 
+  size_t processing_limit;
+  {
+    Mutex::ScopedLock(data_->mutex_);
+    processing_limit = std::max(data_->incoming_messages_.size(),
+                                static_cast<size_t>(1000));
+  }
+
   // data_ can only ever be modified by the owner thread, so no need to lock.
   // However, the message port may be transferred while it is processing
   // messages, so we need to check that this handle still owns its `data_` field
   // on every iteration.
   while (data_) {
+    if (processing_limit-- == 0) {
+      // Prevent event loop starvation by only processing those messages without
+      // interruption that were already present when the OnMessage() call was
+      // first triggered, but at least 1000 messages because otherwise the
+      // overhead of repeatedly triggering the uv_async_t instance becomes
+      // noticable, at least on Windows.
+      // (That might require more investigation by somebody more familiar with
+      // Windows.)
+      TriggerAsync();
+      return;
+    }
+
     HandleScope handle_scope(env()->isolate());
     Context::Scope context_scope(context);