Asyncio Queue и Semaphore: конвейер с ограничением нагрузки

Маркировка деталей конвейера

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

Из этого буфера заготовки забирает бригада маркировщиков. Нанесение гравировки требует лазерной обработки, причём число одновременно работающих лазеров ограничено, поэтому доступ к ним регулируется семафором. После успешной маркировки работник вносит серийный номер детали в общий реестр партии, защищённый замком от одновременной записи.

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

Что уже дано

При выполнении кода уже доступны фабрика станций make_marking_station(buffer_size, laser_count) и сопутствующие вспомогательные структуры. Созданный объект станции station содержит:

• station.queue — очередь asyncio.Queue ограниченной вместимости buffer_size;
• station.slots — семафор asyncio.Semaphore, ограничивающий число параллельно работающих лазеров величиной laser_count;
• station.lock — замок asyncio.Lock для синхронизации записи в реестр;
• station.registry — общий реестр деталей, поддерживающий метод .append(serial), итерирование и взятие длины len();
• готовый метод station.start_workers(worker_factory, worker_count, *, task_group=None), создающий ровно worker_count задач asyncio.Task из переданной фабрики корутин worker_factory() и возвращающий их список. Если передан аргумент task_group, задачи создаются через него, иначе — через asyncio.create_task.

Вам необходимо реализовать асинхронную функцию mark_parts.

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

Напишите асинхронную функцию mark_parts(parts, engrave, station, worker_count) -> list[str].

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

• parts — конечный список строковых серийных номеров деталей (от 0 до 200 строк длиной от 1 до 64 символов);
• engrave — асинхронную функцию гравировки engrave(serial), которая успешно завершается без возвращаемого значения;
• station — готовую станцию маркировки;
• worker_count — целое число маркировщиков в пуле (от 1 до 16, а параметры buffer_size и laser_count станции лежат в диапазоне от 1 до 16).

Правила работы конвейера:

• Загрузка деталей в очередь station.queue (поставщик) должна происходить параллельно с работой маркировщиков. При заполнении буфера добавление приостанавливается (await station.queue.put(...)), создавая противодавление, пока маркировщики не освободят место.
• Фиксированный пул маркировщиков создаётся ровно один раз через предоставленный метод station.start_workers(worker_factory, worker_count, ...) с числом рабочих worker_count. Не создавайте отдельную задачу на каждую отдельную деталь; извлечение данных из очереди и вызовы engrave должны выполняться задачами этого зарегистрированного пула.
• Каждый рабочий держит не более одной извлечённой детали за раз до полного завершения её гравировки, внесения в реестр и подтверждения в очереди.
• Гравировка await engrave(serial) вызывается ровно один раз для каждой детали строго под защитой квоты семафора station.slots.
• Добавление серийного номера в station.registry.append(serial) выполняется только после завершения гравировки этой детали и строго под защитой замка station.lock. Дедупликация не выполняется: каждый повтор серийного номера обрабатывается и записывается отдельно.
• После завершения обработки детали рабочий подтверждает её выполнение вызовом station.queue.task_done().
• Работа завершается строго тогда, когда все переданные детали полностью загружены, отмаркированы, занесены в реестр и подтверждены через очередь. После этого все запущенные рабочие должны быть остановлены, а их задачи завершены и ожидаемы (await). Все захваченные квоты семафора и замка должны оставаться освобождёнными.
• Функция должна вернуть список серийных номеров из station.registry в виде обычного списка list[str]. Порядок записей, обработанных параллельно, может быть произвольным.

Ввод и вывод

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

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

async def example():
    station = make_marking_station(1, 2)
    async def engrave(serial):
        await asyncio.sleep(0)
    registry = await mark_parts(["Q2", "Q1", "Q2"], engrave, station, 2)
    print(sorted(registry))

asyncio.run(example())

Вывод:

['Q1', 'Q2', 'Q2']