Группы

Группа — несколько детей одной задачи, которые идут одновременно, и задача ждёт их всех.

С чего начать, по тому, что нужно веткам:

Когда	Читать
Каждая ветка доходит до конца, что бы ни делали остальные: `[ctx.run(a), ctx.run(b)].wait`	«Ожидание нескольких детей»
Результат без одной из частей ни к чему: `ctx.runAll` отменяет ещё идущие ветки, как только одна кончилась без значения	«Когда один провал делает остальное бессмысленным»
Ветки делят блокировку или пул, или одна ждёт другую	«Ветки, которые ждут друг друга»
Ветка возвращает ресурс для вызывающего	«Что группа отдаёт обратно»
Тело должно закрыть список, который вернул `ctx.runAll`	«Список, который вернула группа»

Строки под кодом — то, о чём сообщает код:

rows closed: источник строк закрыт;
lock acquired, lock waits, lock released: что происходит с блокировкой, которую делят ветки;
outcome: то, чем завершается parent.done.

Ожидание нескольких детей

Экспорту нужны два источника: строки и картинки. Друг друга они не ждут, поэтому открываются сразу оба. Каждая ветка закрывает то, что открыла, если не отдала это наружу. Что бы ни случилось, родитель обязан кончиться правдой об этом, и ничего не должно остаться открытым.

Первая попытка

// Первая попытка: исход решает гонка, а источник утекает.
final parent = Job<void>((ctx) async {
  final rows = Job.deferred<Source>(
    (ctx) => ctx.abandonable(openRows, discard: (source) => source.close()),
  );
  final images = Job.deferred<Source>(
    (ctx) => ctx.abandonable(openImages, discard: (source) => source.close()),
  );

  final sources = await Future.wait([ctx.run(rows), ctx.run(images)]);
  await ctx.join(() => writeArchive(sources));
});

Future.wait — то, чем Dart ждёт несколько future, и здесь у него две беды. Вот что происходит:

Исход решает гонка. Future.wait оставляет себе первую дошедшую до него ошибку, а остальные молча отпускает. Допустим, rows падает, а images кончается Cancelled. Если первым дойдёт провал, родитель кончится Failed. Если первой дойдёт отмена, родитель кончится Cancelled(BodyCancelReason), и провала rows в исходе нет вовсе. Тогда его слышит только наблюдатель ребёнка. Ядро не отдаёт его в зону: тело ждало future от run, поэтому ядро считает, что на провал посмотрели. Если наблюдателя нет ни у родителя, ни у детей, ошибка теряется.
Источник утекает. Допустим, rows падает, а images удаётся. Ветка images возвращает свой источник и кончается Done, поэтому её discard не выполняется: значение дошло до ctx.run. Но Future.wait завершается ошибкой и выбрасывает значения, так что тело источника даже не держит. Его не закрывает никто.

eagerError: true не чинит ни того ни другого:

// Первая попытка с eagerError: та же гонка и та же утечка.
final sources = await Future.wait([
  ctx.run(rows),
  ctx.run(images),
], eagerError: true);

Он меняет только то, когда просыпается тело: на первой ошибке, а не после последней ветки. Если эта ошибка — провал, остановиться остальным веткам никто не предлагает, а детей ядро дожидается в любом случае, поэтому задача кончается не раньше. Если это Cancelled, которым кончилась ветка, тело выпускает чужую отмену, и задача отменяет себя сама: другая ветка кончается Cancelled с ParentCancelReason, а то, что она открывала, уходит в её discard. Исход по-прежнему решает то, чья беда пришла первой, а источник удавшейся ветки по-прежнему пропадает.

Регистрация по приходу и ожидание в одном конверте

final parent = Job<void>((ctx) async {
  final rows = Job.deferred<Source>(
    (ctx) => ctx.abandonable(openRows, discard: (source) => source.close()),
  );
  final images = Job.deferred<Source>(
    (ctx) => ctx.abandonable(openImages, discard: (source) => source.close()),
  );

  final sources = await [
    ctx.run(rows, dispose: (source) => source.close()),
    ctx.run(images, dispose: (source) => source.close()),
  ].wait;
  await ctx.join(() => writeArchive(sources));
});

