Repository navigation
Conversation
cameron314
left a comment
cameron314
left a comment
There was a problem hiding this comment.
Thanks for having a look at this. See inline comments.
| to stop calling wait_dequeue (see [draining a blocking queue](#draining-a-blocking-queue-when-producers-finish) | ||
| below for an alternative). This is necessary because otherwise a consumer |
There was a problem hiding this comment.
| to stop calling wait_dequeue (see [draining a blocking queue](#draining-a-blocking-queue-when-producers-finish) | |
| below for an alternative). This is necessary because otherwise a consumer | |
| to stop calling wait_dequeue (see [draining a blocking queue](#draining-a-blocking-queue-when-producers-finish) | |
| below for an example). This is necessary because otherwise a consumer |
| while (!producersDone.load(std::memory_order_acquire)) { | ||
| if (q.wait_dequeue_timed(item, std::chrono::milliseconds(100))) { | ||
| consumeItem(item); | ||
| } | ||
| } | ||
| // Acquire producer completion before the final dequeue attempts. | ||
| do { | ||
| while (q.try_dequeue(item)) { | ||
| consumeItem(item); | ||
| } | ||
| } while (doneConsumers.fetch_add(1, std::memory_order_acq_rel) + 1 == ConsumerCount); |
There was a problem hiding this comment.
It's not immediately clear to me whether this is equivalent to the non-blocking version. While it may be easier to digest as two dequeueing loops in this example, it is preferable to use a single loop if possible, since that allows the application to put the dequeueing logic in one place (does not require a consumeItem function to literally exist).
| consumers[i] = std::thread([&]() { | ||
| Item item; | ||
| while (!producersDone.load(std::memory_order_acquire)) { | ||
| if (q.wait_dequeue_timed(item, std::chrono::milliseconds(100))) { |
There was a problem hiding this comment.
I really don't like encouraging timeout-based synchronization. Even if correct, it is always a code smell.
Unfortunately, as there is currently no unified waitable API in C++, there's no wait to wait on both the queue and the producer completion condition directly. However, this can be worked around with sentinel values (possibly with a wrapper struct for something more generic).
| const int ConsumerCount = 8; | ||
| std::thread producers[ProducerCount]; | ||
| std::thread consumers[ConsumerCount]; | ||
| std::atomic<bool> producersDone(false); |
There was a problem hiding this comment.
Why use a boolean flag here instead of a count like in the previous example? This requires the application to already be synchronizing its producers in one place in order to apply the example (i.e. it is needlessly restrictive).
|
Updated in 66adf7f to use your per-producer sentinel suggestion from #426. Each producer now sends a completion marker after its final item; after receiving all producer markers, the last consumer sends one stop marker per consumer. This avoids timeout polling and keeps item processing in one loop. It also removes the joined-producers boolean; completion is counted from the individual producers' markers. The explanation distinguishes work being dequeued from work finishing processing: joining every consumer handles the latter. I also changed the link wording to “example.” The exact example passed 16 configurations/scenarios under GCC and Clang ASan/UBSan on Linux ARM64, including no work, delayed producers and slow consumers. No Windows or model-checker validation is claimed. |
2 participants
The blocking producer/consumer example requires knowing the total item count. This adds an example for producers whose work count is not known in advance, addressing #426.
Each producer appends a completion marker to its own queue. After consumers receive all producer markers, the last one sends one stop marker per consumer. The example uses per-producer ordering, a single consumption loop and blocking waits; it does not depend on cross-producer ordering or timeout polling. All consumer threads are joined before queue destruction so in-flight processing also finishes.
Validation: compiled the exact example as C++11 with GCC and Clang ASan/UBSan on Linux ARM64. Both passed 16 cases covering empty and uneven work, delayed production, slow processing, and 8/8, 1/1, 8/1 and 1/8 producer/consumer configurations. Every accepted item was processed exactly once and the queue was empty after joining. Windows and a model checker were not run. The sample documents positive thread counts, nonthrowing Item/application operations and fatal enqueue failure handling.
AI assistance: implemented and locally verified with OpenAI Codex.