Отслеживание асинхронного контекста

Введение

Эти классы используются для связывания состояния и его передачи через обратные вызовы и цепочки промисов. Они позволяют хранить данные в течение всего времени жизни веб-запроса или любого другого асинхронного периода. Это похоже на локальное хранилище потока в других языках.

Классы AsyncLocalStorage и AsyncResource входят в состав модуля node:async_hooks:

import { AsyncLocalStorage, AsyncResource } from 'node:async_hooks';

const { AsyncLocalStorage, AsyncResource } = require('node:async_hooks');

Класс: AsyncLocalStorage

Этот класс создает хранилища, состояние которых остается согласованным при выполнении асинхронных операций.

Хотя можно создать собственную реализацию на основе модуля node:async_hooks, рекомендуется использовать AsyncLocalStorage — это производительная и безопасная для памяти реализация со значительными оптимизациями, которые неочевидно реализовать самостоятельно.

В следующем примере AsyncLocalStorage используется для создания простого средства ведения журнала, которое назначает идентификаторы входящим HTTP-запросам и добавляет их в сообщения, записываемые во время обработки каждого запроса.

import http from 'node:http';
import { AsyncLocalStorage } from 'node:async_hooks';

const asyncLocalStorage = new AsyncLocalStorage();

function logWithId(msg) {
  const id = asyncLocalStorage.getStore();
  console.log(`${id !== undefined ? id : '-'}:`, msg);
}

let idSeq = 0;
http.createServer((req, res) => {
  asyncLocalStorage.run(idSeq++, () => {
    logWithId('start');
    // Imagine any chain of async operations here
    setImmediate(() => {
      logWithId('finish');
      res.end();
    });
  });
}).listen(8080);

http.get('http://localhost:8080');
http.get('http://localhost:8080');
// Prints:
//   0: start
//   0: finish
//   1: start
//   1: finish

const http = require('node:http');
const { AsyncLocalStorage } = require('node:async_hooks');

const asyncLocalStorage = new AsyncLocalStorage();

function logWithId(msg) {
  const id = asyncLocalStorage.getStore();
  console.log(`${id !== undefined ? id : '-'}:`, msg);
}

let idSeq = 0;
http.createServer((req, res) => {
  asyncLocalStorage.run(idSeq++, () => {
    logWithId('start');
    // Imagine any chain of async operations here
    setImmediate(() => {
      logWithId('finish');
      res.end();
    });
  });
}).listen(8080);

http.get('http://localhost:8080');
http.get('http://localhost:8080');
// Prints:
//   0: start
//   0: finish
//   1: start
//   1: finish

Каждый экземпляр AsyncLocalStorage поддерживает независимый контекст хранилища. Несколько экземпляров могут безопасно существовать одновременно без риска взаимного влияния их данных.

new AsyncLocalStorage([options])

• options <Object>
  • defaultValue <any> Значение по умолчанию, используемое, если хранилище не задано.
  • name <string> Имя для значения AsyncLocalStorage.

Создает новый экземпляр AsyncLocalStorage. Хранилище доступно только внутри вызова run() или после вызова enterWith().

Статический метод: AsyncLocalStorage.bind(fn)

• fn <Function> Функция, которую нужно связать с текущим контекстом выполнения.
• Возвращает: <Function> Новая функция, которая вызывает fn в захваченном контексте выполнения.

Связывает указанную функцию с текущим контекстом выполнения.

Статический метод: AsyncLocalStorage.snapshot()

• Возвращает: <Function> Новая функция с сигнатурой (fn: (...args) : R, ...args) : R.

Захватывает текущий контекст выполнения и возвращает функцию, принимающую функцию в качестве аргумента. При каждом вызове возвращенная функция вызывает переданную ей функцию в захваченном контексте.

const asyncLocalStorage = new AsyncLocalStorage();
const runInAsyncScope = asyncLocalStorage.run(123, () => AsyncLocalStorage.snapshot());
const result = asyncLocalStorage.run(321, () => runInAsyncScope(() => asyncLocalStorage.getStore()));
console.log(result);  // returns 123
jscopy

AsyncLocalStorage.snapshot() может заменить использование AsyncResource для простого отслеживания асинхронного контекста, например:

class Foo {
  #runInAsyncScope = AsyncLocalStorage.snapshot();