Когда rows открылась, а images упала, код печатает:

rows closed
outcome: Failed(ParallelWaitError: Bad state: disk)

Изменились две вещи, и каждая чинит одну беду.

ctx.run(child, dispose: ...) чинит утечку. Он регистрирует значение на родителе в тот момент, когда ветка его отдаёт. С этого момента значение закрывает родитель, когда кончается, чем бы он ни кончился. Родитель может кончиться после записи архива, когда другая ветка упала или отменилась, или когда его отменили в промежутке. Ветка, которая не дошла до возврата, закрывает свой источник через discard, с которым его открывала. Под Future.wait это тоже починило бы утечку. Это правило из раздела «Регистрация по приходу» страницы о правилах уборки. Колбэк — dispose, а не discard, потому что источники родитель держит для себя.

.wait чинит исход. Он собирает все ошибки в один ParallelWaitError вместе со значениями тех веток, которые удались. Ядро читает этот конверт. Конверт с настоящим провалом остаётся провалом, со своими ошибками и значениями нетронутыми. Конверт из одних отмен и значений заканчивает родителя первой по списку отменой, так же, как ожидание одной дочерней задачи. Порядок, в котором пришла беда, больше ничего не решает. Значения в конверте уже зарегистрированы, поэтому, если закрыть их ещё и там, они закроются дважды.

.wait не останавливает ни одну ветку раньше срока. Он возвращает управление, когда закончены все ветки, поэтому, если одна ветка упала или была отменена сама по себе, её соседка доигрывает до конца. Отмена родителя их останавливает: она доходит до каждого ребёнка, в том числе до веток .wait. Ресурс, открытый веткой для себя, всё равно освобождается, когда ветка кончается, если ветка зарегистрировала его через dispose: ctx.abandonable(() => open(), dispose: (value) => value.close()). Уборка принадлежит задаче, а не ожиданию.

Когда один провал делает остальное бессмысленным

Бывает, что результат без одной из частей ни к чему, и ветки, которые ещё работают, должны остановиться, как только одна упала. Это делает ctx.runAll:

final values = await ctx.runAll([
  Job.deferred<int>((ctx) => ctx.abandonable(loadRows)),
  Job.deferred<int>((ctx) => ctx.abandonable(loadExtra)),
]);

ctx.runAll запускает каждого ребёнка синхронно, по порядку списка, и возвращает их значения в том же порядке. Как только тело ветки кончается не значением, группа отменяет остальные ветки причиной SiblingCancelReason. Ветку, из которой пришла беда, она не отменяет, иначе ошибка этой ветки превратилась бы в отмену и пропала.

Что бросит runAll, решают итоговые исходы: настоящий провал, если он есть, иначе первая полученная отмена, кроме тех, которые группа запросила сама. Бросается тот самый объект, который несёт исход, как бросил бы await ctx.run(child). Провал, который группа получила, но не бросила, идёт путём ошибки, которую не несёт ни один исход. Движок может руками кончить ветку значением уже после того, как группа услышала о провале её тела. Группа тогда исходит из этого провала, а не из Done ветки.

Когда удаются все ветки, каждая форма отдаёт телу значения в порядке веток. Различаются формы тогда, когда ветка падает:

Форма	Останавливает остальных	Тело просыпается	Значения удавшихся веток
`Future.wait`	нет	после конца последней ветки	выброшены раньше, чем их увидит тело
`Future.wait(eagerError: true)`	нет, если только первая ошибка не `Cancelled` ветки	на первой ошибке	выброшены
`[...].wait`	нет	после конца последней ветки	в `ParallelWaitError.values`
`ctx.runAll`	да, причиной `SiblingCancelReason`	после конца последней ветки	до тела не доходят: группа отменяет их ветки

