Introduction

The Producer-Consumer problem is a classic set of scenarios in concurrent programming, first described by Edsger W. Dijkstra in 1965. The problem shows the need of synchronizing several threads or processes sharing a common resource.

In this problem, there are two kinds of entities that interact with a common shared resource. Producers, who create data, and Consumers that process these data. The difficulty occurs when producers and consumers operate at different rates, potentially leading to circumstances in which producers overrun the buffer (filling it) or consumers exhaust it (emptying it).

To overcome this, synchronization techniques are required to ensure that the shared resource is used safely and orderly. For example, when the buffer is filled, producers must wait for consumers to remove elements from the buffer, allowing more space for newer items. Consumers must wait when the buffer is empty since there are no data points to process. If we let the producers and consumers run without supervision, the threads may encounter race conditions, inconsistent states, or even deadlocks, all of which compromising the reliability of our system.

This challenge replicates real-world situations including work scheduling, logging systems, or data streaming pipelines rather than only a theoretical exercise. In multithreaded systems, the producer-consumer pattern is fundamental since it captures the complexity of organizing independent entities with a dependence.

But where is my ConcurrentQueue<T>?

This post focuses on using lock to build a producer-consumer system, instead of using higher-level tools like ConcurrentQueue<T>. While ConcurrentQueue<T> makes it easy to create thread-safe systems, and is something you should use 99.9999% of the time in your production code, it hides how synchronization actually works. By using lock, you can see how to protect shared resources and prevent threads from interacting with each other.

Understanding the Problem

The Producer-Consumer problem can be boiled down to the following definition. We have a finite-size buffer and two classes of threads, producers and consumers, that put items into the buffer (producers) and take items out of the buffer (consumers). Let's break it down:

What is a Shared Buffer?

A shared buffer is a common data structure, more often than not a queue, that acts as the intermediary between producers and consumers, that allows us to decouple them. Producers add items (or data) in the shared buffer, while consumers remove and process these items. The shared buffer has a fixed size meaning that it does not have infinite space, meaning that only a fixed number of items can exist at any point inside the buffer. Thus a buffer can become full or empty which can cause issues:

• When the Buffer is Full:
  
  A full buffer means no more items can be added until space is freed. If a producer tries to add an item to a full buffer, several things can go wrong. The buffer might throw an exception, disrupting the system and potentially causing a crash. In extreme cases, the producer might overwrite memory it doesn’t own, leading to unpredictable and catastrophic behavior. Alternatively, the buffer might silently drop the item, resulting in data loss and leaving consumers in an inconsistent state. Finally, the producer could block indefinitely while waiting for space, and if consumers are also stalled, this can lead to a deadlock, where neither producers nor consumers can make progress.

• When the Buffer is Empty:
  
  An empty buffer means there are no items available for consumers to process. If a consumer tries to fetch data from an empty buffer, it might throw an exception or return invalid data, potentially crashing the thread or propagating errors. Without proper safeguards, the consumer could also process uninitialized or corrupted data, leading to incorrect behavior. Worse, the consumer might block indefinitely while waiting for items that will never arrive, resulting in a deadlock, as both producers and consumers remain stuck and unable to continue.

What is a Consumer?

A Consumer is a key component in a producer-consumer system, responsible for retrieving items from a shared buffer and perform operations on them. A consumer's role is to process the data or tasks provided by the producer, usually finishing the workflow by providing results, storing information, or initiating downstream actions. Consumers in a logging system, for example, write log entries to a file or database; in a data processing pipeline, they analyze or transform incoming data; and in a task queue, they perform tasks such as image rendering or transaction processing.

A consumer relies only on the shared buffer for input. If the buffer goes empty, the consumer must either wait for more items to arrive or gracefully deal with the lack of data. The way a consumer deals with this is determined by how it is designed. Blocking consumers will pause execution until the items are available, but non-blocking consumers may skip processing or retry later. A badly designed consumer may even crash or consume erroneous data if it is not correctly synced with the buffer.

What is a Producer?

A Producer is an essential component in a producer-consumer system, responsible for generating data or tasks and placing them into a shared buffer for consumers to process. Producers act as the starting point of the workflow, supplying the system with the necessary input to keep it running. In a logging system, a producer may generate log messages; in a data processing pipeline, a producer may read sensor data or retrieve information from an external source. The primary responsibility of a producer is to guarantee that relevant data or tasks are regularly made available to consumers without overwhelming the system.

