lib: create diagnostics_channel module

PR-URL: nodejs#34895
Reviewed-By: Bryan English <[email protected]>
Reviewed-By: Gerhard Stöbich <[email protected]>
Reviewed-By: Vladimir de Turckheim <[email protected]>
Reviewed-By: Rich Trott <[email protected]>
Reviewed-By: Gabriel Schulhof <[email protected]>
Reviewed-By: Michael Dawson <[email protected]>

benchmark/diagnostics_channel/publish.js

@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common.js');
+const dc = require('diagnostics_channel');
+
+const bench = common.createBenchmark(main, {
+  n: [1e8],
+  subscribers: [0, 1, 10],
+});
+
+function noop() {}
+
+function main({ n, subscribers }) {
+  const channel = dc.channel('test');
+  for (let i = 0; i < subscribers; i++) {
+    channel.subscribe(noop);
+  }
+
+  const data = {
+    foo: 'bar'
+  };
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    if (channel.hasSubscribers) {
+      channel.publish(data);
+    }
+  }
+  bench.end(n);
+}

benchmark/diagnostics_channel/subscribe.js

@@ -0,0 +1,19 @@
+'use strict';
+const common = require('../common.js');
+const dc = require('diagnostics_channel');
+
+const bench = common.createBenchmark(main, {
+  n: [1e8],
+});
+
+function noop() {}
+
+function main({ n }) {
+  const channel = dc.channel('channel.0');
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    channel.subscribe(noop);
+  }
+  bench.end(n);
+}

doc/api/diagnostics_channel.md

@@ -0,0 +1,180 @@
+# Diagnostics Channel
+
+<!--introduced_in=REPLACEME-->
+
+> Stability: 1 - Experimental
+
+<!-- source_link=lib/diagnostics_channel.js -->
+
+The `diagnostics_channel` module provides an API to create named channels
+to report arbitrary message data for diagnostics purposes.
+
+It can be accessed using:
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+```
+
+It is intended that a module writer wanting to report diagnostics messages
+will create one or many top-level channels to report messages through.
+Channels may also be acquired at runtime but it is not encouraged
+due to the additional overhead of doing so. Channels may be exported for
+convenience, but as long as the name is known it can be acquired anywhere.
+
+If you intend for your module to produce diagnostics data for others to
+consume it is recommended that you include documentation of what named
+channels are used along with the shape of the message data. Channel names
+should generally include the module name to avoid collisions with data from
+other modules.
+
+## Public API
+
+### Overview
+
+Following is a simple overview of the public API.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+// Get a reusable channel object
+const channel = diagnostics_channel.channel('my-channel');
+
+// Subscribe to the channel
+channel.subscribe((message, name) => {
+  // Received data
+});
+
+// Check if the channel has an active subscriber
+if (channel.hasSubscribers) {
+  // Publish data to the channel
+  channel.publish({
+    some: 'data'
+  });
+}
+```
+
+#### `diagnostics_channel.hasSubscribers(name)`
+
+* `name` {string|symbol} The channel name
+* Returns: {boolean} If there are active subscribers
+
+Check if there are active subscribers to the named channel. This is helpful if
+the message you want to send might be expensive to prepare.
+
+This API is optional but helpful when trying to publish messages from very
+performance-senstive code.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+if (diagnostics_channel.hasSubscribers('my-channel')) {
+  // There are subscribers, prepare and publish message
+}
+```
+
+#### `diagnostics_channel.channel(name)`
+
+* `name` {string|symbol} The channel name
+* Returns: {Channel} The named channel object
+
+This is the primary entry-point for anyone wanting to interact with a named
+channel. It produces a channel object which is optimized to reduce overhead at
+publish time as much as possible.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+```
+
+### Class: `Channel`
+
+The class `Channel` represents an individual named channel within the data
+pipeline. It is use to track subscribers and to publish messages when there
+are subscribers present. It exists as a separate object to avoid channel
+lookups at publish time, enabling very fast publish speeds and allowing
+for heavy use while incurring very minimal cost. Channels are created with
+[`diagnostics_channel.channel(name)`][], constructing a channel directly
+with `new Channel(name)` is not supported.
+
+#### `channel.hasSubscribers`
+
+* Returns: {boolean} If there are active subscribers
+
+Check if there are active subscribers to this channel. This is helpful if
+the message you want to send might be expensive to prepare.
+
+This API is optional but helpful when trying to publish messages from very
+performance-senstive code.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+if (channel.hasSubscribers) {
+  // There are subscribers, prepare and publish message
+}
+```
+
+#### `channel.publish(message)`
+
+* `message` {any} The message to send to the channel subscribers
+
+Publish a message to any subscribers to the channel. This will trigger
+message handlers synchronously so they will execute within the same context.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+channel.publish({
+  some: 'message'
+});
+```
+
+#### `channel.subscribe(onMessage)`
+
+* `onMessage` {Function} The handler to receive channel messages
+  * `message` {any} The message data
+  * `name` {string|symbol} The name of the channel
+
+Register a message handler to subscribe to this channel. This message handler
+will be run synchronously whenever a message is published to the channel. Any
+errors thrown in the message handler will trigger an [`'uncaughtException'`][].
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+channel.subscribe((message, name) => {
+  // Received data
+});
+```
+
+#### `channel.unsubscribe(onMessage)`
+
+* `onMessage` {Function} The previous subscribed handler to remove
+
+Remove a message handler previously registered to this channel with
+[`channel.subscribe(onMessage)`][].
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+function onMessage(message, name) {
+  // Received data
+}
+
+channel.subscribe(onMessage);
+
+channel.unsubscribe(onMessage);
+```
+
+[`diagnostics_channel.channel(name)`]: #diagnostics_channel_diagnostics_channel_channel_name
+[`channel.subscribe(onMessage)`]: #diagnostics_channel_channel_subscribe_onmessage
+[`'uncaughtException'`]: process.md#process_event_uncaughtexception