Под eagerError: true задача всё равно ждёт своих детей, а Cancelled, пришедший первым, отменяет саму задачу, как сказано выше. Ветка ctx.runAll, созданная с cancellable: false, отвергает остановку и кончается Done со своим значением; кто его закроет, показывает раздел «Что группа отдаёт обратно».

Ни одна из форм не закрывает то, что вернули ветки: форма только отдаёт значения или выбрасывает их. Значение закрывает регистрация, которую его получатель сделал, когда оно пришло. У ctx.run это dispose, переданный с каждым вызовом. У ctx.runAll такого аргумента нет, а как зарегистрировать этот список, показывает раздел «Список, который вернула группа».

Ветки, которые ждут друг друга

Две ветки читают строки под одной блокировкой, потому что хранилище пускает одного читателя за раз. Так же ведёт себя пул, в котором свободных мест меньше, чем веток, которым они нужны. Каждая ветка берёт блокировку так, как страница об уборке берёт ресурс: ctx.join(Lock.acquire, dispose: (lock) => lock.release()).

Первая попытка

// Первая попытка: блокировка ждёт группу, а группа — её.
ChildJob<int> reading() {
  return Job.deferred<int>((ctx) async {
    await ctx.join(Lock.acquire, dispose: (lock) => lock.release());
    return ctx.abandonable(loadRows);
  });
}

final parent = Job<List<int>>((ctx) => ctx.runAll([reading(), reading()]));

lock acquired
lock waits

Больше ничего не приходит: группа висит без ошибки и без таймера. Вот что происходит:

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

Ветка, которая ждёт Job.value или Job.done соседки, вешает группу так же: у соседки нет исхода, пока группа не решила.

Отмена здесь не помогает. Она снимает такое зависание, только если кончает то, чего ветка ещё ждёт: тело или его ребёнка, которые ждут через ctx.abandonable или на операции, которая слышит ctx.onCancel. Вторая ветка ждёт в ctx.join на Lock.acquire, а тот отмены не слышит, поэтому висит даже отменённый родитель. Голый await отмена тоже не кончает, как и ожидание внутри ctx.uncancellable или в задаче с cancellable: false. Отмена ветки, которая держит блокировку, ничего не даёт: её тело уже кончилось.

Блокировка в ребёнке ветки

ChildJob<int> reading() {
  return Job.deferred<int>((ctx) async {
    final locked = Job.deferred<int>((ctx) async {
      await ctx.join(Lock.acquire, dispose: (lock) => lock.release());
      return ctx.abandonable(loadRows);
    });
    return ctx.run(locked);
  });
}

lock acquired
lock waits
lock released
lock acquired
lock released
outcome: Done([21, 21])

Блокировку держит ребёнок locked, а кончается он внутри тела ветки. Тогда же его уборка отпускает блокировку, ещё до того, как группа что-то решила, и вторая ветка её берёт. Для веток, которые зависят друг от друга, есть и другой выход, [...].wait: под ним каждая ветка разматывается сама.

То, что ребёнок locked открывает для вызывающего, ветка регистрирует по приходу, через ctx.run(locked, discard: ...), и этот discard выполняется уже без блокировки.

Два барьера

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

Стадия	Что делает каждая ветка
Тело	возвращает значение и ждёт своих детей
Первый барьер	ждёт тел и детей остальных
Уборка	выполняет свой стек уборки; `discard` откладывается, и при успехе он не выполняется никогда
Второй барьер	ждёт уборки остальных
Передача	`runAll` возвращает значения; потом ветка кончается `Done`

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

Первый барьер держит и колбэки dispose ветки. Поэтому ветка, которая ждёт соседку или то, что соседка освобождает в своей уборке, вешает группу, как блокировка из раздела «Первая попытка» выше. Тело, которое кончилось без значения, вместо этого кончает ожидание: группа отменяет остальные ветки, и ctx.runAll ждёт конца каждой ветки. К приходу ошибки всё, что ветки взяли, закрыто.

Что группа отдаёт обратно

Свой это ресурс или отданный наружу, ветка говорит тем членом, через который регистрирует его освобождение:

Job.deferred<Source>((ctx) async {
  final cache = await ctx.abandonable(
    openCache,
    dispose: (cache) => cache.close(),
  );
  final rows = await ctx.abandonable(
    openRows,
    discard: (source) => source.close(),
  );
  await ctx.join(() => cache.warm(rows));

  return rows;
});

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

Через что ветка регистрирует освобождение	Группа полностью удалась	Группа кончилась чем-то другим
`dispose` у `ctx.abandonable` и `ctx.join` или `onDispose`: ресурс ветка держит у себя	закрывается, когда кончается ветка	закрывается, когда кончается ветка
`discard` у `ctx.abandonable` и `ctx.join` или `onDiscard`: ветка отдаёт ресурс наружу	достаётся вызывающему открытым	закрывает ветка, которая принимает остановку, ещё до возврата из группы
`discard` у `ctx.run(child, discard: ...)`: ресурс от ребёнка ветки	достаётся вызывающему открытым	закрывает ветка, как в строке выше
любой `discard` в ветке, созданной с `cancellable: false`	достаётся вызывающему открытым	открыт: ветка отвергает остановку и кончается `Done`

Ресурс ветка может получить и от своего ребёнка:

Job.deferred<Source>((ctx) async {
  final opening = Job.deferred<Source>(
    (ctx) => ctx.abandonable(openRows, discard: (source) => source.close()),
  );
  return ctx.run(opening, discard: (source) => source.close());
});

Собственный discard ребёнка источник не закрывает. Ребёнок отдаёт источник и кончается Done внутри ветки, ещё до того, как группа что-то решила, а discard срабатывает, только если его задача кончилась не Done. Здесь источник дошёл до ветки. Поэтому ветка регистрирует источник заново в момент, когда он пришёл, через discard, переданный в ctx.run. Это собственная регистрация ветки, третья строка таблицы. Этот вызов показан в разделе «Регистрация по приходу» страницы о правилах уборки.

Последняя строка следует из раздела «Выбор колбэка» страницы об уборке. Ветка, созданная с cancellable: false, остановку отвергает и кончается Done. Её значение считается отданным через её собственный Job.value: discard, зарегистрированный веткой, не выполнился бы, и больше ресурс никто не закрывает. Тело по-прежнему держит rows, задачу, которую передало в runAll, и закрывает значение через неё:

final rows = Job.deferred<Source>(cancellable: false, (ctx) => openRows());
try {
  await ctx.runAll([rows, images]);
} on Object {
  if (rows.outcome case Done(:final value)) value.close();
  rethrow;
}

Список, который вернула группа

ctx.runAll отдаёт телу список открытых источников, и тело должно закрыть каждый из них, что бы ни случилось после возврата группы. Список ничем не отличается от любого другого значения, и получатель у него тело. В отличие от run, у runAll нет dispose или discard, чтобы зарегистрировать список по приходу, поэтому тело регистрирует его само.

Первая попытка

// Первая попытка: исключение до цикла оставляет все источники открытыми.
final sources = await ctx.runAll(branches);
await ctx.join(() => writeArchive(sources));
for (final source in sources) {
  source.close();
}

Тело закрывает источники там, где они ему больше не нужны, как любой код на Dart закрывает файл после последней записи. Цикл выполняется, только если тело до него дойдёт. Если отмена приходит, пока пишется архив, ctx.join дожидается архива и бросает отмену, вместо того чтобы вернуться. Упавший writeArchive так же бросает свою ошибку. В обоих случаях тело до цикла не доходит. Список пропадает на руках у тела, и об утечке не объявляет ничто: вызывающий видит обычную отмену или ошибку архива.

`try` следующей строкой

final sources = await ctx.runAll(branches);
try {
  await ctx.join(() => writeArchive(sources));
} finally {
  for (final source in sources) {
    source.close();
  }
}

