Imagine a kitchen where cooks prepare dishes and servers carry them away. A shared counter lets both sides work at their own pace, but it must not overflow, and servers must not take from an empty counter. The producer–consumer pattern coordinates tasks that generate data with tasks that process it through a shared buffer. section 9.8
One or more producers add items while one or more consumers remove and process them. Synchronization prevents three failures: producers adding to a full buffer, consumers reading an empty buffer, and concurrent tasks conflicting while accessing the buffer. section 9.8
A standard solution assigns each synchronization tool a focused responsibility:
- The empty semaphore counts available buffer slots. Producers acquire it before inserting and wait when the buffer is full. section 9.8
- The full semaphore counts produced items. Consumers acquire it before removing and wait when the buffer is empty. section 9.8
- A mutex protects the critical section so only one thread accesses the shared buffer at a time. section 9.8
# Chapter 9/producer_consumer.py
import time
from threading import Thread, Semaphore, Lock
SIZE = 5
BUFFER = ["" for i in range(SIZE)] #1
producer_idx: int = 0
mutex = Lock()
empty = Semaphore(SIZE)
full = Semaphore(0)
class Producer(Thread):
def __init__(self, name: str, maximum_items: int = 5):
super().__init__()
self.counter = 0
self.name = name
self.maximum_items = maximum_items
def next_index(self, index: int) -> int:
return (index + 1) % SIZE
def run(self) -> None:
global producer_idx
while self.counter < self.maximum_items:
empty.acquire() #2
mutex.acquire() #3
self.counter += 1
BUFFER[producer_idx] = f"{self.name}-{self.counter}"
print(f"{self.name} produced: "
f"'{BUFFER[producer_idx]}' into slot {producer_idx}")
producer_idx = self.next_index(producer_idx)
mutex.release() #3
full.release() #4
time.sleep(1)
class Consumer(Thread):
def __init__(self, name: str, maximum_items: int = 10):
super().__init__()
self.name = name
self.idx = 0
self.counter = 0
self.maximum_items = maximum_items
def next_index(self) -> int: #5
return (self.idx + 1) % SIZE #5
def run(self) -> None:
while self.counter < self.maximum_items:
full.acquire() #6
mutex.acquire() #7
item = BUFFER[self.idx]
print(f"{self.name} consumed item: "
f"'{item}' from slot {self.idx}")
self.idx = self.next_index()
self.counter += 1
mutex.release() #8
empty.release() #9
time.sleep(2)
if __name__ == "__main__":
threads = [
Producer("SpongeBob"),
Producer("Patrick"),
Consumer("Squidward")
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
Explore a Python implementation using threads, a fixed-size buffer, one mutex, and semaphores for empty slots and available items. section 9.8