A producer relies only on the shared buffer for input. It must also handle situations in which the buffer is full. Depending on the design, the producer may block until space becomes available, drop the item, or activate backpressure mechanisms to slow its generation rate. The choice is based on the system's requirements and tolerance for delays or data loss. For example, a real-time system may favor speed over dependability and drop data, but a critical system may stall the producer in order to ensure that every item is processed.

What is a Semaphore?

A Semaphore is a synchronization primitive used in computer science to control access to a common resource by multiple threads and avoid critical section problems in a concurrent system. It's like a manager that orchestrates which threads can enter a section of code that shouldn't be accessed by multiple threads at the same time. A semaphore uses a counter to track how many threads or processes can access the resource simultaneously. The counter begins with a specified initial value, representing the initial number of "permits" for accessing the resource. When a thread wants to use a resource, it decrements the counter, effectively "acquiring" a permit. If the counter is greater than zero, the thread can proceed. However, if the counter reaches zero, the thread blocks and waits until another thread releases a permit by incrementing the counter.

To understand how it works in practice, consider a semaphore initialized with a count of 3, representing three available slots for accessing a shared resource.

When processes interact with the semaphore, the count is decremented or incremented based on whether they are acquiring or releasing a resource. For instance:

Initial Semaphore Count: 3

Process A --wait()-> Decrease Count: 2
Process B --wait()-> Decrease Count: 1
Process C --wait()-> Decrease Count: 0 (No more available slots)

Process A --signal()-> Increase Count: 1 (Resource released) 

Let’s say three processes—Process A, Process B, and Process C—attempt to access the resource sequentially. When Process A calls wait(), it decrements the semaphore’s count from 3 to 2 and proceeds to access the resource. Similarly, Process B calls wait(), further decreasing the count to 1, and gains access. When Process C calls wait(), it reduces the count to 0, taking the last available slot. At this point, the semaphore’s counter indicates that no slots are left, and any subsequent process calling wait() will block until a slot becomes available.

Once Process A completes its work, it calls signal() to release its slot, increasing the semaphore’s count from 0 to 1. This signals that a resource is now free, allowing another waiting process to proceed. The flow of these operations ensures that only up to three processes can access the resource simultaneously, while additional processes are queued and served in turn as resources are freed.

In a producer-consumer system, semaphores play a critical role in managing the flow of data. For instance, a semaphore can track the number of items in the buffer, blocking consumers if the buffer is empty, or track the available slots, blocking producers if the buffer is full. This coordination ensures that producers and consumers interact with the shared resource in an orderly and efficient manner, preventing race conditions or data corruption.

What is a Mutex?

A mutex (short for mutual exclusion) is a synchronization tool that allows only one thread or process to access a shared resource at a time. Unlike a semaphore, which can manage multiple simultaneous accesses, a mutex ensures exclusive access by locking the resource when one thread acquires it. Other threads attempting to acquire the mutex will block until the resource is released. This behavior prevents race conditions and ensures thread safety when accessing critical sections of code or shared data.

To understand how a mutex works, imagine a scenario where multiple threads compete to access a shared resource, such as a log file or a database connection. When a thread acquires the mutex, it effectively locks the resource, preventing other threads from interfering. Once the thread completes its task, it releases the mutex, unlocking the resource for the next waiting thread.

Initial Mutex State: Unlocked

Thread A --acquire()-> Mutex is Locked
Thread B --acquire()-> Mutex is Locked, wait
Thread C --acquire()-> Mutex is Locked, wait

Thread A --release()-> Mutex is Unlocked

This flow ensures that only one thread can access the shared resource at any given time. Unlike a semaphore, which tracks permits with a counter, a mutex is binary—it is either locked or unlocked. Additionally, mutexes are typically owned by the thread that locks them, meaning only the owning thread can release the mutex. This ownership guarantees that other threads cannot accidentally unlock a mutex they did not acquire, ensuring safe and predictable behavior.

Using Semaphores and Locks

To solve the producer-consumer problem, we can use two semaphores and a lock. Each plays a distinct role in ensuring safe and efficient synchronization.

