Свой движок

Эта страница строит движок на задаче со страницы «Своё поверх ядра»: на MyJob, её контексте MyContext и обёртках _launch и _untilFinished, какими их показывает раздел «Своя задача». Движок — очередь задач с правилом:

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

Строки под кодом показывают, что он печатает при запуске. Наблюдатель задачи printer печатает то, что до него дошло: onError: для ошибки. cancel и sign out отмечают моменты, когда пользователь отменяет задачу или выходит из аккаунта, а outcome: показывает, чем завершается job.done.

Своя очередь

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

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

Очередь хранит задачи, запускает каждую через _launch и пропускает задачу, отменённую в ожидании:

// Первая попытка: задача, отказавшаяся от отмены, тоже пропускается.
final class MyQueue {
  final _waiting = <MyJob<Object?>>[];

  void add(MyJob<Object?> job) => _waiting.add(job);

  Future<void> run() async {
    while (_waiting.isNotEmpty) {
      final job = _waiting.removeAt(0);
      if (job.isFinished) continue;
      job._launch();
      await job._untilFinished;
    }
  }
}

Вторая задача создана с cancellable: false, и пользователь отменяет её, пока она ждёт в очереди:

final first = MyJob<void>(key: 'first', (ctx) => ctx.abandonable(upload));
final second = MyJob<void>(
  key: 'second',
  cancellable: false,
  (_) async => print('second runs'),
);
final third = MyJob<void>(key: 'third', (_) async => print('third runs'));
final queue = MyQueue()
  ..add(first)
  ..add(second)
  ..add(third);
final running = queue.run();
await second.cancel();
await running;
print('second: ${await second.done}');

third runs
second: Cancelled(manual)

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

Пользователь отменяет вторую задачу, пока она ждёт. Задачу, которая не стартовала, ядро заканчивает сразу, что бы ни говорил cancellable: cancellable: false не даёт оборвать тело, которое уже началось, а до старта обрывать нечего.
Проверка в цикле видит, что задача кончилась, и пропускает её.

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

Отказ, пока задача ждёт

Любая отмена, о которой просят задачу, приходит в cancelWith, среди них cancel(), каскад от родителя и отмена, о которой движок просит сам через cancelOwnJob контекста. Чтобы отмена прошла, переопределение передаёт её в super, и анализатор требует, чтобы этот вызов в нём был. Чтобы отклонить отмену, переопределение выходит до этого вызова. MyJob помнит очередь, в которую его добавили. Пока он ждёт в ней, он отклоняет то же, что отклонил бы в работе, а отмена, которую он пропустил, вынимает его из очереди:

  MyQueue? _queue;

  @override
  void cancelWith(Cancelled cancelled, {bool rejectable = true}) {
    final waits = _queue?._waiting.contains(this) ?? false;
    if (waits && !cancellable && rejectable) return;
    _queue?._waiting.remove(this);
    super.cancelWith(cancelled, rejectable: rejectable);
  }

Очередь сообщает каждой задаче, чья она, если задача к этому времени ещё не кончилась:

  void add(MyJob<Object?> job) {
    if (job._queue != null) throw StateError('$job is in a queue already');
    if (job.isRunning) throw StateError('$job runs already');
    _waiting.add(job);
    if (!job.isFinished) job._queue = this;
  }

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

  @override
  void finished() {
    _body = null;
    _queue = null;
  }

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

Отмену отклоняет только задача, которая ждёт в очереди. Кто отвечает на отмену задачи с cancellable: false, зависит от того, где она:

Где задача	Кто отвечает на отмену
Ещё не добавлена в очередь	ядро: задача кончается `Cancelled`, и проверка в цикле её пропускает
Ждёт в очереди	`MyJob`, который её отклоняет
Никем не поставлена в очередь, например её не принял в свой `ctx.run` отменённый родитель	ядро: отклони задача отмену там, она не стартовала бы и не кончилась бы никогда
Забрана из очереди, чтобы стартовать	ядро, которое её отклоняет

Проверка в цикле остаётся для задачи, которую отменили до того, как добавили в очередь.

