使用multiprocessing.pool.map传递kwargs

2024-05-12 23:48:32 发布

您现在位置:Python中文网/ 问答频道 /正文

我想用Pool.map()将关键字参数传递给我的worker函数。在搜索论坛时,我找不到明确的例子。

示例代码:

import multiprocessing as mp

def worker((x,y), **kwargs):
    kwarg_test = kwargs.get('kwarg_test', False)
    print("kwarg_test = {}".format(kwarg_test))     
    if kwarg_test:
        print("Success")
    return x*y

def wrapper_process(**kwargs):
    jobs = []
    pool=mp.Pool(4)
    for i, n in enumerate(range(4)):
        jobs.append((n,i))
    pool.map(worker, jobs) #works
    pool.map(worker, jobs, kwargs) #how to do this?   

def main(**kwargs):
    worker((1,2),kwarg_test=True) #accepts kwargs
    wrapper_process(kwarg_test=True)

if __name__ == "__main__":    
    main()

输出:

kwarg_test = True
Success
kwarg_test = False
kwarg_test = False
kwarg_test = False
kwarg_test = False
TypeError: unsupported operand type(s) for //: 'int' and 'dict'

类型错误与在multiprocessing.Pool或Queue中解析参数有关,我尝试了其他几种语法,比如列出kwargs;[kwargs,kwargs,kwargs,kwargs],还尝试了几次将kwarg包含在作业列表中,但都没有成功。我在multiprocessing.pool中跟踪了一个映射到另一个映射异步的代码,并得到了 task_batches = Pool._get_tasks(func, iterable, chunksize) 在pool.py中遇到生成器结构时。我很高兴以后能了解更多,但现在我只想知道:

是否有一个简单的语法允许通过pool.map传递kwargs?


Tags: 代码testfalsetruemapmaindefjobs
2条回答

如果要遍历其他参数,请使用@ArcturusB的答案。

如果您只想传递它们,每次迭代都有相同的值,那么您可以这样做:

from functools import partial
pool.map(partial(worker, **kwargs), jobs)

Partial将参数“绑定”到函数。但是,旧版本的Pythoncannot序列化了部分对象。

^{} doc声明:

A parallel equivalent of the map() built-in function (it supports only one iterable argument though). It blocks until the result is ready.

我们只能通过一个有争议的论点。故事结束了。但我们可以幸运地想到一个解决方法:定义worker_wrapper函数,它接受一个参数,将其解包为args和kwargs,并将它们传递给worker

def worker_wrapper(arg):
    args, kwargs = arg
    return worker(*args, **kwargs)

在你的wrapper_process中,你需要从 jobs(甚至在构造作业时直接调用)并调用worker_wrapper

arg = [(j, kwargs) for j in jobs]
pool.map(worker_wrapper, arg)

这是一个工作实现,尽可能接近原始的 代码:

import multiprocessing as mp

def worker_wrapper(arg):
    args, kwargs = arg
    return worker(*args, **kwargs)

def worker(x, y, **kwargs):
    kwarg_test = kwargs.get('kwarg_test', False)
    # print("kwarg_test = {}".format(kwarg_test))     
    if kwarg_test:
        print("Success")
    else:
        print("Fail")
    return x*y

def wrapper_process(**kwargs):
    jobs = []
    pool=mp.Pool(4)
    for i, n in enumerate(range(4)):
        jobs.append((n,i))
    arg = [(j, kwargs) for j in jobs]
    pool.map(worker_wrapper, arg)

def main(**kwargs):
    print("=> calling `worker`")
    worker(1, 2,kwarg_test=True) #accepts kwargs
    print("=> no kwargs")
    wrapper_process() # no kwargs
    print("=> with `kwar_test=True`")
    wrapper_process(kwarg_test=True)

if __name__ == "__main__":    
    main()

通过测试:

=> calling `worker`
Success
=> no kwargs
Fail
Fail
Fail
Fail
=> with `kwar_test=True`
Success
Success
Success
Success

相关问题 更多 >