Lock, Wait, Notify: A Python Monitor Saga

Introduction

Sometimes in a multi-threaded program, there is a need to protect a resource from concurrent access, that is access by multiple threads. One way of doing this is by implementing a Monitor Object.

Implementing this pattern usually involves the use of the locks such as threading.Lock or threading.RLock as well as condition variables such as threading.Condition.

Basically a monitor is a synchronization construct, which ensures that only a single thread can execute a particular section of code at a particular time. This piece of the critical section. A monitor also provides mechanisms for threads for certain conditions to become true. In the example in this article it is the presence of items in the buffer.

Implementation in Python

A use case for this pattern would be a buffer, which is shared by a number of producers and consumers. To start, import the following:

import threading
import time
import random

Line by line:

1. The threading module provides the concurrency primitives, like the Lock and Condition classes which are needed to implement the Monitor pattern. It also enables the creation and the management of the producer and consumer threads.
2. For the simulation in this example the sleep() function from the time modules is needed. This is needed to create deliberate pauses in the threads.
3. Finally the random module is used to create variable and pseudorandom delays in the threads.

Now on to the Buffer class itself. Start with the constructor:

    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = []
        self.lock = threading.Lock()
        self.not_empty = threading.Condition(self.lock)
        self.not_full = threading.Condition(self.lock)

Line by line:

1. The Buffer has a set capacity.
2. The buffer itself is a list.
3. The lock ensures mutual exclusion when accessing the buffer. Remember that a Monitor allows only a single thread at a time to access the resource.
4. The not_empty is used to by the consumers of the buffer to make they can read from the buffer.
5. Conversely the not_full ensures producers wait with adding to the buffer when it is full.

The produce() method allows adding to the buffer.

    def produce(self, item):
        with self.lock:
            while len(self.buffer) >= self.capacity:
                print(f"Producer: Buffer is full, waiting for space. Current buffer: {self.buffer}")
                self.not_full.wait()

            self.buffer.append(item)
            print(f"Producer: Added {item}. Current buffer: {self.buffer}")
            self.not_empty.notify()

Line by line:

1. Start by acquiring the lock to the Buffer. This ensures that only a single thread can access the critical section. Doing this prevents race conditions when multiple producers access the buffer simultaneously
2. The while loop checks if the buffer at capacity. Why a loop and not an if-statement? A while loop allows handling wake-up events, i.e. the buffer becomes filled below capacity. What does the while-loop do?
   • It outputs the the current state of the buffer.
   • The wait() call does the following:
     • It atomically releases the lock.
     • It blocks the thread until it receives a notification.
     • And it re-acquires the lock before returning.
   • The wait() call allows other threads, particularly consumers, to access the buffer while the producer waits.
3. When space becomes available in the buffer, append the item to the buffer. Remember that this in the critical section so only a single thread will have access to the buffer at this point.
4. Call the notify() method on the not_empty condition, to wake up a single consumer thread waiting for items. Mind that although the notification is a signal the buffer’s state has changed, it does not transfer control of the buffer directly. This only happens when the methoed exits the with block.

The Buffer class also contains a consume() method to fetch items from the buffer.

    def consume(self):
        with self.lock:
            while not self.buffer:
                print(f"Consumer: Buffer is empty, waiting for items.")
                self.not_empty.wait()

            item = self.buffer.pop(0)
            print(f"Consumer: Removed {item}. Current buffer: {self.buffer}")
            self.not_full.notify()
            return item

Line by line:

1. Like the producer() method, the consumer() method needs access to the buffer through the use of a lock.
2. Also like the producer() method a while loop makes sure it handles wake-ups where a thread resumes without notification. It also ensures the condition is still valid after the thread regains the lock.
3. In the while loop:
   • Output a warning message, since the buffer is empty.
   • The call to the wait() method does the following:
     • It atomically releases the lock and suspends the thread.
     • The thread remains blocked till a producers adds an item and notifies the thread.
     • When the thread is woken up, the lock is automatically re-acquired before proceeding. This ensures only a single thread at a time can access the buffer.