rejectable равен false у отмены, которую не может отклонить ни одна задача, например у той, которую движок шлёт через обёртку вокруг cancelWith, чтобы снять задачу, что бы ни говорил cancellable, и такая отмена уходит в super. В том же запуске вторая задача теперь выполняется в свой черёд:

second runs
third runs
second: Done(null)

cancel() возвращает future, которая ждёт конца задачи, отклонена отмена или нет, поэтому здесь она завершается, когда вторая задача отработала.

Один цикл и никакого другого старта

  Future<void>? _running;

  // Один цикл за раз: второй вызов получает future цикла, который идёт.
  // Запущенный в микротаске, цикл обнуляет `_running` после того, как её
  // присвоили, даже когда выполнять нечего.
  Future<void> run() => _running ??= Future.microtask(_loop);

  Future<void> _loop() async {
    try {
      while (_waiting.isNotEmpty) {
        final job = _waiting.removeAt(0);
        if (job.isFinished) continue;
        job._launch();
        await job._untilFinished;
      }
    } finally {
      _running = null;
    }
  }

Второй вызов run(), пока цикл идёт, получает future этого цикла. Два цикла брали бы каждый следующую задачу из _waiting, и две задачи работали бы одновременно. Запущенный сразу, цикл, которому нечего выполнять, кончился бы раньше, чем _running получила свою future, и оставил бы эту завершённую future каждому следующему вызову.

MyJob — это ChildJob, поэтому тело, у которого есть задача очереди, может передать её в ctx.run. Прежде чем запустить ребёнка, ядро спрашивает его в adoptedBy, и задача, которая ждёт в своей очереди, отказывает:

  @override
  void adoptedBy(JobContextBase parent) {
    if (_queue?._waiting.contains(this) ?? false) {
      throw StateError('$this waits in its queue');
    }
  }

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

Движок целиком

Движок — это задача со страницы «Своё поверх ядра» и код разделов выше в одной библиотеке Dart, которая импортирует package:async_job/engine.dart: задача и очередь обращаются к приватным членам друг друга.

Класс	Члены	Показаны в разделе
`MyJob`	`_body`, конструктор, `createContext`, `execute`, `_launch`, `_untilFinished`	«Своя задача»
`MyJob`	`_queue`, `cancelWith`, `finished()`	«Отказ, пока задача ждёт»
`MyJob`	`adoptedBy`	«Один цикл и никакого другого старта»
`MyQueue`	`_waiting`	«Первая попытка»
`MyQueue`	`add`	«Отказ, пока задача ждёт»
`MyQueue`	`_running`, `run`, `_loop`	«Один цикл и никакого другого старта»
`MyContext`	только конструктор	«Своя задача»

Где член показан дважды, в силе последний: finished(), который отпускает и очередь, заменяет тот, что отпускает только тело, а add и run заменяют свои версии из первой попытки.

Два вызова run() делят один цикл, и задачи выполняются одна за другой:

final queue = MyQueue();
for (final name in ['photo', 'video']) {
  final job = MyJob<void>((ctx) async {
    print('$name starts');
    await ctx.abandonable(upload);
    print('$name ends');
  });
  queue.add(job);
}
await Future.wait([queue.run(), queue.run()]);

photo starts
photo ends
video starts
video ends

Раздел «Своё правило» даёт этому MyContext правило.

Своё правило

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

check() служит контрольной точкой контекста, и правило движка ставится туда. Пока тело работает, её спрашивают такие вызовы:

Вызов	Спрашивает `check()`
`abandonable`, `pause`, `uncancellable`	перед действием
`join`	перед действием и ещё раз после него
`run`	когда пришло значение ребёнка
`runAll`	перед тем, как отдать значения

ctx.child в таблице нет: он делает ребёнка и ничего не спрашивает. MyContext оставляет child ядра, поэтому ребёнок, сделанный через ctx.child в задаче этого движка, — ребёнок ядра, и его контекст правило не спрашивает: его спрашивает родитель, когда пришло значение ребёнка. Движок, которому правило нужно внутри детей, переопределяет child, как сказано в разделе «Своя задача».

Задача движка скачивает строки через соединение и сохраняет их:

final job = MyJob<void>(observer: printer, (ctx) async {
  ctx.onCancel(() => print('close the connection'));
  final rows = await ctx.join(download);
  print('downloaded $rows rows');
  await ctx.join(() => save(rows));
  print('saved');
}).._launch();

Отменённая, пока идёт скачивание, задача закрывает соединение и кончается Cancelled:

cancel
close the connection
outcome: Cancelled(manual)

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

MyContext переопределяет check(). Сперва он спрашивает super.check(), контрольную точку ядра, вызова которой анализатор требует от переопределения, а потом правило:

  // Первая попытка: нарушенное правило завершает задачу как Failed.
  @override
  void check() {
    super.check();
    if (!account.signedIn) throw const SignedOut();
  }

Пользователь выходит из аккаунта, пока идёт скачивание:

sign out
onError: SignedOut
outcome: Failed(SignedOut)

Правило, которое перестало выполняться, не должно быть провалом. Вот что происходит:

check() бросает SignedOut внутри join, и тело выпускает его наружу.
Задача кончается Failed с ошибкой самого движка, и наблюдатель слышит ошибку.
Соединение никто не закрывает: onCancel вызывается только при отмене.
Ребёнок задачи, если бы он был, доработал бы до конца.

Собственная отмена движка

Правило бросает Cancelled с причиной самого движка:

final class SignedOutReason extends CancelReason {
  const SignedOutReason();

  @override
  String get name => 'signed out';
}

  @override
  void check() {
    super.check();
    if (!account.signedIn) {
      throw Cancelled.by(
        reason: const SignedOutReason(),
        stackTrace: StackTrace.current,
      );
    }
  }

Пользователь выходит из аккаунта, пока идёт скачивание:

sign out
close the connection
outcome: Cancelled(signed out)

После выхода из аккаунта задача теперь кончается так же, как после отмены: соединение закрыто, исход Cancelled. Тело выпускает Cancelled правила наружу и тем отменяет само себя, а задача, тело которой отменило само себя, останавливается так же, как отменённая: её колбэки onCancel вызываются, а дети останавливаются. Об этой отмене задачу никто не просил, поэтому в cancelWith она не приходит: cancellable: false её не отклоняет, а ctx.uncancellable её не придерживает, ведь check() бросает внутри того вызова, который спросил правило, здесь внутри join.

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

Не дожидаясь контрольной точки

С abandonable на месте первого join тело идёт дальше и после выхода из аккаунта:

  final rows = await ctx.abandonable(download);

sign out
downloaded 42 rows
close the connection
outcome: Cancelled(signed out)

abandonable спрашивает правило перед скачиванием, а после него не спрашивает, поэтому тело получило строки, а соединение оставалось открытым, пока правило не спросил join. Движок, который не может ждать следующей контрольной точки или должен остановить задачу, что бы ни поймало её тело, отменяет задачу сам, когда пользователь выходит из аккаунта, тем же Cancelled, который бросает его правило. MyJob получает обёртку вокруг cancelWith, как раньше вокруг start:

  void _stop(Cancelled cancelled) => cancelWith(cancelled, rejectable: false);

Движок зовёт её из account.onSignOut, который аккаунт вызывает, когда пользователь выходит. На странице одна задача; движок, у которого их несколько, зовёт _stop у каждой своей идущей задачи и строит для каждой свой Cancelled:

account.onSignOut = () => job._stop(
  Cancelled.by(
    reason: const SignedOutReason(),
    stackTrace: StackTrace.current,
  ),
);

sign out
close the connection
outcome: Cancelled(signed out)

Два способа, которыми движок останавливает задачу при выходе из аккаунта:

	Правило в `check()`	`_stop` из `onSignOut`
Задача принимает отмену	когда она выходит из тела	внутри выхода
Колбэки `onCancel` вызываются	когда она выходит из тела	внутри выхода, а `abandonable` бросает сразу
Тело, которое её поймало и пошло дальше	не отменено	всё равно кончается `Cancelled`
`cancellable: false`	её не отклоняет	её не отклоняет: `rejectable: false`
`ctx.uncancellable`	её не придерживает	её не придерживает

Обёртка вокруг cancelOwnJob контекста останавливает задачу так же, как _stop: rejectable: false она передаёт сама.