Multiprocessing Pipe in Python
You can use a pipe between processes by multiprocessing.Pipe class.
In this tutorial you will discover how to use a multiprocessing pipe in Python.
Need for a Pipe
A process is a running instance of a computer program.
Every Python program is executed in a Process, which is a new instance of the Python interpreter. This process has the name MainProcess and has one thread used to execute the program instructions called the MainThread. Both processes and threads are created and managed by the underlying operating system.
Sometimes we may need to create new child processes in our program in order to execute code concurrently.
Python provides the ability to create and manage new processes via the multiprocessing.Process class.
In multiprocessing programming, we often need to share data between processes.
One approach to sharing data is to use a pipe.
What is the pipe and how can we use it in Python?
Run your loops using all CPUs, download my FREE book to learn how.
What is a Pipe
In multiprocessing, a pipe is a connection between two processes in Python.
It is used to send data from one process which is received by another process.
Under the covers, a pipe is implemented using a pair of connection objects, provided by the multiprocessing.connection.Connection class.
Creating a pipe will create two connection objects, one for sending data and one for receiving data. A pipe can also be configured to be duplex so that each connection object can both send and receive data.
Pipe vs Queue
Both a multiprocessing.Pipe and a multiprocessing.Queue can be used to send and receive objects and data between processes.
A Pipe is simpler than a queue. It is a lower-level mechanism, requiring first the explicit creation of the connections between a pair of processes, then the explicit sending and receiving of data between processes.
A Queue is a high-level constructor that can be treated like a local data structure that just so happens to be shared among processes.
Importantly, a Queue is designed to be used with multiple producers and multiple consumers in mind, whereas a Pipe is intended for a pair of processes only.
The targeted and simpler nature of pipes can make them more efficient and potentially faster in sharing data between two processes.
- Both pipe and queue can be used to shared data between processes
- Pipe is simple and low-level, queue is more capable and higher-level
- Pipe is between two processes, and the queue has multiple producers and consumers.
Confused by the multiprocessing module API?
Download my FREE PDF cheat sheet
How to Use the Pipe
Let’s take a closer look at how to use the pipe class.
Create a Pipe
A pipe can be created by calling the constructor of the multiprocessing.Pipe class, which returns two multiprocessing.connection.Connection objects.
By default, the first connection (conn1) can only be used to receive data, whereas the second connection (conn2) can only be used to send data.
The connection objects can be made duplex or bidirectional.
This can be achieved by setting the “duplex” argument to the constructor to True.
In this case, both connections can be used to send and receive data.
Share Objects With Pipe
Objects can be shared between processes using the Pipe.
The Connection.send() function can be used to send objects from one process to another.
The objects sent must be picklable.
The Connection.recv() function can be used to receive objects in one process sent by another.
The objects received will be automatically un-pickled.
The function call will block until an object is received.
Share Bytes With Pipe
Data can be shared between processes using the Pipe.
This can be achieved by sending and receiving data in the form of packages of bytes.
Bytes can be sent from one process to another via the Connection.send_bytes() function.
If byte data is held in a buffer data structure, then an “offset” and “size” arguments can be specified when sending bytes.
The function call will block until there are bytes to receive.
A single message of bytes will be read.
A maximum length of bytes can be specified via the “maxlength” argument.
Byte data can also be received into an existing byte buffer, with an offset.
This can be achieved via the Connection.recv_bytes_into() function with an optional “offset” argument.
Status of Pipe
The status of the pipe can be checked via the Connection.poll() function.
This will return a boolean as to whether three is data to be received and read from the pipe.
A timeout can be set via the “timeout” argument. If specified, the call will block until data is available. If no data is available before the timeout number of seconds has elapsed, then the function will return.
Now that we know how to use the multiprocessing.Pipe, let’s look at some worked examples.
Free Python Multiprocessing Course
Download my multiprocessing API cheat sheet and as a bonus you will get FREE access to my 7-day email course.
Discover how to use the Python multiprocessing module including how to create and start child processes and how to use a mutex locks and semaphores.
Example of Using a Pipe
We can explore how to use a multiprocessing.Pipe to share data between processes.
In this example we will create a sender process that will generate random numbers and send them to another process via the pipe. We will also create a receiver process that will receive numbers sent from the other process and report them.
We can first define the sender process.
We can define a new function named sender() that takes a connection as an argument on which it will send objects. It then loops ten times and each iteration it will generate a random number between 0 and 1 via random.random(), block for a fraction of a second to simulate work, then send the value to the other process via the pipe.
Once done, the sender will send a special value, called a sentinel value to indicate that no more values will be sent. In this case we will use the value “None“.
The sender() function below implements this.
Next, we can define the receiver process.
We can define a new receiver() function that takes a connection on which to receive objects.
The function will loop forever. Each iteration, it will receive an object on the pipe and block until an object is received. It will then report the value. If the received value was the special sentinel value, it will break the loop and the process will terminate.
The receive() function below implements this.
Finally, the main process will create the process and wait for them to finish.
First, a new pipe is created that will be used to send and receive objects between the processes.
Next, we can create a child process that will execute the sender() function and take the conn2 that can only send data along the pipe. Once created and configured the child process is started.
We can then create another child process that will execute the receiver() function and take the conn1 that can only receive data via the pipe. This process can also then be started.
The main process can then block until both child processes have finished.
Tying this together, the complete example is listed below.
Running the example first creates the pipe, then creates and starts both child processes.
The main process then blocks until the child processes finish.
The sender child process then runs in a loop, generating and sending ten random values along the pipe. Once all values are generated and sent, the sender process terminates.
The child process loops, receiving objects from the pipe each iteration. It blocks until an object appears each iteration. Received values are reported, and the loop is broken once the sentinel value is received.
Note, your specific results will differ given the use of random numbers.
This highlights how to use a default pipe for sending data from one process to another.
Next, let’s look at a duplex or bidirectional pipe between two processes.
Overwheled by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Using a Duplex Pipe
A multiprocessing.Pipe can be used to both send and receive data between two processes.
This is called a duplex or bidirectional pipe and can be achieved by setting the “duplex” argument to True when creating a pipe.
In this example, we will play ping pong between two processes, player1 and player2. Player1 will start the game by generating a random value between 0 and 1 and send it to player2. Player2 will receive the value, add a new random value to the received value and send it back to player1. Player1 will receive the value and perform the same action of adding a random value to the received value and sending it back.
This process is repeated until a value above 10 is received, after which both player processes will terminate.
First, we can define a function that takes a connection and a value as arguments, adds a random value to the value and sends it along the connection. The value passed in as an argument will be the value received along the pipe.
This function can be used by both players in the ping pong game, and can be used by player1 when starting the game.
The generate_send() function listed below implements this.