Use a function object: class Copier(object): def __init__(self, tgtdir): self.target_dir = tgtdir def __call__(self, src): copy_file(src, self.target_dir) To run your Pool.map: p.map(Copier(target_dir), file_list)
You can use functools.partial for this (as you suspected): from functools import partial def target(lock, iterable_item): for item in iterable_item: # Do cool stuff if (… some condition here …): lock.acquire() # Write to stdout or logfile, etc. lock.release() def main(): iterable = [1, 2, 3, 4, 5] pool = multiprocessing.Pool() l = multiprocessing.Lock() func … Read more
These problems usually boil down to the following: The function you are trying to parallelize doesn’t require enough CPU resources (i.e. CPU time) to rationalize parallelization! Sure, when you parallelize with multiprocessing.Pool(8), you theoretically (but not practically) could get a 8x speed up. However, keep in mind that this isn’t free – you gain this … Read more
You could use the blocking capabilities of queue to spawn multiple process at startup (using multiprocessing.Pool) and letting them sleep until some data are available on the queue to process. If your not familiar with that, you could try to “play” with that simple program: import multiprocessing import os import time the_queue = multiprocessing.Queue() def … Read more
The following code seems to work: import multiprocessing, time def task(args): count = args[0] queue = args[1] for i in xrange(count): queue.put(“%d mississippi” % i) return “Done” def main(): manager = multiprocessing.Manager() q = manager.Queue() pool = multiprocessing.Pool() result = pool.map_async(task, [(x, q) for x in range(10)]) time.sleep(1) while not q.empty(): print q.get() print result.get() … Read more
The trick is to pass the Queue as an argument to the initializer. Appears to work with all the Pool dispatch methods. import multiprocessing as mp def f(x): f.q.put(‘Doing: ‘ + str(x)) return x*x def f_init(q): f.q = q def main(): jobs = range(1,6) q = mp.Queue() p = mp.Pool(None, f_init, [q]) results = p.imap(f, … Read more
Regarding the fact that all of your processes are running on the same CPU, see my answer here. During import, numpy changes the CPU affinity of the parent process, such that when you later use Pool all of the worker processes that it spawns will end up vying for for the same core, rather than … Read more
Trying again as I just saw the bounty 😉 Basically I think the error message means what it said – multiprocessing shared memory Arrays can’t be passed as arguments (by pickling). It doesn’t make sense to serialise the data – the point is the data is shared memory. So you have to make the shared … Read more
The multiprocessing.pool.Pool class creates the worker processes in its __init__ method, makes them daemonic and starts them, and it is not possible to re-set their daemon attribute to False before they are started (and afterwards it’s not allowed anymore). But you can create your own sub-class of multiprocesing.pool.Pool (multiprocessing.Pool is just a wrapper function) and … Read more
This is a Python bug. When waiting for a condition in threading.Condition.wait(), KeyboardInterrupt is never sent. Repro: import threading cond = threading.Condition(threading.Lock()) cond.acquire() cond.wait(None) print “done” The KeyboardInterrupt exception won’t be delivered until wait() returns, and it never returns, so the interrupt never happens. KeyboardInterrupt should almost certainly interrupt a condition wait. Note that this … Read more