new 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 copy

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 copy

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

Этот переход будет действовать в течение всего синхронного выполнения. Это означает, что, например, если контекст был активирован в обработчике события, последующие обработчики событий также будут выполняться в этом контексте, если только они явно не связаны с другим контекстом с помощью 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 copy

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
} copy

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
} copy

Использование с 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
  });
} copy

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

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

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

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

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

new AsyncResource(type[, options])

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

Статический метод: 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 before, вызовет функцию, вызовет обратные вызовы AsyncHooks after, а затем восстановит исходный контекст выполнения.

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