Отмена

cancel() запрашивает; тело останавливается в контрольной точке. От того, какая это точка, зависит, что станет с операцией за ней:

Job<void>((ctx) async {
  // Собственный сигнал остановки клиента базы, отменяется вместе с задачей.
  final stop = CancelToken();
  ctx.onCancel(stop.cancel);

  // Ожидание кончается сразу; чтение идёт дальше, а значение отбрасывается.
  final rows = await ctx.abandonable(database.readAll);

  // Ждёт, пока миграция не кончится или не остановится по токену,
  // и только потом задача сдаётся.
  await ctx.join(() => database.migrate(stop));

  // Отмена ждёт конца всего шага, и запущенный им ребёнок
  // не отменяется; её бросит следующая точка.
  await ctx.uncancellable(() async {
    await database.writeVersion();
    await ctx.run(database.readyFlag());
  });

  // Между шагами вычисления оборачивать нечего.
  ctx.check();
  use(rows);
});

Метод	Если отмена приходит во время ожидания
ctx.abandonable(action)	Оставляет действие: перестаёт его ждать и сразу бросает Cancelled, а действие идёт дальше. Его результат отбрасывается, а если вызову передан dispose или discard, уходит туда.
ctx.join(action)	Присоединяется к действию: ждёт его дальше, пока оно не кончится, затем бросает Cancelled вместо возврата значения либо собственную ошибку действия, если оно упало. Если вызову передан dispose или discard, join сперва отдаёт туда значение и дожидается конца освобождения, а потом бросает.
ctx.uncancellable(action)	Придерживает запрос до конца секции: ни onCancel, ни передачи отмены детям, пока она идёт.
ctx.pause(duration)	Сразу бросает Cancelled и снимает свой таймер.
ctx.check()	Бросает, если задача уже приняла отмену.

Работающая задача принимает запрос прямо внутри cancel(), если его не придерживает секция uncancellable и задача не создана с cancellable: false. Приняв его, задача становится отменённой: отмена передаётся запущенным ею детям, вызываются её колбэки onCancel, и задача кончится Cancelled, что бы тело ни делало дальше. Что делают члены контекста на отменённой задаче:

• С этого момента check, abandonable, join, pause, uncancellable, run, start, runAll и each бросают Cancelled на своих контрольных точках.
• onCancel тоже бросает: его колбэки уже вызваны, и зарегистрированный сейчас не был бы вызван никогда.
• onDispose, onDiscard, disown и unattended остаются доступны, чтобы тело могло организовать уборку после отмены.

Тело может отменить само себя, бросив Cancelled или выпустив наружу отмену ребёнка. Оно принимает отмену в момент броска, и задача отменяется так же: отмена передаётся её детям, её колбэки onCancel вызываются, и она кончается Cancelled.

Внутри действия, переданного контексту, обычный await уместен: контекст не добавляет контрольных точек между шагами этого действия. Шагу, которому нужна своя проверка, нужно своё обращение к контексту. Раздел «Шаг, который должен закончиться» ставит каждую из двух записей в свой join, а потом обе в один и показывает, что делает каждая форма, когда отмена приходит во время первой записи.

У паузы и срока своя страница: «Время» ставит ctx.pause рядом с обычной задержкой и даёт задаче предел через timeout.

Строки под кодом показывают, что он печатает при запуске: база говорит, что пишет, cancel отмечает момент, когда пользователь отменяет, а outcome: показывает, чем завершается job.done. printing, наблюдатель каждой задачи ниже, печатает то, что до него дошло: log: для ctx.log, onError: для ошибки.

Остановка операции

Задача открывает базу, мигрирует её и отдаёт; discard закрывает базу, если задача кончилась не Done. Миграция идёт в три шага по 10 мс, и перед каждым она читает CancelToken, сигнал остановки клиента базы, а не тип этого пакета; найдя его отменённым, она останавливается, бросая DatabaseStopped. Когда пользователь отменяет, миграция должна остановиться, а база закрыться, когда в неё уже никто не пишет. Пользователь отменяет на 15-й миллисекунде, во время второго шага.

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