First we are going to use two semaphores for tracking the items in the buffer. The ConsumerSemaphore tracks the number of available items in the buffer. Consumers wait on this semaphore when the buffer is empty. The ProducerSemaphore tracks the number of available slots in the buffer for producers. Producers wait on this semaphore when the buffer is full.

The BufferLock is used to ensure mutual exclusion when we are modifying our shared buffer. Both producers and consumers must acquire the lock before accessing or modifying the buffer.

The solution works by coordinating the actions of producers and consumers using semaphores and locks. Semaphores ensure that threads only proceed when specific conditions are met. For example, a producer is blocked from adding items if the buffer is full, as the space semaphore’s count will be zero. Similarly, a consumer cannot remove an item from an empty buffer, as the item semaphore’s count will prevent them from proceeding. These semaphores act as gatekeepers, ensuring producers and consumers interact with the buffer only when it is in a valid state.

The lock ensures mutual exclusion when modifying the shared buffer. Without it, multiple threads could access the buffer simultaneously, leading to race conditions, data corruption, or other undefined behavior. By requiring each thread to acquire the lock before modifying the buffer, we ensure that only one thread at a time can perform operations like adding or removing items. This protects the integrity of the buffer and prevents conflicts between producers and consumers.

Finally, the combination of semaphores and locks maintains a balanced workflow between producers and consumers. The semaphores dynamically manage the flow of threads, allowing producers to proceed only when there is space in the buffer and consumers to act only when there are items to process. This synchronization mechanism prevents deadlocks and ensures that both producers and consumers can operate efficiently without wasting resources.

Now, lets see the Produce method, that is used by producers to add data to the buffer:

PROCEDURE Produce();
BEGIN
    Wait(ProducerSemaphore);            # Wait for space in the buffer.
    Lock(BufferLock);                   # Lock the buffer for safe access.

    Add(item);                          # Add an item in the buffer.

    Unlock(BufferLock);                 # Unlock the buffer.
    Signal(ConsumerSemaphore);          # Signal that an item is available.
END;

Here's how it works:

1. Wait for Space:
   • Wait(ProducerSemaphore) ensures that there is available space in the buffer before proceeding. If the buffer is full, the producer will block until space becomes available.
2. Lock the Buffer:
   • Lock(BufferLock) acquires a lock on the shared buffer to prevent simultaneous access by other threads (producers or consumers). This ensures mutual exclusion.
3. Add an Item:
   • Add(item) safely places a new item into the buffer while the lock is held.
4. Unlock the Buffer:
   • Unlock(BufferLock) releases the lock, allowing other threads to access the buffer.
5. Signal Availability:
   • Signal(ConsumerSemaphore) increments the semaphore, notifying consumers that a new item is now available in the buffer.

The Consume method, is used by consumers to get data from the buffer:

PROCEDURE Consume();
BEGIN
    WAIT(ConsumerSemaphore);            # Wait for items in the buffer.
    Lock(BufferLock);                   # Lock the buffer for safe access.

    Remove(item);                       # Remove an item from the buffer.

    Unlock(BufferLock);                 # Unlock the buffer.
    Signal(ProducerSemaphore);          # Signal that a slot is available
END;

Here's how it works:

1. Wait for Items:
   • WAIT(ConsumerSemaphore) ensures that there is at least one item in the buffer before proceeding. If the buffer is empty, the consumer will block until an item becomes available.
2. Lock the Buffer:
   • Lock(BufferLock) acquires a lock on the shared buffer to ensure mutual exclusion, preventing other threads (producers or consumers) from accessing the buffer simultaneously.
3. Remove an Item:
   • Remove(item) safely retrieves an item from the buffer while holding the lock. This operation ensures that no other thread modifies the buffer during the removal.
4. Unlock the Buffer:
   • Unlock(BufferLock) releases the lock, allowing other threads to access the buffer.
5. Signal Slot Availability:
   • Signal(ProducerSemaphore) increments the semaphore, notifying producers that a slot in the buffer is now free and available for new items.

With that out of the way, lets move to the actual C# implementation.

Implementation

Now that we've discussed the concept of solving the producer-consumer problem using semaphores and locks, let's dive into the implementation. Our goal is to create a system where multiple producers can generate items and add them into a shared buffer, while multiple consumers can retrieve and process those items, all in a thread-safe manner.

