Last Updated on September 12, 2022
You can join a process pool by calling join() on the pool after calling close() or terminate() in order to wait for all processes in the pool to be shutdown.
In this tutorial you will discover how to join a process pool in Python.
Let’s get started.
Need to Wait for Process Pool to Close
The multiprocessing.pool.Pool in Python provides a pool of reusable processes for executing ad hoc tasks.
A process pool can be configured when it is created, which will prepare the child workers.
A process pool object which controls a pool of worker processes to which jobs can be submitted. It supports asynchronous results with timeouts and callbacks and has a parallel map implementation.
— multiprocessing — Process-based parallelism
We can issue one-off tasks to the process pool using functions such as apply() or we can apply the same function to an iterable of items using functions such as map(). Results for issued tasks can then be retrieved synchronously, or we can retrieve the result of tasks later by using asynchronous versions of the functions such as apply_async() and map_async().
The process pool must be shutdown once we are finished with it in order to release the child worker processes.
We often need to wait for the process pool to close completely and release all resources before continuing on in our application.
How can we safely know when the process pool is shut down completely?
Run loops using all CPUs, download your FREE book to learn how.
What is Joining?
Joining a resource is a common pattern in concurrent programming.
It is a mechanism on an active concurrency primitive that allows the caller to wait for the target primitive to finish.
It is implemented using a join() function on the target object.
This pattern is used with thread-based and process-based concurrency.
Joining Threads
For example, it is common for one thread to call join() another target thread to wait for the target thread to finish before continuing on in the application.
For example:
|
1 2 3 |
... # wait for the target thread to finish thread.join() |
You can learn more about joining threads in the tutorial:
It is also common for a thread to call join() on a thread-safe queue.Queue to wait for all tasks to be completed.
For example:
|
1 2 3 |
... # wait for all tasks in the target queue to be marked complete queue.join() |
You can learn more about joining thread-safe queues in the tutorial:
Joining Processes
It is also common for a parent process to call join() a target child process to wait for the target process to terminate before continuing on in the application.
|
1 2 3 |
... # wait for the target process to finish process.join() |
You can learn more about joining processes in the tutorial:
We can also call join() on a process-safe queue via the multiprocessing.JoinableQueue class.
This allows a process to wait until all tasks on the queue have been marked as done.
For example:
|
1 2 3 |
... # wait for all tasks to be marked as done queue.join() |
Next, let’s look at how we might also join a process pool.
How to Join the Process Pool
We can join the process pool by calling the Pool.join() function.
For example:
|
1 2 3 |
... # join the process pool pool.join() |
Calling join() on the process pool will allow the caller to wait for all worker processes in the process pool to be closed completely.
If all child processes in the pool have already terminated, then the call to join() will return immediately, otherwise it will block and return only once all child processes have terminated.
Wait for the worker processes to exit. One must call close() or terminate() before using join().
— multiprocessing — Process-based parallelism
Why Join a Process Pool?
We may want to wait for the process pool to be completely shut down for many reasons, such as:
- Wait for all issued tasks to finish, before exiting.
- Wait for the resources to be released before continuing on with an application.
- Wait for the pool to be completely closed before closing and releasing other resources during a program exit.
For example, if we issue tasks from the main process then exit the main process without joining the process pool, then the process pool will be forcefully closed and the tasks in the pool will not complete.
When to Join a Process Pool?
We can only join a process pool after it has been shut down.
Recall that a process pool can be shutdown by explicitly calling the Pool.close() function or the Pool.terminate() function.
The close() function will wait for all issued tasks to finish before closing the worker processes, whereas the terminate() function will immediately terminate all worker processes, even if they are currently processing tasks.
For example:
|
1 2 3 |
... # close the process pool pool.close() |
You can learn more about shutting the process pool down in the tutorial:
Only once the pool has been closed can we join the process pool.
Therefore the idiom for joining a process pool is to first shutdown the pool and then join.
For example:
|
1 2 3 4 5 |
... # close the process pool pool.close() # join the process pool pool.join() |
If we do not close the pool first before joining the process pool, then an error will be raised.
Now that we know how to join the process pool, let’s look at some worked examples.
Free Python Multiprocessing Pool Course
Download your FREE Process Pool PDF cheat sheet and get BONUS access to my free 7-day crash course on the Process Pool API.
Discover how to use the Multiprocessing Pool including how to configure the number of workers and how to execute tasks asynchronously.
Example of Joining a Process Pool After Close
We can explore how to join a process pool.
In this example we will create a process pool, issue a task, then close the process pool and wait for all processes to close.
Firstly, we can define a new custom function to execute as a task in the pool which will report a message and sleep for a moment.
The task() function below implements this.
|
1 2 3 4 5 6 7 8 |
# task executed in a worker process def task(): # report a message print(f'Task executing', flush=True) # block for a moment sleep(1) # report a message print(f'Task done', flush=True) |
Next, in the main process we can create a process pool with a default configuration.
|
1 2 3 |
... # create and configure the process pool pool = Pool() |
Next, we can issue our task() function to the process pool asynchronously.
Notice that we don’t have to wait for the task to complete.
|
1 2 3 |
... # issue tasks to the process pool result = pool.apply_async(task) |
Next, we can close the process pool.
This will prevent any further tasks from being issued to the process pool, but will allow currently executing tasks in the pool to complete.
|
1 2 3 |
... # close the process pool pool.close() |
We can then call join() to wait for all tasks to complete and all child worker processes to close.
|
1 2 3 |
... # wait a moment pool.join() |
Finally, we can report a message that the application is exiting.
|
1 2 3 |
... # report a message print(f'Main done', flush=True) |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
# SuperFastPython.com # example of joining a process pool after calling close from time import sleep from multiprocessing.pool import Pool # task executed in a worker process def task(): # report a message print(f'Task executing', flush=True) # block for a moment sleep(1) # report a message print(f'Task done', flush=True) # protect the entry point if __name__ == '__main__': # create and configure the process pool pool = Pool() # issue a task to the process pool pool.apply_async(task) # close the process pool pool.close() # wait a moment pool.join() # report a message print(f'Main done', flush=True) |
Running the example first creates the process pool then issues the task to the process pool.
The process pool begins executing the task in a child worker process.
The main process then closes the process pool while the task is running.
This prevents the pool from taking any further tasks, then closes all child worker processes once all tasks are completed.
The main process then joins the process pool, blocking until all child processes are closed and released.
The task in the process pool finishes and the worker processes in the pool are closed.
The main process carries on and reports a final message.
|
1 2 3 |
Task executing Task done Main done |
Next, let’s take a closer look at joining the process pool after terminating it.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Joining a Process Pool After Terminate
We can explore how to join the process pool after calling terminate().
Recall that the terminate() function will forcefully close all processes in the process pool immediately.
Nevertheless, it may still take a moment for the pool to shut down immediately. We can call join() to be sure that all child processes have finished after calling terminate(), before continuing on.
|
1 2 3 4 5 |
... # terminate the process pool pool.terminate() # wait a moment pool.join() |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
# SuperFastPython.com # example of joining a process pool after calling terminate from time import sleep from multiprocessing.pool import Pool from multiprocessing import active_children # task executed in a worker process def task(): # report a message print(f'Task executing', flush=True) # block for a moment sleep(1) # report a message print(f'Task done', flush=True) # protect the entry point if __name__ == '__main__': # create and configure the process pool pool = Pool() # issue a task to the process pool pool.apply_async(task) # terminate the process pool pool.terminate() # wait a moment pool.join() # report a message print(f'Main done', flush=True) |
Running the example first creates the process pool then issues the task to the process pool.
The process pool begins executing the task in a child worker process. In this case, it does not get a chance to print the first message.
The main process then terminates the process pool while the task is running.
This prevents the pool from taking any further tasks, then closes all child worker processes (almost) immediately.
The main process then joins the process pool, blocking until all child processes are closed and released.
The task in the process pool does not get a chance to start completely, let alone finish, and the worker processes in the pool are terminated.
The main process carries on and reports a final message.
|
1 |
Main done |
Next, let’s take a look at what happens if we attempt to join a process pool before shutting it down.
Error When Joining a Process Pool Without Shutting Down
We can explore the case where we attempt to join the process pool without shutting it down first.
|
1 2 3 |
... # wait a moment pool.join() |
In this case, we expect an error to be raised indicating that this is not valid.
The complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# SuperFastPython.com # example of joining a process pool without shutting it down from time import sleep from multiprocessing.pool import Pool # task executed in a worker process def task(): # report a message print(f'Task executing', flush=True) # block for a moment sleep(1) # report a message print(f'Task done', flush=True) # protect the entry point if __name__ == '__main__': # create and configure the process pool pool = Pool() # issue a task to the process pool pool.apply_async(task) # wait a moment pool.join() # report a message print(f'Main done', flush=True) |
Running the example first creates the process pool then issues the task to the process pool.
The process pool begins executing the task in a child worker process.
The main process then attempts to join the process pool.
An exception is raised immediately, as expected, indicating “Pool is still running“.
|
1 2 3 |
Traceback (most recent call last): ... ValueError: Pool is still running |
What if We Don’t Join the Process Pool
We can explore the case of what happens if we do not join the process pool after shutting it down.
For example, we may call close() to shutdown the process pool and allow the issued tasks to complete first.
|
1 2 3 |
... # close the process pool pool.close() |
We then do not join the pool and instead report a message and allow the main process to exit.
|
1 2 3 |
... # report a message print(f'Main done', flush=True) |
In this case, we expect the Python garbage collector to terminate the process pool and immediately stop all worker processes, not allowing the issued task to complete.
The complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# SuperFastPython.com # example of not joining a process pool after calling close from time import sleep from multiprocessing.pool import Pool # task executed in a worker process def task(): # report a message print(f'Task executing', flush=True) # block for a moment sleep(1) # report a message print(f'Task done', flush=True) # protect the entry point if __name__ == '__main__': # create and configure the process pool pool = Pool() # issue a task to the process pool pool.apply_async(task) # close the process pool pool.close() # report a message print(f'Main done', flush=True) |
Running the example first creates the process pool then issues the task to the process pool.
The process pool begins executing the task in a child worker process.
The main process then does not join the process pool, instead, it reports a message and exits.
The Python garbage collector finalizes the process pool which ultimately results in the terminate() function being called automatically on the pool.
This terminates all processes immediately, preventing the issued task from finishing.
This example highlights an important case of why we might need to join the process pool, specifically to allow issued tasks to finish before exiting the application.
|
1 |
Main done |
Further Reading
This section provides additional resources that you may find helpful.
Books
- Multiprocessing Pool Jump-Start, Jason Brownlee (my book!)
- Multiprocessing API Interview Questions
- Pool Class API Cheat Sheet
I would also recommend specific chapters from these books:
- Effective Python, Brett Slatkin, 2019.
- See: Chapter 7: Concurrency and Parallelism
- High Performance Python, Ian Ozsvald and Micha Gorelick, 2020.
- See: Chapter 9: The multiprocessing Module
- Python in a Nutshell, Alex Martelli, et al., 2017.
- See: Chapter: 14: Threads and Processes
Guides
- Python Multiprocessing Pool: The Complete Guide
- Python ThreadPool: The Complete Guide
- Python Multiprocessing: The Complete Guide
- Python ProcessPoolExecutor: The Complete Guide
APIs
References
Takeaways
You now know how to join a process pool in Python.
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Tim Mossholder on Unsplash
Do you have any questions?