  get() { return this.#runInAsyncScope(() => asyncLocalStorage.getStore()); }
}

const foo = asyncLocalStorage.run(123, () => new Foo());
console.log(asyncLocalStorage.run(321, () => foo.get())); // returns 123
jscopy

asyncLocalStorage.disable()

Отключает экземпляр AsyncLocalStorage. Все последующие вызовы asyncLocalStorage.getStore() будут возвращать undefined, пока снова не будет вызван asyncLocalStorage.run() или asyncLocalStorage.enterWith().

При вызове asyncLocalStorage.disable() будут завершены все текущие контексты, связанные с экземпляром.

Перед сборкой мусора для asyncLocalStorage необходимо вызвать asyncLocalStorage.disable(). Это не относится к хранилищам, предоставленным asyncLocalStorage, поскольку эти объекты собираются сборщиком мусора вместе с соответствующими асинхронными ресурсами.

Используйте этот метод, когда asyncLocalStorage больше не используется в текущем процессе.

asyncLocalStorage.getStore()

• Возвращает: <any>

Возвращает текущее хранилище. Если этот метод вызван вне асинхронного контекста, созданного вызовом asyncLocalStorage.run() или asyncLocalStorage.enterWith(), он возвращает undefined.

asyncLocalStorage.enterWith(store)

• store <any>

Переходит в контекст на оставшуюся часть текущего синхронного выполнения, а затем сохраняет хранилище во всех последующих асинхронных вызовах.

Пример:

const store = { id: 1 };
// Replaces previous store with the given store object
asyncLocalStorage.enterWith(store);
asyncLocalStorage.getStore(); // Returns the store object
someAsyncOperation(() => {
  asyncLocalStorage.getStore(); // Returns the same object
});
jscopy

Этот переход продолжается в течение всего синхронного выполнения. Это означает, что если, например, контекст был установлен внутри обработчика события, последующие обработчики событий также будут выполняться в этом контексте, если только они явно не привязаны к другому контексту с помощью AsyncResource. Поэтому следует отдавать предпочтение run(), а не enterWith(), если нет веских причин использовать последний метод.

const store = { id: 1 };

emitter.on('my-event', () => {
  asyncLocalStorage.enterWith(store);
});
emitter.on('my-event', () => {
  asyncLocalStorage.getStore(); // Returns the same object
});

asyncLocalStorage.getStore(); // Returns undefined
emitter.emit('my-event');
asyncLocalStorage.getStore(); // Returns the same object
jscopy

asyncLocalStorage.name

• Тип: <string>

Имя экземпляра AsyncLocalStorage, если оно задано.

asyncLocalStorage.run(store, callback[, ...args])

• store <any>
• callback <Function>
• ...args <any>

Синхронно выполняет функцию в контексте и возвращает ее результат. Хранилище недоступно за пределами функции обратного вызова. Оно доступно для любых асинхронных операций, созданных внутри обратного вызова.

Необязательные args передаются функции обратного вызова.

Если функция обратного вызова вызывает ошибку, run() также выбрасывает ее. Этот вызов не влияет на трассировку стека, а контекст завершается.

Пример:

const store = { id: 2 };
try {
  asyncLocalStorage.run(store, () => {
    asyncLocalStorage.getStore(); // Returns the store object
    setTimeout(() => {
      asyncLocalStorage.getStore(); // Returns the store object
    }, 200);
    throw new Error();
  });
} catch (e) {
  asyncLocalStorage.getStore(); // Returns undefined
  // The error will be caught here
}
jscopy

asyncLocalStorage.exit(callback[, ...args])

• callback <Function>
• ...args <any>

Синхронно выполняет функцию вне контекста и возвращает ее результат. Хранилище недоступно внутри функции обратного вызова и асинхронных операций, созданных в ней. Любой вызов getStore() внутри функции обратного вызова всегда будет возвращать undefined.

Необязательные args передаются функции обратного вызова.

Если функция обратного вызова вызывает ошибку, exit() также выбрасывает ее. Этот вызов не влияет на трассировку стека, а контекст восстанавливается.

Пример:

// Within a call to run
try {
  asyncLocalStorage.getStore(); // Returns the store object or value
  asyncLocalStorage.exit(() => {
    asyncLocalStorage.getStore(); // Returns undefined
    throw new Error();
  });
} catch (e) {
  asyncLocalStorage.getStore(); // Returns the same object or value
  // The error will be caught here
}
jscopy

asyncLocalStorage.withScope(store)

• store <any>
• Возвращает: {RunScope}

Создает освобождаемую область, которая устанавливает заданное хранилище и автоматически восстанавливает предыдущее значение хранилища при освобождении области. Этот метод предназначен для работы с явным управлением ресурсами в JavaScript (синтаксис using).

Пример:

import { AsyncLocalStorage } from 'node:async_hooks';

const asyncLocalStorage = new AsyncLocalStorage();

{
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // Prints: my-store
}

console.log(asyncLocalStorage.getStore()); // Prints: undefined

const { AsyncLocalStorage } = require('node:async_hooks');

const asyncLocalStorage = new AsyncLocalStorage();

{
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // Prints: my-store
}

console.log(asyncLocalStorage.getStore()); // Prints: undefined

Метод withScope() особенно полезен для управления контекстом в синхронном коде, когда нужно убедиться, что при выходе из блока восстанавливается предыдущее значение хранилища, даже если возникает ошибка.

import { AsyncLocalStorage } from 'node:async_hooks';

const asyncLocalStorage = new AsyncLocalStorage();

try {
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // Prints: my-store
  throw new Error('test');
} catch (e) {
  // Store is automatically restored even after error
  console.log(asyncLocalStorage.getStore()); // Prints: undefined
}

const { AsyncLocalStorage } = require('node:async_hooks');

const asyncLocalStorage = new AsyncLocalStorage();

try {
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // Prints: my-store
  throw new Error('test');
} catch (e) {
  // Store is automatically restored even after error
  console.log(asyncLocalStorage.getStore()); // Prints: undefined
}

Важно: При использовании withScope() в асинхронных функциях до первого await учитывайте, что изменение области повлияет на контекст вызывающего кода. Синхронная часть асинхронной функции (до первого await) выполняется сразу при вызове. Когда функция доходит до первого await, она возвращает промис вызывающему коду. В этот момент изменение области становится видимым в контексте вызывающего кода и сохраняется в последующем синхронном коде, пока значение области не будет изменено. Для асинхронных операций предпочтительно использовать run(), который корректно изолирует контекст при переходах между асинхронными операциями.

import { AsyncLocalStorage } from 'node:async_hooks';

const asyncLocalStorage = new AsyncLocalStorage();

async function example() {
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // Prints: my-store
  await someAsyncOperation(); // Function pauses here and returns promise
  console.log(asyncLocalStorage.getStore()); // Prints: my-store
}

// Calling without await
example(); // Synchronous portion runs, then pauses at first await
// After the promise is returned, the scope 'my-store' is now active in caller!
console.log(asyncLocalStorage.getStore()); // Prints: my-store (unexpected!)
mjscopy

Использование с async/await

Если внутри асинхронной функции только один вызов await должен выполняться в контексте, следует использовать следующий шаблон:

async function fn() {
  await asyncLocalStorage.run(new Map(), () => {
    asyncLocalStorage.getStore().set('key', value);
    return foo(); // The return value of foo will be awaited
  });
}
jscopy

В этом примере хранилище доступно только в функции обратного вызова и функциях, вызываемых foo. За пределами run вызов getStore вернет undefined.

Устранение неполадок: потеря контекста

В большинстве случаев AsyncLocalStorage работает без проблем. В редких ситуациях текущее хранилище теряется в одной из асинхронных операций.

Если ваш код использует обратные вызовы, достаточно преобразовать его в промисы с помощью util.promisify(), чтобы он начал работать с нативными промисами.

Если необходимо использовать API на основе обратных вызовов или ваш код предполагает собственную реализацию thenable, используйте класс AsyncResource, чтобы связать асинхронную операцию с правильным контекстом выполнения. Найдите вызов функции, ответственный за потерю контекста, записывая содержимое asyncLocalStorage.getStore() после вызовов, которые, по вашему мнению, вызывают потерю. Если в журнале появляется undefined, вероятно, за потерю контекста отвечает последний вызванный обратный вызов.

Класс: RunScope

Освобождаемая область, возвращаемая методом asyncLocalStorage.withScope(), которая автоматически восстанавливает предыдущее значение хранилища при освобождении. Этот класс реализует протокол явного управления ресурсами и предназначен для работы с синтаксисом using в JavaScript.

При выходе из блока using область автоматически восстанавливает предыдущее значение хранилища — как при обычном завершении, так и при возникновении ошибки.

scope.dispose()

Явно завершает область и восстанавливает предыдущее значение хранилища. Этот метод идемпотентен: его многократный вызов дает тот же результат, что и однократный.

Метод [Symbol.dispose]() делегирует выполнение dispose().

Если withScope() вызывается без ключевого слова using, для восстановления предыдущего значения хранилища необходимо вручную вызвать dispose(). Если не вызвать dispose(), значение хранилища сохранится до конца текущего контекста выполнения:

import { AsyncLocalStorage } from 'node:async_hooks';

const storage = new AsyncLocalStorage();

// Without using, the scope must be disposed manually
const scope = storage.withScope('my-store');
// storage.getStore() === 'my-store' here

scope.dispose(); // Restore previous value
// storage.getStore() === undefined here

const { AsyncLocalStorage } = require('node:async_hooks');

const storage = new AsyncLocalStorage();

// Without using, the scope must be disposed manually
const scope = storage.withScope('my-store');
// storage.getStore() === 'my-store' here

scope.dispose(); // Restore previous value
// storage.getStore() === undefined here

Класс: AsyncResource

Класс AsyncResource предназначен для расширения асинхронными ресурсами среды выполнения. С его помощью пользователи могут легко инициировать события жизненного цикла собственных ресурсов.

Хук init срабатывает при создании AsyncResource.

Ниже приводится обзор API AsyncResource.

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])

• type <string> Тип асинхронного события.
• options <Object>
  • triggerAsyncId <number> Идентификатор контекста выполнения, в котором было создано это асинхронное событие. По умолчанию: executionAsyncId().
  • requireManualDestroy <boolean> Если задано значение true, вызов emitDestroy при сборке мусора для объекта отключается. Обычно этот параметр не нужно задавать (даже если emitDestroy вызывается вручную), кроме случаев, когда получено значение asyncId ресурса, а с его помощью вызывается чувствительный к безопасности API emitDestroy. Если задано значение false, вызов emitDestroy при сборке мусора выполняется только при наличии хотя бы одного активного хука destroy. По умолчанию: false.

Пример использования:

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();
  }
}
jscopy

Статический метод: AsyncResource.bind(fn[, type[, thisArg]])

• fn <Function> Функция, которую нужно связать с текущим контекстом выполнения.
• type <string> Необязательное имя, связываемое с базовым AsyncResource.
• thisArg <any>

Связывает указанную функцию с текущим контекстом выполнения.

asyncResource.bind(fn[, thisArg])

• fn <Function> Функция, которую нужно связать с текущим AsyncResource.
• thisArg <any>

Связывает указанную функцию с областью действия этого AsyncResource.

asyncResource.runInAsyncScope(fn[, thisArg, ...args])