The Shared Buffer

In this implementation, we are going to use a queue as the shared buffer to hold items produced by the producers until they are consumed. The queue allows us to demonstrate the concepts of synchronization and coordination in a straightforward and easily understandable manner. Specifically, the queue serves as a simple first-in, first-out (FIFO) data structure, making it intuitive to see how producers add items and consumers retrieve them.

However, it’s important to note that in production environments, you should almost always use a concurrent data structure, such as ConcurrentQueue in .NET. These data structures are optimized for multi-threaded scenarios and handle thread safety internally, making them more efficient and less error-prone. While our implementation adds locks and semaphores to make a standard queue thread-safe, using a ConcurrentQueue would eliminate the need for such explicit synchronization logic. This approach ensures better performance and maintainability in real-world applications.

For the purpose of this demonstration, we are intentionally using a manually synchronized queue to showcase how semaphores and locks work together to solve the producer-consumer problem. This allows you to understand the mechanics behind concurrency before relying on higher-level abstractions in production code.

The shared resource is designed to manage a queue in a thread-safe way, allowing producers to add items and consumers to retrieve them without conflicts. To achieve this, we use a combination of a queue, semaphores, and a lock.

public class SharedResource
{
    /// <summary>  
    /// The underlying queue that holds the data.  
    /// Access is synchronized with the <see cref="_lock"/> object.  
    /// </summary>  
    private readonly Queue<int> _queue;  

    /// <summary>  
    /// A Semaphore that tracks the number of items currently available in the queue.  
    /// Ensures consumers block when the queue is empty.  
    /// </summary>  
    private readonly SemaphoreSlim _emptyQueueSemaphore;  

    /// <summary>  
    /// A Semaphore that tracks the number of slots currently available in the queue.  
    /// Ensures producers block when the queue is full.  
    /// </summary>  
    private readonly SemaphoreSlim _fullQueueSemaphore;  

    /// <summary>  
    /// Lock object to ensure thread-safe access to the <see cref="_queue"/>.  
    /// This prevents race conditions during enqueue and dequeue operations.  
    /// </summary>  
    private readonly object _lock = new object();  

    /// <summary>  
    /// The capacity of our queue.  
    /// </summary>  
    private readonly int _maxQueueSize;}

The _queue variable is the actual queue where items are stored. It acts as the shared buffer between producers and consumers. Since multiple threads can access this queue at the same time, we need to protect it using a lock to avoid issues like race conditions.

The _emptyQueueSemaphore keeps track of how many items are currently in the queue. If the queue is empty, consumers will block on this semaphore until a producer adds an item. It ensures consumers don’t try to take something that isn’t there.

The _fullQueueSemaphore manages the available space in the queue. If the queue is full, producers will block on this semaphore until a consumer removes an item, making space available. It ensures the queue never overflows.

The _lock is a simple mechanism to ensure only one thread accesses the queue at a time. This is critical when adding or removing items to avoid data corruption or unexpected behavior.

Finally, the _maxQueueSize is the maximum number of items the queue can hold. It defines the size of the buffer and ensures that the system respects the queue’s limits.

public class Shared Resource
{
    // Variable declaration...

    public SharedResource(int capacity)  
    {  
        if (capacity <= 0)  
        {  
            throw new ArgumentOutOfRangeException(nameof(capacity), "Capacity must be greater than zero.");  
        }  
        _maxQueueSize = capacity;  
        _queue = new Queue<int>(capacity);  
        _emptyQueueSemaphore = new SemaphoreSlim(0);
        _fullQueueSemaphore = new SemaphoreSlim(capacity);
    }
}

The constructor sets everything up to make the queue functional and safe for concurrent use.

The constructor ensures the capacity is greater than zero. If someone tries to create a queue with zero or negative capacity, it throws an error. After all, a queue with no capacity wouldn’t make sense, and the _queue is created with the given capacity. This prepares the queue to handle items up to the specified limit.

Also the two semaphores are initialized. The _emptyQueueSemaphore starts with a count of 0, meaning that the queue is initially empty, so consumers will wait until producers add items. The _fullQueueSemaphore starts with a count equal to the queue's capacity, indicating that the queue has all its slots available for producers to use.

