Python multiprocessing with iterable and multiple arguments

Question

Using multiprocessing, I want to pass an iterable and multiple arguments:

a) to a function running on n_core cpu b) yield (or return) n_core results at a time c) in any order of completion

from multiprocessing import Pool 

def func(iterable, args):
    this, that, other = args[0], args[1], args[2]

    for s in iterable:
        return ' '.join([s, this, that, other])        

def main():
    iterable = ['abc', 'bcd', 'cde', 'def', 'efg', 'fgh', 'ghi', 'hij']
    args = ['this', 'that', 'other']
    n_core = 2

    p = Pool(n_core)
    for r in p.imap_unordered(func, iterable, args):
        print(r)

if __name__ == '__main__':
    main()

The expected results are:

"abc this that other"
"bcd this that other"
"cde this that other"
"def this that other" 
"efg this that other" 
"fgh this that other"
"ghi this that other" 
"hij this that other"

What is the correct way to make this work?

Secondly, would concurrent.futures.ProcessPoolExecutor be a better option for this problem?

Answer 1

You can create a new_iterable that combines the values in iterable with args:

from multiprocessing import Pool

def func(args):
    iterable, this, that, other = args[0], args[1][0], args[1][1], args[1][2]
    return ' '.join([iterable, this, that, other])

def main():
    iterable = ['abc', 'bcd', 'cde', 'def', 'efg', 'fgh', 'ghi', 'hij']
    args = ['this', 'that', 'other']
    new_iterable = ([x, args] for x in iterable)
    n_core = 2

    p = Pool(n_core)
    for r in p.imap_unordered(func, new_iterable):
        print(r)

if __name__ == '__main__':
    main()

Output

abc this that other
bcd this that other
cde this that other
def this that other
efg this that other
fgh this that other
ghi this that other
hij this that other

This solution uses a generator expression to create a new iterable that combines the entries from iterable with the desired args. You can also use a generator function to do the same thing.

Update: I modified func to produce the expected results that you mentioned in the comments and added to your question.

Answer 2

The code in the question seems wrong. func should expect a single item and not the whole iterable.

Instead of:

def func(iterable, args):
    this, that, other = args[0], args[1], args[2]

    for s in iterable:
        return ' '.join([s, this, that, other])        

You may use:

def func(item, args):
    this, that, other = args[0], args[1], args[2]
    return ' '.join([item, this, that, other])        

Apart from this error imap_unordered does not accept more than one argument.

This code would do what you expect:

try:
    from itertools import izip
except ImportError:  # Python 3 built-in zip already returns iterable
    izip = zip

from itertools import repeat
from multiprocessing import Pool

def func_star_single(func_item_args):
    """Equivalent to:
       func = func_item_args[0]
       item = func_item_args[1]
       args = func_item_args[2:]
       return func(item,args[0],args[1],...)
    """
    return func_item_args[0](*[func_item_args[1]] + func_item_args[2])


def func(item, args):
    this, that, other = args[0], args[1], args[2]
    return ' '.join([item, this, that, other])    


def main():
    iterable = ['abc', 'bcd', 'cde', 'def', 'efg', 'fgh', 'ghi', 'hij']
    args = ['this', 'that', 'other']
    n_core = 2

    p = Pool(n_core)
    for r in p.imap_unordered(func_star_single, izip(repeat(func), iterable, repeat(list([args])))):
        print(r)

if __name__ == '__main__':
    main()