上下文

我正在接收流：

symbols = ['ABC', 'DFG', ...]  # 52 of these

handlers = { symbol: Handler(symbol) for symbol in symbols }

async for symbol, payload in lines:  # 600M of these
    handlers[symbol].feed(payload)

我需要使用多个CPU核心来加快速度。

handler['ABC'](例如)保持状态，但它与handler['DFG']的状态不相交

基本上我不能同时运行两个内核，例如handler['ABC']。

我到目前为止的方法

我提出了以下解决方案，但它是部分伪代码，因为我看不到如何实现它。

NCORES = 4
symbol_curr_active_on_core = [None]*NCORES

NO_CORES_FREE = -1
def first_free_core():
    for i, symbol in enumerate(symbol_curr_active_on_core):
        if not symbol:
            return i
    return NO_CORES_FREE

for symbol, payload in lines:
    # wait for avail core to handle it

    while True:
        sleep(0.001)
        if first_free_core() == NO_CORES_FREE:
            continue
        if symbol in symbol_curr_active_on_core:
            continue
        core = first_free_core()
        symbol_curr_active_on_core[core] = symbol

        cores[core].execute(
            processor[symbol].feed(payload),
            on_complete=lambda core_index: 
                symbol_curr_active_on_core[core_index] = None
        )

所以我的问题具体是：如何将最后一条语句转换为有效的Python代码？

        cores[core].execute(
            processor[symbol].feed(payload),
            on_complete=lambda core_index: 
                symbol_curr_active_on_core[core_index] = None
        )

PS更广泛地说，我的方法是最优的吗？

推荐答案

假设以下方法是可行的：

您的Handler类可以进行腌制和
Handler类不携带太多状态信息，从而使其与每个辅助进程调用之间的序列化开销过大。

主进程创建一个handlers字典，其中键是52个符号中的一个，值是具有两个键的字典：‘Handler’，其值是符号的处理程序；以及‘process’，其值为True或False，具体取决于进程当前是否正在处理该符号的一个或多个有效负载。

池中的每个进程使用另一个queue_dict字典进行初始化，该字典的关键字是52个符号之一，其值是一个multiprocessing.Queue实例，该实例将保存要为该符号处理的有效负载实例。

主进程迭代输入的每一行以获得下一个符号/有效负载对。有效载荷被排队到用于当前符号的适当队列中。通过检查当前符号的processing标志，访问handlers词典以确定任务是否已排队到处理池中以处理当前符号的特定符号处理程序。如果该标志为True，则不需要进一步执行任何操作。否则，将processing标志设置为True，并将apply_async作为参数传递给该符号的处理程序。

每次主任务将有效负载写入52个处理程序队列之一时，排队任务(即有效负载)的计数都会保持并递增。指定为apply_async参数的Worker函数接受其处理程序参数，并从中推导出需要处理的队列。对于它在队列中找到的每个有效负载，它都会调用处理程序的feed方法。然后，它返回一个元组，该元组由更新的处理程序和从队列中删除的有效负载消息数的计数组成。apply_async方法的回调函数(1)更新handlers字典中的处理程序，(2)将相应符号的processing标志重置为False。最后，它将排队的任务数减去已删除的有效负载消息数。

在将有效负载排入队列后，主进程检查当前是否有进程正在运行该符号的处理程序，并且发现processing标志为True，并且在此基础上没有通过apply_async提交新任务，此时有一个小窗口，该工作进程已经完成了对其队列中的所有有效负载的处理，并且即将返回或已经返回，而回调函数只是还没有将processing标志设置为False。在这种情况下，有效载荷将在队列中处于未处理状态，直到从输入中读取该符号的下一个有效载荷并进行处理。但是，如果该符号没有更多的输入行，那么当所有任务都完成时，我们将有未处理的有效负载。但我们也会有一个非零的排队任务计数，这向我们表明我们遇到了这种情况。因此，与尝试实现复杂的多处理同步协议相比，检测这种情况并通过重新创建一个新的池并检查52个队列中的每个队列来处理它会更简单。

