Python异步缓冲区和进程数据
问题描述
我在异步事件循环内的一些CPU密集型任务中遇到了问题。在维护传入数据的缓冲区并从它构建数据包时,我遇到了麻烦。我曾尝试使用执行器来执行CPU限制的工作,但在从缓冲区中删除数据包时,维护缓冲区的顺序时遇到了问题。
我正在寻找一种最佳实践方法来实现以下功能,而无需在事件循环内执行CPU限制的任务。
import asyncio
import struct
class Reader(asyncio.Protocol):
def __init__(self):
self.extra = bytearray()
def data_received(self, data):
self.extra.extend(data)
packet = get_packet(bytes(self.extra))
if packet:
del self.extra[:len(packet)]
if verify_hash(packet): # CPU intensive
asyncio.async(distribute(packet)) # Some asyncio fan-out callback
def get_packet(data): # CPU intensive
if len(data) > HEADER_SIZE:
payload_size, = struct.unpack_from(HEADER_FORMAT, data)
if len(data) >= HEADER_SIZE + payload_size:
return data[:HEADER_SIZE + payload_size]
return None
loop = asyncio.get_event_loop()
loop.run_until_complete(loop.create_server(Reader, '0.0.0.0', 8000))
loop.run_forever()
解决方案
您希望能够尽可能快地处理进入Reader
的所有数据,但也不能让多个线程/进程尝试并行处理这些数据;这就是以前使用Executor时遇到争用情况的原因。相反,您应该启动一个工作进程,该进程可以处理所有数据包数据,一次一个,使用multiprocessing.Queue
将数据从父进程传递到工作进程。然后,当辅助进程构建、验证了有效的数据包并准备分发时,它会使用另一个multiprocessing.Queue
将其发送回父进程中的线程,该线程可以使用线程安全的call_soon_threadsafe
方法安排distribute
运行。
这里有一个未经测试的示例,应该可以让您了解如何做到这一点:
import asyncio
import struct
from concurrent.futures.ProcessPoolExecutor
import threading
def handle_result_packets():
""" A function for handling packets to be distributed.
This function runs in a worker thread in the main process.
"""
while True:
packet = result_queue.get()
loop.call_soon_threadsafe(asyncio.async, distribute(packet))
def get_packet(): # CPU intensive
""" Handles processing all incoming packet data.
This function runs in a separate process.
"""
extra = bytearray()
while True:
data = data_queue.get()
extra.extend(data)
if len(data) > HEADER_SIZE:
payload_size, = struct.unpack_from(HEADER_FORMAT, data)
if len(data) >= HEADER_SIZE + payload_size:
packet = data[:HEADER_SIZE + payload_size]
del extra[:len(packet)]
if verify_hash(packet):
result_queue.put(packet)
class Reader(asyncio.Protocol):
def __init__(self):
self.extra = bytearray()
self.t = threading.Thread(target=handle_result_packets)
self.t.start()
def data_received(self, data):
data_queue.put(data)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
data_queue = multiprocessing.Queue()
result_queue = multiprocessing.Queue()
p = multiprocessing.Process(target=get_packet)
p.start()
loop.run_until_complete(loop.create_server(Reader, '0.0.0.0', 8000))
loop.run_forever()
相关文章