With the shared resource in place, we can now define the methods for producing and consuming items. These methods handle the core logic of adding and removing items from the queue, ensuring thread safety and synchronization through semaphores and locks.

Both methods are implemented as asynchronous to prevent blocking the execution of other tasks. Without this, producers or consumers waiting for semaphores to release (e.g., when the queue is full or empty) would block the entire thread, reducing the overall responsiveness and scalability of the system. By using asynchronous methods, threads can perform other work while waiting, making the application more efficient, especially in scenarios involving multiple producers and consumers.

The ProduceAsync method

The ProduceAsync method is responsible for generating items and adding them to the queue, waiting if necessary for space to become available.

public async Task ProduceAsync(int value, CancellationToken cancellationToken = default)  
{  
    await _fullQueueSemaphore.WaitAsync(cancellationToken); // Wait for space.  
    lock (_lock) // Enter critical section.  
    {  
        _queue.Enqueue(value);  
        Console.WriteLine($"Produced: {value}");  
    } // Exit critical section.  
    _emptyQueueSemaphore.Release(); // Signal that there are items available in the queue.  
}

The ProduceAsync method is the core logic for adding items to the shared queue in a thread-safe and synchronized manner. It ensures that producers only add items when there is space available and that access to the queue is properly coordinated with other threads.

The method starts by awaiting _fullQueueSemaphore.WaitAsync. This semaphore tracks the available slots in the queue. If the queue is full, the producer will wait until a consumer removes an item, freeing up space. The asynchronous nature of WaitAsync ensures that the thread isn’t blocked while waiting, allowing other operations to proceed.

A lock on _lock is used to enter the critical section where the producer interacts with the queue. This guarantees that only one thread can modify the queue at a time, preventing race conditions. Inside the critical section, the item (value) is added to the queue using _queue.Enqueue(value).

Once the item is added, the lock is released, allowing other threads (producers or consumers) to access the queue.

Finally, the producer calls _emptyQueueSemaphore.Release to signal that an item is now available in the queue. This semaphore notifies waiting consumers, allowing them to proceed with consuming the item.

The ConsumeAsync method

The ConsumeAsync method handles retrieving items from the shared queue in a thread-safe manner. It ensures that consumers only access the queue when there are items available and coordinates access using semaphores and locks.

public async Task<int> ConsumeAsync(CancellationToken cancellationToken = default)  
{  
    await _emptyQueueSemaphore.WaitAsync(cancellationToken);  
    var value = 0;  
    lock (_lock)  
    {  
        value = _queue.Dequeue();  
        Console.WriteLine($"Consumed: {value}");  
    }  
    _fullQueueSemaphore.Release();  

    return value;  
}

The method starts by awaiting _emptyQueueSemaphore.WaitAsync. This semaphore tracks the number of items currently available in the queue. If the queue is empty, the consumer will block here, waiting for a producer to add an item. By using the asynchronous version (WaitAsync), the thread is not blocked, allowing other tasks to execute while the consumer waits.

A lock on _lock ensures mutual exclusion while interacting with the queue. Once inside the critical section, the consumer safely retrieves an item using _queue.Dequeue(). This ensures no other threads (producers or consumers) can modify the queue while the current thread is working. After retrieving the item, the lock is released, allowing other threads to access the queue.

The consumer calls _fullQueueSemaphore.Release, signaling that a slot in the queue is now available for producers to use. This step ensures producers don’t remain blocked when the queue has room for new items.

Finally, the method returns the value retrieved from the queue, allowing the caller to process the consumed item.

Finally, we have two helper methods AvailableSpaces and AvailableElements, that return the current count of the semaphores:

public int AvailableSpaces => _fullQueueSemaphore.CurrentCount;
public int AvailableElements => _emptyQueueSemaphore.CurrentCount;

The Consumer Class

The Consumer class represents an entity that retrieves and processes items from the shared resource in a thread-safe and synchronized manner. Its role is to repeatedly consume items as they become available, ensuring no race conditions occur during the interaction with the shared resource.

private readonly SharedResource _resource;  

public Consumer(SharedResource queue)  
{  
    _resource = queue;   
}

The constructor takes a single parameter of the SharedResource type. This is the shared buffer (queue) that the consumer will interact with.

