Is it cleaner to decrement a sleeping process counter in the waking process or the woken process?

Question

Say I have a reader and a producer that share a buffer. I have a wait queue for the reader to put itself in if there is no contents in the buffer when it reads. I also have an atomic variable for keeping track of the number of readers.

The producer wakes readers when contents are added to the buffer.

Is it cleaner to decrement the reader counter in the producer, after it does the waking, or the reader, after it wakes?

Example code

void producer(){
    add_to_buffer();
    if(atomic_read(num_waiting_readers) > 0){
        wake_event(consumer_queue);
        // Cleaner here?
        atomic_decrement(num_waiting_readers);
    }
}

void consumer(){
    if(buffer_isempty()){
        atomic_increment(num_waiting_readers);
        wait_event(consumer_queue, 0);
        // Or here, after wake?
        // atomic_decrement(num_waiting_readers);
    }
    read_buffer();
}

Edit: In my specific case, there is only one reader, but the producer might be called multiple times. The libraries are Linux kernel 4.9 (c language), although in my actual code I use the interruptible versions of wait_event and wake_event

Answer 1

Is it cleaner to decrement the reader counter in the producer, after it does the waking, or the reader, after it wakes?

The cleanest code is code where you can read it and tell confidently if it is correct or not.
For multi-threaded code, this means that after every statement where there is a possible interaction with another thread, you analyze what the effects are if your current thread got interrupted by that other thread at exactly that point of execution.

I will give an example of such an analysis on the consumer thread in your code. Each liner of code is prefixed with the thread it executes on and in comments are some observations from the analysis.

/*C*/ void consumer(){
/*C*/     if(buffer_isempty()){
/*C*/         atomic_increment(num_waiting_readers);
/*P*/ void producer(){
/*P*/     add_to_buffer();
/*P*/     if(atomic_read(num_waiting_readers) > 0){
/*P*/         wake_event(consumer_queue);            // Event fizzles, as there are no threads waiting on it!!
/*P*/         atomic_decrement(num_waiting_readers); // This prevents the next producer to re-trigger the event
/*P*/     }
/*P*/ }
/*C*/        wait_event(consumer_queue, 0);          // Consumer will wait forever to be woken
/*C*/     }
/*C*/     read_buffer();
/*C*/ }

The fix to this race condition is to do both the increment and the decrement in the consumer thread and accept that there might be multiple wake_event events before the consumer actually wakes up.

Answer 2

I'm not a parallelism expert, but here's what concerns me.  If there are multiple producers, then there is a race condition by the > 0 test and the wake/decrement.

Let's say there is only one consumer actually waiting on the queue, so the count is 1.  When the race condition happens, multiple producers will add to the buffer and both detect that there is at least 1 waiting consumer, both then waking & decrementing — taking the counter to -1.

Moving the decrement to before the wake shrinks window of the race condition but doesn't eliminate it.

I submit that the best approach is doing an operation that atomically both decrements if the value is > 0 and also let's you know that the decrement happened (or didn't b/c was 0 to start).

Answer 3

I don't know what language/library you're using here, but for most, you're working at least two levels of abstraction too low. For keeping a count and blocking when the count is too low, you would typically use a semaphore. Additionally, there is a common even higher-level abstraction called a channel or concurrent queue where this is all handled for you. If it's not in the standard library, you should be able to find a third-party library.

As Erik's excellent answer points out, this sort of code is difficult to get correct. Another issue your code doesn't address is back pressure, which keeps producers from producing faster than consumers can consume. There are probably other unhandled boundary conditions I can't pinpoint. In production code, you should use the above higher-level abstractions whenever possible, because they've already thought it through and baked the solutions into the library.

Back to your original question, if you are wanting to work at this lower level as a learning exercise, or if there are other factors limiting the abstractions you can use, correctly addressing those tricky boundary cases will typically result in one thread doing the increment and the other thread doing the decrement.