Процесс и потоки Python: двухфазная обработка матрицы

Двухфазная нормализация матрицы

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

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

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

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

Что уже дано

В среде выполнения доступна вспомогательная фабрика make_matrix_workspace(matrix, worker_count) и строковая константа UNSCALABLE = 'UNSCALABLE'. Создаваемый объект рабочего пространства workspace берёт на себя арифметические операции и проверку корректности этапов:

• workspace.sum_group(group) — подсчитывает и возвращает сумму строк с индексами range(group, len(matrix), worker_count) в текущем потоке (пустая группа даёт сумму 0);
• workspace.add_sum(group, subtotal) — добавляет подсчитанную подсумму к общему итогу; вызов должен выполняться строго под защитой общего объекта threading.Lock;
• workspace.total — свойство, возвращающее текущую общую сумму матрицы (доступно после завершения первой фазы);
• workspace.normalize_group(group) — делит элементы назначенных строк на workspace.total в текущем потоке; вызывается только после добавления сумм всеми группами и строго при ненулевом значении total;
• workspace.normalized_matrix() — возвращает итоговую преобразованную матрицу после того, как все потоки завершат нормализацию.

Также в окружении объявлены функции observe_matrix_probe и make_matrix_example_observation для проверки хода вычислений. В редакторе вам предстоит реализовать функции matrix_worker и run_matrix_pipeline.

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

Реализуйте две функции: matrix_worker(matrix, worker_count, result_queue) и run_matrix_pipeline(matrix, worker_count).

1. matrix_worker(matrix, worker_count, result_queue):

• Выполняется в отдельном дочернем процессе multiprocessing.Process.
• Создаёт рабочее пространство через workspace = make_matrix_workspace(matrix, worker_count).
• Создаёт ровно worker_count нативных потоков threading.Thread (по одному на каждую группу 0 <= group < worker_count) и координирует их с помощью общего объекта threading.Lock и барьера фаз.
• Фаза 1: каждый поток вычисляет подсумму своей группы через workspace.sum_group(group), захватывает общий threading.Lock и вызывает workspace.add_sum(group, subtotal). После добавления поток ожидает завершения первой фазы всеми остальными потоками.
• Если итоговая сумма workspace.total == 0, нормализация полностью пропускается: процесс немедленно отправляет в result_queue значение UNSCALABLE и завершает работу.
• Фаза 2: если сумма не равна нулю, те же самые потоки вызывают workspace.normalize_group(group) для своих групп. После завершения всех потоков воркер отправляет в result_queue результат вызова workspace.normalized_matrix().

2. run_matrix_pipeline(matrix, worker_count):

• Принимает прямоугольную матрицу целых чисел (от 0 до 256 строк, от 0 до 32 столбцов, элементы от -1000 до 1000) и количество потоков worker_count (от 1 до 4).
• Создаёт очередь multiprocessing.Queue и запускает настоящий дочерний процесс multiprocessing.Process(target=matrix_worker, args=(matrix, worker_count, queue)).
• Считывает результат из очереди до ожидания завершения процесса, после чего финализирует процесс вызовом join() и возвращает полученное значение.
• Результатом является нормализованная матрица (числа сверяются с точностью rel_tol=1e-12, abs_tol=1e-12) либо строка UNSCALABLE, если сумма элементов матрицы равна нулю.

Ввод и вывод

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

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

observed = make_matrix_example_observation()
print([list(row) for row in observed['result']])

Вывод:

[[0.1, 0.2], [0.3, 0.4]]