Ждать сигнала: Deferred и Latch

Ref из прошлого урока отвечает на вопрос «что сейчас». У Рыдкулла другой вопрос: «скажи, когда». Файберу нужно не прочитать значение, а уснуть до тех пор, пока его не разбудят. Для этого у Effect два простых инструмента: одноразовая ячейка и ворота.

Deferred

Deferred ты уже встречал в уроке 53: там ленивая книга клала себя в такую ячейку, а wasSent ждал её. Три операции:

• Deferred.make<A>() заводит пустую ячейку;
• Deferred.succeed(cell, value) кладёт значение и будит всех, кто ждал; повторный вызов отвечает false и ничего не меняет;
• Deferred.await(cell) ждёт значения.

Кроме succeed есть fail и die: ячейку можно заполнить и провалом, и тогда все ждущие упадут им.

Одноразовость делает из Deferred колокольчик для гонки из урока 64. Соберём raceAll сами:

export const firstReply = <A, R>(asks: readonly Effect.Effect<A, never, R>[]): Effect.Effect<A, never, R> =>
  Effect.gen(function* () {
    const answer = yield* Deferred.make<A>();
    const fibers = yield* Effect.forEach(asks, (ask) =>
      Effect.forkChild(Effect.flatMap(ask, (value) => Deferred.succeed(answer, value))),
    );
    const value = yield* Deferred.await(answer);
    yield* Fiber.interruptAll(fibers);
    return value;
  });

Холм

Все трое спрашивают сразу, каждый, получив ответ, пытается положить его в ячейку. Кладёт первый, остальные succeed ничего не делают, а мы их сразу прерываем. Это и есть устройство гонки: колокольчик, в который звонит первый добежавший.

Latch

Latch это ворота. Latch.make(false) заводит закрытые. latch.await ждёт, пока откроют. latch.open открывает и будит всех разом. Ворота можно закрыть снова, latch.close, и тогда следующие снова будут ждать. Этим ворота отличаются от Deferred: колокольчик звонит один раз, ворота открываются и закрываются сколько угодно.

export class Gate extends Context.Service<
  Gate,
  {
    readonly open: Effect.Effect<void>;
    readonly pass: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
  }
>()('bashnya/Gate') {
  static readonly layer = Layer.effect(
    Gate,
    Effect.map(Latch.make(false), (latch) =>
      Gate.of({
        open: Effect.asVoid(latch.open),
        pass: (self) => Effect.andThen(latch.await, self),
      }),
    ),
  );
}

[
  '300 мс ворота открыты',
  '300 мс прошёл кадр 5',
  '300 мс прошёл кадр 6'
]

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

Свой мьютекс

Из Ref и Deferred можно собрать мьютекс, замок, который пускает внутрь по одному. В ячейке два поля: занят ли замок и кто ждёт. Войти: если свободно, занять; если занято, встать в очередь со своим колокольчиком и ждать. Выйти: если кто-то ждёт, позвонить в его колокольчик и передать замок ему; если никто, освободить.

const makeMutex = Effect.gen(function* () {
  const state = yield* Ref.make<{ readonly busy: boolean; readonly waiting: readonly Deferred.Deferred<void>[] }>({ busy: false, waiting: [] });
  const lock = Effect.flatMap(Deferred.make<void>(), (me) =>
    Effect.flatten(Ref.modify(state, (s) =>
      s.busy ? [Deferred.await(me), { busy: true, waiting: [...s.waiting, me] }] : [Effect.void, { busy: true, waiting: s.waiting }])));
  const unlock = Effect.flatten(Ref.modify(state, (s) => {
    const [next, ...rest] = s.waiting;
    return next === undefined ? [Effect.void, { busy: false, waiting: [] }] : [Deferred.succeed(next, undefined), { busy: true, waiting: rest }];
  }));
  return <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.andThen(lock, Effect.ensuring(self, unlock));
});

[
  '400 мс ночь вошла',
  '500 мс ночь вышла',
  '500 мс утро вошла',
  '600 мс утро вышла',
  '600 мс вечер вошла',
  '700 мс вечер вышла'
]

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

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

Под капотом

Deferred.await на пустой ячейке не крутит цикл и не спит по таймеру. Файбер записывает себя в список ждущих ячейки и останавливается: исполнитель о нём забывает, пока ячейку не заполнят. succeed проходит по списку и ставит каждого обратно в очередь исполнителя. Latch устроен так же, только список у него обнуляется при каждом open. Сто ждущих стоят ровно столько, сколько сто записей в списке.

Где тут дыра

let ready = false;
let wakeups = 0;
const waitReady: Effect.Effect<void> = Effect.suspend(() => {
  wakeups++;
  return ready ? Effect.void : Effect.andThen(Effect.sleep('50 millis'), waitReady);
});

Сто кадров ждут, флажок поднимают через секунду:

прошли через 1058 мс, проснулись впустую 2006 раз

Две тысячи лишних пробуждений, чтобы узнать «нет, ещё нет». И кадры прошли не когда флажок подняли, а на полсотни миллисекунд позже, на следующей проверке. Чем чаще проверять, тем меньше опоздание и тем больше лишней работы; честного выбора тут нет. Ворота на Latch не просыпаются ни разу, пока их не откроют, и пропускают в ту же миллисекунду.

Башня, этап 31

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

Практика

Дальше

Утром склад Магистрали прислал жалобу.

Не больше стольких-то сразу, разберём в следующем уроке.