cameron314 / cameron314/concurrentqueue

How to drain a multi-producer queue properly?

Open
#426 1 comment 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
C++
Stars
12.5k
Forks
1.9k
PR merge metrics
No merged PRs in 30d

Description

I struggle to figure out how to drain a queue properly.
tl;dr Sentinel-based approach fails because sending the sentinel last doesn't guarantee it's read last.

I have a multi-producer queue.
I have multiple consumers, however they (blockingly) dequeue under a mutex, so from the queue perspective it's single-consumer:

(This is simplified - I actually use wait_dequeue_bulk with multiple iterations per under single mutex lock, but the core logic is the same)

void consumer() {
  while (true) {
    Item item;
    {
       const std::lock_guard<std::mutex> lock(gMutex);
       wait_dequeue(&item);
    }
    process(item);
  }
}

Now, I want to drain the queue and exit gracefully.

My initial idea was to:

  1. Make sure producers are stopped.
  2. Add the sentinel to the queue (from the "main" thread rather than producer).
  3. When a consumer reads sentinel, it sets the global "exit_now" flag.
bool gExitNow = false;

void consumer() {
  while (true) {
    Item item;
    {
       const std::lock_guard<std::mutex> lock(gMutex);
       if (gExitNow) return;
       wait_dequeue(&item);
       if (item == kSentinel) {
         gExitNow = true;
         return;
       }
    }
    process(item);
  }
}

However, the documentation says that the order is not guaranteed in multiple-producer use case, so if there are some items in the queue when I add sentinel, it may happen that the sentinel would come up ahead of time and there are some other items in the queue.

So far I have a few options to work around it, but they don't look elegant.

  1. After stopping all producers and before sending the sentinel, busy wait until size_approx() == 0. Doesn't sound exiting, and also I'm not sure whether I can rely on size_approx() not returning 0 when there's something in the queue.

  2. Before sending sentinel, set another bool shutdown_mode to true. In shutdown_mode, consumers would start to read non-blockingly rather than blockingly, and exit if there's nothing to read. When a consumer reads sentinel, it skips it but doesn't return (it will return on the next iteration when discovers non-blockingly that there's nothing to read).

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the queue documentation for wait_dequeue, wait_dequeue_bulk, and size_approx, then inspect the documented ordering guarantees for multi-producer use. Done means establishing and documenting a reliable graceful-drain approach that handles queued items and consumer shutdown without relying on an unsupported ordering assumption.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp
Domain
backend
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.