The ConsumeAsync method performs the actual consumption of items in an infinite loop, which is gracefully terminated using a CancellationToken.

public async Task ConsumeAsync(CancellationToken cancellationToken)  
{  
    while (!cancellationToken.IsCancellationRequested)  
    {   
        if (_resource.AvailableElements == 0)  
        {  
            Console.WriteLine($"Consumer: No available elements, waiting for semaphore access");  
        }  

        var item = await _resource.ConsumeAsync(cancellationToken);  
        Console.WriteLine($"Consumer: Processed {item}");  
    }  
}

Before attempting to consume an item, the method checks the shared resource’s AvailableElements property. If the queue is empty, it logs a message indicating that the consumer is waiting for semaphore access.

The consumer calls the shared resource’s ConsumeAsync method to dequeue an item. This method handles all synchronization (waiting for the semaphore, locking the queue, and signaling availability to producers). Once an item is successfully retrieved, the consumer logs the action with its identifier (Consumer) and the item value. This provides clear visibility into the actions of consumers.

The loop continues until the CancellationToken signals a request to stop. This allows the system to gracefully terminate consumers when needed.

The Producer Class

The Producer class represents an entity that generates and adds items to the shared resource. Its role is to repeatedly produce items while ensuring thread-safe interactions with the shared buffer. Like the Consumer class, it operates continuously and can be gracefully terminated using a CancellationToken.

public class Producer  
{  
    private readonly SharedResource _resource;  

    public Producer(SharedResource resource)  
    {  
        _resource = resource;  
    }  

    public async Task ProduceAsync(CancellationToken token)  
    {  
        var item = 0;  
        while (!token.IsCancellationRequested)  
        {    
            if (_resource.AvailableSpaces == 0)  
            {  
                Console.WriteLine($"Producer: No available spaces, waiting for semaphore access");  
            }  

            Console.WriteLine($"Producer: Adding {item}");  
            await _resource.ProduceAsync(item++, token);  
        }  
    }
}

The ProduceAsync method contains the logic for generating and adding items to the shared resource. The method uses a local variable item to track the current value being produced. This value increments with each iteration of the loop, ensuring each item added to the queue is unique.

Before attempting to produce an item, the method checks the shared resource’s AvailableSpaces property to determine if the queue has room. If the queue is full, it logs a message indicating that the producer is waiting for space. The method logs the item being added, providing a clear view of the producer's activity in the system.

The producer calls the shared resource’s ProduceAsync method, passing the item and the cancellation token. This method handles synchronization (waiting for the semaphore, locking the queue, and signaling availability to consumers).

The loop continues producing items until the CancellationToken signals a request to stop. This ensures the producer can be gracefully terminated when the system needs to shut down or adjust workload.

Demonstrating the Producer-Consumer System

To see the producer-consumer system in action, we need a main method to tie everything together. This method will:

1. Initialize the Shared Resource: Set up the queue with a fixed capacity.
2. Create Producers and Consumers: Instantiate multiple producers and consumers to simulate real-world behavior.
3. Run Asynchronous Tasks: Start the producers and consumers on separate tasks so they can work concurrently.
4. Graceful Shutdown: Use a CancellationTokenSource to allow stopping the system by user input.

public static async Task Main(string[] args)
{
    const int bufferSize = 5; // Define the size of the queue
    const int producerCount = 2; // Number of producers
    const int consumerCount = 3; // Number of consumers

    // Create the shared resource
    var sharedResource = new SharedResource(bufferSize);

    // Create a cancellation token for stopping the system
    var cts = new CancellationTokenSource();

    // Create and start producer tasks
    var producers = Enumerable.Range(0, producerCount)
        .Select(_ => new Producer(sharedResource))
        .Select(producer => Task.Run(() => producer.ProduceAsync(cts.Token)))
        .ToArray();

    // Create and start consumer tasks
    var consumers = Enumerable.Range(0, consumerCount)
        .Select(i => new Consumer(sharedResource, i))
        .Select(consumer => Task.Run(() => consumer.ConsumeAsync(cts.Token)))
        .ToArray();

    Console.WriteLine("Press any key to stop the system...");
    Console.ReadKey();
    cts.Cancel(); // Signal cancellation to stop all tasks

    // Wait for all tasks to complete
    await Task.WhenAll(producers.Concat(consumers));
    Console.WriteLine("System shut down gracefully.");
}

