Обработка синхронного чтения сокетов и асинхронных событий в Python ⇐ Python

[Anonymous]

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

• Каждый поток постоянно читает данные из своего выделенного сокета.
• Мне нужно, чтобы каждый поток обрабатывал «личные» события и «широковещательные» события. Широковещательное событие должно быть получено всеми потоками одновременно, и ни один из них не должен его пропустить из-за того, что другие прочитают его первыми.

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

• Используйте os.pipe() для создания каналов связи для личных и широковещательных событий.< /li>
  Используйте select.select() для одновременного ожидания сокетов и этих каналов.

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

Код: Выделить всё

import select
import socket
import os
import threading

# Setup for sockets and pipes

def thread_function(sock, rfd):
while True:
readable, _, _ = select.select([sock, rfd], [], [])
for r in readable:
if r == sock:
data = sock.recv(1024)  # Handle socket data
elif r == rfd:
os.read(rfd, 1024)  # Handle event

# Threads setup and event handling logic here

Мои вопросы:

[*]Как реализовать механизм широковещательных событий, который гарантирует, что все потоки получат событие без одного потреблять его раньше других?
[*]Есть ли лучший способ структурировать эту систему для обработки как выделенных, так и широковещательных событий, сохраняя при этом блокировку операций чтения сокетов?
< /ul>
Будем очень признательны за любые идеи, предложения или примеры. Спасибо!

Подробнее здесь: https://stackoverflow.com/questions/785 ... -in-python