Очереди: Queue и PubSub

До сих пор у каждого кадра был свой файбер от прихода до ответа. Так можно, пока кадров немного. Когда их тысячи в секунду, а башня отвечает сотне, лучше развести две роли: производитель кладёт кадры в ящик, потребитель берёт и отвечает. Между ними очередь, и всё интересное происходит, когда она полна.

Queue

Queue.bounded(2) заводит очередь на два места. Queue.offer(queue, item) кладёт, Queue.take(queue) берёт, а если пусто, ждёт. Главное в том, что делает offer, когда мест нет:

[
  '0 мс положил a',
  '0 мс положил b',
  '200 мс положил c',
  '200 мс взял a',
  '400 мс положил d',
  '400 мс взял b',
  '600 мс взял c',
  '800 мс взял d'
]

Приёмник положил два кадра сразу и на третьем остановился: мест нет. Он ждал, пока потребитель не взял a, и только тогда положил c. Это противодавление: медленный конец сообщает быстрому «подожди», и работа не копится без предела. Подвигай скорости производителя и потребителя и размер очереди:

producer

flowing

buffer

0/8

consumer

0

0 produced0 consumed0 paused0 tps

Подними producer выше consumer, буфер заполняется. Когда длина достигает hwm, producer переходит в paused: это и есть backpressure. Опустится буфер ниже половины hwm, producer возвращается в flowing. Это упрощённая модель: в Node hwm, это порог для возврата false у write(), в Web Streams похожая логика реализована через возврат промиса из pull.

Иногда ждать нельзя: приёмник висит на проводе Магистрали, и если он остановится, Магистраль решит, что башня упала. Тогда очередь отбрасывает лишнее, а не ждёт. Два способа:

{
  accepted: [ true, true, false, false, false ],
  dropping: [ 1, 2 ],
  sliding: [ 4, 5 ]
}

Queue.dropping(2) отбрасывает новое: в полную очередь offer не кладёт и отвечает false. Остаются первые. Queue.sliding(2) отбрасывает старое: новое кладётся, а самое старое выпадает. Остаются последние. Для статуса соседа, где важен только последний, подходит скользящая. Для кадров, где важен каждый, ни та, ни другая: им нужна ограниченная с ожиданием или отказ с честной ошибкой.

Приёмник башни собран так:

export const viaInbox = <B, R>(
  lines: readonly string[],
  capacity: number,
  handle: (line: string) => Effect.Effect<B, never, R>,
): Effect.Effect<B[], never, R> =>
  Effect.gen(function* () {
    const inbox = yield* Queue.bounded<string>(capacity);
    yield* Effect.forkChild(Effect.forEach(lines, (line) => Queue.offer(inbox, line), { discard: true }));
    return yield* Effect.forEach(lines, () => Effect.flatMap(Queue.take(inbox), handle));
  });

[ 'A', 'B', 'C' ]

Производитель в фоне, forkChild, кладёт строки и ждёт, когда места нет. Потребитель берёт по одной и обрабатывает. Даже очередь на одно место пропускает всё, по порядку: производитель чаще ждёт.

PubSub

Очередь отдаёт каждый кадр одному потребителю: взял один, у остальных его нет. Для объявлений нужно другое: башня меняет статус, «на связи», «занята», и узнать должны все, кто слушает. Это PubSub:

export const announce = (statuses: readonly string[], subscribers: number): Effect.Effect<string[][]> =>
  Effect.scoped(
    Effect.gen(function* () {
      const hub = yield* PubSub.unbounded<string>();
      const subscriptions = yield* Effect.forEach(Array.from({ length: subscribers }), () =>
        PubSub.subscribe(hub),
      );
      yield* PubSub.publishAll(hub, statuses);
      return yield* Effect.forEach(subscriptions, (subscription) =>
        Effect.forEach(statuses, () => PubSub.take(subscription)),
      );
    }),
  );

[ [ 'на связи', 'занята' ], [ 'на связи', 'занята' ] ]

PubSub.subscribe это ресурс: подписка живёт в области и снимается, когда область закрыли, поэтому вокруг Effect.scoped. Подписываться надо до публикации: сообщения, опубликованные раньше, новый подписчик не получит. У PubSub те же размеры, что у очереди: bounded, dropping, sliding, unbounded.

Под капотом

Queue.bounded устроена из тех же частей, что ворота и колокольчики из урока 66: буфер на capacity мест и два списка ждущих файберов. take из пустой записывает себя в «ждут кадра» и спит. offer в полную записывает себя в «ждут места». Каждая операция, которая меняет буфер, будит одного ждущего с другой стороны. У PubSub у каждого подписчика своя очередь, а публикация кладёт сообщение в каждую.

Где тут дыра

const inbox = yield* Queue.unbounded<number>();
yield* Effect.forkChild(Effect.forever(Effect.andThen(Effect.sleep('1 millis'), Queue.offer(inbox, 1))));
yield* Effect.forkChild(Effect.forever(Effect.andThen(Effect.sleep('10 millis'), Queue.take(inbox))));

размер неограниченной очереди каждые 250 мс: [ 177, 358, 542, 725 ]

Никто не ждёт, это правда. Приёмник кладёт раз в миллисекунду, обработчик берёт раз в десять, и очередь растёт на семьсот кадров в секунду. За ночь это двадцать пять миллионов, и башня упадёт, когда кончится память. А кадры в конце очереди к тому времени устарели на часы: ответ на них уже никому не нужен.

Лечится тем, что очередь ограничена, и выбор сделан явно: ждать, отбрасывать новое, отбрасывать старое. Неограниченная очередь не решает, что делать с лишней работой, а откладывает решение до момента, когда принимать его поздно.

Башня, этап 33

У башни теперь есть приёмник: кадры ждут в ограниченной очереди, обработчик берёт по одному и отвечает по порядку. И доска объявлений, где каждый подписчик слышит каждый статус.

Практика

Глубже в архиве

Очереди и противодавление без Effect, на итераторах и потоках Node, разобраны в уроке «Потоки и итераторы».

Дальше

Рыдкулл держит две ведомости, и цифры на них не сходятся.

Транзакции ждут в следующем уроке.