From 4be08f7bdd77991e9e453c1cda863c3f20c338d5 Mon Sep 17 00:00:00 2001 From: Navan Chauhan Date: Thu, 2 Jul 2020 20:48:33 +0530 Subject: initial commit --- plip/basic/parallel.py | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 plip/basic/parallel.py (limited to 'plip/basic/parallel.py') diff --git a/plip/basic/parallel.py b/plip/basic/parallel.py new file mode 100644 index 0000000..1f71943 --- /dev/null +++ b/plip/basic/parallel.py @@ -0,0 +1,52 @@ +import itertools +import multiprocessing +from builtins import zip +from functools import partial + +from numpy import asarray + + +class SubProcessError(Exception): + def __init__(self, e, exitcode=1): + self.exitcode = exitcode + super(SubProcessError, self).__init__(e) + + pass + + +def universal_worker(input_pair): + """This is a wrapper function expecting a tiplet of function, single + argument, dict of keyword arguments. The provided function is called + with the appropriate arguments.""" + function, arg, kwargs = input_pair + return function(arg, **kwargs) + + +def pool_args(function, sequence, kwargs): + """Return a single iterator of n elements of lists of length 3, given a sequence of len n.""" + return zip(itertools.repeat(function), sequence, itertools.repeat(kwargs)) + + +def parallel_fn(f): + """Simple wrapper function, returning a parallel version of the given function f. + The function f must have one argument and may have an arbitray number of + keyword arguments. """ + + def simple_parallel(func, sequence, **args): + """ f takes an element of sequence as input and the keyword args in **args""" + if 'processes' in args: + processes = args.get('processes') + del args['processes'] + else: + processes = multiprocessing.cpu_count() + + pool = multiprocessing.Pool(processes) # depends on available cores + + result = pool.map_async(universal_worker, pool_args(func, sequence, args)) + pool.close() + pool.join() + cleaned = [x for x in result.get() if x is not None] # getting results + cleaned = asarray(cleaned) + return cleaned + + return partial(simple_parallel, f) -- cgit v1.2.3