python multiprocessing Pool with map_async

Question

I trying to use the multiprocessing package in python with a Pool.

I have the function f which is called by the map_async function:

from multiprocessing import Pool

def f(host, x):
    print host
    print x

hosts = ['1.1.1.1', '2.2.2.2']
pool = Pool(processes=5)
pool.map_async(f,hosts,"test")
pool.close()
pool.join()

This code has the next error:

Traceback (most recent call last):
  File "pool-test.py", line 9, in <module>
    pool.map_async(f,hosts,"test")
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 290, in map_async
    result = MapResult(self._cache, chunksize, len(iterable), callback)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 557, in __init__
    self._number_left = length//chunksize + bool(length % chunksize)
TypeError: unsupported operand type(s) for //: 'int' and 'str'

I don't know how to pass more than 1 argument to the f function. Are there any way?

You just can use pool.map and drop the "test" dummy variable altogether. — danodonovan
– danodonovan, Commented Jul 13, 2017 at 14:57

F.X. · Accepted Answer · 2013-05-14 18:45:17Z

13

"test" is interpreted as map_async's chunksize keyword argument (see the docs).

Your code should probably be (here copy-pasted from my IPython session) :

from multiprocessing import Pool

def f(arg):
    host, x = arg
    print host
    print x

hosts = ['1.1.1.1', '2.2.2.2']
args = ((host, "test") for host in hosts)
pool = Pool(processes=5)
pool.map_async(f, args)
pool.close()
pool.join()
## -- End pasted text --

1.1.1.1
test
2.2.2.2
test

Note: In Python 3 you can use starmap, which will unpack the arguments from the tuples. You'll be able to avoid doing host, x = arg explicitely.

edited May 14, 2013 at 18:45

answered May 14, 2013 at 11:46

F.X.

7,5156 gold badges53 silver badges74 bronze badges

Sign up to request clarification or add additional context in comments.

9 Comments

dseira Over a year ago

I tested it but the result is not good; it print both hosts but only the "t" and the "e" of the "test" word.

F.X. Over a year ago

Weird. It definitely does not do that on my computer. See update for my results -- I copy-pasted and checked them again.

dseira Over a year ago

With x=["test","test"] it works but it has no sense because imagine that the hosts list is about 10000 and I only want one x for compare the results. It not viable to have an x list with 10000 entries with the same result. Anyway, thanks.

F.X. Over a year ago

That was just a quick'n'dirty way of doing it in your simple example. Maybe see the updated version?

dseira Over a year ago

Finally I used a global option. Really is a static variable.

|

Russel Winder · Accepted Answer · 2015-05-17 12:44:21Z

6

Pool returns a context manager in Python 3 and so a with statement can be used. This avoids problems with exceptions and means no necessity to close and join. In this case the function is always receiving a constant for the x variable and so this can be handled with a partial evaluation. map_async is lazy and so we need to get the result for the actions to happen, might as well just use map. Thus:

from multiprocessing import Pool
from functools import partial

def f(host, x):
    print(host)
    print(x)

hosts = ('1.1.1.1', '2.2.2.2')
with Pool(processes=5) as pool:
    pool.map(partial(f, x='test'), hosts)

results in:

1.1.1.1
test
2.2.2.2
test

answered May 17, 2015 at 12:44

Russel Winder

3,9042 gold badges20 silver badges12 bronze badges

Comments

yoder · Accepted Answer · 2014-09-18 18:44:31Z

as I recall, the Pool().map() and .map_async() specifically accept only a single argument. this limitation can be worked around by passing a list, but of course then you need a customized function designed to take a list(like) object as an argument.

one approach is to write the custom code once -- aka, a general "function + args" wrapper. i worked up something like this (note: this is only partially tested):

def tmp_test():
    # a short test script:
    #
    A=[[1,2], [2,3], [4,5], [6,7]]
    P=mpp.Pool(mpp.cpu_count())
    X=P.map_async(map_helper, [[operator.eq]+a for a in A])
    #
    return X.get()


def null_funct(args=[], kwargs={}):
    # a place-holder 
    pass
#
def map_helper(args_in = [null_funct, [], {}]):
    # helper function for pool.map_async(). pass data as a list(-like object):
    # [function, [args], {kwargs}] (though we'll allow for some mistakes).
    #
    funct = args_in[0]
    #
    # allow for different formatting options:
    if not (isinstance(args_in[1], list) or isinstance(args_in[1], tuple) or isinstance(args_in[1], dict)):
        # probably passed a list of parameters. just use them:
        args = args_in[1:]
        #
        return funct(*args)
    #
    # if the args are "properly" formatted:
    args=[]
    kwargs = {}
    for arg in args_in[1:]:
        # assign list types to args, dict types to kwargs...
        if isinstance(arg, list) or isinstance(arg, tuple): args += arg
        if isinstance(arg, dict): kwargs.update(arg)
    return funct(*args, **kwargs)

Collectives™ on Stack Overflow

python multiprocessing Pool with map_async

3 Answers 3

9 Comments

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

9 Comments

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related