Multiprocessing Python: ограничение очереди семафором

Ограничение очереди матриц семафором

Главный процесс непрерывно формирует тяжёлые вычислительные матрицы и передаёт их на обработку в рабочий процесс через multiprocessing.Queue. Чтобы оперативная память не переполнялась скопившимися в системном буфере объектами, подача задач контролируется межпроцессным семафором свободных слотов: перед отправкой очередной матрицы процесс обязан дождаться свободного разрешения.

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

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

Что уже дано

В редакторе представлена заготовка функции feed_matrices.

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

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

Реализуйте функцию feed_matrices(matrices, work_queue, slots, capacity, worker) -> None.

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

• matrices — конечную последовательность прямоугольных целочисленных матриц (от 0 до 16 матриц; от 0 до 8 строк и от 0 до 12 столбцов; элементы — целые числа от -1000 до 1000, включая булевы значения);
• work_queue — предоставленную межпроцессную очередь multiprocessing.Queue;
• slots — межпроцессный семафор multiprocessing.Semaphore(capacity);
• capacity — начальное число свободных слотов (целое число от 1 до 4);
• worker — уже запущенный рабочий процесс multiprocessing.Process.

Правила отправки и синхронизации:

• Перед каждой отправкой матрицы в work_queue необходимо занять разрешение семафора вызовом slots.acquire().
• Все матрицы из matrices должны быть переданы в исходную очередь строго по порядку с сохранением всех повторений без подмены очереди, семафора или процесса.
• Маркер завершения None разрешено отправлять в очередь только после того, как дочерний процесс завершит обработку абсолютно всех отправленных матриц и вернёт все выданные разрешения в семафор. Факт извлечения матрицы из очереди или фиксированная пауза подтверждением завершения обработки не являются.
• После отправки маркера None функция должна дождаться штатного завершения рабочего процесса вызовом worker.join() и вернуть None.
• Требований к конкретному итоговому значению счётчика семафора не предъявляется: корректность оценивается по порядку захвата слотов, отправки данных, возврата разрешений и финализации воркера.

Ввод и вывод

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

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

probe = make_slot_probe([[[1, 2], [3, 4]], [[5, 6]]], 2)
observed = probe.run()
print(observed['boundary']['returned_none'])
print(observed['aggregates'])

Вывод:

True
[[4, 6], [5, 6]]