Modify multiprocess function to accept a list of args

user_4981251 0
Viewed 69

How do I update the multi_proc_parallel_functions function below to accept a list of args. This is using the multiprocess module.

Please note I will be using this with AWS Lambda and other multiprocessing modules can have issues on Lambda.

adder functions below are simply toy functions used to demo the issue.

import multiprocess as mp

def parallel_functions(function,send_end):

    send_end.send(function())


def multi_proc_parallel_functions(function_list,target_func):
    jobs = []
    pipe_list = []

    for function in function_list:

        recv_end, send_end = mp.Pipe(False)
        p = mp.Process(target=target_func, args=(function,send_end))

        jobs.append(p)

        pipe_list.append(recv_end)
        p.start()

    result_list = [x.recv() for x in pipe_list]

    for proc in jobs:
        proc.join()

    return result_list
    

def adder10():
    return np.random.randint(5) + 10

def adder1000():
    return np.random.randint(5) + 1000

Create a list of functions

function_list = [adder10,adder10,adder10,adder1000]

Run all functions

multi_proc_parallel_functions(function_list,parallel_functions)

[13, 13, 13, 1003]

How do I update the multi_proc_parallel_functions to accept a varied length list of args which will vary per function as follows:

def adder10(x,y):
    return np.random.randint(5) + 10 + x * y

def adder1000(a,b, c):
    return np.random.randint(5) + 1000 -a + b +c

I think this will require *args.

python
2 Answers

This would be one way of doing it (with positional argument support):

import multiprocessing as mp

def parallel_functions(function, send_end, *args):
    send_end.send(function(*args))

def multi_proc_parallel_functions(function_list, target_func):
    jobs = []
    pipe_list = []

    for (function, *args) in function_list:
        recv_end, send_end = mp.Pipe(False)
        p = mp.Process(target=target_func, args=(function, send_end, *args))
        jobs.append(p)
        pipe_list.append(recv_end)
        p.start()

    result_list = [x.recv() for x in pipe_list]

    for proc in jobs:
        proc.join()

    return result_list

import numpy as np

def adder10(x, y):
    return np.random.randint(5) + 10 + x * y

def adder1000(a, b, c):
    return np.random.randint(5) + 1000 -a + b +c

multi_proc_parallel_functions(
  [ (adder10, 5, 4),
    (adder10, 1, 2),
    (adder1000, 5, 6, 7) ],
  parallel_functions
)

Note that how the multiprocessing module works will depend on whether you are on Windows, macOS or Linux.

On Linux, the default way of creating a mp.Process is by using the fork-syscall, which means the function/its arguments does not need to be serializable/possible to pickle. The child process will inherit memory from the parent. macOS supports fork, Windows doesn't.

On Windows/macOS, the spawn syscall is used by default instead. This requires that everything sent to the child process is serializable/possible to pickle. This means you won't be able to send lambda expressions or dynamically created functions for example.

Example of something that would work on Linux (with your original implementation), but not on Windows (or macOS by default):

multi_proc_parallel_functions(
  [ lambda: adder10(5, 4),
    lambda: adder10(1, 2),
    lambda: adder1000(5, 6, 7) ],
  parallel_functions
)

# spawn: _pickle.PicklingError: Can't pickle <function <lambda> at 0x7fce7cc43010>: attribute lookup <lambda> on __main__ failed
# fork: [30, 12, 1008]
user_80083960

I would suggest using the operator module, which has functions for the math operations. This way, you can send a list of operators and values to modify the initial value in a flexible way.

Example where each argument is a tuple of (operator, value):

import operator

import numpy as np


np.random.seed(123)


def adder(*args):
    x = np.random.randint(5)

    print(x)

    for operator, value in args:
        x = operator(x, value)
        print(x)
   
    return x


adder((operator.add, 5), (operator.mul, 10))

This operation (2 + 5 * 10) outputs:

2
7
70
user_33045360
Related