Семафор

Предел в forEach из урока 63 ограничивает одну рассылку. Но строки в линию шлёт не только рассылка: ответы на кадры, отложенные ответы из урока 62, гонки ретрансляторов. У каждой части свой предел, а линия одна, и все пределы складываются. Предел нужно ставить не на того, кто шлёт, а на то место, куда шлют.

Свой семафор

Замок из прошлого урока пускал по одному. Замени в нём «занят или свободен» на «сколько свободных мест», и выйдет семафор. Взять: если мест больше нуля, уменьшить и пройти; если ноль, встать в очередь с колокольчиком. Вернуть: если кто-то ждёт, позвонить ему и отдать место; если никто, увеличить счётчик. Ячейка с числом и очередью, Ref.modify на входе и выходе, и те же оговорки про прерывание, что у мьютекса. Собирать его второй раз не будем: в Effect он готовый.

Semaphore

export const throttled = (wire: Wire['Service'], permits: number): Effect.Effect<Wire['Service']> =>
  Effect.map(Semaphore.make(permits), (semaphore) =>
    Wire.of({ send: (line) => semaphore.withPermits(1)(wire.send(line)) }),
  );

Semaphore.make(3) заводит семафор на три разрешения. semaphore.withPermits(n)(self) берёт n разрешений, исполняет self и возвращает их при любом исходе: успех, ошибка, дефект, прерывание. Взятие и возврат стоят рядом, как у acquireRelease из урока 55, и по той же причине.

throttled ставит предел на сам провод. Кто бы ни слал через этот провод, больше трёх строк сразу не уйдёт:

200:a 200:b 200:c 400:d 400:e 400:f 600:g одновременно 3

Семь строк пришли разом, провод отвечает за двести миллисекунд. Три ушли сразу, три во второй волне, последняя в третьей. Шлющий об этом не знает: для него send иногда отвечает позже.

У семафора ещё несколько ручек. withPermit это withPermits(1). withPermitsIfAvailable(n) не ждёт: если разрешений нет, отвечает Option.none() и ничего не делает. resize(n) меняет число разрешений на ходу. А take и release берут и отдают руками, и о них ниже.

Подвигай: сколько разрешений и какой длины строки, и когда уйдёт последняя.

RateLimiter

Семафор ограничивает, сколько одновременно. Иногда нужен другой предел: сколько за время. Магистраль берёт плату за каждую строку и разрешает не больше ста в минуту, сколько бы их ни шло сразу. Это ограничитель частоты, и в Effect он лежит в effect/persistence, модуль RateLimiter: настройки limit, window, алгоритм fixed-window или token-bucket, а при превышении onExceeded: 'delay' ждёт, 'fail' отказывает. Счётчики он хранит в хранилище: в памяти, layerStoreMemory, или в Redis, чтобы предел был общим на все башни.

Нестабильно. Весь модуль RateLimiter в effect 4.1 помечен @stability unstable: имена настроек и устройство хранилища могут поменяться в минорной версии. Урок написан на 4.1.0. Задачи на RateLimiter в разделе нет, браузерный раннер этот модуль не грузит.

Под капотом

Semaphore в Effect это ровно то, что описано выше: счётчик и очередь ждущих, только аккуратно собранный с прерыванием. Ждущий, которого прервали, уходит из очереди; разрешение, которое уже успели отдать прерванному, возвращается следующему. withPermits берёт разрешения в неделимом куске, как acquireRelease, и возвращает в финализаторе. Поэтому семафор на одно разрешение и есть правильный мьютекс: Semaphore.make(1) и withPermit.

Где тут дыра

const semaphore = yield* Semaphore.make(1);
const send = (line: string) =>
  Effect.gen(function* () {
    yield* semaphore.take(1);
    if (line === 'битая') return yield* Effect.fail('провод отказал');
    yield* semaphore.release(1);
    return line;
  });
const first = yield* Effect.exit(send('битая'));
const second = yield* Effect.timeoutOption(send('целая'), '500 millis');

[ 'Failure', 'None' ]

Битая строка упала, и release не дошёл. Разрешение ушло вместе с ней. Целая строка ждала полсекунды и не дождалась: на семафоре на одно разрешение больше нет ни одного, и не будет никогда. Это утечка, как незакрытая линия Ринцвальда из EF6, только утекает не линия, а право ей пользоваться.

Лечится withPermits: возврат стоит в финализаторе и случится при любом исходе. take и release руками нужны редко, и тогда их кладут в acquireRelease.

Башня, этап 32

Провод башни получает предел: не больше стольких-то строк сразу, кто бы их ни слал. Остальные ждут очереди.

Практика

Дальше

Пондер стоит у входной трубки Гекса, куда сыплются кадры.

Очереди ждут в следующем уроке.