Тело ждёт миграцию через abandonable, чтобы при отмене задача кончалась сразу:

// Первая попытка: база закрывается под идущей миграцией.
final job = Job<Database>(observer: printing, (ctx) async {
  final database = await ctx.join(
    Database.open,
    discard: (database) => database.close(),
  );
  final stop = CancelToken();

  await ctx.abandonable(() => database.migrate(stop));

  return database;
});

step 1
cancel
database closed
outcome: Cancelled(manual)
step 2 on a closed database
step 3 on a closed database

abandonable кончает ожидание, а не миграцию. Вот что происходит:

1. Отмена выходит из abandonable сразу, и задача кончается.
2. discard закрывает базу, когда миграции остаётся записать ещё два шага.
3. Всё, что миграция бросит потом, уйдёт в onError и по умолчанию дальше в зону.

abandonable подходит операции, от которой задача может уйти: например, чтению, результат которого уже никому не нужен.

Результат, пришедший так поздно, отбрасывается, а если abandonable передан dispose или discard, уходит туда. Пока тело или дети задачи ещё идут, этот колбэк выполняется сразу, и задача его дожидается. Если задача уже выполняет стек уборки, колбэк добавляется в стек. Если задача уже завершилась, колбэк выполняется отдельно. Этот порядок разобран в разделе «Значение, которое приходит поздно» страницы о правилах уборки.

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

join остаётся с миграцией до её конца:

// Вторая попытка: ненужная миграция идёт до конца.
await ctx.join(() => database.migrate(stop));

step 1
cancel
step 2
step 3
database closed
outcome: Cancelled(manual)

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

Токен через onCancel

Миграция не знает об отмене задачи: она слушает только свой токен. Отменить токен вместе с задачей можно через onCancel:

final stop = CancelToken();
ctx.onCancel(stop.cancel);

await ctx.join(() => database.migrate(stop));

step 1
cancel
step 2
migration stopped
onError: DatabaseStopped
database closed
outcome: Cancelled(manual)

Колбэк, переданный в ctx.onCancel, вызывается синхронно, когда задача принимает отмену, до того как тело дойдёт до контрольной точки. Здесь он отменяет токен, миграция читает его перед третьим шагом и бросает DatabaseStopped, а join дожидается этого, прежде чем задача закроет базу. Запрос, который нужно прервать, или подписка, которую нужно отменить, проходят через onCancel так же. Остановке, которая занимает время, этого мало, и что ей нужно, показывает раздел «Остановка, которая занимает время» в конце страницы.

Строка onError: и есть тот самый DatabaseStopped. join бросает ошибку своего действия как есть, даже после отмены, а тело её не ловит. Задача всё равно кончается Cancelled, поэтому ни один исход эту ошибку не несёт. Она идёт путём любого такого провала тела: к наблюдателю, затем в зону, если наблюдатель не перехватит её в onUnanswered. Каждая остановка попадала бы в отчёты приложения о сбоях как сбой. Поэтому приложение один раз, в main, называет остановку своего клиента базы, и каждая задача на этой странице идёт с этой строкой:

Job.isCancellationError = (error) => error is DatabaseStopped;

Названная им ошибка доходит до наблюдателя и на нём останавливается. Фильтр разобран в разделе «Ошибки отмены» страницы об ошибках. Что видит catch вокруг этого join, разобрано ниже, в разделе «Перехват ошибок операции».

Шаг, который должен закончиться

Миграция позади, и последний шаг помечает базу готовой: пишет версию схемы, а затем флаг готовности. База с версией, но без флага не старая и не готовая. Две записи делаются двумя вызовами базы, writeVersion и writeReadyFlag. Пользователь отменяет, пока пишется версия.

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

join ждёт своё действие, поэтому каждая запись идёт через него:

// Первая попытка: после отмены во время первой записи второй не будет.
await ctx.join(database.writeVersion);
await ctx.join(database.writeReadyFlag);

cancel
version written
outcome: Cancelled(manual)

Шаг не доходит до конца: база остаётся с версией, но без флага готовности. Каждый join служит отдельной контрольной точкой. Первый дожидается версии и бросает отмену, и до второго тело не доходит.

Один join на шаг

await ctx.join(() async {
  await database.writeVersion();
  await database.writeReadyFlag();
});

cancel
version written
ready flag written
outcome: Cancelled(manual)

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

Шаг, который запускает ребёнка

Шаг тот же, но флаг теперь пишет отдельная задача: её возвращает database.readyFlag(), а шаг запускает её дочерней через ctx.run. Пользователь отменяет, пока пишется версия.

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

У шага остаётся его единственный join:

// Первая попытка: внутри join ctx.run бросает отмену.
await ctx.join(() async {
  await database.writeVersion();
  await ctx.run(database.readyFlag());
});

cancel
version written
outcome: Cancelled(manual)

Флаг снова потерян. join ждёт своё действие, но задача принимает отмену сразу, и дальше ctx.run бросает её вместо того, чтобы запустить ребёнка; уже запущенный ребёнок был бы отменён.

Придержать отмену

uncancellable придерживает отмену, пока шаг не кончится:

await ctx.uncancellable(() async {
  await database.writeVersion();
  await ctx.run(database.readyFlag());
});

cancel
version written
ready flag written
outcome: Cancelled(manual)

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

return ctx.uncancellable(...) тоже не выносит значение наружу: задача принимает отмену раньше, чем тело вернёт значение, оно не доходит ни до исхода, ни до value, и регистрации onDiscard срабатывают. Шаг, который сделал что-то ценное, сохраняет это внутри секции, отменена задача или нет:

final receipt = await ctx.uncancellable(() async {
  final receipt = await api.pay(order);
  // Сохраняем в любом случае: задача ещё может кончиться Cancelled.
  await store.save(receipt);
  return receipt;
});

Если сохранять, только когда ctx.heldCancellation показывает отмену, чек пропадёт. Геттер говорит, придержана ли отмена в момент чтения, а отмена может прийти и после него: до закрытия секции, пока задача ждёт детей, пока убирает за собой. Отмена, которую задача отклонить не может, например правило движка, не придерживается вовсе: она сразу помечает задачу, и показывает её только ctx.job.isCancelled.

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

Всегда ожидайте ctx.uncancellable. Секция открывается при вызове, даже если её future не ждут. Без await секция может пережить тело; если задача завершится раньше, придержанная отмена потеряется: cancel() вернётся, а исходом будет Done. Если же, когда секция закрывается, ещё работает ребёнок, отмена срабатывает тогда и становится исходом поверх значения тела. Секция придерживает запрос на отмену, а не отмену, которую бросает само тело. Если тело пошло дальше, не дождавшись секции, и потом отменило само себя, бросив Cancelled или выпустив наружу отмену ребёнка, задача принимает эту отмену сразу. onCancel вызывается, пока секция ещё открыта, и abandonable внутри неё бросает.

Задача, которая отклоняет отмену

Задача пишет резервную копию, а оборванная копия хуже никакой: всё тело — шаг, который должен закончиться. Если нужно защитить всё тело, а не одну секцию, создайте Job(body, cancellable: false). Она отклоняет обычную отмену после запуска тела, а до него её ещё можно отменить:

Когда приходит отмена	Что делает задача
До запуска тела	Кончается исходом Cancelled с started: false, и тело не запускается вовсе
После запуска тела	Отклоняет обычную отмену: тело идёт дальше, а cancel() ждёт, пока задача кончится

