Инструменты конкурентности · очереди, семафоры, гонки

Прошлые два урока были про модель: чем конкурентность отличается от параллелизма и по каким осям различаются контракты. Этот урок прикладной. Берём конкретные инструменты, которыми управляют конкурентностью в Node: примитивы синхронизации, очереди и пулы, адаптеры между моделями, потоки и реактивные стримы. И главное, разбираем гонки, ради которых половина этих инструментов и существует.

1. Гонка там, где её не ждут

Начнём с проблемы. В Node один поток JavaScript, поэтому кажется, что гонок не бывает. Это самое опасное заблуждение в асинхронном коде.

let balance = 100;

async function withdraw(amount: number): Promise<void> {
  const current = balance; // 1. прочитали баланс
  await persistAudit(amount); // 2. УСТУПИЛИ управление event loop
  balance = current - amount; // 3. записали по старому current
}

// два списания почти одновременно
await Promise.all([withdraw(100), withdraw(1)]);
// ожидали 100 - 101 = -1, получили balance = 99 или 0, деньги потерялись

Между шагами 1 и 3 стоит await. На нём функция уступает управление, и второй withdraw успевает прочитать тот же current = 100. Оба пишут по устаревшему снимку. Это race condition, и она настоящая, несмотря на один поток.

Лечение бывает трёх сортов:

• убрать разделяемое состояние (сделать владение exclusive, см. урок 09);
• сделать операцию атомарной, без await в середине критической секции;
• защитить критическую секцию примитивом синхронизации.

Первый способ лучший, но не всегда возможен. Поэтому познакомимся с примитивами.

2. Mutex и Semaphore на промисах

Mutex пропускает в защищённый участок ровно одного. Остальные ждут. На промисах он пишется в десяток строк: очередь ожидающих и флаг занятости.

class Mutex {
  private locked = false;
  private waiters: Array<() => void> = [];

  async acquire(): Promise<void> {
    if (!this.locked) {
      this.locked = true;
      return;
    }
    await new Promise<void>((resolve) => this.waiters.push(resolve));
  }

  release(): void {
    const next = this.waiters.shift();
    if (next) next();
    else this.locked = false;
  }
}

Теперь критическая секция withdraw защищена: пока один держит mutex, второй ждёт на acquire, и снимок баланса не устаревает.

Semaphore, это обобщение: пропускает не одного, а до N. Mutex это семафор с лимитом 1. Семафор это ровно тот инструмент, которым ограничивают параллелизм: “не больше пяти запросов к этому API разом”. Покрути лимит на демо ниже и посмотри, как семафор держит активных под потолком, а остальных в очереди.

Лимит 1, это последовательность. Лимит, равный числу задач, это полный параллелизм без тормоза. Между ними весь спектр управляемой нагрузки.

Третий классический примитив, RWLock, различает два вида работы с общим состоянием. Читатели ничего не ломают, их можно пускать толпой; писатель меняет состояние, ему нужна эксклюзивность. Кэш, который читают сотни запросов и раз в минуту обновляет один, под мьютексом выстроил бы читателей в очередь зря. Поверх уже написанного RWLock собирается просто: счётчик активных читателей и mutex для писателей, первый читатель захватывает его, последний отпускает. Один нюанс надо решить явно: при постоянном потоке читателей писатель может ждать вечно (writer starvation), поэтому зрелые реализации перестают пускать новых читателей, как только писатель встал в очередь. В однопоточном JS этот примитив нужен редко, но в Go (sync.RWMutex), Rust (RwLock) и в блокировках своей СУБД ты встретишь его под этим же именем.

3. Очереди, пулы, коллекторы

Из семафора вырастают три родственных инструмента.

AsyncQueue, это буфер между быстрым источником и медленным потребителем. Потребитель тянет через await queue.next(), а если пусто, ждёт. Так push-поток превращается в pull, не теряя элементов.

Пул, это N исполнителей, разбирающих задачи из общей очереди. Освободился, взял следующую. Так устроен pLimit, так устроен и worker pool для CPU-задач. Демо ниже показывает пул из N воркеров и очередь перед ним: подвигай число воркеров и скорость продюсера, увидишь, как заполняется буфер и где появляется блокировка.

producer

8

/ s

queue

0/16

0 completed0 dropped0 ms avg0 tpsrunning

Подними producer выше суммарной мощности воркеров, и очередь наполняется до maxQueue, после чего новые задачи падают в dropped. Без лимита очередь росла бы до OOM, с лимитом сервис быстро ловит overload, и оркестратор поднимает реплику. Добавь воркеров, очередь снова пустеет, латентность падает.

Коллектор, это обратная задача: собрать значения из многих источников, пока не выполнится условие (набралось N, истёк таймаут, выполнился предикат), и отдать пачкой. Полезно для сбора запросов пачками и для дебаунса событий.

Идея ограничения параллелизма помещается в крошечную функцию:

function pLimit(concurrency: number) {
  let active = 0;
  const queue: Array<() => void> = [];

  const next = () => {
    active -= 1;
    queue.shift()?.();
  };

  return async function run<T>(task: () => Promise<T>): Promise<T> {
    if (active >= concurrency) {
      await new Promise<void>((resolve) => queue.push(resolve));
    }
    active += 1;
    try {
      return await task();
    } finally {
      next();
    }
  };
}

Есть ещё один инструмент из той же семьи, который в проде экономит больше всего: склейка одинаковых запросов, пока предыдущий в полёте. Ниже редактор с такой задачей, тесты считают походы в источник. Гоняются прямо в браузере, ставить ничего не нужно.