1 producer 14 Consumers problem

[wysota]

The idea underlying my class is to have a circular buffer able to serve one producer and several consumers. All the consumers must be able to read each byte produced by the producer, and the producer must wait until all the consumers have retrieved the produced data. I do not see anything like that in your implementation.

You forgot to mention that in your post

What you want is broadcast, not a synchronized access to data (and you can do it with events). And you don't need semaphores for that. If you want to keep it the way it is, there is QAtomicInt that can be incremented atomically which is sufficient for your task.

Still you need a single semaphore to protect the buffer itself but then you only need to make sure all consumers accessed the data before overwriting it. You can do it in one of two ways:

1. each item in the buffer has its own semaphore that gets released (with the value of 1) by each consumer after reading the data and gets acquired (with the value of 14 or whatever the amount of consumers you have per cell) by the producer. This makes sure the producer overwrites data in a cell only if all consumers have already accessed it.

2. each item in the buffer has a counter that gets incremented when a consumer accessed it. The producer has to wait until the value of that counter equals the number of consumers (+1 if the producer itself also increments the buffer to provide consistency). This can also be implemented using a mutex and a wait condition to make sure the producer only reads the counter if its value has changed recently.

Right now I can think of at least one more way involving two semaphores and a bunch of QAtomicInt objects to solve the task. One semaphore protects the buffer from overrunning, the other from underrunning. One is released by the producer each time it writes an item, the other is released by the last consumer accessing the item. Consumers use the atomic integer to check if they are the last one to read the data and if they should release the semaphore.

Here is the supposed code for implementing the consumer:

int currentIndex = ...;
 
underrunSemaphore.acquire(1);
consumeData(data[currentIndex]);
int val = data[currentIndex].counter.fetchAndAddOrdered(1);
if(val==1){
  // I'm the second (last) consumer to access the data
  overrunSemaphore.release(1);
}
currentIndex++;

int currentIndex = ...;

underrunSemaphore.acquire(1);
consumeData(data[currentIndex]);
int val = data[currentIndex].counter.fetchAndAddOrdered(1);
if(val==1){
  // I'm the second (last) consumer to access the data
  overrunSemaphore.release(1);
}
currentIndex++;

To copy to clipboard, switch view to plain text mode