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 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.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
} 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 перед выполнением обратных вызовов, вызывается функция, запускаются хуки 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. Однако создание объектов Workers не связано с созданием задач и не позволяет определить, когда задачи были запланированы.

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

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