In " ", I wrote about the notorious GIL in Python and did a simple experiment on existing concurrency mechanism in Python. Today, I'll describe another common scenario in concurrent programming, the Producer/Consumer problem, with Python. Concurrent Programming in Python is not what you think it is What is the producer/consumer scenario? Imagine you are at a fast-food chain, in which there are an order line and a serving line. The staff in the serving line prepare the meals while their coworkers in the order line take orders from customers. In this scenario, the staff in the order line are known as , who produce order chits, whereas their coworkers in the serving line are , who consume order chits and prepare the meals. You may see producers and consumers as two distinct groups of entities, connected with the list of orders (or a better term in software engineering, a ) producers consumers queue For human-being, coordinating these groups of workers (consumers and producers) seems relatively straightforward and intuitive. However, for computer sense, it's unfortunately not the case. So, what's the catch of the Producer/Consumer problem? Let's put our fast-food chain scenario into code: queue threading

orders = queue.Queue() : 
        new_order = orders.get() orders.task_done() _ range( ):
        orders.put( ) order_line = [threading.Thread(target=order_line_or_producer) _ range( )] serving_line = [threading.Thread(target=serving_line_or_consumer) _ range( )] [t.start() t order_line]
[t.start() t serving_line] orders.join() [t.join() t order_line]
[t.join() t serving_line] """
Implementation 1: Infinite Loop in Consumers
""" import import : def serving_line_or_consumer () while True # prepare meals from `new_order`, assuming GIL is released while preparing meals # Invoke this to indicate the "order" in the Queue is processed : def order_line_or_producer () # Each staff in the serving line produces 200 orders for in 200 "Order" # Let's put 4 staff into the order line for in 4 # Let's assign 6 staff into the serving line for in 6 # Put all staff to work for in for in # "join" the order, block until all orders are cleared # "join" the threads, ending all threads for in for in Let's focus on the consumer logic in : serving_line_or_consumer() : new_order = orders.get() orders.task_done() : def serving_line_or_consumer () while True # PROBLEM: Wait for orders # prepare meals from `new_order`, assuming GIL is released while preparing meals # Invoke this to indicate the "order" in the Queue is processed The consumer is implemented with an infinite while loop. With an infinite loop, when there aren't any orders to be prepared, the consumer thread will be . If you're unsure why is an anti-pattern, it is because we are simply spinning in an infinite loop, waiting for an event to occur. While busy waiting seems simple and straightforward to  implement, CPU cycles are wasted. A comprehensive explanation of busy waiting can be found . busy waiting busy waiting here What if we terminate consumers once they finished preparing all orders? Let's put it into code again: queue threading

orders = queue.Queue() has_order = has_order:
        new_order = orders.get() new_order : has_order = orders.task_done() _ range( ):
        orders.put( ) order_line = [threading.Thread(target=order_line_or_producer) _ range( )] serving_line = [threading.Thread(target=serving_line_or_consumer) _ range( )] [t.start() t order_line]
[t.start() t serving_line] [orders.put( ) _ range(len(serving_line))] orders.join() [t.join() t order_line]
[t.join() t serving_line] """
Implementation 2: Use a Sentinel Value to Stop Busy Waiting
""" import import : def serving_line_or_consumer () True while if is None # Check for Sentinel Value False # prepare meals from `new_order`, assuming GIL is released while preparing meals # Invoke this to indicate the "order" in the Queue is processed : def order_line_or_producer () # Each staff in the serving line produces 200 orders for in 200 "Order" # Let's put 4 staff into the order line for in 4 # Let's assign 6 staff into the serving line for in 6 # Put all staff to work for in for in # ADDED THIS: Inform serving line no more orders None for in # "join" the order, block until all orders are cleared # "join" the threads, ending all threads for in for in In the second implementation, we use a sentinel value to inform the consumers when there aren't any new orders. We added a line to put into : None None orders [orders.put( ) _ range(len(serving_line))] None for in We then updated the consumer logic to regard the sentinel value: has_order = has_order:
        new_order = orders.get() new_order : has_order = orders.task_done() : def serving_line_or_consumer () True while if is None # Check for Sentinel Value False # prepare meals from `new_order`, assuming GIL is released while preparing meals # Invoke this to indicate the "order" in the Queue is processed Okay, with the new logic, whenever the consumer encounters the sentinel value , it will break from the infinite loop. However, the caveat is, we have to explicitly recreate a consumer whenever the system gets new orders after consumers are terminated. This isn't really what desired in our scenario. You wouldn't want to dismiss the staff in the serving line when there aren't any new orders. None A Much Better Solution: Use a Semaphore In a producer/consumer problem, it is best to use a concurrency control object: Lock, Mutex, Semaphore, etc. If you find these terms foreign, let just focus on for now. semaphore What is a Semaphore? A semaphore is a variable used to control access to a critical section in a multi-processing environment. In a multi-processing environment (a program that involves more than 1 thread/process), a critical section is a group of resources (variables, data, etc) shared among threads/processes. In Python, Semaphore is a class with an internal counter, which value is always an integer, and at least 2 methods: , and . I will explain what and perform in the latter part. acquire() release() acquire release In our fast-food chain scenario, the use of semaphore is not for controlling access to shared resources, but to allow consumers to while for an event. Let's put semaphore into our scenario: sleep waiting queue threading

