[Python] Why should I pass a function using initializer and can I use shared memory instead?

Take this MWE:

from multiprocessing import Pool
from time import perf_counter as now
import numpy as np


def make_func():
n = 20000
np.random.seed(7)
M = np.random.rand(n, n)
return lambda x, y: M[x, x] + M[y, y]


class ParallelProcessor:
def __init__(self):
pass

def process_task(self, args):
"""Unpack arguments internally"""
index, integer_arg = args
print(f(index, integer_arg))

def run_parallel(self, tasks, num_cores=None):
"""Simplified parallel execution without partial"""
num_cores = num_cores

task_args = [(idx, val) for idx, val in enumerate(tasks)]
start = now()
global f
f = make_func()
print(f"************** {now() - start} seconds to make f")
start = now()
with Pool(num_cores) as pool:
results = pool.map( self.process_task, task_args)
print(f"************** {now() - start} seconds to run all jobs")
return results


if __name__ == "__main__":
processor = ParallelProcessor()
processor.run_parallel(tasks=[1, 2, 3, 4, 5], num_cores=2)


I have declared f to be global. I think that means that a copy of the large numpy array will be made in each worker.

Alternatively I could use initializer with:

from multiprocessing import Pool
from time import perf_counter as now
import time
import os
import numpy as np

def make_func():
n = 20000
np.random.seed(7)
M = np.random.rand(n, n)
return lambda x, y: M[x, x] + M[y, y]

def init_worker():
global f
f = make_func()

class ParallelProcessor:
def __init__(self):
pass

def process_task(self, args):
"""Unpack arguments internally"""
index, integer_arg = args
print(f(index, integer_arg))

def run_parallel(self, tasks, num_cores=None):
"""Parallel execution with proper initialization"""
num_cores = num_cores or len(os.sched_getaffinity(0))
task_args = [(idx, val) for idx, val in enumerate(tasks)]

start = now()
with Pool(num_cores, initializer=init_worker) as pool:
results = pool.map(self.process_task, task_args)
print(f"************** {now() - start} seconds to run all jobs")
return results

if __name__ == "__main__":
processor = ParallelProcessor()
processor.run_parallel(tasks=[1, 2, 3, 4, 5], num_cores=2)


I am told this is better style but I can't see what the advantage is. I am not sure why f has to be declared global in `init_worker. In any case a copy of the large numpy array is still sent to each worker. Overall it also seems to be slower.

I am using Linux.


Ideally I would like not to make a copy of the array at each worker. Is there a fast way to use shared memory to avoid that?

Continue reading...