try открывается следующей же строкой после возврата из группы, раньше всего, что может бросить. finally выполняется, как бы тело ни покинуло try: на отмене, которую бросает join, на ошибке архива или после последней строки. Список, который тело держит для себя, так закрывается целиком, если ни один close не бросает. close, который бросил, обрывает цикл, и источники после него остаются открытыми.

Отдельная регистрация на каждый источник закрывает остальные и в этом случае. Стек уборки вызывает каждый колбэк, что бы ни сделал предыдущий. Ошибка одного уходит в onError и по умолчанию дальше в зону, как описано в разделе «Выбор колбэка» страницы об уборке. Цикл регистраций встаёт на место try, следующей же строкой:

final sources = await ctx.runAll(branches);
for (final source in sources) {
  ctx.onDispose(source.close);
}
await ctx.join(() => writeArchive(sources));

Список, который тело отдаёт

final exported = Job.deferred<List<Source>>((ctx) async {
  final sources = await ctx.runAll(branches);
  try {
    await ctx.join(() => writeArchive(sources));
  } on Object {
    for (final source in sources) {
      source.close();
    }
    rethrow;
  }
  return sources;
});

Эта ветка пишет архив, а потом возвращает источники открытыми, чтобы вызывающий работал с ними. finally закрыл бы их и при успехе, поэтому закрытие переезжает в catch. Он покрывает каждую ошибку тела и ничего после него, а возвращённый список всё ещё может не дойти ни до кого. Пусть exported стоит веткой в группе ctx.runAll, и соседка падает уже после того, как тело вернулось. Группа отменяет ветку, как описано в разделе «Что группа отдаёт обратно», и до вызывающего список не доходит. Тело вышло из try задолго до этого, и источники остаются открытыми.

ctx.onDiscard выполняется ровно тогда, когда ветка кончается не Done: она упала или её остановила группа. Поэтому он закрывает список и в этом случае, а когда список доходит, не мешает. Каждый источник получает свою регистрацию, как в разделе выше:

final exported = Job.deferred<List<Source>>((ctx) async {
  final sources = await ctx.runAll(branches);
  for (final source in sources) {
    ctx.onDiscard(source.close);
  }
  await ctx.join(() => writeArchive(sources));
  return sources;
});

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

Чего `runAll` не обещает

Ветка, тело которой вернуло значение, ещё не кончилась: она ждёт остальную группу, как показывает раздел «Два барьера». Трёх вещей, которых можно ждать, не происходит:

Чего можно ждать	Что происходит
Ветка, которую группа останавливает, прекращает работу	Прекращается только её ожидание
Детей упавшей ветки отменяют	Они доигрывают: провал никогда не переходит к детям, и группа их ждёт
У ветки, тело которой вернуло значение, есть исход	Не раньше, чем группа решит: до тех пор её `outcome` равен `null`

Остановленная ветка, которая ждёт через ctx.abandonable, кончается, а операция за ней доигрывает. Чтобы остановилась сама работа, передайте ей отмену через ctx.onCancel и дождитесь её через ctx.join. Ветка, созданная с cancellable: false, отвергает остановку, и группа её дожидается.

Пока группа не решила, у ветки, тело которой вернуло значение, isCancelled равен false, и она ещё может кончиться Cancelled: когда другая ветка кончается без значения, группа отменяет её причиной SiblingCancelReason, и isCancelled становится true.

Когда группа отдала список, решение окончательно. Каждая ветка кончается Done с тем значением, которое отдала. Отмену, которая застаёт ветку, пока та ещё разматывается, ветка отвергает, и каскад от родителя тоже.

Группы

Ожидание нескольких детей

Первая попытка

Регистрация по приходу и ожидание в одном конверте

Когда один провал делает остальное бессмысленным

Ветки, которые ждут друг друга

Первая попытка

Блокировка в ребёнке ветки

Два барьера

Что группа отдаёт обратно

Список, который вернула группа

Первая попытка

try следующей строкой

Список, который тело отдаёт

Чего runAll не обещает

`try` следующей строкой

Чего `runAll` не обещает