Queue и Lock в Python: дедупликация заявок между рабочими потоками

Потоковый аудит дедупликации заявок

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

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

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

Что уже дано

В среде уже объявлены вспомогательные классы и функции, включая make_transaction_probe, для проверки работы конвейера. Дочерний сервис запускается как полноценный процесс и передаёт записи в параллельные потоки, разделяющие реестр идентификаторов и блокировку. Вам предстоит реализовать две функции: admit_transaction и collect_transactions.

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

Реализуйте две функции верхнего уровня для фильтрации и координации конвейера.

Функция admit_transaction(record, seen_ids, lock, accepted_queue) -> bool

Принимает словарь записи record, разделяемое множество идентификаторов seen_ids, объект lock (threading.Lock) и межпроцессную очередь accepted_queue (multiprocessing.Queue).

Правила проверки записи:

• Под защитой блокировки lock атомарно проверьте наличие record["transaction_id"] в seen_ids и, если такого идентификатора ещё нет, добавьте его в seen_ids.
• Все ранее присутствовавшие в seen_ids идентификаторы считаются уже принятыми; сохраняйте их. Для идентификаторов действуют стандартные правила равенства и хеширования скалярных типов Python (включая равенство False и 0).
• Если идентификатор уже встречался, функция возвращает False и ничего не отправляет в очередь.
• Если идентификатор встретился впервые, функция помещает исходный полученный объект record в accepted_queue без каких-либо изменений его данных и возвращает True.

Функция collect_transactions(records, input_queue, accepted_queue, report_queue, service) -> dict

Принимает список исходных записей records, входную очередь input_queue, очередь принятых заявок accepted_queue, очередь отчёта report_queue и запущенный дочерний процесс service (multiprocessing.Process).

Правила координации:

• Отправьте каждую запись из records в input_queue ровно один раз, после чего отправьте один маркер остановки None.
• Заберите из очереди report_queue финальное агрегированное число отклоненных дубликатов $D$ (целое число от $0$ до $N$, где $N=\text{len}(records)$).
• Заберите из accepted_queue все $N-D$ подтверждённых записей.
• Дождитесь завершения дочернего процесса с помощью вызова service.join(). Вызывать service.join() разрешено только после того, как все ожидаемые элементы и отчёт вычитаны из очередей, чтобы не вызвать дедлок.
• Верните словарь со строгой структурой: {"accepted": [...], "duplicates": D}, где значением ключа "accepted" является список фактически полученных подтверждённых объектов записей, а "duplicates" — целочисленное количество дубликатов $D$.

Ограничения:

• Количество записей в records составляет от 0 до 64.
• Суммарный объём JSON-данных записей не превышает 512 КиБ.
• Число потоков воркера — от 1 до 8.
• В исходящих очередях нет маркеров конца; создавать искусственные отчёты или генерировать фиктивные записи запрещено.
• Порядок записей во входном списке, порядок в итоговом списке принятых заявок и выбор того, какая из двух одинаковых заявок признаётся первой, не фиксируются.

Ввод и вывод

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

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

probe = make_transaction_probe([{'transaction_id': 'A', 'value': 7}, {'transaction_id': 'A', 'value': 7}], 2)
observed = probe.run()
print(len(observed['summary']['accepted']))
print(observed['summary']['duplicates'])

Вывод:

1
1