4

我在处理异步事件循环中的一些 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()
4

2 回答 2

5

您希望能够尽快处理所有传入的数据Reader，但也不能让多个线程/进程尝试并行处理这些数据；这就是您之前使用执行程序遇到竞争条件的方式。相反，您应该启动一个可以处理所有数据包数据的工作进程，一次一个，使用 amultiprocessing.Queue将数据从父进程传递给工作进程。然后，当 worker 有一个有效的数据包被构建、验证并准备好分发时，它使用另一个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()
于 2014-08-01T20:57:06.763 回答
0

我会尝试总结整个数据包处理逻辑，并将繁重的任务分成几部分。以 MD5 哈希为例：

@asyncio.coroutine
def verify_hash(packet):
    m = hashlib.md5()
    for i in range(len(packet) // 4096 + 1):
        yield m.update(packet[i:i+4096])
    return m.digest() == signature


@asyncio.coroutine
def handle_packet(packet):
    verified = yield from verify_hash(packet)
    if verified:
        yield from distribute(packet)


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)]
            asyncio.async(handle_packet(packet))

请注意，数据包的传入速度可能比处理速度快得多，Reader因此请确保监控系统负载并在需要时停止接收。但那是另一个故事：）

于 2014-10-09T13:01:40.933 回答