doc/api/index.md

@@ -23,6 +23,7 @@
 * [Crypto](crypto.md)
 * [Debugger](debugger.md)
 * [Deprecated APIs](deprecations.md)
+* [Diagnostics Channel](diagnostics_channel.md)
 * [DNS](dns.md)
 * [Domain](domain.md)
 * [Errors](errors.md)

lib/diagnostics_channel.js

@@ -0,0 +1,122 @@
+'use strict';
+
+const {
+  ArrayPrototypeIndexOf,
+  ArrayPrototypePush,
+  ArrayPrototypeSplice,
+  ObjectCreate,
+  ObjectGetPrototypeOf,
+  ObjectSetPrototypeOf,
+  SymbolHasInstance,
+  WeakRefPrototypeGet
+} = primordials;
+
+const {
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+  }
+} = require('internal/errors');
+
+const { triggerUncaughtException } = internalBinding('errors');
+
+const { WeakReference } = internalBinding('util');
+
+// TODO(qard): should there be a C++ channel interface?
+class ActiveChannel {
+  subscribe(subscription) {
+    if (typeof subscription !== 'function') {
+      throw new ERR_INVALID_ARG_TYPE('subscription', ['function'],
+                                     subscription);
+    }
+    ArrayPrototypePush(this._subscribers, subscription);
+  }
+
+  unsubscribe(subscription) {
+    const index = ArrayPrototypeIndexOf(this._subscribers, subscription);
+    if (index >= 0) {
+      ArrayPrototypeSplice(this._subscribers, index, 1);
+
+      // When there are no more active subscribers, restore to fast prototype.
+      if (!this._subscribers.length) {
+        // eslint-disable-next-line no-use-before-define
+        ObjectSetPrototypeOf(this, Channel.prototype);
+      }
+    }
+  }
+
+  get hasSubscribers() {
+    return true;
+  }
+
+  publish(data) {
+    for (let i = 0; i < this._subscribers.length; i++) {
+      try {
+        const onMessage = this._subscribers[i];
+        onMessage(data, this.name);
+      } catch (err) {
+        process.nextTick(() => {
+          triggerUncaughtException(err, false);
+        });
+      }
+    }
+  }
+}
+
+class Channel {
+  constructor(name) {
+    this._subscribers = undefined;
+    this.name = name;
+  }
+
+  static [SymbolHasInstance](instance) {
+    const prototype = ObjectGetPrototypeOf(instance);
+    return prototype === Channel.prototype ||
+           prototype === ActiveChannel.prototype;
+  }
+
+  subscribe(subscription) {
+    ObjectSetPrototypeOf(this, ActiveChannel.prototype);
+    this._subscribers = [];
+    this.subscribe(subscription);
+  }
+
+  get hasSubscribers() {
+    return false;
+  }
+
+  publish() {}
+}
+
+const channels = ObjectCreate(null);
+
+function channel(name) {
+  let channel;
+  const ref = channels[name];
+  if (ref) channel = ref.get();
+  if (channel) return channel;
+
+  if (typeof name !== 'string' && typeof name !== 'symbol') {
+    throw new ERR_INVALID_ARG_TYPE('channel', ['string', 'symbol'], name);
+  }
+
+  channel = new Channel(name);
+  channels[name] = new WeakReference(channel);
+  return channel;
+}
+
+function hasSubscribers(name) {
+  let channel;
+  const ref = channels[name];
+  if (ref) channel = WeakRefPrototypeGet(ref);
+  if (!channel) {
+    return false;
+  }
+
+  return channel.hasSubscribers;
+}
+
+module.exports = {
+  channel,
+  hasSubscribers,
+  Channel
+};

node.gyp

@@ -51,6 +51,7 @@
       'lib/constants.js',
       'lib/crypto.js',
       'lib/cluster.js',
+      'lib/diagnostics_channel.js',
       'lib/dgram.js',
       'lib/dns.js',
       'lib/dns/promises.js',