• fn <Function> Функция, вызываемая в контексте выполнения этого асинхронного ресурса.
• thisArg <any> Получатель, используемый при вызове функции.
• ...args <any> Необязательные аргументы, передаваемые функции.

Вызывает указанную функцию с заданными аргументами в контексте выполнения асинхронного ресурса. При этом устанавливается контекст, инициируются хуки AsyncHooks перед обратными вызовами, вызывается функция, инициируются хуки AsyncHooks после обратных вызовов, а затем восстанавливается исходный контекст выполнения.

asyncResource.emitDestroy()

• Возвращает: <AsyncResource> Ссылка на asyncResource.

Вызывает все хуки destroy. Этот метод следует вызывать только один раз. При повторном вызове будет выброшена ошибка. Его необходимо вызвать вручную. Если ресурс будет собран сборщиком мусора, хуки destroy вызваны не будут.

asyncResource.asyncId()

• Возвращает: <number> Уникальный идентификатор asyncId, назначенный ресурсу.

asyncResource.triggerAsyncId()

• Возвращает: <number> Тот же triggerAsyncId, который передается конструктору AsyncResource.

Использование AsyncResource для пула потоков Worker

В следующем примере показано, как использовать класс AsyncResource для правильного отслеживания асинхронных операций в пуле Worker. Аналогичный подход можно использовать для других пулов ресурсов, например пулов подключений к базе данных.

Предположим, что задача заключается в сложении двух чисел. Используется файл с именем task_processor.js, содержащий следующий код:

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);
});

Пул Worker для этой задачи можно организовать следующим образом:

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;

Без явного отслеживания, добавленного объектами WorkerPoolTaskInfo, можно было бы предположить, что обратные вызовы связаны с отдельными объектами Worker. Однако создание объектов Worker не связано с созданием задач и не позволяет определить, когда задачи были запланированы.

Этот пул можно использовать следующим образом:

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();
  });
}

Интеграция AsyncResource с EventEmitter

Обработчики событий, вызванные объектом EventEmitter, могут выполняться в контексте, отличном от того, который был активен при вызове eventEmitter.on().

В следующем примере показано, как использовать класс AsyncResource, чтобы правильно связать обработчик события с нужным контекстом выполнения. Такой же подход можно применить к Stream или аналогичному классу, управляемому событиями.

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);