Флаг не даёт оборвать тело, которое уже началось, а до старта обрывать нечего. Задача, созданная через Job(body), стартует на следующей микротаске, поэтому не запущенной её застаёт только отмена, которая пришла раньше. Задача, созданная через Job.deferred, ждёт, пока её запустят: через start() или через ctx.run родителя. Запустят ли её вообще, ядро не знает. Откажи такая задача в отмене и не запусти её потом никто, она не кончилась бы никогда: её done не завершился бы, а cancel(), которому она отказала, ждал бы вечно. Более сильной формы, чтобы снять задачу, у cancel() нет.

Движок, который держит задачи до старта, знает то, чего не знает ядро: он запускает их сам. solo, построенный на этом ядре, сам запускает задачи своей очереди и отклоняет отмену за задачу с cancellable: false, пока она ждёт там. Такой отказ строит раздел «Своя очередь» страницы о движке. Ещё solo отменяет задачу тогда, когда состояние контроллера перестаёт ей подходить: задача там сама объявляет, в каких состояниях она работает. Такую отмену не придерживают ни секция, ни cancellable: false.

Перехват ошибок операции

Неудачная миграция здесь не смертельна: тело пишет её в лог и идёт дальше помечать базу готовой. Отменённая задача всё равно должна остановиться, а в логе должны быть сбои, а не отмены. Пользователь отменяет на 15-й миллисекунде.

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

Отмена бросает Cancelled, поэтому ветка для него стоит первой и пробрасывает его дальше:

// Первая попытка: ошибка, вызванная токеном, записывается как сбой.
final stop = CancelToken();
ctx.onCancel(stop.cancel);

try {
  await ctx.join(() => database.migrate(stop));
} on Cancelled {
  rethrow;
} on Exception catch (error) {
  ctx.log('migration failed: $error');
}
await ctx.join(() async {
  await database.writeVersion();
  await database.writeReadyFlag();
});

step 1
cancel
step 2
migration stopped
log: migration failed: DatabaseStopped
outcome: Cancelled(manual)

Лог называет отмену неудачной миграцией. Вот что происходит:

1. Токен останавливает миграцию, и из join выходит собственный DatabaseStopped миграции, а не Cancelled: join бросает ошибку своего действия как есть.
2. Ветка для Cancelled его пропускает, и его ловит on Exception. Код ветки on Exception выполняется на уже отменённой задаче.
3. Задача всё равно кончается Cancelled: join следующего шага бросает, не начав своё действие.

Без токена ветка для Cancelled работает: миграция доходит до конца, и join бросает Cancelled. Без этой ветки on Exception ловит и этот Cancelled, потому что Cancelled реализует Exception, и пишет migration failed: Cancelled(manual).

Спросить задачу

Отменена ли задача, спрашивают у задачи, а не у ошибки:

try {
  await ctx.join(() => database.migrate(stop));
} on Exception catch (error) {
  ctx.check();
  ctx.log('migration failed: $error');
}

step 1
cancel
step 2
migration stopped
outcome: Cancelled(manual)

ctx.check() бросает отмену задачи, если та её приняла, что бы ни поймал catch. Миграция, упавшая сама, пишется в лог, и тело идёт дальше. В отличие от раздела «Токен через onCancel», строки onError: DatabaseStopped здесь нет: DatabaseStopped ловит on Exception, и из тела он не выходит. Тело кончается отменой задачи из ctx.check(), а отмена в onError не попадает. Если ветка должна ловить всё, включая Error, как on Object catch (error), первой строкой в ней нужен тот же ctx.check().

Платит ветка за это провалами, которые приходят после отмены. Когда задача её приняла, ctx.check() бросает отмену, что бы ни поймал catch, даже настоящий провал. Миграция, упавшая сама уже после отмены пользователя, в лог не попадает, а тело кончается отменой, так что о провале не слышат ни onError, ни зона. Если такой провал надо услышать, ветка на отменённой задаче бросает пойманное дальше вместо ctx.check():

try {
  await ctx.join(() => database.migrate(stop));
} on Exception catch (error) {
  if (ctx.job.isCancelled) rethrow;
  ctx.log('migration failed: $error');
}

