Дочерние задачи и цепочки

Корневая Job может разделить работу на дочерние:

// Создан контроллером, но ещё никем не запущен: очередь возьмёт его
// через `sync` ниже, родитель через `ctx.run`.
SoloJob<String> _sync(int item) =>
    job<Ready, String>(key: _Op.sync, (ctx) async {
      // Ребёнок на стриме: прогресс загрузки, который API закрывает,
      // когда загрузка кончилась. Родитель ждёт ребёнка и без await,
      // а отмена родителя отменяет и его.
      ctx.each(
        api.progress(item),
        (childCtx, sent) =>
            childCtx.emit(childCtx.state.copyWith(sent: sent)),
      );

      // Создан контроллером, запущен родителем и минует очередь:
      // пока он работает, очередь держит родитель.
      final upload = job<Ready, String>(
        key: _Op.upload,
        (childCtx) => childCtx.join(() => api.push(item)),
      );
      return ctx.run(upload);
    });

Job<String> sync(int item) => enqueueJob(_sync(item));

job(...) создаёт задачу и ничего не запускает. enqueueJob даёт ей место в очереди, а ctx.run внутри тела делает её ребёнком; _sync написан один раз и дальше на этой странице идёт обоими путями.

Ребёнок запускается сразу, минуя очередь, с проверкой собственных стартовых правил: рабочего типа, canStart и keepWhile, тех самых, что описывает раздел «Правила» страницы о состоянии. Родитель удерживает очередь до завершения всех детей, даже если его тело вернулось раньше: sync кончается, когда загрузка вернула путь, а API закрыл прогресс.

Что происходит между родителем и ребёнком сверх показанного на этой странице, принадлежит ядру, и описывает это страница async_job о детях. Она показывает, что принятая отмена родителя передаётся детям, а упавшее тело даёт им доработать и ждёт их. Там же сказано, почему child.ignoreFailure() не заменяет обработку future от ctx.run. Несколько детей бок о бок составляют группу ctx.runAll, о ней раздел «Ожидание нескольких детей» страницы о группах. Там, где эти страницы создают ребёнка через Job.deferred, контроллер создаёт его через job(...) или ctx.child. ctx.child — для кода, у которого есть только контекст: он делает задачу того же контроллера с рабочим типом и keepWhile той задачи, которая его сделала. Других задач ctx.run и ctx.runAll контроллера не берут, а каждая ветка группы отвечает по своим стартовым правилам, как ребёнок ctx.run.

ctx.each(stream, onData) подписывается в ребёнке, которому принадлежит подписка, и отдаёт колбэку каждое событие вместе с контекстом ребёнка: так _sync выше следит за прогрессом загрузки. О таких детях в контроллере страница «Стримы»: что делает подписка, сделанная в теле руками, сколько ребёнок держит очередь и как следить за другим контроллером.

api и analytics принадлежат приложению, из которого взяты примеры, _Op — его перечисление ключей задач, а в состоянии Ready, с полями sent, path и paused, работает его контроллер.

Future от ctx.run

// Родитель сам отвечает за упавшую загрузку.
Job<bool> trySync(int item) =>
    enqueue<Ready, bool>(key: _Op.trySync, (ctx) async {
      try {
        final path = await ctx.run(_sync(item));
        ctx.emit(ctx.state.copyWith(path: path));
        return true;
      } on ApiException {
        return false;
      }
    });

ctx.run(child) возвращает Future<T> с результатом ребёнка. Она ждёт ребёнка, его детей и его уборку: catch выше срабатывает, когда API закрыл прогресс, за которым следит _sync. При успехе она проверяет отмену и правила состояния родителя, прежде чем вернуть значение. Поэтому значение, которое родитель обязан освободить, регистрируют на самом вызове, ctx.run(child, dispose: ...), а не строкой ниже: почему, говорит раздел «discard для значения, которое уходит» страницы о ресурсах. Ошибка или отмена ребёнка бросается в тело родителя со своим стеком вызовов.

Ветка выше называет ApiException намеренно. Cancelled реализует Exception, поэтому on Exception или on Object на её месте поймали бы вместе с ошибками API и отмену, как ребёнка, так и самой trySync. Чего это стоит и что широкая ветка обязана сделать первым делом, показывает раздел «Перехват ошибок внутри тела» страницы об ошибках.

Работа рядом с ребёнком

Job<String> syncAndAnnounce(int item) =>
    enqueue<Ready, String>(key: _Op.syncAndAnnounce, (ctx) async {
      // Сохранена, но не ожидается: загрузка идёт, пока родитель работает.
      // Ошибка загрузки здесь игнорируется, а тело упадёт с ней
      // на `return`.
      final uploading = ctx.run(_sync(item))..ignore();
      await ctx.join(() => analytics.send('started $item'));
      return uploading;
    });

