多处理池和生成器

Question

首先看下面的代码：

pool = multiprocessing.Pool(processes=N)
batch = []
for item in generator():
    batch.append(item)
    if len(batch) == 10:
        pool.apply_async(my_fun, args=(batch,))
        batch = []
# leftovers
pool.apply_async(my_fun, args=(batch,))

基本上我是从生成器中检索数据，收集到一个列表中，然后生成一个消耗这批数据的进程。

这可能看起来不错，但是当消费者（又名池进程）比生产者（又名生成器）慢时，主进程的内存使用量会增长，直到生成器停止或...系统内存不足。

如何避免这个问题？

Answer 1

在这种情况下，您可能希望使用有限大小的队列。

q = multiprocessing.Queue(maxSize).

当与最大值一起使用时。大小，这将为您提供必要的计数并在调用 q.put() 已满时阻塞线程，因此您永远不能在其上发布超过一定数量的工作项，从而限制存储所需的内存待处理的项目。

或者，您可以使用计数信号量（例如，multiprocessing.BoundedSemaphore(maxSize)）。每次从生成器获取工作项时获取它，并在处理项后将其释放到工作函数 (my_fun) 中。这样，等待处理的工作项的最大数量永远不会超过信号量的初始值。

Answer 2

使用grouper itertools 配方将生成器中的数据分块。

使用concurrent futures 中的基础架构，通过流程处理任务提交和检索。

你可以

• 提交一组任务；等待他们完成；然后提交另一个组，或
• 通过在每次完成任务时提交一个新任务来保持管道满载。

---

设置（尝试模拟您的过程）：

import concurrent.futures
import itertools, time, collections, random
from pprint import pprint

# from itertools recipes
def grouper(iterable, n, fillvalue=None):
    "Collect data into fixed-length chunks or blocks"
    # grouper('ABCDEFG', 3, 'x') --> ABC DEF Gxx"
    args = [iter(iterable)] * n
    return itertools.zip_longest(*args, fillvalue=fillvalue)

# generator/iterator facsimile
class G:
    '''Long-winded range(n)'''
    def __init__(self, n=108):
        self.n = n
        self.a = []
    def __iter__(self):
        return self
    def __next__(self):
        #self.a.append(time.perf_counter())
        if self.n < 0:
            raise StopIteration
        x = self.n
        self.n -= 1
        return x

def my_func(*args):
    time.sleep(random.randint(1,10))
    return sum(*args)

---

等待任务组完成

if __name__ == '__main__':
    nworkers = 4
    g = G()
    # generate data three-at-a-time
    data = grouper(g, 3, 0)
    results = []
    fs = []
    with concurrent.futures.ProcessPoolExecutor(max_workers=nworkers) as executor:
        for args in data:
            print(f'pending:{len(executor._pending_work_items)}')
            # block submission - limit pending tasks to conserve resources (memory) 
            if len(executor._pending_work_items) == nworkers:
                # wait till all complete and get the results
                futures = concurrent.futures.wait(fs, return_when=concurrent.futures.ALL_COMPLETED)
                #print(futures)
                results.extend(future.result() for future in futures.done)
                fs = list(futures.not_done)
            # add a new task
            fs.append(executor.submit(my_func, args))
        # data exhausted - get leftover results as they finish
        for future in concurrent.futures.as_completed(fs):
            print(f'pending:{len(executor._pending_work_items)}')
            result = future.result()
            results.append(result)

    pprint(results)

---

保持进程池满。

if __name__ == '__main__':
    nworkers = 4
    g = G()
    # generate data three-at-a-time
    data = grouper(g, 3, 0)
    results = []
    fs = []
    with concurrent.futures.ProcessPoolExecutor(max_workers=nworkers) as executor:
        for args in data:
            print(f'pending:{len(executor._pending_work_items)}')
            # block submission - limit pending tasks to conserve resources (memory) 
            if len(executor._pending_work_items) == nworkers:
                # wait till one completes and get the result
                futures = concurrent.futures.wait(fs, return_when=concurrent.futures.FIRST_COMPLETED)
                #print(futures)
                results.extend(future.result() for future in futures.done)
                fs = list(futures.not_done)
            # add a new task
            fs.append(executor.submit(my_func, args))
        # data exhausted - get leftover results as they finish
        for future in concurrent.futures.as_completed(fs):
            print(f'pending:{len(executor._pending_work_items)}')
            result = future.result()
            results.append(result)

    pprint(results)