Python has a Queue class that can be very quick to implement a producer/consumer queue, and it is thread-safe.
Each worker should call get() on the queue to retrieve a task. This call will block if no tasks are available, causing the worker to go idle until one becomes available. Then the worker should execute the task and finally call task_done() on the queue.
You would put tasks in the queue by calling put() on the queue.
From the main thread, you can call join() on the queue to wait until all pending tasks have been completed.
This approach has the benefit that you are not creating and destroying threads, which is expensive. The worker threads will run continuously, but will be asleep when no tasks are in the queue, using zero CPU time.
You would put tasks in the queue by calling put() on the queue.
From the main thread, you can call join() on the queue to wait until all pending tasks have been completed.
This approach has the benefit that you are not creating and destroying threads, which is expensive. The worker threads will run continuously, but will be asleep when no tasks are in the queue, using zero CPU time.
multiprocessing simulates threads with child processes, then builds extra features (like pools, explicit shared data, etc.) on top of it, and also (in multiprocessing.dummy) provides those same extra features for threads. (Not ideal stdlib design, but historical reasons…) futures runs on top of either threading or multiprocessing (depending on which executor you use), providing the same interface either way.
So what has been my latest learning is no need to build your own thread pool implementation when you already have a perfectly good one in the standard library.
thread-based Pool interface in the multiprocessing module
It is implemented using a dummy Process class wrapping a python thread. This thread-based Process class can be found in multiprocessing.dummy which is mentioned briefly in the docs. This dummy module supposedly provides the whole multiprocessing interface based on threads.
A simple way of implementing multiprocessing in python is
map() returns the direct return value of a function called on the iterable through the processes or threads or you can think of concurrent.map as just a parallel version of the builtin map function.
One nice advantage of this existing implementation in multiprocessing, is that it should make any such threading patch much easier to write
Another example of built in thread based pool is concurrent.futures module. Concurrent.futures has a minimalistic API. It's easy to use for very straightforward problems. Below is implementation based on futures of example above.
though, according to the documentation ProcessPoolExecutor is built on top of multiprocessing. Here is another example, i got from web:
Changing this to use multiple processes instead of multiple threads is just a matter of s/ThreadPoolExecutor/ProcessPoolExecutor.
submit is used to generate a Futureobject for a single function call with its associated arguments.
submit is used to generate a Futureobject for a single function call with its associated arguments.
To take advantage of process level parallelism you still have to have a pickle-able function, i.e. defined at the top level in a module.
ProcessPoolExecutor is doing the exact same thing as multiprocessing.Pool with a simpler (and more limited) API. If you can get away with using ProcessPoolExecutor, use that, because I think it's more likely to get enhancements in the long-term.
Note that you can use all the helpers from multiprocessing with ProcessPoolExecutor, like Lock, Queue, Manager, etc. The main reasons to use multiprocessing.Pool is if you need initializer/initargs (though there is an open bug to get those added to ProcessPoolExecutor), or maxtasksperchild. Or you're running Python 2.7 or earlier, and don't want to install (or require your users to install) the backport of concurrent.futures.
Apart from that, an obvious way to proceed using multiprocessing is to use the Pool.apply_async() method, put the async result objects on a bounded Queue.Queue, and have threads in your main program pull those off the Queue and wait for the results to show up. This is easy enough, but it's not magic. It solves your problem because bounded Queues are the canonical way to mediate between producers and consumers that run at different speeds. Nothing in concurrent.futures addresses that problem directly, and it's at the heart of your "massive amounts of memory" problem.
However, worth noting, multiprocessing.Pool.map outperforms ProcessPoolExecutor.map. Note that the performance difference is very small per work item, so you'll probably only notice a large performance difference if you're using map on a very large iterable. See this closed bug filed against ProcessPoolExecutor for more info. The reason for the performance difference is that multiprocessing. Pool will batch the iterable passed to map into chunks, and then pass the chunks to the worker processes, which reduces the overhead of IPC between the parent and children. ProcessPoolExecutor always passes one item from the iterable at a time to the children, which can lead to much slower performance with large iterables, due to the increased IPC overhead.
ProcessPoolExecutor is doing the exact same thing as multiprocessing.Pool with a simpler (and more limited) API. If you can get away with using ProcessPoolExecutor, use that, because I think it's more likely to get enhancements in the long-term.
Note that you can use all the helpers from multiprocessing with ProcessPoolExecutor, like Lock, Queue, Manager, etc. The main reasons to use multiprocessing.Pool is if you need initializer/initargs (though there is an open bug to get those added to ProcessPoolExecutor), or maxtasksperchild. Or you're running Python 2.7 or earlier, and don't want to install (or require your users to install) the backport of concurrent.futures.
Apart from that, an obvious way to proceed using multiprocessing is to use the Pool.apply_async() method, put the async result objects on a bounded Queue.Queue, and have threads in your main program pull those off the Queue and wait for the results to show up. This is easy enough, but it's not magic. It solves your problem because bounded Queues are the canonical way to mediate between producers and consumers that run at different speeds. Nothing in concurrent.futures addresses that problem directly, and it's at the heart of your "massive amounts of memory" problem.
However, worth noting, multiprocessing.Pool.map outperforms ProcessPoolExecutor.map. Note that the performance difference is very small per work item, so you'll probably only notice a large performance difference if you're using map on a very large iterable. See this closed bug filed against ProcessPoolExecutor for more info. The reason for the performance difference is that multiprocessing. Pool will batch the iterable passed to map into chunks, and then pass the chunks to the worker processes, which reduces the overhead of IPC between the parent and children. ProcessPoolExecutor always passes one item from the iterable at a time to the children, which can lead to much slower performance with large iterables, due to the increased IPC overhead.
Concurrent.futures provides a more convenient programming model for long-running task submission and monitoring situations. A program I recently wrote using concurrent.futures involved monitoring a directory for incoming files over a 2-3 hour window, translating each file as it arrives to a task, submitting it and so on. Future objects returned by the ProcessPoolExecutor allow for tracking task status, providing intermediate status reports etc in a convenient way.
References:
- http://stackoverflow.com/questions/3033952/python-thread-pool-similar-to-the-multiprocessing-pool
- http://www.geeksforgeeks.org/forums/topic/multi-processes-or-single-process-with-multiple-threads/
- https://news.ycombinator.com/item?id=7690885
