Multiprocessing Queue in Python

1. Python Multiprocessing Queue
2. Python Multiprocessing Queue Methods
3. Use a Multiprocessing Queue With Multiple Processes
4. Conclusion

While programming, you can run two or more programs parallelly. However, it becomes a tedious task if you need to communicate between the programs.

This article discusses how we can use multiprocessing queues in Python for communication between two python programs.

Python Multiprocessing Queue

Python provides us with the multiprocessing module to create, run, and manage two or more python programs parallelly. You can import the multiprocessing module into your program using the following import statement.

import multiprocessing

After importing the module, create a multiprocessing queue using the Queue() method. The multiprocessing.Queue() method returns a multiprocessing queue.

Code:

import multiprocessing as mp

myQueue = mp.Queue()
print("The multiprocessing Queue is:")
print(myQueue)

Output:

The multiprocessing Queue is:
<multiprocessing.queues.Queue object at 0x7fa48f038070>

You can see that a Python multiprocessing queue has been created in the memory at the given location. After creating the Python multiprocessing queue, you can use it to pass data between two or more processes.

Python Multiprocessing Queue Methods

There are various multiprocessing queue methods with the help of which we can perform various operations.

Insert Element Into the Python Multiprocessing Queue

We can use the put() method to insert an element into the multiprocessing queue. When invoked on a multiprocessing queue, the method takes an element as its input argument and adds the element to the queue, and after execution, it returns None.

Code:

import multiprocessing as mp

myQueue = mp.Queue()
return_value = myQueue.put(1)
print(return_value)

Output:

None

If no input arguments are given to the put() method, the program runs into the TypeError exception, as shown below.

Code:

import multiprocessing as mp

myQueue = mp.Queue()
return_value = myQueue.put()
print(return_value)

Output:

Traceback (most recent call last):
  File "/home/aditya1117/PycharmProjects/pythonProject/string12.py", line 4, in <module>
    return_value= myQueue.put()
TypeError: put() missing 1 required positional argument: 'obj'

Here, we haven’t provided any input argument to the put() method. Due to this, the program has raised the TypeError exception saying that the required positional argument is missing.

Extract an Element From a Python Multiprocessing Queue

You can use the get() method to extract an element from a multiprocessing queue. The get() method, when invoked on a multiprocessing queue, it returns the front element of the queue after removing it from the queue.

Code:

import multiprocessing as mp

myQueue = mp.Queue()
myQueue.put(1)
myQueue.put(2)
myQueue.put(3)
myQueue.put(4)
myQueue.put(5)
return_value = myQueue.get()
print(return_value)

Output:

1

We have first enqueued five elements into the multiprocessing queue. After that, we obtained an element using the get() method.

Observe that the get() method has returned the value 1 inserted into the multiprocessing queue at first. This is because queues follow the First In First Out(FIFO) order for accessing elements.

Get the Size of Python Multiprocessing Queue

We can get the size of a multiprocessing queue using the qsize() method. The qsize() method returns the approximate size of the python multiprocessing queue.

Code:

import multiprocessing as mp

myQueue = mp.Queue()
myQueue.put(1)
myQueue.put(2)
myQueue.put(3)
myQueue.put(4)
myQueue.put(5)
return_value = myQueue.qsize()
print("The size of multiprocessing queue is:")
print(return_value)

Output:

The size of multiprocessing queue is:
5

In the above example, we’ve used the term "approximate size" instead of "size" of the queue. This is because the queue is shared between multiple processes.

Due to this, another process might add an element to the queue or delete an element from the queue just after we have obtained its size. Therefore, the size returned by the qsize() method is not reliable.

Check if a Multiprocessing Queue Is Empty

The empty() method checks if a multiprocessing queue is empty if the method returns True if the queue is empty. Otherwise, it returns False.

Code:

import multiprocessing as mp

myQueue = mp.Queue()
myQueue.put(1)
myQueue.put(2)
myQueue.put(3)
myQueue.put(4)
myQueue.put(5)
return_value = myQueue.empty()
print("The multiprocessing queue is empty:")
print(return_value)

Output:

The multiprocessing queue is empty:
False

The multiprocessing queue has five elements in it. Therefore, the empty() method returns False.

Close a Python Multiprocessing Queue

If you want no process should write into a multiprocessing queue, you can close the queue using the close() method. The close() method, when invoked on a multiprocessing queue in any of the processes, closes the queue.

After this, no process can insert an element into the queue. Let’s now tackle how we can use a multiprocessing queue in a Python program.

Use a Multiprocessing Queue With Multiple Processes

Define Functions To Create Processes

To use a multiprocessing queue between different processes in Python, we first need to create multiple processes. We will first define two functions.

The first function will take the multiprocessing queue as an input argument. While execution, it will add positive numbers from 1 to 1000 to the Python multiprocessing queue.

def addPositive(queue):
    print("I am in addPositive.")
    for i in range(1, 1001):
        queue.put(i)

The second function will also take the multiprocessing queue as an input argument. However, it will add negative numbers from -1000 to -1 to the multiprocessing queue.

def addNegative(queue):
    print("I am in addNegative.")
    for i in range(-1000, 0):
        queue.put(i)

Create Processes To Write Data Into Multiprocessing Queue

After creating the functions, we will create two separate processes using these two functions. We can use the Process() method to create a process.

The Process() method takes a function as its first input argument assigned to the target parameter. It also takes a tuple containing the function’s input arguments provided in the target.

The tuple is assigned to the args parameter of the Process() method. After execution, the Process() method returns a Process object.

We will create a process for adding positive and negative numbers to the multiprocessing queue.

myQueue = mp.Queue()
process1 = mp.Process(target=addPositive, args=(myQueue,))
process2 = mp.Process(target=addNegative, args=(myQueue,))

Start the Processes To Write Data Into the Multiprocessing Queue

After creating the process, we can use the start() method to start the execution of the process. Once the processes get executed, the numbers will be written to the multiprocessing queue.

process1.start()
process2.start()

If any of the processes are terminated abruptly using the terminate() command or due to exceptions, the multiprocessing queue might get corrupt. After that, you cannot read from the queue or write to the queue in any process.

Therefore, all the processes must get executed smoothly.

Wait in the Main Process for the Child Processes To Finish

The parent process in which we have created other processes might finish its execution before the child processes. In such cases, zombie processes are created and always remain present in the computer’s memory.

To avoid this situation, we can pause the execution of the parent process until the child processes finish their execution. We can use the join() method to make the parent process wait for the child process to finish its execution.

process1.join()
process2.join()

Print the Contents of the Multiprocessing Queue

We can print the contents of the multiprocessing using the get() method, empty() method and the print() function. We will check if the multiprocessing queue is empty or not using the empty() method.

If the queue is not empty, we will extract an element from the queue using the get() method and print the result. Otherwise, we will close the multiprocessing queue using the close() method to finish the program’s execution.

Code:

import multiprocessing as mp


def addPositive(queue):
    print("I am in addPositive.")
    for i in range(1, 100):
        queue.put(i)


def addNegative(queue):
    print("I am in addNegative.")
    for i in range(-100, 0):
        queue.put(i)


myQueue = mp.Queue()
process1 = mp.Process(target=addPositive, args=(myQueue,))
process2 = mp.Process(target=addNegative, args=(myQueue,))
process1.start()
process2.start()
process1.join()
process2.join()
while myQueue:
    print(myQueue.get(), end=",")
myQueue.close()

Output:

1,2,3,4,5,6,7,8,9,10,-1001,11,12,13,-1000,-999,-998,-997,-996,-995,-994,-993,-992,-991,-990,-989,-988,-987,-986,-985,-984,-983,-982,-981,14,-980,15,-979,16,17,18,19,20,21,22,23,24,25,26,-978,-977,-976,-975,-974,-973,-972,-971,-970...

The code will run until the queue is empty.

Observe that the queue randomly contains positive and negative numbers. This proves that the data was being written into the multiprocessing queue in a parallel manner using two different processes.

Conclusion

In this article, we have discussed python multiprocessing queues. The multiprocessing module provides high-level functions to create a child process.

We suggest using the multiprocessing module instead of the fork() method to create child processes. You can use Pipe and SimpleQueue objects to share data between processes.

You can read more about them in this documentation.