Тогда провал доходит до onError и до зоны, а остановка миграции не идёт дальше onError, потому что её называет фильтр, поставленный в main. Выбор такой: ветка с ctx.check() теряет провал, который приходит после отмены, а ветке, которая бросает дальше, нужен Job.isCancellationError в main, иначе до зоны дойдёт и остановка.

Если задача уже приняла отмену, перехват её Cancelled эту отмену не снимает. Код после catch выполняется, но следующая контрольная точка снова бросает, и исходом всё равно будет Cancelled, даже если тело вернёт значение.

Cancelled, который задаче не принадлежит

Cancelled, перехваченный в теле, не всегда принадлежит самой задаче. Операция бросает его, когда ждёт value отменённой задачи. Дочерняя задача, запущенная через ctx.run и отменённая напрямую, а не через родителя, бросает свой Cancelled из run. ctx.check() тогда проходит, потому что сама задача не отменена, и тело идёт дальше и может кончиться Done. Ветка из раздела «Спросить задачу» пишет такой Cancelled в лог как сбой.

Страница показывает миниатюру, когда та отрисовалась, а когда нет, уходит без неё; упавшая миниатюра попадает в лог, а отменённая — нет. Раз ctx.check() прошёл, пойманный Cancelled задаче не принадлежит, и проверка типа отличает отменённую необязательную дочернюю задачу от упавшей:

final thumbnail = Job.deferred<String>(renderThumbnail);
try {
  return 'page with ${await ctx.run(thumbnail)}';
} on Exception catch (error) {
  ctx.check();
  if (error is! Cancelled) ctx.log('thumbnail failed: $error');
  return 'page without a thumbnail';
}

Ветка, которая пробрасывает любой Cancelled, кончает и саму задачу: с причиной BodyCancelReason, если это Cancelled ребёнка, и с собственной причиной чужой задачи, если это Cancelled операции. Эти причины разбирает раздел «Почему задачу отменили» страницы об исходах.

Остановка, которая занимает время

Задача ведёт запись с устройства и освобождает его в onDispose. Устройство останавливается не сразу, а пользователь отменяет задачу во время записи.

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

Остановка запускается работой, которую задача не ждёт:

// Первая попытка: устройство освобождается, пока ещё останавливается.
ctx.onDispose(device.release);
ctx.onCancel(() => ctx.unattended(device.stop));
await ctx.abandonable(device.record);

cancel
device released
outcome: Cancelled(manual)
device stopped

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

Дождаться остановки

Чтобы кончиться только после остановки устройства, задача ждёт через join операцию, которую остановка прерывает. Здесь это запись, и она кончается только тогда, когда устройство остановилось:

ctx.onDispose(device.release);
ctx.onCancel(() => ctx.unattended(device.stop));
await ctx.join(device.record);

cancel
device stopped
device released
outcome: Cancelled(manual)

join ждёт, пока остановка не закончит запись, так же, как ждёт миграцию в разделе «Токен через onCancel». Если операция кончается, как только остановка началась, join ждать нечего, и задача кончается, пока устройство ещё останавливается.

Остановка идёт через ctx.unattended, а не прямо в async-колбэк. onCancel принимает void Function(). Dart пропустит туда и async-функцию, но задача не узнает, что с ней стало:

1. Future, которую она вернёт, не ждёт никто.
2. Её ошибка уходит мимо наблюдателя в зону, где вызван колбэк. Для отмены, принятой внутри cancel(), это зона кода, который вызвал cancel().
3. В onError попадает только то, что колбэк бросит синхронно, а async-функция синхронно не бросает никогда: даже ошибка до её первого await уходит в её future.

Работа в изоляте

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

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

Isolate.run выполняет функцию в новом изоляте и возвращает её результат, а join его ждёт:

// Первая попытка: изолят ничто не останавливает.
ctx.onDiscard(index.removeFile);
await ctx.join(() => Isolate.run(index.build));

