До сих пор у башни был один поток: одна линия, по одной строке, один шаг за другим. Ночь Магистрали так не устроена. Строки приходят по двум линиям сразу. Отправлять их дальше по одной долго, а все разом нельзя. Магистраль берёт кадры пачками дешевле, чем поштучно, а если бить её чаще, чем она успевает, она ломается. И одну и ту же ночь надо и записать в архив, и отправить дальше. Сегодня пять операторов, по одному на каждую из этих забот.
Две линии в одну: merge
Stream.merge(north, south) даёт поток, в котором строки обеих линий идут по мере прихода: какая раньше пришла, та раньше и прошла. Северная линия отдаёт строку раз в 300 миллисекунд, южная раз в 200:
ю1@200 с1@300 ю2@400 с2@600 ю3@600Каждая линия читается своим файбером, поэтому медленная северная не тормозит южную. Порядок внутри линии сохраняется, порядка между линиями нет: его и не было, линии не договаривались.
Когда кончается общий поток? По умолчанию, когда кончатся обе линии: haltStrategy: 'both'. Можно 'either', конец по первой замолчавшей, и тогда южная строка ю3 в примере выше не пройдёт: северная уже кончилась. Можно 'left' и 'right': ждать конца только одной. Для двух линий Магистрали правильный ответ 'both': тишина на одной линии не значит, что на другой нечего читать.
Если линий больше двух, Stream.mergeAll берёт массив потоков и concurrency: сколько читать сразу.
Отправлять разом: mapEffect с concurrency
Stream.mapEffect из урока 85 делает эффект для каждого значения. По умолчанию по одному: следующий кадр уходит, когда ответил предыдущий. С параметром concurrency уходят до concurrency кадров сразу:
export const relayEach =
<A, B, E2, R2>(send: (frame: A) => Effect.Effect<B, E2, R2>, concurrency: number) =>
<E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<B, E | E2, R | R2> =>
Stream.mapEffect(self, (frame) => send(frame), { concurrency });Четыре кадра, отправка которых занимает 3, 1, 2 и 1 секунду, и concurrency: 2:
ушёл 3@3000 ушёл 1@3000 ушёл 2@5000 ушёл 1@5000Второй кадр ушёл через секунду, но в поток попал на третьей: ответы идут в порядке кадров, а не в порядке ответов. Поток держит готовые ответы, пока не ответил кадр перед ними. Для журнала башни это то, что нужно: строка N в журнале это ответ на кадр N. Если порядок не важен, unordered: true выпускает ответы по мере готовности:
ушёл 1@1000 ушёл 3@3000 ушёл 2@3000 ушёл 1@4000Тот же concurrency, что у Effect.forEach из урока 63, и тот же смысл: не больше стольких сразу. Противодавление никуда не делось: пока заняты два места, третий кадр не уходит, а линия не читается дальше, чем нужно.
Пачками: grouped и groupedWithin
Магистраль принимает пачку кадров за один вызов, и пачка из десяти дешевле десяти вызовов. Stream.grouped(10) собирает значения в массивы по десять. Последняя пачка бывает короче: пять кадров по два дают [1, 2], [3, 4], [5].
У grouped одна беда: пачка уходит, когда полна. Ночью, когда кадров мало, девятый кадр может ждать десятого час. Почтамт так не работает: мешок с письмами уходит, когда полон или когда пришло время отправки. Stream.groupedWithin(size, within) делает то же: пачка уходит, как только наберётся size, или через within после первого кадра пачки, сколько бы их ни набралось.
export const batches =
(size: number, within: Duration.Input) =>
<A, E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<A[], E, R> =>
Stream.groupedWithin(self, size, within);Три кадра пришли сразу, четвёртый через сотню миллисекунд, пятый через три секунды, batches(3, '1 second'): пачка abc уходит сразу, полная; d ждёт секунду и уходит одна; e уходит, когда поток кончился. Без within d ждала бы e три секунды.
Не бить молотком: throttle
Stream.throttle ограничивает скорость потока. Он считает стоимость: функция cost говорит, сколько единиц стоит кусок, units за duration это лимит. Стратегия 'shape' придерживает лишнее, пока не освободится лимит. Стратегия 'enforce' выбрасывает лишнее; для писем она не годится, разве что для сводок.
Теперь вспомни «Под капотом» из урока 85: поток гоняет не значения, а куски. throttle считает стоимость куска. Шесть кадров, которые пришли одним куском, с cost: (chunk) => chunk.length и лимитом два в секунду стоят шесть единиц: throttle честно ждёт две секунды и выпускает все шесть разом. Скорость в среднем соблюдена, а Магистраль получила молоток. Поэтому сначала поток режут по одному кадру, Stream.rechunk(1), и только потом считают:
export const paced =
(perSecond: number) =>
<A, E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<A, E, R> =>
self.pipe(
Stream.rechunk(1),
Stream.throttle({
cost: (chunk) => chunk.length,
units: perSecond,
duration: '1 second',
strategy: 'shape',
}),
);1@0 2@0 3@500 4@1000 5@1500 6@2000Первые два уходят сразу: лимит в начале полон. Дальше по кадру каждые полсекунды. Два в секунду, без «в среднем».
Двум читателям: broadcast
Ночь надо и записать в архив, и отправить дальше. Два раза прочитать линию нельзя: каждый запуск потока это новая подписка, и вторая подписка начнёт с того места, где линия сейчас, а не с начала ночи. Нужен один поток и два читателя, и каждый получает всё. Это Stream.broadcast и его брат Stream.broadcastN:
export const tee = <A, E, R, E2, R2>(
self: Stream.Stream<A, E, R>,
archive: (value: A) => Effect.Effect<void, E2, R2>,
forward: (value: A) => Effect.Effect<void, E2, R2>,
): Effect.Effect<void, E | E2, R | R2> =>
Effect.scoped(
Effect.gen(function* () {
const [toArchive, toForward] = yield* Stream.broadcastN(self, { n: 2, capacity: 16 });
yield* Effect.all(
[Stream.runForEach(toArchive, archive), Stream.runForEach(toForward, forward)],
{ concurrency: 2, discard: true },
);
}),
);broadcastN возвращает эффект, который даёт кортеж из n потоков, и требует Scope: внутри живёт PubSub из урока 68 и файбер, который читает исходный поток и раздаёт значения подписчикам. Effect.scoped закрывает это всё, когда оба читателя кончили. capacity это размер буфера PubSub. Быстрый читатель может уйти вперёд медленного не больше чем на capacity значений, дальше suspend по умолчанию: исходный поток ждёт самого медленного. Противодавление опять на месте.
broadcastN начинает читать исходный поток, только когда подписались все n. broadcast без N возвращает один поток, на который можно подписаться сколько угодно раз, и начинает сразу. Тот, кто подписался позже, начало не увидит, если не попросить replay: сколько последних значений показать новым подписчикам. Когда читателей знаешь заранее, бери broadcastN.
Где тут дыра
const paced = Stream.throttle(frames, {
cost: () => 1,
units: 2,
duration: '1 second',
strategy: 'shape',
});На его проверке кадры приходили по одному из провода, fromWire, и каждый был отдельным куском. Всё сходилось: два в секунду. Ночью кадры пришли из Stream.fromIterable пачкой из ста, одним куском, и cost: () => 1 оценил этот кусок в одну единицу. Сто кадров ушли в Магистраль в ту же миллисекунду. Функция cost получает кусок, и () => 1 значит «кусок любой длины стоит одну единицу». Правильный cost считает длину, а правильный paced ещё и режет поток по одному, чтобы длина всегда была единицей.
Магистраль, этап 12
Пять приспособлений для ночи на двух линиях: слить линии, отправлять с ограничением, собирать пачки со сроком, держать скорость, раздать ночь архиву и линии. Тесты двигают часы и смотрят, когда что ушло.
Практика
Дальше
Под утро в журнале башни ровные пачки, два кадра в секунду и архив, где есть всё. Липвиг пролистывает журнал до конца и находит строку «мусор: 14».
Комната мёртвых писем ждёт в следующем уроке.