orders = queue.Queue()
has_order = threading.Semaphore(value= ) has_order.acquire(): new_order = orders.get() orders.task_done() _ range( ):
        orders.put( )
        has_order.release() order_line = [threading.Thread(target=order_line_or_producer) _ range( )] serving_line = [threading.Thread(target=serving_line_or_consumer) _ range( )] [t.start() t order_line]
[t.start() t serving_line] orders.join() [t.join() t order_line]
[t.join() t serving_line] """
Implementation 3: Use Semaphores
""" import import 0 # ADDED THIS : def serving_line_or_consumer () while # ADDED THIS: Acquire a Semaphore, or sleep until the counter of semaphore is larger than zero # prepare meals from `new_order`, assuming GIL is released while preparing meals : def order_line_or_producer () # Each staff in the serving line produces 200 orders for in 200 "Order" # ADDED THIS: Release the Semaphore, increment the internal counter by 1 # Let's put 4 staff into the order line for in 4 # Let's assign 6 staff into the serving line for in 6 # Put all staff to work for in for in # "join" the order, block until all orders are cleared # "join" the threads, ending all threads for in for in In the latest implementation, the consumers wait to a semaphore. The simplified logic of would be: acquire acquire() If internal counter is :
     allow thread sleep waiting be larger than once internal counter is more than , decrement True internal counter is larger than :
     decrement internal counter True the zero the to while for it to zero the zero it by 1 return if the zero the by 1 return <code> </code> With , our consumers will be idle when there isn't any order the internal counter of the semaphore is zero). acquire() In our case, we updated our to make use of semaphore as well. When a producer receives an order, it the semaphore. In short, doesn't do anything but increment the internal counter of the semaphore. order_line_or_producer() releases release() With the semaphore being incremented, will then act accordingly. This is a much better solution among all. With semaphore, consumers will be put to idle/sleep while they wait for new orders, and producers inform consumers by incrementing the semaphore. acquire() Conclusion In this article, we implemented a producer/consumer problem with 3 different implementations: using an infinite loop to wait for events using a sentinel value to terminate consumers when they are not needed temporarily (the better solution) using a concurrency control object: a semaphore to allow consumers to sleep while waiting Among these, a concurrency control object is much preferred. In this scenario, we used a semaphore. In general, there are other control objects than just semaphore. I highly recommend you explore on your own. Anecdote: In Private File Saver , I implemented a desktop client to sync local files to AWS S3 bucket in Python. The performance was satisfactory until I had to sync more than 12,000 files. It warranted a need to rewrite the logic to be more efficient. One of the obvious improvements would be implementing producer/consumer solution. (The details of this problem can be found here ). First published at melvinkoh.me

Target

Solving Producer/Consumer Problem of Concurrent Programming in Python

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

Asynchronous Tasks with Celery + Redis in Django

The Essential Guide to Relearning Java Thread Primitives

A Guide to Improving Your Python Performance Speed

A simple introduction to Python’s asyncio

Assume It Worked and Fix Later

Asynchronous Tasks with Celery + Redis in Django

The Essential Guide to Relearning Java Thread Primitives

A Guide to Improving Your Python Performance Speed

A simple introduction to Python’s asyncio

Assume It Worked and Fix Later

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps