AsyncResource
History
The class AsyncResource is designed to be extended by the embedder's async
resources. Using this, users can easily trigger the lifetime events of their
own resources.
The init hook will trigger when an AsyncResource is instantiated.
The following is an overview of the AsyncResource API.
import { AsyncResource, executionAsyncId } from 'node:async_hooks'; // AsyncResource() is meant to be extended. Instantiating a // new AsyncResource() also triggers init. If triggerAsyncId is omitted then // async_hook.executionAsyncId() is used. const asyncResource = new AsyncResource( type, { triggerAsyncId: executionAsyncId(), requireManualDestroy: false }, ); // Run a function in the execution context of the resource. This will // * establish the context of the resource // * trigger the AsyncHooks before callbacks // * call the provided function `fn` with the supplied arguments // * trigger the AsyncHooks after callbacks // * restore the original execution context asyncResource.runInAsyncScope(fn, thisArg, ...args); // Call AsyncHooks destroy callbacks. asyncResource.emitDestroy(); // Return the unique ID assigned to the AsyncResource instance. asyncResource.asyncId(); // Return the trigger ID for the AsyncResource instance. asyncResource.triggerAsyncId();
const { AsyncResource, executionAsyncId } = require('node:async_hooks'); // AsyncResource() is meant to be extended. Instantiating a // new AsyncResource() also triggers init. If triggerAsyncId is omitted then // async_hook.executionAsyncId() is used. const asyncResource = new AsyncResource( type, { triggerAsyncId: executionAsyncId(), requireManualDestroy: false }, ); // Run a function in the execution context of the resource. This will // * establish the context of the resource // * trigger the AsyncHooks before callbacks // * call the provided function `fn` with the supplied arguments // * trigger the AsyncHooks after callbacks // * restore the original execution context asyncResource.runInAsyncScope(fn, thisArg, ...args); // Call AsyncHooks destroy callbacks. asyncResource.emitDestroy(); // Return the unique ID assigned to the AsyncResource instance. asyncResource.asyncId(); // Return the trigger ID for the AsyncResource instance. asyncResource.triggerAsyncId();
new AsyncResource(type, options?): AsyncResource
stringObjectnumberexecutionAsyncId().booleantrue, disables emitDestroy
when the object is garbage collected. This usually does not need to be set
(even if emitDestroy is called manually), unless the resource's asyncId
is retrieved and the sensitive API's emitDestroy is called with it.
When set to false, the emitDestroy call on garbage collection
will only take place if there is at least one active destroy hook.
Default: false.Example usage:
class DBQuery extends AsyncResource { constructor(db) { super('DBQuery'); this.db = db; } getInfo(query, callback) { this.db.get(query, (err, data) => { this.runInAsyncScope(callback, null, err, data); }); } close() { this.db = null; this.emitDestroy(); } }
AsyncResource.bind(fn, type?, thisArg?): void
Binds the given function to the current execution context.
asyncResource.bind(fn, thisArg?): void
Binds the given function to execute to this AsyncResource's scope.
asyncResource.runInAsyncScope(fn, thisArg?, ...args?): void
Call the provided function with the provided arguments in the execution context of the async resource. This will establish the context, trigger the AsyncHooks before callbacks, call the function, trigger the AsyncHooks after callbacks, and then restore the original execution context.
asyncResource.emitDestroy(): AsyncResource
AsyncResourceasyncResource.Call all destroy hooks. This should only ever be called once. An error will
be thrown if it is called more than once. This must be manually called. If
the resource is left to be collected by the GC then the destroy hooks will
never be called.
asyncResource.asyncId(): number
numberasyncId assigned to the resource.asyncResource.triggerAsyncId(): number
numbertriggerAsyncId that is passed to the
AsyncResource constructor.The following example shows how to use the AsyncResource class to properly
provide async tracking for a Worker pool. Other resource pools, such as
database connection pools, can follow a similar model.
Assuming that the task is adding two numbers, using a file named
task_processor.js with the following content:
import { parentPort } from 'node:worker_threads'; parentPort.on('message', (task) => { parentPort.postMessage(task.a + task.b); });
const { parentPort } = require('node:worker_threads'); parentPort.on('message', (task) => { parentPort.postMessage(task.a + task.b); });
a Worker pool around it could use the following structure:
import { AsyncResource } from 'node:async_hooks'; import { EventEmitter } from 'node:events'; import { Worker } from 'node:worker_threads'; const kTaskInfo = Symbol('kTaskInfo'); const kWorkerFreedEvent = Symbol('kWorkerFreedEvent'); class WorkerPoolTaskInfo extends AsyncResource { constructor(callback) { super('WorkerPoolTaskInfo'); this.callback = callback; } done(err, result) { this.runInAsyncScope(this.callback, null, err, result); this.emitDestroy(); // `TaskInfo`s are used only once. } } export default class WorkerPool extends EventEmitter { constructor(numThreads) { super(); this.numThreads = numThreads; this.workers = []; this.freeWorkers = []; this.tasks = []; for (let i = 0; i < numThreads; i++) this.addNewWorker(); // Any time the kWorkerFreedEvent is emitted, dispatch // the next task pending in the queue, if any. this.on(kWorkerFreedEvent, () => { if (this.tasks.length > 0) { const { task, callback } = this.tasks.shift(); this.runTask(task, callback); } }); } addNewWorker() { const worker = new Worker(new URL('task_processor.js', import.meta.url)); worker.on('message', (result) => { // In case of success: Call the callback that was passed to `runTask`, // remove the `TaskInfo` associated with the Worker, and mark it as free // again. worker[kTaskInfo].done(null, result); worker[kTaskInfo] = null; this.freeWorkers.push(worker); this.emit(kWorkerFreedEvent); }); worker.on('error', (err) => { // In case of an uncaught exception: Call the callback that was passed to // `runTask` with the error. if (worker[kTaskInfo]) worker[kTaskInfo].done(err, null); else this.emit('error', err); // Remove the worker from the list and start a new Worker to replace the // current one. this.workers.splice(this.workers.indexOf(worker), 1); this.addNewWorker(); }); this.workers.push(worker); this.freeWorkers.push(worker); this.emit(kWorkerFreedEvent); } runTask(task, callback) { if (this.freeWorkers.length === 0) { // No free threads, wait until a worker thread becomes free. this.tasks.push({ task, callback }); return; } const worker = this.freeWorkers.pop(); worker[kTaskInfo] = new WorkerPoolTaskInfo(callback); worker.postMessage(task); } close() { for (const worker of this.workers) worker.terminate(); } }
const { AsyncResource } = require('node:async_hooks'); const { EventEmitter } = require('node:events'); const path = require('node:path'); const { Worker } = require('node:worker_threads'); const kTaskInfo = Symbol('kTaskInfo'); const kWorkerFreedEvent = Symbol('kWorkerFreedEvent'); class WorkerPoolTaskInfo extends AsyncResource { constructor(callback) { super('WorkerPoolTaskInfo'); this.callback = callback; } done(err, result) { this.runInAsyncScope(this.callback, null, err, result); this.emitDestroy(); // `TaskInfo`s are used only once. } } class WorkerPool extends EventEmitter { constructor(numThreads) { super(); this.numThreads = numThreads; this.workers = []; this.freeWorkers = []; this.tasks = []; for (let i = 0; i < numThreads; i++) this.addNewWorker(); // Any time the kWorkerFreedEvent is emitted, dispatch // the next task pending in the queue, if any. this.on(kWorkerFreedEvent, () => { if (this.tasks.length > 0) { const { task, callback } = this.tasks.shift(); this.runTask(task, callback); } }); } addNewWorker() { const worker = new Worker(path.resolve(__dirname, 'task_processor.js')); worker.on('message', (result) => { // In case of success: Call the callback that was passed to `runTask`, // remove the `TaskInfo` associated with the Worker, and mark it as free // again. worker[kTaskInfo].done(null, result); worker[kTaskInfo] = null; this.freeWorkers.push(worker); this.emit(kWorkerFreedEvent); }); worker.on('error', (err) => { // In case of an uncaught exception: Call the callback that was passed to // `runTask` with the error. if (worker[kTaskInfo]) worker[kTaskInfo].done(err, null); else this.emit('error', err); // Remove the worker from the list and start a new Worker to replace the // current one. this.workers.splice(this.workers.indexOf(worker), 1); this.addNewWorker(); }); this.workers.push(worker); this.freeWorkers.push(worker); this.emit(kWorkerFreedEvent); } runTask(task, callback) { if (this.freeWorkers.length === 0) { // No free threads, wait until a worker thread becomes free. this.tasks.push({ task, callback }); return; } const worker = this.freeWorkers.pop(); worker[kTaskInfo] = new WorkerPoolTaskInfo(callback); worker.postMessage(task); } close() { for (const worker of this.workers) worker.terminate(); } } module.exports = WorkerPool;
Without the explicit tracking added by the WorkerPoolTaskInfo objects,
it would appear that the callbacks are associated with the individual Worker
objects. However, the creation of the Workers is not associated with the
creation of the tasks and does not provide information about when tasks
were scheduled.
This pool could be used as follows:
import WorkerPool from './worker_pool.js'; import os from 'node:os'; const pool = new WorkerPool(os.availableParallelism()); let finished = 0; for (let i = 0; i < 10; i++) { pool.runTask({ a: 42, b: 100 }, (err, result) => { console.log(i, err, result); if (++finished === 10) pool.close(); }); }
const WorkerPool = require('./worker_pool.js'); const os = require('node:os'); const pool = new WorkerPool(os.availableParallelism()); let finished = 0; for (let i = 0; i < 10; i++) { pool.runTask({ a: 42, b: 100 }, (err, result) => { console.log(i, err, result); if (++finished === 10) pool.close(); }); }
Event listeners triggered by an EventEmitter may be run in a different
execution context than the one that was active when eventEmitter.on() was
called.
The following example shows how to use the AsyncResource class to properly
associate an event listener with the correct execution context. The same
approach can be applied to a Stream or a similar event-driven class.
import { createServer } from 'node:http'; import { AsyncResource, executionAsyncId } from 'node:async_hooks'; const server = createServer((req, res) => { req.on('close', AsyncResource.bind(() => { // Execution context is bound to the current outer scope. })); req.on('close', () => { // Execution context is bound to the scope that caused 'close' to emit. }); res.end(); }).listen(3000);
const { createServer } = require('node:http'); const { AsyncResource, executionAsyncId } = require('node:async_hooks'); const server = createServer((req, res) => { req.on('close', AsyncResource.bind(() => { // Execution context is bound to the current outer scope. })); req.on('close', () => { // Execution context is bound to the scope that caused 'close' to emit. }); res.end(); }).listen(3000);