Чтобы родитель работал одновременно с ребёнком, сохраните future, которую вернул ctx.run, и дождитесь её позже. Тогда их записи состояния могут чередоваться, и каждая задача отвечает своим правилам за каждую из них. Запись ребёнка, которую не принимает рабочий тип или keepWhile родителя, отменяет родителя, а с ним и ребёнка.

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

Три способа игнорировать ребёнка

В родителе	Ошибка, которую бросает future от ctx.run	Провал тела ребёнка, который кончился Cancelled
ctx.run(child).ignore()	выброшена: за неё никто не отвечает	уходит в onUnanswered
child.ignoreFailure()	не обработана, уходит в зону	погашена
ctx.start(child)	future нет: провал ребёнка уходит в onUnanswered	уходит в onUnanswered

Там, где в таблице стоит onUnanswered, ошибка, которую называет isExpectedError ребёнка, дальше onError не идёт, и ответа за неё не просят ни у кого. Future от ctx.run всё равно бросает эту ошибку, поэтому при child.ignoreFailure() она по-прежнему уходит в зону необработанной.

ctx.run(child).ignore() явно игнорирует результат этой future, при этом родитель продолжает ждать детей. Одного child.ignoreFailure() недостаточно для обработки ошибок future, которую вернул ctx.run. ctx.start(child) запускает ребёнка и не возвращает future: его провал уходит в onUnanswered контроллера, как любой провал, который никто не наблюдал, а родитель всё так же ждёт ребёнка. ctx.start берите для ребёнка, за провал которого кто-то всё же должен ответить, а ctx.run(child).ignore() — только для того, чей провал никому не нужен: onError контроллера его слышит, и никто за него не отвечает.

Если ребёнок кончился Cancelled, а его тело упало, эта ошибка в тело родителя не бросается. Так бывает, например, когда отмена остановила операцию, а тело бросило ошибку этой операции. await ctx.run(child) бросает Cancelled, а ошибка уходит в onUnanswered контроллера: по умолчанию в Solo.unansweredHandler, а без него в зону. child.ignoreFailure() её гасит, а ctx.run(child).ignore() нет: он обрабатывает то, что бросает future, а она бросает отмену. Ошибка, которую приложение назвало в Job.isCancellationError, дальше onError контроллера не идёт, как в разделе «Токен» страницы об отмене.

Ребёнок, запущенный в теле, берёт isExpectedError задачи, которая его запустила, если у него нет собственного, а собственный бывает только у ребёнка из job(...): у ctx.child и ctx.each такого параметра нет. Широкая проверка вроде error is Exception в предикате родителя гасит и баги каждого ребёнка, поэтому держите предикат проверкой типов, которые бросает API: об этом раздел «Назвать ожидаемые ошибки» страницы об ошибках.

Ребёнок, которого отвернули правила

Job<String> resend(int item) =>
    enqueue<Ready, String>(key: _Op.resend, (ctx) {
      // Ребёнок со своим правилом: на паузе загрузки нет.
      final upload = job<Ready, String>(
        key: _Op.upload,
        canStart: (state) => !state.paused,
        (childCtx) => childCtx.join(() => api.push(item)),
      );
      return ctx.run(upload);
    });

Когда состояние на паузе, ничего не загружается, и resend кончается так:

Cancelled(body: Job(_Op.upload): Cancelled(rules: canStart))

Ребёнок, которого отвернули стартовые правила, не запускается вовсе. Сначала он получает родителя и уровень вложенности, и только потом его завершают Cancelled с RulesCancelReason. Наблюдатель слышит отказ как исход задачи этого дерева, а job.level говорит ему, насколько глубоко под родителем стоит эта задача; child.done хранит саму отмену. В список ожидания он не попадает, поэтому родитель никого не ждёт; но future от ctx.run эту отмену всё равно несёт, и именно она кончает resend выше.

body: в этой строке — это BodyCancelReason: такую причину ядро даёт телу, выпустившему наружу отмену своего ребёнка, а после неё идут ребёнок и отмена, которой он кончился. В разделе «Сведения об отмене» есть все причины и то, как каждая печатается.

Другое дело, когда правила canStart и keepWhile сами бросают ошибку. Ошибка принадлежит правилу, ребёнок завершается Failed с ней, а ctx.run бросает её синхронно: строка после вызова не выполняется.

Цепочки завершённой работы

// Запускается после успеха `sync`, вне очереди, с обычным JobContext.
Job<void> syncAndReport(int item) =>
    sync(item).then((ctx, path) => analytics.send(path));

job.then((ctx, value) => ...) создаёт продолжение: Job, которая запускается после успеха источника, включая его детей и уборку, а для sync это момент, когда API закрыл прогресс. Колбэк получает результат и новый контекст ядра JobContext, а вернуть может значение или future. Если sync упала, колбэк не вызывается, а Job, которую вернул then, кончается Failed с той же ошибкой.

