Противодавление

Липвиг описал противодавление из урока 68: медленный конец говорит быстрому «подожди», и работа не копится там, где её некуда положить. В уроке 68 его давала ограниченная очередь. У тянущего потока из прошлого урока оно встроено, и сегодня видно, где оно работает само, где его надо настроить и где его не бывает вовсе.

Даром

Провод, из которого берут, и башня, которая пишет каждую строку секунду:

let reads = 0;
const wire = Stream.fromEffectRepeat(Effect.sync(() => ++reads));

const night = wire.pipe(
  Stream.take(3),
  Stream.runForEach((line) => Effect.andThen(Effect.log(`пишу ${line}`), Effect.sleep('1 second'))),
);

Пока башня пишет первую строку, reads равно единице. Провод отдал одну строку и стоит: вторую никто не просил. Через секунду башня просит вторую, ещё через секунду третью. Всего три чтения на три строки. Скорость всей цепочки задаёт самый медленный шаг, и никакой очереди между ними нет. Это противодавление, и написать для него не пришлось ни строчки.

Пока шаги тянут друг у друга, так работает любая цепочка: map, filter, mapEffect никогда не берут у шага перед собой больше, чем у них попросили.

Буфер: читать вперёд

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

Stream.buffer ставит такую стопку между двумя шагами:

const ahead =
  (capacity: number) =>
  <A, E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<A, E, R> =>
    Stream.buffer(self, { capacity });

Ниже буфера всё по-прежнему тянет. Выше него свой файбер тянет из провода и складывает в очередь на capacity мест, пока они есть. Возьми тот же провод с ahead(4): пока башня пишет первую строку, reads равно пяти. Одна строка у башни в руках, четыре лежат в буфере, а шестую провод не отдаст, пока место не освободится. Противодавление осталось, только отодвинулось на capacity строк.

Подвигай скорости и размер буфера, как в уроке 68:

producer

flowing

buffer

0/4

consumer

0

0 produced0 consumed0 paused0 tps

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

Когда стопка полна

Полный буфер может делать три вещи, и выбор это strategy:

Stream.buffer(statuses, { capacity: 2 }); // suspend: провод ждёт
Stream.buffer(statuses, { capacity: 2, strategy: 'dropping' }); // новое выпадает
Stream.buffer(statuses, { capacity: 2, strategy: 'sliding' }); // старое выпадает

Быстрый провод из десяти строк, Stream.range(1, 10), и башня, которая пишет строку в секунду:

suspend   [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
dropping  [ 1, 2 ]
sliding   [ 9, 10 ]

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

Линия, которая не умеет ждать

Всё выше работало, потому что провод можно попросить подождать. С линией, которая толкает сама, так нельзя. Обработчик onLine(line) вызывают, когда строка пришла. Ответить «приходи позже» он не может: линия не спрашивает.

Stream.callback из прошлого урока прячет это за очередью, и по умолчанию очередь без предела. Пока башня успевает, разницы нет. Когда не успевает, строки копятся в памяти всю ночь, и к утру башня падает, не потеряв ни строки. Почтмейстер, который берёт всё, что суют в окошко, кончает так.

Значит, очередь надо ограничить, и тогда снова встаёт вопрос, что делать, когда она полна:

Stream.callback<string>(subscribe, { bufferSize: 3, strategy: 'dropping' });

Тут есть подвох, и в документации он записан мелким шрифтом. Queue.offerUnsafe из обработчика синхронный: он не умеет ждать. Поэтому strategy: 'suspend' у толкающей линии не ждёт, а вежливо возвращает false, и строка пропадает. dropping делает то же самое, но хотя бы честно об этом говорит. sliding возвращает true и молча выбрасывает самую старую.

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

export const fromBusyFeed = (
  feed: Feed,
  capacity: number,
  onDrop: (line: string) => void,
): Stream.Stream<string> =>
  Stream.callback<string>(
    (queue) =>
      Effect.acquireRelease(
        Effect.sync(() =>
          feed.listen(
            (line) => {
              if (!Queue.offerUnsafe(queue, line)) onDrop(line);
            },
            () => Queue.endUnsafe(queue),
          ),
        ),
        (unsubscribe) => Effect.sync(unsubscribe),
      ),
    { bufferSize: capacity, strategy: 'dropping' },
  );

onDrop здесь не украшение. Потеря, о которой знаешь, это беда, которую можно чинить: утром Липвиг попросит Магистраль прислать эти строки ещё раз. Потеря, о которой не знаешь, это клякса в отчёте. Лучше всего, конечно, не терять вовсе, и для этого у толкающей линии один способ: перестать быть толкающей. Если у источника есть «прочитай следующее», бери его, а не подписку. Вся защита этого урока нужна для тех, у кого такого нет.

Stream.fromEventListener для событий браузера устроен так же: у него есть bufferSize, и при полной очереди лишние события он отбрасывает. Об этом сказано в его документации, и теперь понятно почему.

Где тут дыра

const lines = Stream.callback<string>(
  (queue) =>
    Effect.acquireRelease(
      Effect.sync(() =>
        magistral.listen(
          (line) => Queue.offerUnsafe(queue, line),
          () => Queue.endUnsafe(queue),
        ),
      ),
      (unsubscribe) => Effect.sync(unsubscribe),
    ),
  { bufferSize: 3, strategy: 'suspend' },
);

Ночью линия тихая, башня успевает, очередь не наполняется ни разу, и Ринцвальд прав. Под утро линия присылает десять строк разом, a до j. В журнале башни a, b, c. Остальных семи нет нигде: ни в журнале, ни в ошибке, ни в логе. offerUnsafe вернул false семь раз, и никто этого не прочитал. Ждать умеет Queue.offer, эффект; обработчик линии эффект не исполняет. Ринцвальд написал «ждать» там, где ждать некому.

Магистраль, этап 11

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

Практика

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

Противодавление в стримах Node и браузера разобрано в уроке про стримы раздела про асинхронность.

Дальше

К вечеру у башни есть стопка, сводка и счёт выброшенных писем. Ринцвальд долго смотрит на свою строку с suspend, потом на счётчик onDrop в журнале.

Две линии в одну башню ждут в следующем уроке.