First, it’s not clear when a pipeline like this would actually be useful. There is no way for the producer and consumer to simultaneously make progress, and this code will have problems if there are multiple producer or consumer threads.

Also, in general it’s probably not the best design to acquire a lock in one function and release it in another. It makes it more difficult to reason about the code. Further, this precludes the use of a context manager, so one needs to think about exceptions.

Of course, for the “real” implementation one wants semaphores, which are discussed later, although for a “toy” example one could doing polling with “sleep()s”.

malbert137 on June 2, 2020

import random
import concurrent.futures
import time
import threading

FINISH = 'THE END'

class ToyQueueWithPollingNotForProduction:
    def __init__(self, capacity, polling_interval=0.05):
        self.capacity = capacity
        self.messages = []
        self.lock = threading.Lock()
        self.polling_interval = polling_interval
    def send_message(self, message):
        print(f'sending message of {message}')
        while True:
            with self.lock:
                if len(self.messages) < self.capacity:
                    self.messages.append(message)
                    return
            time.sleep(self.polling_interval)
    def recv_message(self):
        while True:
            with self.lock:
                if self.messages:
                    msg = self.messages.pop(0)
                    print(f'consuming message of {msg}')
                    return msg
            time.sleep(self.polling_interval)

producer_pipeline = []
consumer_pipeline = []

def producer(pipeline):
    for _ in range(pipeline.capacity):
        message = random.randint(1, 100)
        producer_pipeline.append(message)
        pipeline.send_message(message)
    pipeline.send_message(FINISH)

def consumer(pipeline):
    message = None
    while message is not FINISH:
        message = pipeline.recv_message()
        if message is not FINISH:
            consumer_pipeline.append(message)
            time.sleep(random.random())
        else:
            break


if __name__ == '__main__':
    pipeline = ToyQueueWithPollingNotForProduction(10)
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as ex:
        ex.submit(producer, pipeline)
        ex.submit(consumer, pipeline)
    print(f'producer: {producer_pipeline}')
    print(f'consumer: {consumer_pipeline}')

Manish Sharma on Oct. 7, 2023

I couldn’t understand why producer and consumer methods are symbiotically related, like why locking in other and releasing in other.

Bartosz Zaczyński RP Team on Oct. 8, 2023

@Manish Sharma The producer and consumer share a common resource—the pipeline. In order to ensure that only one of them can access the shared resource at a time—so that the producer doesn’t override unconsumed data or the consumer doesn’t try to start reading incomplete data—they must cooperate with each other through locks.

Whenever the producer finishes writing the data, it notifies the consumer that it’s safe to read from the pipeline by releasing the corresponding lock. Conversely, when the consumer clears the pipeline, it notifies the producer by releasing the producer’s lock.

In that sense, the producer and consumer are in a symbiotic relationship because they use each other’s states to coordinate and synchronize their operations.

I hope that clears it up for you.

Tony Ngok on Feb. 11, 2024

In this example, are these two ex.submit() creating a producer and a consumer thread respectively?

Bartosz Zaczyński RP Team on Feb. 12, 2024

@Tony Ngok The idea is to allocate a pool of threads upfront to pay the cost of their creation only once. When you submit a task to the executor, it assigns one of the available threads to your job or puts it on a waitlist. So, calling ex.submit() never creates a new thread. Not sure if this is what you were asking about, though.

Become a Member to join the conversation.