Placeholder: investigate switching from a channel to locks and convars

[rkuris]

But the worker is still blocking on the channel. I'm not sure I understand the difference between blocking on a channel versus blocking on a condition variable.

In an effort of separating concerns that are coupled, the code introduces an implicit dependency between the counter in the semaphore and a secondary counter that can lead to a deadlock if there are any "math" errors. Instead of trying to catch this through comprehensive code review, it would be better to use a synchronization approach where there is a single counter (if possible).

Originally posted by @bernard-avalabs in #1685 (comment)

[bernard-avalabs]

The current solution uses a combination of a modified semaphore and a channel to apply back pressure, with the goal of bounding the staleness of the most recent persisted revision. In general, mixing semaphores and channels (or any other synchronization primitive) can result in tricky and difficult to reason about solutions. In this case, there is an implicit dependence between the values of the counter inside the semaphore, and a separate counter managed by the event loop to determine when to "release permits" in the modified semaphore. This can lead to a deadlock if the semaphore counter reaches zero before the event loop counter falls below the threshold to reset the semaphore. The current solution may work, but it is fragile since a change to just one of the counter thresholds (max permits or count value before reset) could lead to a deadlock that might be difficult to find during standard tests.

This back pressure problem is actually very similar to the standard textbook bounded buffer problem, with the only difference relevant to synchronization being when to signal the consumer of the buffer, which is a common variant of the problem.

The textbook solution to the bounded buffer-based producer/consumer problem uses 1 lock, and 2 condition variables, which are synchronization primitives that are designed to work together. Here is C-like pseudo-code for the modified solution that meets the basic requirements of our problem:

// Assume 0 < THRESHOLD < MAX_COUNT. 
// Example values can be THRESHOLD = 2, MAX_COUNT = 4
int shared_count = MAX_COUNT
int shared_item;

void produce(int item) {
    lock(&mutex);
    while (shared_count == 0) {
        cond_wait(&commit_not_full, &mutex);
    }

    // Update shared item
    shared_item = item;
    
    // Decrement and check threshold trigger
    if (--shared_count <= THRESHOLD) {
        cond_signal(&persist_ready);  // Assuming just one consumer
    }
    unlock(&mutex);
}

int consume() {
    lock(&mutex);
    while (shared_count > THRESHOLD) {
        cond_wait(&persist_ready, &mutex);
    }
   
    // Save a local copy of the shared_item. Optionally clear shared_item.
    int local_item = shared_item;
   
    // Reset shared_count to initial value.
    shared_count = MAX_COUNT;

    // Wakeup all (if any) producers that are blocked
    cond_broadcast(&commit_not_full);
    
    unlock(&mutex);
    return local_item;
}

This is based on the solution outlined in Silberschatz's "Operating System Concepts". A similar producer/consumer solution is also in Remzi's "Operating Systems: Three Easy Pieces" which is available for free online: https://pages.cs.wisc.edu/~remzi/OSTEP/threads-cv.pdf

In this solution, there is a single shared counter. Producers produce until that counter goes to 0. When it does, it blocks on commit_not_full. It signals the consumer to wakeup when the shared counter is 2 or less after it decrements it. The consumer sleeps on "persist_ready" while the shared counter is greater than 2. When it does wakes up, it consumes the shared item, resets the counter to its initial value, and calls broadcast to wake up any producers that are current sleeping. A broadcast is used here instead of signal because the consumer has effectively consumed multiple items instead of just one.

In my opinion, this textbook solution is easier to reason about than one based on mixing semaphores and channels. It also doesn't require sending messages and waking up the event loop thread even when there is nothing for it to do other than to update a counter.

[rkuris]

The snippet

   // Reset shared_count to initial value.
    shared_count = MAX_COUNT;

is incorrect. It needs to increase the shared count back to however many it persisted, which means it has to take the lock again.

The semaphore permit system records how many commits were actually persisted, but since that takes a while, it can't increase the count until the persist is done.

In particular, imagine that it takes 10s, so add a sleep right before you update shared_count. Now you can see that this can (and will) conflict with the producer.

[bernard-avalabs]

The implementation shouldn't be persisting inside the critical section. The code snippet takes out the item (representing a revision), and that revision should be persisted outside of the critical section.

If you need to maintain an invariant that there is a max number of commits that can occur before a persist, then you can just move the counter update and broadcast into a separate lock/unlock block after persisting.

void consume_completed(int num_commits_persisted) {
    lock(&mutex);
    shared_count += num_commits_persisted;
    cond_broadcast(&commit_not_full);
    unlock(&mutex);
}

I believe num_commits_persisted is just MAX_COUNT minus the shared count value at the time of consume. Just return a local copy of shared count in consume in the same way that the code is returning a copy of shared_item.

[rkuris]

I don't understand how this doesn't deadlock. Imagine if the consumer is very slow to start up, and the producer produces the maximum count before blocking on the cond_wait. Then, the consumer starts up, and blocks on the very first line (the lock).

It looks to me like the commit_not_full condvar is never sent unless you already have the lock, unless I'm missing something.

[bernard-avalabs]

Blocking on cond_wait releases the mutex. This is why the mutex is taken as a parameter to cond_wait. The consumer would not block on the lock in your scenario.

Unblocking from a cond_wait automatically re-acquires the mutex.

https://linux.die.net/man/3/pthread_cond_wait

[rkuris]

Ah yes, that makes sense. I do like the fact that every commit isn't working it's way down the pipe, but it's now the producer's choice as to when to persist instead of the consumer. There is definitely less data to move and the window where the shared data is incorrect is much smaller.

We would need to clear shared_item. We can't keep a reference to a revision longer than we need it.

I think this code is more traditional and avoids having a queue, but I'm not convinced this is less deadlock-prone or easier to reason about than the existing queue approach. We can discuss this option a bit more in person.l

[rkuris]

I created an implementation of this as a prototype. It passes the tests but I'm not convinced this is simpler still. The branch is here:

#1697