from multiprocessing import Pool, Queue
import time
from queue import Empty
from threading import Lock

# This class needs to be Pickle-able:
class Handler:
    def __init__(self, symbol):
        self.symbol = symbol
        self.counter = 0

    def feed(self, payload):
        # For testing just increment counter by payload:
        self.counter += payload


def init_pool(the_queue_dict):
    global queue_dict
    queue_dict = the_queue_dict


def worker(handler):
    symbol = handler.symbol
    q = queue_dict[symbol]
    tasks_removed = 0
    while True:
        try:
            payload = q.get_nowait()
            handler.feed(payload)
            tasks_removed += 1
        except Empty:
            break
    # return updated handler:
    return handler, tasks_removed

def callback_result(result):
    global queued_tasks
    global lock

    handler, tasks_removed = result
    # show done processing this symbol by updating handler state:
    d = handlers[handler.symbol]
    # The order of the next two statements matter:
    d['handler'] = handler
    d['processing'] = False
    with lock:
        queued_tasks -= tasks_removed

def main():
    global handlers
    global lock
    global queued_tasks

    symbols = [
        'A','B','C','D','E','F','G','H','I','J','K','L','M','AA','BB','CC','DD','EE','FF','GG','HH','II','JJ','KK','LL','MM',
        'a','b','c','d','e','f','g','h','i','j','k','l','m','aa','bb','cc','dd','ee','ff','gg','hh','ii','jj','kk','ll','mm'
    ]

    queue_dict = {symbol: Queue() for symbol in symbols}

    handlers = {symbol: {'processing': False, 'handler': Handler(symbol)} for symbol in symbols}

    lines = [
        ('A',1),('B',1),('C',1),('D',1),('E',1),('F',1),('G',1),('H',1),('I',1),('J',1),('K',1),('L',1),('M',1),
        ('AA',1),('BB',1),('CC',1),('DD',1),('EE',1),('FF',1),('GG',1),('HH',1),('II',1),('JJ',1),('KK',1),('LL',1),('MM',1),
        ('a',1),('b',1),('c',1),('d',1),('e',1),('f',1),('g',1),('h',1),('i',1),('j',1),('k',1),('l',1),('m',1),
        ('aa',1),('bb',1),('cc',1),('dd',1),('ee',1),('ff',1),('gg',1),('hh',1),('ii',1),('jj',1),('kk',1),('ll',1),('mm',1)
    ]


    def get_lines():
        # Emulate 52_000 lines:
        for _ in range(10_000):
            for line in lines:
                yield line

    POOL_SIZE = 4

    queued_tasks = 0
    lock = Lock()

    # Create pool of POOL_SIZE processes:
    pool = Pool(POOL_SIZE, initializer=init_pool, initargs=(queue_dict,))
    for symbol, payload in get_lines():
        # Put some limit on memory utilization:
        while queued_tasks > 10_000:
            time.sleep(.001)
        d = handlers[symbol]
        q = queue_dict[symbol]
        q.put(payload)
        with lock:
            queued_tasks += 1
        if not d['processing']:
            d['processing'] = True
            handler = d['handler']
            pool.apply_async(worker, args=(handler,), callback=callback_result)
    # Wait for all tasks to complete
    pool.close()
    pool.join()

    if queued_tasks:
        # Re-create pool:
        pool = Pool(POOL_SIZE, initializer=init_pool, initargs=(queue_dict,))
        for d in handlers.values():
            handler = d['handler']
            d['processing'] = True
            pool.apply_async(worker, args=(handler,), callback=callback_result)
        pool.close()
        pool.join()
        assert queued_tasks == 0

    # Print results:
    for d in handlers.values():
        handler = d['handler']
        print(handler.symbol, handler.counter)


if __name__ == "__main__":
    main()

打印：

这篇关于如何在CPU核心上执行函数，并在完成时获得回调？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！