Общие ресурсы: RcMap, RcRef и Pool

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

Линия на каждый кадр

Так пишет Ринцвальд:

const sendOnce = (to: string, line: string) =>
  withResources([`линия ${to}`], Effect.sync(() => void log.push(`> ${to}: ${line}`)));

await Effect.runPromise(
  Effect.forEach(['a', 'b', 'c'], (line) => sendOnce('T-3', line)).pipe(Effect.provide(Depot.layerTest(log))),
);

[
  'открыт линия T-3',
  '> T-3: a',
  'закрыт линия T-3: ok',
  'открыт линия T-3',
  '> T-3: b',
  'закрыт линия T-3: ok',
  'открыт линия T-3',
  '> T-3: c',
  'закрыт линия T-3: ok'
]

Три строки, три линии. Всё честно закрыто, ничего не течёт, но склад работает в три раза больше, чем башня. А если две смены пишут к T-3 одновременно, у каждой своя линия, хотя хватило бы одной.

Счётчик ссылок

Решение старое, как библиотеки: счётчик ссылок. Первый, кому нужна линия, её открывает. Каждый следующий берёт ту же и увеличивает счётчик. Каждый, кто закончил, уменьшает. Когда счётчик дошёл до нуля, линию можно закрыть. Можно, но не обязательно сразу: если через миг к T-3 снова будут писать, дешевле подождать.

RcMap из Effect это словарь таких ресурсов по ключу, со счётчиком у каждого и временем простоя:

export class Neighbors extends Context.Service<
  Neighbors,
  { readonly send: (to: string, line: string) => Effect.Effect<void, WireDown> }
>()('bashnya/Neighbors') {
  static readonly layer = Layer.effect(
    Neighbors,
    Effect.gen(function* () {
      const lines = yield* RcMap.make({
        lookup: (to: string) => resource(`линия ${to}`),
        idleTimeToLive: '1 second',
      });
      const journal = yield* Journal;
      return Neighbors.of({
        send: (to, line) =>
          Effect.scoped(Effect.andThen(RcMap.get(lines, to), journal.write(`> ${to}: ${line}`))),
      });
    }),
  );
}

RcMap.make получает lookup, как открыть ресурс по ключу. Это наш resource из урока 55: взятие и возврат вместе. idleTimeToLive говорит, сколько ресурс живёт после того, как счётчик дошёл до нуля. RcMap.get(lines, to) берёт линию к соседу и требует область: пока она открыта, счётчик учитывает этого пользователя. Effect.scoped вокруг одной отправки и есть «попользовался и отдал».

const neighbors = yield* Neighbors;
for (const line of ['a', 'b', 'c']) yield* neighbors.send('T-3', line);
yield* neighbors.send('T-5', 'd');
console.log('после отправок:', shared);
yield* Effect.sleep('1100 millis');
console.log('через 1,1 секунды:', shared);

после отправок: [ 'открыт линия T-3', 'открыт линия T-5' ]
через 1,1 секунды: [
  'открыт линия T-3',
  'открыт линия T-5',
  'закрыт линия T-3: ok',
  'закрыт линия T-5: ok'
]

Три строки к T-3 прошли по одной линии. Через секунду простоя обе линии закрылись сами. А сам словарь живёт в области слоя: когда слой разберут, RcMap закроет и те линии, чьё время простоя ещё не вышло.

Подвигай, когда две смены берут и возвращают линию к T-3, и посмотри, сколько раз её откроют:

RcRef

Если ключ один, словарь не нужен. RcRef это та же мысль для одного ресурса: RcRef.make({ acquire, idleTimeToLive }) и RcRef.get(ref) внутри области. Подходит для линии к архиву, книги отправленных из урока 53, любого дорогого ресурса, который нужен то часто, то никому.

Pool

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

const lines = yield* Pool.make({ acquire: Effect.suspend(() => resource(`линия архива ${++next}`)), size: 2 });
yield* Effect.forEach(
  ['ночь', 'утро', 'вечер'],
  (shift) =>
    Effect.scoped(
      Effect.gen(function* () {
        const line = yield* Pool.get(lines);
        log.push(`${at()} ${shift} взяла ${line}`);
        yield* Effect.sleep('200 millis');
      }),
    ),
  { concurrency: 'unbounded' },
);

[
  'открыт линия архива 1',
  'открыт линия архива 2',
  '0 мс ночь взяла линия архива 1',
  '0 мс утро взяла линия архива 2',
  '200 мс вечер взяла линия архива 2',
  'закрыт линия архива 1: ok',
  'закрыт линия архива 2: ok'
]

Три смены пришли одновременно (concurrency: 'unbounded' значит «все сразу», подробно в EF7). Две линии в пуле, двум сменам по линии, третья ждала, пока одна освободится. Пул нужен там, где ресурс нельзя делить, но держать один на всех мало: соединения с базой, рабочие процессы.

Под капотом

У каждой записи RcMap есть ресурс, его область, счётчик и, когда счётчик ноль, файбер ожидания. get увеличивает счётчик и кладёт в твою область финализатор, который счётчик уменьшит. Дошёл до нуля: запускается файбер, который проспит idleTimeToLive и закроет область записи. Новый get до того прерывает файбер, и всё начинается сначала. Сон идёт по Clock из контекста, поэтому в тестах время простоя крутят TestClock, как ленивую книгу в уроке 53.

Где тут дыра

Дыра урока уже разобрана, это линия на каждый кадр в начале. Её младшая сестра тише: RcMap без idleTimeToLive. Тогда линия закрывается в ту же секунду, когда счётчик дошёл до нуля, и башня, которая пишет к соседу раз в десять миллисекунд по одной строке, открывает и закрывает линию на каждой строке, как Ринцвальд. Время простоя подбирают по тому, как часто пишут: чуть больше обычного промежутка между строками.

Башня, этап 24

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

Практика

Дальше

Рыдкулл перечитывает записи склада за неделю и откладывает их.

Заказ ждёт в следующем уроке.