step 1
cancel
step 2
step 3
index file removed
outcome: Cancelled(manual)

Сборка доходит до конца, как миграция в разделе «Вторая попытка»: join её ждёт, и никто не велит ей остановиться. И велеть нечем: Isolate.run возвращает future и держит изолят при себе, так что нет ни изолята, который можно убить, ни токена, который сборка прочла бы. abandonable вместо join сразу кончает задачу, и файл удаляется, пока изолят ещё пишет в него.

Убить изолят

Isolate.spawn отдаёт изолят, который запускает, а onCancel его убивает. Функция ниже встаёт на место Isolate.run для синхронной работы:

/// Выполняет синхронную [work] в отдельном изоляте, который отмена [ctx]
/// убивает.
Future<T> runInIsolate<T>(JobContext ctx, T Function() work) async {
  final port = ReceivePort();
  void Function()? unregister;
  final Object? message;
  try {
    final isolate = await ctx.join(
      () => Isolate.spawn(
        _runAndExit<T>,
        (work, port.sendPort),
        paused: true,
        onError: port.sendPort,
        onExit: port.sendPort,
      ),
      discard: (isolate) => isolate.kill(priority: Isolate.immediate),
    );
    unregister = ctx.onCancel(() => isolate.kill(priority: Isolate.immediate));
    isolate.resume(isolate.pauseCapability!);
    // Первое сообщение и есть ответ: значение, ошибка или выход изолята,
    // убитого раньше, чем он ответил.
    message = await ctx.join(() => port.first);
  } finally {
    unregister?.call();
    port.close();
  }
  return switch (message) {
    (final T value,) => value,
    [final error, final stackTrace] => throw RemoteError(
      '$error',
      '$stackTrace',
    ),
    _ => throw StateError('the isolate ended without an answer'),
  };
}

void _runAndExit<T>((T Function(), SendPort) task) {
  final (work, port) = task;
  Isolate.exit(port, (work(),));
}

ctx.onDiscard(index.removeFile);
await runInIsolate(ctx, index.build);

step 1
cancel
index file removed
outcome: Cancelled(manual)

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

1. Isolate.spawn запускает изолят с работой и портом. Изолят шлёт туда значение в записи, чтобы его нельзя было принять за список, который шлёт onError, и выходит. onError и onExit шлют в тот же порт, поэтому первое сообщение и есть ответ: значение, ошибка работы или выход изолята, который умер, не ответив.
2. Изолят запускается на паузе, через join с discard, который его убивает. Отмена, пришедшая, пока изолят запускается, застаёт join в ожидании, и, как сказано в таблице в начале страницы, join отдаёт изолят в discard, как только тот появился, и только после этого бросает отмену. Изолят умирает, не выполнив ни строки работы; без discard он так и остался бы на паузе.
3. onCancel убивает изолят с Isolate.immediate, и тот останавливается посреди шага Dart-кода: своих контрольных точек работе не нужно. Вызов нативного кода, например sleep, сначала доходит до конца. resume пускает работу только тогда, когда onCancel уже на месте, а finally его снимает, чтобы задача, которая запускает изоляты один за другим, не копила по убийству на каждый.
4. Второй join ждёт ответа, а убитый изолят отвечает своим выходом. Задача кончается только тогда, когда изолят мёртв, и onDiscard удаляет файл уже после него. Ответ — сообщение, а не ошибка, поэтому на отменённой задаче join бросает её Cancelled, и ctx.check() не нужен, в отличие от раздела «Спросить задачу».

В отличие от Isolate.run, ошибка работы выходит из runInIsolate как RemoteError с текстом и стеком ошибки: onError шлёт и то и другое строками, а сам объект ошибки остаётся в изоляте. Всё, что держит work, копируется в новый изолят вместе с ней. Замыкание, которое держит ReceivePort, скопировать нельзя: изолят не запускается, и runInIsolate бросает ошибку Isolate.spawn.