Продолжение не наследует контекст состояния, правила, наблюдателя или позицию в очереди контроллера. У него есть собственный необязательный наблюдатель, а имя ему даёт сам then: Job(then), если вызов не передал key или describe. Оно принадлежит тому, кто вызвал then, а не контроллеру: хуки контроллера и Solo.observer его не слышат. Провал достаётся тому, кто читает исход продолжения, в том числе провал sync, пришедший по цепочке. За провал, который никто не читает, и за ошибку без исхода продолжение отвечает само: в onUnanswered своего наблюдателя, если тот подмешивает JobAnswerer, а иначе в зоне, где вызвали then. Solo.unansweredHandler их тоже не слышит.

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

Отмена идёт по цепочке в обе стороны, как описывает страница «Цепочки» async_job. Для контроллера это значит, что close() достаёт продолжение через незавершённый источник, но не владеет тем, которое уже работает после его завершения.

Ребёнком продолжение не бывает. Внутри тела контроллера ctx.run принимает Job этого контроллера, те, что делает job(...) или ctx.child и никто не поставил в очередь, а продолжение не из них. Оно корневая Job ядра, простая Job, а не ChildJob, поэтому ctx.run(tail) не компилируется:

The argument type 'Job<void>' can't be assigned to the parameter type 'ChildJob<dynamic>'.

Job ядра, созданная вручную через Job.deferred, — это ChildJob, и она компилируется, а контроллер отказывает ей ошибкой ArgumentError:

Invalid argument (job): was not created by this Solo (make it with job(...) or ctx.child of this controller): "Job()"

Шаги подряд

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

// То же разделение, что у `sync`: сам шаг и вход через очередь.
SoloJob<void> _recordPath(String path) => job<Ready, void>(
  key: _Op.record,
  (ctx) async => ctx.emit(ctx.state.copyWith(path: path)),
);

Job<void> recordPath(String path) => enqueueJob(_recordPath(path));

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

// Первая попытка: продолжение встаёт в очередь за чужой работой.
Job<void> syncAndRecord(int item) =>
    sync(item).then((ctx, path) => recordPath(path).value);

then читается как «а затем». Его колбэк получает обычный JobContext ядра: emit на нём нет и состояния за ним тоже. Так что само продолжение состояние контроллера не пишет. Оно зовёт recordPath, обычную Job контроллера: emit делает она, и она же встаёт в конец его очереди. Конец источника делает сразу две вещи: освобождает место в очереди и запускает продолжение. Поэтому всё, что поставили в очередь, пока работал источник, стоит впереди новой Job: если другой метод контроллера, save(), позвали, пока шла sync, порядок получается sync, save, recordPath. А контроллер, который закрывается дренажом, новых Job не берёт вовсе: путь не записывается, а продолжение кончается Cancelled(closed).

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

// Вторая попытка: sync ждёт конца этой задачи, а задача ждёт sync.
Job<void> syncAndRecord(int item) =>
    enqueue<Ready, void>(key: _Op.syncAndRecord, (ctx) async {
      final path = await sync(item).value;
      await recordPath(path).value;
    });

Теперь очередь на оба шага держит одна задача, и зовёт она методы, которые у контроллера уже есть. Но метод, который ставит в очередь, создаёт корневую Job, откуда бы его ни позвали, и она ждёт место, которое держит сам вызвавший. А вызвавший ждёт её: загрузка не начинается, и с места ничего не сдвинется, пока close() не отменит обе. Без await такой метод ловушкой не будет: поставленная им Job выполнится после этой задачи и после всего, что поставили раньше.

Само зависание ничего не показывает, но Solo.debug предупреждает о нём, когда задачу ставят в очередь:

enqueue Job(_Op.sync) while Job(_Op.syncAndRecord) is running: it starts after Job(_Op.syncAndRecord) ends

Метод с droppable, вызванный изнутри своей же задачи, заводит в ловушку, из которой не выручает даже close(). Он отдаёт эту самую задачу: если sync ставит задачу с droppable, а её тело повторяет попытку через await sync(item).value, задача ждёт сама себя. close() помечает задачу отменённой, но простой await от этого не просыпается, а значение придёт только с концом тела, поэтому и close() ждёт вечно. Solo.debug показывает это в строке дубликата:

enqueue Job(_Op.sync): duplicate of Job(_Op.sync), which is running: if awaited from inside Job(_Op.sync), it waits for itself

Два ребёнка подряд

// Снова `_sync` и `_recordPath`, теперь детьми. Родитель держит очередь,
// пока не закончат дети, поэтому `save()`, поставленная тем временем,
// ждёт обоих шагов.
Job<void> syncAndRecord(int item) =>
    enqueue<Ready, void>(key: _Op.syncAndRecord, (ctx) async {
      final path = await ctx.run(_sync(item));
      await ctx.run(_recordPath(path));
    });

У шага, написанного через job(...), открыты оба пути: enqueueJob даёт ему собственное место в очереди, ctx.run делает его ребёнком задачи, у которой это место уже есть. Используйте детей одного родителя, если вся последовательность должна удерживать очередь без запуска другой корневой Job между шагами.