Enhancements

The producer-consumer implementation we’ve discussed is robust and demonstrates key synchronization principles. However, there are a few enhancements worth noting to make the example more realistic and to clarify its behavior under certain conditions.

Adding Random Delays to Simulate Real-Life Behavior

In real-world scenarios, producers and consumers rarely operate at consistent, predictable rates. For example:

• A producer might experience delays due to external factors like data acquisition time or resource availability.
• A consumer might take longer to process certain items depending on their complexity.

To simulate this variability, we can introduce random delays into the producer and consumer loops. This creates a more dynamic system and allows us to observe how the semaphores and locks handle non-deterministic behavior.

Here's how we can modify the methods:

Producer

var random = new Random();
while (!token.IsCancellationRequested)
{
    if (_resource.AvailableSpaces == 0)
    {
        Console.WriteLine($"Producer: No available spaces, waiting for semaphore access");
    }

    Console.WriteLine($"Producer: Adding {item}");
    await _resource.ProduceAsync(item++, token);

    await Task.Delay(random.Next(100, 1000)); // Random delay between 100ms and 1000ms
}

Consumer

var random = new Random();
while (!cancellationToken.IsCancellationRequested)
{   
    if (_resource.AvailableElements == 0)
    {
        Console.WriteLine($"Consumer: No available elements, waiting for semaphore access");
    }

    var item = await _resource.ConsumeAsync(cancellationToken);
    Console.WriteLine($"Consumer: Processed {item}");

    await Task.Delay(random.Next(500, 1500)); // Random delay between 500ms and 1500ms
}

By introducing these delays, the producer-consumer system more closely resembles real-world conditions where production and consumption rates fluctuate. It also showcases how the synchronization mechanisms handle unpredictable workloads.

The Illusion of First-Come, First-Served

While the implementation might seem like it operates on a First-Come, First-Served (FCFS) basis, this is not strictly true. The order in which threads are scheduled and granted access to semaphores is determined by the operating system’s thread scheduler, which prioritizes threads based on several factors, such as:

• Thread priority.
• CPU time availability.
• The system’s overall load.

This means that even if Producer A tries to enqueue an item before Producer B, the OS might schedule Producer B’s thread first, allowing it to acquire the semaphore and the lock before Producer A. Similarly, consumers might not process items in the exact order they arrive, depending on how the OS schedules their threads.

If two consumers are waiting for items, you might observe that the one which started waiting last gets to consume first, even though the queue maintains the correct order of items. This is because semaphore access is not inherently fair—it does not guarantee that threads are served in the order they began waiting.

Understanding this nuance is important when designing systems where strict fairness or predictable thread execution order is critical. If FCFS behavior is a strict requirement, additional mechanisms, such as a per-thread semaphore or queueing requests, would need to be implemented.

If strict First-Come, First-Served (FCFS) behavior is critical for your use case, additional mechanisms such as per-thread semaphores or request queues can be implemented. If you’re interested in exploring solutions to enforce true FCFS order in a producer-consumer system, let me know in the comments—I’d be happy to dive into it in a future post!

Conclusion

The producer-consumer problem is a cornerstone of concurrent programming, demonstrating how multiple threads can safely interact with shared resources. In this post, we explored how to solve this problem using semaphores and locks, breaking down the core concepts into an intuitive, step-by-step approach. We built a shared resource with a thread-safe queue, implemented producers and consumers that operate asynchronously, and demonstrated how semaphores and locks work together to ensure synchronization and prevent race conditions.

Beyond the implementation, we delved into enhancements that bring the system closer to real-world scenarios, such as adding random delays to simulate varying workloads. We also discussed the nuances of thread scheduling and why, despite appearances, our solution isn’t strictly First-Come, First-Served (FCFS)—a limitation determined by the operating system’s thread scheduler.

This journey through the producer-consumer problem has provided both practical and conceptual insights. Whether you’re implementing this pattern in a real-world application or simply looking to deepen your understanding of concurrency, this post should serve as a solid foundation.

If you have questions, suggestions, or would like to see specific aspects—such as enforcing true FCFS order—covered in future posts, let me know in the comments.