4. The method can now get an item from the buffer and does so using the pop() method.
5. A notification is printed out.
6. Because an item has been removed from the buffer it is not full, so the not_full condition is notified, so a waiting producer thread can add an item to the buffer.
7. Finally the item is returned to the caller.
8. Note that the lock is only released after the method exits the with statement.

The full Buffer implementation looks like this:

class Buffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = []
        self.lock = threading.Lock()
        self.not_empty = threading.Condition(self.lock)
        self.not_full = threading.Condition(self.lock)

    def produce(self, item):
        with self.lock:  # Acquire the monitor's lock
            while len(self.buffer) >= self.capacity:
                print(f"Producer: Buffer is full, waiting for space. Current buffer: {self.buffer}")
                self.not_full.wait()

            self.buffer.append(item)
            print(f"Producer: Added {item}. Current buffer: {self.buffer}")
            self.not_empty.notify()

    def consume(self):
        with self.lock:  # Acquire the monitor's lock
            while not self.buffer:
                print(f"Consumer: Buffer is empty, waiting for items.")
                self.not_empty.wait()

            item = self.buffer.pop(0)
            print(f"Consumer: Removed {item}. Current buffer: {self.buffer}")
            self.not_full.notify()
            return item

Next implement a function to produce items:

def producer_task(buffer:Buffer, num_items:int):
    for i in range(num_items):
        item = f"item_{i}"
        buffer.produce(item)
        time.sleep(random.uniform(0.1, 0.5))

Line by line:

1. The function has two parameters: the shared buffer, and a number indicating the number of items to be produced.
2. Iterate over a range determined by the num_items parameter.
3. It creates named items, and adds those to the buffer.
4. Finally it deliberately pauses execution for a random number of seconds.

The consumer_task() works in a similar fashion:

def consumer_task(buffer:Buffer, num_items:int):
    for i in range(num_items):
        buffer.consume()
        time.sleep(random.uniform(0.2, 0.6))

Line by line:

1. The function receives two parameters, the shared buffer and a number indicating how many items the consumer with retrieve.
2. For each item the function calls the consume() method, which allows thread-safe retrieval through the monitor-pattern.
3. After an item is consumed the thread pauses execution for a random number of seconds. This is done because:
   • It creates natural variability in consumption rates.
   • Pausing prevents the consumer from continuously polling the buffer and thereby prevents wasting CPU cycles.
   • Also it simulates processing time for the consumed item.

Testing time

Now it is time to test the code:

if __name__ == "__main__":
    buffer_capacity = 5
    shared_buffer = Buffer(buffer_capacity)

    num_producers = 2
    num_consumers = 3
    items_per_producer = 10
    items_to_consume = num_producers * items_per_producer // num_consumers 

    producers = []
    consumers = []

    for i in range(num_producers):
        p = threading.Thread(target=producer_task, args=(shared_buffer, items_per_producer))
        producers.append(p)
        p.start()

    for i in range(num_consumers):
        c = threading.Thread(target=consumer_task, args=(shared_buffer, items_to_consume))
        consumers.append(c)
        c.start()

    for p in producers:
        p.join()

    for c in consumers:
        c.join()

    print("\nAll tasks finished.")
    print(f"Final buffer state: {shared_buffer.buffer}")

Line by line:

1. Start by setting up a shared buffer with a capacity of five.
2. Then come the concurrency variables:
   • There will be two producer threads.
   • Each will produce ten 10 items.
   • Three consumer threads will retrieve these items.
   • To distribute the workload the number of items each consumer needs to consume is calculate using integer division.
3. The pattern to created producer and consumer tasks is roughly the same:
   • Create a thread object with appropiate function and the correct arguments.
   • Store the thread in a list for later reference and management.
   • Start the thread. This launches the thread in a separate thread of control.
4. After all the threads have been launched, the program uses the join() method to wait for completion of each of these threads. It first waits for the producers to finish producing items, finally it waits for the consumer to finish processing. This joining strategy is not strictly needed, as the consumers will naturally wait for the items to be produced.
5. Finally the program prints out a completion message and the current state of the buffer.

Conclusion

Coding this pattern in Python is not hard, and the code is clear and easy to maintain. It also makes clear that writing concurrent code must be done carefully.