Asyncio: восстановление пула воркеров после ошибки

Восполнение пула воркеров рендеринга

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

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

Смена штатно финализируется вызовом queue.join(). Постройте диспетчер с динамическим восполнением упавших исполнителей и определите итоговое количество успешно отрендеренных кадров вместе со списком зафиксированных сбоев.

Что уже дано

При выполнении решения в среде доступен класс RenderContext, предоставляющий все необходимые компоненты конвейера:

• context.queue — асинхронная очередь кадров, изначально заполненная всеми объектами на обработку;
• context.sem — семафор, ограничивающий число одновременно выполняемых отрисовок;
• context.lock — замок для синхронизации разделяемых операций;
• context.pool — реестр задач воркеров с методами start(coroutine) для регистрации новой задачи и stop() для остановки пула;
• context.counter — защищённый счётчик с методом increment();
• context.render(frame) — асинхронная функция отрисовки кадра, которая либо успешно завершается, либо выбрасывает стандартное исключение;
• context.failed_frame(task) — метод, возвращающий объект кадра, на котором упала завершившаяся с ошибкой задача воркера.

Вам необходимо реализовать только корутину replace_failed_workers.

Что нужно сделать

Реализуйте асинхронную функцию replace_failed_workers(context, worker_count).

Функция принимает два аргумента:

• context — экземпляр RenderContext, содержащий от 0 до 1000 конечных уникальных хешируемых кадров и настроенный лимит параллелизма от 1 до 20;
• worker_count — положительное целое число от 1 до 20, задающее начальный и неизменный размер пула работающих воркеров.

Функция должна вернуть пару (success_count, failures):

• success_count — целое число успешно отрендеренных кадров;
• failures — список кортежей вида (original_frame, original_error) для всех упавших задач (порядок записей о сбоях произвольный).

Правила работы воркеров и супервизора:

• Каждый воркер запускается и регистрируется в пуле через context.pool.start(worker_coroutine). Суммарное количество одновременно работающих воркеров ни в какой момент времени не должно превышать worker_count.
• Воркер извлекает кадры из context.queue и выполняет вызов await context.render(frame) под защитой семафора context.sem (async with context.sem:).
• При успешном завершении context.render(frame) воркер обязан ровно один раз увеличить счётчик вызовом context.counter.increment() строго под защитой замка context.lock (async with context.lock:), после чего подтвердить завершение кадра вызовом context.queue.task_done().
• Если context.render(frame) выбрасывает исключение, ошибка обязана беспрепятственно выйти наружу из корутины воркера, чтобы его задача asyncio.Task завершилась в аварийном состоянии с этим исключением. Подавлять ошибку внутри воркера, перезапускать обработку этого кадра или продолжать работу того же воркера запрещено.
• Супервизор отслеживает упавшие задачи воркеров. Для каждой завершившейся с ошибкой задачи супервизор извлекает исходное исключение через task.exception(), определяет сбойный кадр с помощью context.failed_frame(task), сохраняет пару (frame, error), подтверждает выбывание упавшего кадра из очереди вызовом context.queue.task_done() и запускает ровно одного нового воркера взамен выбывшего. Остальные воркеры продолжают работу.
• Очередь считается полностью обработанной после завершения ожидания await context.queue.join(). После этого все оставшиеся активные воркеры должны быть корректно остановлены (например, через await context.pool.stop()).
• Все слоты семафора и замки к моменту возврата из функции должны быть полностью освобождены.

Ввод и вывод

Стандартные потоки ввода stdin и вывода stdout в решении не используются: считывать ввод и вызывать print() не нужно. После выполнения кода учащегося проверяющая система считывает из пространства имён модуля функцию верхнего уровня replace_failed_workers и вызывает её с проверочными аргументами.

Пример использования

import asyncio
async def render(frame):
    await asyncio.sleep(0)
    if frame == "bad":
        raise RuntimeError("renderer failed")
context = RenderContext(["bad", "ok"], render, 1)
count, failures = asyncio.run(replace_failed_workers(context, 1))
print(count)
print([(frame, type(error).__name__) for frame, error in failures])

Вывод:

1
[('bad', 'RuntimeError')]