Ресурсы и освобождение

Job, которая что-то взяла, будь то хэндл базы, подписка, блокировка или временный файл, обязана это вернуть. Отмена приходит на await и из кода, вызванного оператором, но никогда не приходит из очереди между двумя операторами. Освобождение регистрируют методами контекста, и выбранный метод решает, кто освобождает ресурс и когда:

(ctx) async {
  // Создано здесь, на месте: освобождение регистрирует следующая строка,
  // даже если `listen` отменил задачу.
  final sub = device.events.listen(onEvent);
  ctx.onDispose(sub.cancel);

  // Получено вызовом: освобождение едет на вызове, а не строкой ниже.
  final db = await ctx.join(Database.open, dispose: (db) => db.close());

  // Уходит с результатом, поэтому освобождается, только если Job
  // кончилась отменой или ошибкой.
  final file = await ctx.join(openTemp, discard: (file) => file.delete());

  // Передано состоянию, и дальше владеет им оно.
  ctx
    ..check()
    ..disown(db)
    ..emit(Ready(db));

  return file;
},

Метод	Что и когда освобождает
dispose: у ctx.abandonable, ctx.join или ctx.run	значение этого вызова, при любом исходе
discard: у ctx.abandonable, ctx.join или ctx.run	значение этого вызова, и только если Job кончилась отменой или ошибкой
ctx.onDispose(callback)	то, что закрывает колбэк, при любом исходе
ctx.onDiscard(callback)	то, что закрывает колбэк, и только если Job кончилась отменой или ошибкой
ctx.disown(value)	ничего: снимает регистрацию, которую один из этих трёх вызовов сделал для значения

Database, device, openTemp и archive принадлежат приложению, из которого взяты примеры, а Idle, Loaded(rows) и Ready(db) входят в состояния его контроллера.

Получение ресурса вызовом

Job открывает базу, читает строки и публикует их. Отменённая в любой момент, она не должна оставить открытую базу.

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

// Первая попытка: отмена влезает между открытием и onDispose.
Job<void> load() => enqueue<Idle, void>(key: 'load', (ctx) async {
  final db = await ctx.join(Database.open);
  ctx.onDispose(db.close);

  final rows = await ctx.join(db.readAll);
  ctx.emit(Loaded(rows));
});

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

Отмените эту Job, пока Database.open в полёте. Вот что происходит:

1. join остаётся с вызовом, как и должен, и открытие доходит до конца.
2. Тогда join бросает Cancelled вместо значения, и до строки ниже тело не доходит.
3. База открыта, а единственная ссылка на неё ушла вместе с броском.
4. Стек уборки пуст: в этой Job никто не знает, что база вообще есть.

Освобождение едет на вызове

Job<void> load() => enqueue<Idle, void>(key: 'load', (ctx) async {
  final db = await ctx.join(Database.open, dispose: (db) => db.close());

  final rows = await ctx.join(db.readAll);
  ctx.emit(Loaded(rows));
});

abandonable и join принимают колбэк dispose для ресурса, который принадлежит Job, и правило тут одно:

• значение, не дошедшее до тела, освобождается на месте;
• дошедшее ложится на стек уборки и освобождается, когда Job заканчивается.

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

join дожидается этого освобождения до того, как бросить, поэтому следующая Job в очереди стартует с уже закрытой базой. Регистрируйте каждое освобождение один раз: добавление ctx.onDispose(db.close) в это тело закрыло бы одну базу дважды.

У ресурса, который тело создаёт само, такого зазора нет, потому что между созданием и регистрацией освобождения нет await. В блоке, которым открывается эта страница, ctx.onDispose(sub.cancel) стоит прямо под listen. Код, вызванный из listen, например onListen источника, может отменить задачу ещё до возврата из listen; onDispose всё равно принимает регистрацию на отменённой, но ещё работающей задаче, и освобождение выполняется вместе с остальными. Эта регистрация возвращает функцию, которую можно сохранить как removeDisposer, чтобы снять регистрацию, не вызывая колбэк.

Ничто не мешает и такому созданию поехать на вызове:

final sub = await ctx.join(
  () => device.events.listen(onEvent),
  dispose: (sub) => sub.cancel(),
);

abandonable и join принимают действие, которое возвращает без ожидания, поэтому синхронное создание регистрируется на вызове, как и любое другое. По сравнению с парой listen и onDispose под ним это даёт контрольную точку, которую вызов делает перед своим действием. Сначала он спрашивает, не отменена ли Job, и бросает, не начав действия. Возьмём отмену, которая уже стоит, например её только что отпустила секция uncancellable выше. В этом случае подписки не возникает вовсе, тогда как та пара её создаёт и отменяет во время уборки. Этим контрольным точкам посвящена страница «Отмена».

Возврат ресурса вызывающему

Теперь база стала результатом, а не средством: open ниже открывает её и возвращает. С этого момента ею владеет тот, кто взял значение, например через await open().value.

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

// Первая попытка: при успехе база закрывается раньше, чем её отдают.
Job<Database> open() => enqueue<Idle, Database>(key: 'open', (ctx) async {
  final db = await ctx.join(Database.open, dispose: (db) => db.close());
  await ctx.run(job<Idle, void>((child) => child.join(db.readAll)));
  return db;
});

Базу всё ещё освобождает эта Job, если она упадёт, поэтому колбэк, который срабатывает при любом исходе, выглядит безопасным. Но успех тоже исход:

1. Тело возвращает хэндл.
2. Уборка выполняется до доставки исхода, и dispose закрывает базу прямо там.
3. await вызывающего затем завершается хэндлом, закрытым мгновением раньше.

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

discard для значения, которое уходит

Job<Database> open() => enqueue<Idle, Database>(key: 'open', (ctx) async {
  final db = await ctx.join(Database.open, discard: (db) => db.close());
  await ctx.run(job<Idle, void>((child) => child.join(db.readAll)));
  return db;
});

discard срабатывает, только когда Job заканчивается не Done: отменённой или упавшей. Соответствующий метод регистрации называется ctx.onDiscard. Вызов принимает либо dispose, либо discard: если передать оба, он бросит ArgumentError, ничего не начав. Всё, что Job оставляет себе, будь то блокировка, временный файл или подписка, берёт dispose. Иначе его не освободит никто на успешном пути, куда не смотрит ни один тест на отмену.

Ребёнок выше и показывает, чем это различение окупается. ctx.run стоит между базой и return, то есть значение уже на руках, а отменить Job ещё можно. Отмена, пришедшая во время этого ожидания, заканчивает Job раньше, чем значение уйдёт: вызывающий получает Cancelled вместо базы, и discard её закрывает.

Регистрация принадлежит той Job, которая её сделала, и вместе со значением не передаётся: сработает она или нет, зависит от исхода этой Job, а не от того, куда ушло значение. Job, закончившаяся Done, базу уже передала, читает кто-нибудь её value или нет, и вызывающий, который так его и не прочитал, оставляет базу открытой. Тот, кто взял базу, владеет ею с этого момента и регистрирует освобождение сам. Родитель, который берёт её через ctx.run(child, discard: ...), регистрирует на вызове. Регистрация строкой ниже опоздала бы. ctx.run проверяет родителя, когда значение ребёнка уже на руках, и контрольная точка, бросившая там, уносит значение с собой. Так что до строки, которая зарегистрировала бы освобождение, дело не доходит. Такую передачу базы шаг за шагом разбирает раздел «Ресурс, который передают дальше» страницы о правилах уборки пакета async_job.

Передача ресурса состоянию

Базу держит сам экран. Job открывает её и мигрирует, причём этот шаг нельзя обрывать на середине. Затем Ready(db) становится состоянием, и с этого момента закрывать базу должен контроллер, а не Job.

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

// Первая попытка: dispose закрывает базу, которую держит состояние.
final db = await ctx.join(Database.open, dispose: (db) => db.close());
await ctx.uncancellable(db.migrate);
ctx.emit(Ready(db));

Вот что происходит:

1. Запись проходит, и экран показывает Ready.
2. Потом тело возвращается, и Job разматывает стек уборки.
3. dispose закрывает базу, которую держит экран.
4. Объект состояния при этом не меняется. У экрана остаётся хэндл, выглядящий ровно как прежде, а отказ всплывает на первом чтении из него.

Вторая попытка

// Вторая попытка: disown стоит до emit, который может бросить.
await ctx.uncancellable(db.migrate);
ctx
  ..disown(db)
  ..emit(Ready(db));

disown снимает регистрацию, которую join сделал для базы, и Job больше не закрывает то, что собирается отдать. Снимает он её и до того, как хоть что-то проверит, состоится ли запись вообще.

emit относится к контрольным точкам: для Job, принявшей отмену, он бросает Cancelled вместо записи.

Отмените эту Job, пока идёт миграция. Вот что происходит:

1. Секция uncancellable держит отмену до конца миграции.
2. Job принимает её, когда секция закрывается, и отмена выходит на первой контрольной точке после этого. Здесь ею оказывается emit.
3. К моменту, когда emit бросает, регистрации уже нет.
4. Состоянию база не досталась, стек уборки о ней больше не знает, и не закрывает её никто.

check, disown, emit

await ctx.uncancellable(db.migrate);
ctx
  ..check()
  ..disown(db)
  ..emit(Ready(db));

Три вызова каскада синхронны, поэтому между ними ничто не приходит. Если check бросил, а бросает он первым, когда Job отменена или её правила больше не держатся, база ещё зарегистрирована, и уборка её закроет. Если emit записал состояние и затем бросил, потому что Job отменил синхронный слушатель, базой уже владеет состояние. Стек уборки ею больше не владеет. В обоих случаях владелец остаётся ровно один.

disown находит регистрацию по значению, которое вернул join. Регистрацию, сделанную через ctx.onDispose, он так не найдёт: disown вернёт false, а колбэк всё равно выполнится. Её снимает функция, которую вернул onDispose, и зовут эту функцию там же, где здесь стоит disown.

После передачи базу не освобождает ни одна Job, не освобождает её и close(). Закрыть её должен сам контроллер. Для этого у него заводят метод, чья Job закрывает базу и переводит состояние дальше, и дожидаются этого метода до close(), как dispose() в разделе «Дождаться освобождения» примера камеры.

При асинхронной передаче саму передачу и disown ставят внутрь одной ожидаемой секции uncancellable:

await ctx.uncancellable(() async {
  await archive.take(file);
  ctx.disown(file);
});

В секции нет ни одной контрольной точки: передача ждётся обычным await, а disown только снимает регистрацию, так что отмене негде выйти между ними. Секция держит обычную отмену, пока не выполнятся обе строки. Правила состояния самой Job, которые никакая секция не держит, проверяются там, где секция открывается, до начала передачи.

Без секции пара распадается. Возьмём await ctx.join(() => archive.take(file)) с ctx.disown(file) строкой ниже: join проверяет Job ещё раз после передачи, и отмена, принятая за это время, заставляет его бросить. Файл тогда уже у архива, а его регистрация ещё стоит, так что уборка удалит то, что держит архив.

Когда случится освобождение

Job пишет временный файл и удаляет его при любом исходе. Следующая Job в очереди не должна его застать, и close() к своему возврату тоже.

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

// Первая попытка: файл удаляется поздно и в одиночку.
// Удалён при любом исходе: dispose срабатывает всегда.
await ctx.abandonable(openTemp, dispose: (file) => file.delete());

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

1. Job заканчивается, и следующая Job стартует, или возвращается close(), пока openTemp ещё работает.
2. Файл появляется после этого, а удаление идёт следом, поздно и в одиночку, и ни того, ни другого никто не ждёт.

О поздней ошибке вызова или самого колбэка узнаёт хук onError контроллера, а затем onUnanswered просят за неё ответить, и по умолчанию он передаёт её дальше, в Solo.unansweredHandler или в зону.

Ожидание, которое остаётся с вызовом

// Удалён до старта следующей Job и до возврата из close().
await ctx.join(openTemp, dispose: (file) => file.delete());

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

Порядок уборки и поздние результаты

Шаг	Что делает Job
Дети	ждёт завершения каждого ребёнка
Стек уборки	разматывает его с последней регистрации, ожидая каждый колбэк
Вторая волна	выполняет discard, который отмена сделала нужным задним числом
Конец	получает исход, выполняет свой обработчик состояния, даёт очереди запустить следующую Job

Первые три шага принадлежат ядру, async_job, и поздние результаты, которые они приносят, тоже его. Порядок описан и хранится в разделе «Порядок уборки и поздние результаты» страницы о правилах уборки пакета async_job. Контроллер добавляет последнюю строку:

1. Сначала фиксируется исход.
2. Затем обработчик состояния, переданный в enqueue, вычисляет состояние, которое оставляет после себя упавшая или отменённая Job, как описано в разделе «Состояние после ошибки или отмены» страницы о состоянии.
3. Затем очередь запускает следующую Job.

Код, который ждёт done или value, продолжается уже после этого запуска.

Обычный try/finally нужен для того, что не переживает тело: блокировки, взятой на один шаг и отпущенной раньше, чем тело пойдёт дальше, или временного файла одного витка цикла. Контрольная точка, бросившая внутри try, выполняет finally так же, как любой другой бросок. Регистрация вместо этого держала бы эти освобождения до конца Job и копила бы по одной на виток. Зарегистрированная уборка нужна для того, что должно жить столько же, сколько Job, и только она охватывает время после возврата из тела и до завершения его детей.

Об ошибке, брошенной уборкой, узнают хук onError контроллера и Solo.observer. Затем onUnanswered просят за неё ответить, и по умолчанию он передаёт её дальше, в Solo.unansweredHandler или в зону. Весь маршрут описан в разделе «Ответ за ошибку» страницы об ошибках. О Cancelled, брошенном уборкой, узнают те же двое, и дальше он не идёт: отмена означает чьё-то решение, а не отказ. Исход Job не меняет ни то, ни другое: упавшая уборка оставляет Done там, где был Done.

Никогда не ждите из уборки done, value или cancel() своей же Job. Каждый из них завершается только после этой уборки, поэтому такое ожидание не кончается, как не кончается и close(), который ждёт эту Job. Более поздняя Job той же очереди не может стартовать, пока не закончится эта, поэтому уборка, которая её ждёт, держит очередь, пока та Job в ней стоит. pending показывает такую Job как in its cleanup, а более позднюю называет среди стоящих за ней в queued. Как читать эту строку, показано в разделе «Что удерживает контроллер» страницы о диагностике.