Passing multiple iterable args to ProcessPoolExecutor.map using a partial

user_8763097 0
Viewed 413

Setup

I have a function set up to take in multiple keyword args:

def process(image, folder, param1, param2, param3):
    do_things
    return

And I have several combinations of param1/2/3 that I want to iterate through in a full permutation fashion (i.e. 1,4,7 then 1,4,8 until 3,6,9):

param1 = [1,2,3]
param2 = [4,5,6]
param3 = [7,8,9]

However, the image and folder remain the same, so I've set up a partial like so and wrapped it inside of a ProcessPoolExecutor map function:

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        ...
    )

Question

When I have 1 param for the function, and remove param2 and param3 from process(), the following single iterable works:

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        [q for q in param1]
    )

But how do I pass in multiple iterable for the executor.map to work with?

I've tried

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        [q for q in param1],
        [q for q in param2],
        [q for q in param3],
    )

Both with and without specifying the function keyword name, but in either case I get a TypeError: process() got multiple values for argument 'image'

Also tried:

    with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        [i for i in product(param1, param2, param3)]
    )

The above givesthe error: BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

Expected outcome

I want the executor.map to iterate through multiple iterables and pass them to the correct function keyword.

Code for Reproducible Error:

from functools import partial
from concurrent import futures
from itertools import product

def new_fn(a, b, c):
    print(a, b, c)
    return (a, b, c)

a_static = 'Hi: '
b_it = ['b1', 'b2', 'b3']
c_it = ['c1', 'c2', 'c3']
p = [i for i in product(b_it, c_it)]

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(new_fn, a=a_static), p
    )

data
python
1 Answers

You never say exactly how you want to iterate through param1/2/3, so this is just a guess.

You can do what I think you want by changing the calling sequence of the process() function slightly and using the built-in zip() function to unzip the data (as mentioned in the documentation).

Note how the built-in partial() function is being used — trying to pass it keyword arguments for positional ones, as you were attempting, was incorrect.

I also added the if __name__ == '__main__': guard needed to make Python multiprocessing code (also) work on Windows.

import concurrent.futures as futures
from functools import partial


def process(image, folder, params):
    param1, param2, param3 = params
    print(f'param1={param1}, param2={param2}, param3={param3}')


if __name__ == '__main__':

    image = 'image'
    folder = 'folder'

    param1 = [1, 2, 3]
    param2 = [4, 5, 6]
    param3 = [7, 8, 9]

    with futures.ProcessPoolExecutor(max_workers=4) as executor:
        data = executor.map(
            partial(process, image, folder),
            zip(*zip(param1, param2, param3))
        )

        for _ in data:
            ...

Output:

param1=1, param2=2, param3=3
param1=4, param2=5, param3=6
param1=7, param2=8, param3=9
user_3552300
Related