worker: make MessagePort inherit from EventTarget

benchmark/worker/messageport.js

@@ -4,6 +4,7 @@ const common = require('../common.js');
 const { MessageChannel } = require('worker_threads');
 const bench = common.createBenchmark(main, {
   payload: ['string', 'object'],
+  style: ['eventtarget', 'eventemitter'],
   n: [1e6]
 });
 
@@ -25,14 +26,26 @@ function main(conf) {
   const { port1, port2 } = new MessageChannel();
 
   let messages = 0;
-  port2.onmessage = () => {
+  function listener() {
     if (messages++ === n) {
       bench.end(n);
       port1.close();
     } else {
       write();
     }
-  };
+  }
+
+  switch (conf.style) {
+    case 'eventtarget':
+      port2.onmessage = listener;
+      break;
+    case 'eventemitter':
+      port2.on('message', listener);
+      break;
+    default:
+      throw new Error('Unsupported listener type');
+  }
+
   bench.start();
   write();
 

lib/internal/event_target.js

@@ -31,8 +31,13 @@ const kEvents = Symbol('kEvents');
 const kStop = Symbol('kStop');
 const kTarget = Symbol('kTarget');
 
+const kHybridDispatch = Symbol.for('nodejs.internal.kHybridDispatch');
+const kCreateEvent = Symbol('kCreateEvent');
 const kNewListener = Symbol('kNewListener');
 const kRemoveListener = Symbol('kRemoveListener');
+const kIsNodeStyleListener = Symbol('kIsNodeStyleListener');
+const kMaxListeners = Symbol('kMaxListeners');
+const kMaxListenersWarned = Symbol('kMaxListenersWarned');
 
 // Lazy load perf_hooks to avoid the additional overhead on startup
 let perf_hooks;
@@ -157,7 +162,7 @@ Object.defineProperty(Event.prototype, SymbolToStringTag, {
 // the linked list makes dispatching faster, even if adding/removing is
 // slower.
 class Listener {
-  constructor(previous, listener, once, capture, passive) {
+  constructor(previous, listener, once, capture, passive, isNodeStyleListener) {
     this.next = undefined;
     if (previous !== undefined)
       previous.next = this;
@@ -166,6 +171,7 @@ class Listener {
     this.once = once;
     this.capture = capture;
     this.passive = passive;
+    this.isNodeStyleListener = isNodeStyleListener;
 
     this.callback =
       typeof listener === 'function' ?
@@ -185,13 +191,17 @@ class Listener {
   }
 }
 
+function initEventTarget(self) {
+  self[kEvents] = new Map();
+}
+
 class EventTarget {
   // Used in checking whether an object is an EventTarget. This is a well-known
   // symbol as EventTarget may be used cross-realm. See discussion in #33661.
   static [kIsEventTarget] = true;
 
   constructor() {
-    this[kEvents] = new Map();
+    initEventTarget(this);
   }
 
   [kNewListener](size, type, listener, once, capture, passive) {}
@@ -206,7 +216,8 @@ class EventTarget {
     const {
       once,
       capture,
-      passive
+      passive,
+      isNodeStyleListener
     } = validateEventListenerOptions(options);
 
     if (!shouldAddListener(listener)) {
@@ -228,7 +239,7 @@ class EventTarget {
     if (root === undefined) {
       root = { size: 1, next: undefined };
       // This is the first handler in our linked list.
-      new Listener(root, listener, once, capture, passive);
+      new Listener(root, listener, once, capture, passive, isNodeStyleListener);
       this[kNewListener](root.size, type, listener, once, capture, passive);
       this[kEvents].set(type, root);
       return;
@@ -247,7 +258,8 @@ class EventTarget {
       return;
     }
 
-    new Listener(previous, listener, once, capture, passive);
+    new Listener(previous, listener, once, capture, passive,
+                 isNodeStyleListener);
     root.size++;
     this[kNewListener](root.size, type, listener, once, capture, passive);
   }
@@ -290,39 +302,63 @@ class EventTarget {
     if (event[kTarget] !== null)
       throw new ERR_EVENT_RECURSION(event.type);
 
-    const root = this[kEvents].get(event.type);
+    this[kHybridDispatch](event, event.type);
+
+    return event.defaultPrevented !== true;
+  }
+
+  [kHybridDispatch](nodeValue, type) {
+    let event;
+    const createEvent = () => {
+      if (event === undefined) {
+        event = this[kCreateEvent](nodeValue, type);
+        event[kTarget] = this;
+      }
+      return event;
+    };
+
+    const root = this[kEvents].get(type);
     if (root === undefined || root.next === undefined)
       return true;
 
-    event[kTarget] = this;
-
     let handler = root.next;
     let next;
 
     while (handler !== undefined &&
-           (handler.passive || event[kStop] !== true)) {
+           (handler.passive || (event?.[kStop] !== true))) {
       // Cache the next item in case this iteration removes the current one
       next = handler.next;
 
       if (handler.once) {
         handler.remove();
         root.size--;
+        const { listener, capture } = handler;
+        this[kRemoveListener](root.size, type, listener, capture);
       }
 
       try {
-        const result = handler.callback.call(this, event);
+        let arg;
+        if (handler.isNodeStyleListener) {
+          arg = nodeValue;
+        } else {
+          arg = createEvent();
+        }
+        const result = handler.callback.call(this, arg);
         if (result !== undefined && result !== null)
-          addCatch(this, result, event);
+          addCatch(this, result, createEvent());
       } catch (err) {
-        emitUnhandledRejectionOrErr(this, err, event);
+        emitUnhandledRejectionOrErr(this, err, createEvent());
       }
 
       handler = next;
     }
 
-    event[kTarget] = undefined;
+    if (event !== undefined)
+      event[kTarget] = undefined;
+  }
 
-    return event.defaultPrevented !== true;
+  [kCreateEvent](event) {
+    return event;
   }
 
   [customInspectSymbol](depth, options) {
@@ -350,15 +386,19 @@ Object.defineProperty(EventTarget.prototype, SymbolToStringTag, {
   value: 'EventTarget',
 });
 
-const kMaxListeners = Symbol('maxListeners');
-const kMaxListenersWarned = Symbol('maxListenersWarned');
+function initNodeEventTarget(self) {
+  initEventTarget(self);
+  // eslint-disable-next-line no-use-before-define
+  self[kMaxListeners] = NodeEventTarget.defaultMaxListeners;
+  self[kMaxListenersWarned] = false;
+}
+
 class NodeEventTarget extends EventTarget {
   static defaultMaxListeners = 10;
 
   constructor() {
     super();
-    this[kMaxListeners] = NodeEventTarget.defaultMaxListeners;
-    this[kMaxListenersWarned] = false;
+    initNodeEventTarget(this);
   }
 
   [kNewListener](size, type, listener, once, capture, passive) {
@@ -410,17 +450,18 @@ class NodeEventTarget extends EventTarget {
   }
 
   on(type, listener) {
-    this.addEventListener(type, listener);
+    this.addEventListener(type, listener, { [kIsNodeStyleListener]: true });
     return this;
   }
 
   addListener(type, listener) {
-    this.addEventListener(type, listener);
+    this.addEventListener(type, listener, { [kIsNodeStyleListener]: true });
     return this;
   }
 
   once(type, listener) {
-    this.addEventListener(type, listener, { once: true });
+    this.addEventListener(type, listener,
+                          { once: true, [kIsNodeStyleListener]: true });
     return this;
   }
 
@@ -470,6 +511,7 @@ function validateEventListenerOptions(options) {
     once: Boolean(options.once),
     capture: Boolean(options.capture),
     passive: Boolean(options.passive),
+    isNodeStyleListener: Boolean(options[kIsNodeStyleListener])
   };
 }
 
@@ -520,4 +562,9 @@ module.exports = {
   EventTarget,
   NodeEventTarget,
   defineEventHandler,
+  initEventTarget,
+  initNodeEventTarget,
+  kCreateEvent,
+  kNewListener,
+  kRemoveListener,
 };

lib/internal/per_context/messageport.js

@@ -1,12 +1,31 @@
 'use strict';
+const {
+  SymbolFor,
+} = primordials;
+
 class MessageEvent {
-  constructor(data, target) {
+  constructor(data, target, type) {
     this.data = data;
     this.target = target;
+    this.type = type;
   }
 }
 
-exports.emitMessage = function(data) {
-  if (typeof this.onmessage === 'function')
-    this.onmessage(new MessageEvent(data, this));
+const kHybridDispatch = SymbolFor('nodejs.internal.kHybridDispatch');
+
+exports.emitMessage = function(data, type) {
+  if (typeof this[kHybridDispatch] === 'function') {
+    this[kHybridDispatch](data, type);
+    return;
+  }
+
+  const event = new MessageEvent(data, this, type);
+  if (type === 'message') {
+    if (typeof this.onmessage === 'function')
+      this.onmessage(event);
+  } else {
+    // eslint-disable-next-line no-lonely-if
+    if (typeof this.onmessageerror === 'function')
+      this.onmessageerror(event);
+  }
 };