In multiprocessing, processes often need to work together and coordinate their actions to avoid conflicts or to wait for certain events. This is where synchronization tools come in handy. One important synchronization primitive is the Condition Variable.
A Condition Variable lets one or more processes pause execution until another process signals them to continue. It acts like a “wait here until something changes” tool, helping processes communicate changes in shared state.
In this article, we will learn how to use Python’s multiprocessing.Condition to coordinate multiple processes effectively. We will explore how to make processes wait and notify each other, enabling smooth process cooperation.
Creating a Condition Variable
To start coordinating processes with a Condition Variable, you first create a Condition object from the multiprocessing module. This object lets processes wait and notify each other about changes.
The key methods are .wait() and .notify(). A process that calls .wait() pauses and releases the lock until another process calls .notify(). When .notify() is called, it wakes one waiting process to continue.
from multiprocessing import Process, Condition
import time
def waiter(cond):
print("Waiting for the condition...")
with cond:
cond.wait()
print("Condition met! Proceeding...")
def notifier(cond):
time.sleep(1)
with cond:
print("Notifying the condition.")
cond.notify()
if __name__ == "__main__":
condition = Condition()
p1 = Process(target=waiter, args=(condition,))
p2 = Process(target=notifier, args=(condition,))
p1.start()
p2.start()
p1.join()
p2.join()In this example, the waiter process calls cond.wait() inside the with cond block. This causes it to pause and release the condition lock, waiting for a notification. After a short pause, the notifier process acquires the same condition, calls cond.notify(), and signals the waiting process. The waiter then resumes and prints its message. This simple interaction shows how processes can coordinate their flow using condition variables.
Coordinating Multiple Waiters
A condition variable can coordinate many processes waiting for the same event. Each waiting process calls .wait() on the shared condition, pausing until notified. To wake all waiting processes at once, you use .notify_all(). This releases every waiting process, letting them continue together.
from multiprocessing import Process, Condition
import time
def waiter(name, cond):
print(f"{name} is waiting.")
with cond:
cond.wait()
print(f"{name} received notification!")
def notifier(cond):
time.sleep(2)
with cond:
print("Notifying all waiters.")
cond.notify_all()
if __name__ == "__main__":
condition = Condition()
names = ["Fred", "George", "Hermione"]
waiters = [Process(target=waiter, args=(name, condition)) for name in names]
for w in waiters:
w.start()
notifier_process = Process(target=notifier, args=(condition,))
notifier_process.start()
for w in waiters:
w.join()
notifier_process.join()In this example, three processes (Fred, George, and Hermione) wait on the same condition. When the notifier process calls notify_all(), all waiting processes resume their work and print their messages. This method efficiently wakes every process waiting for the same signal, allowing synchronized progress.
Using Condition Variables with Shared State
Condition variables are often used alongside shared variables to coordinate processes based on specific state changes. Instead of just waiting for a signal, processes wait until a condition on the shared state becomes true. This usually means waiting inside a loop that checks the state before proceeding.
from multiprocessing import Process, Condition, Value
import time
def waiter(flag, cond):
with cond:
while not flag.value:
print("Waiting for flag to be set...")
cond.wait()
print("Flag is set! Proceeding...")
def setter(flag, cond):
time.sleep(3)
with cond:
flag.value = True
print("Setting flag to True and notifying.")
cond.notify()
if __name__ == "__main__":
flag = Value('b', False)
condition = Condition()
p1 = Process(target=waiter, args=(flag, condition))
p2 = Process(target=setter, args=(flag, condition))
p1.start()
p2.start()
p1.join()
p2.join()In this example, the waiter process repeatedly checks the shared flag. It waits inside a loop until the flag becomes True. The setter process changes the flag and calls notify() to wake the waiter. Using .wait() inside a loop ensures that no notifications are missed and that the process only continues when the condition is truly met. This pattern is essential to safely coordinate shared state changes.
Practical Example: Producer-Consumer with Condition Variables
This example demonstrates how to coordinate a producer and a consumer using condition variables and a shared list. The producer generates items and adds them to a shared queue, while the consumer waits until items are available to consume.
from multiprocessing import Process, Condition, Manager
import time
import random
def producer(queue, cond):
for i in range(5):
time.sleep(random.uniform(0.5, 1.5))
with cond:
item = f"item-{i}"
queue.append(item)
print(f"Produced {item}")
cond.notify()
print("Producer finished producing.")
def consumer(queue, cond):
while True:
with cond:
while not queue:
print("Consumer waiting for items...")
cond.wait()
item = queue.pop(0)
print(f"Consumed {item}")
if item == "item-4":
break
if __name__ == "__main__":
manager = Manager()
queue = manager.list()
condition = Condition()
p1 = Process(target=producer, args=(queue, condition))
p2 = Process(target=consumer, args=(queue, condition))
p1.start()
p2.start()
p1.join()
p2.join()Here, the consumer process waits inside a loop until the queue has items. When the producer adds an item, it calls notify() to wake the consumer. This coordination ensures the consumer only consumes when data is available, and both processes run smoothly together. The condition variable synchronizes their actions based on the shared queue’s state.
Conclusion
Condition variables are powerful tools for coordinating processes in Python’s multiprocessing. They allow one or more processes to wait for a certain event or state change, using the main methods .wait(), .notify(), and .notify_all() to manage this signaling. A common pattern involves waiting on a condition inside a loop that checks shared state, ensuring processes proceed only when the right conditions are met. These primitives are especially useful for scenarios like signaling between processes and implementing producer-consumer patterns where smooth coordination is key.
