4

I have a classical one producer - multiple consumers queue, each item can be processed independently. On rare occasions (<1%) there can be multiple queue items related to the same object the consumer is processing. It is perfectly fine to have that with the only constraint being that the object (potentially referenced by more than one queue item) cannot be processed simultaneously by more than 1 consumer. The second queue item would then have to wait until the first queue item is done processing the common object.

Obviously, I don't want to force the consumer to be completely single threaded (99% of the queue items can be processed independently), but at the same time I don't want to create a critical section or a mutex for each object (100 or so critical sections stored in a dictionary does not look good). Is there a better design pattern?

Improve this question
6
  • Depends what you're trying to optimize for, I guess. E.g., can you increase latency? Are you sure you're never going to have much contention of this sort (2 events close in time same object) as you currently claim? What's the throughput (and thus, contention between consumers for getting their end of the queue)? I can think of a solution that only uses as many mutexes as there are consumers but requires a synchronized (i.e., locked) read and write to a (separate) queue (of mutexes) for each item taken off the main queue. That may or may not be acceptable to you.
    – davidbak
    Dec 16, 2021 at 23:28
  • There is never a fixed number of consumers - it is pretty much fire-and-forget situation: a producer receives a notification and fires a worker (using async/await). It does not care what happens afterwards. But it is possible (but rare) to receive a notification about the same object twice, thus resulting in two threads processing the same object on separate threads. That's the situation I want to avoid without havign a keyed critical section for each object (there can be an unlimited number of unique objects).
    – Dmitry Streblechenko
    Dec 16, 2021 at 23:53
  • Well, some applications have a fixed number of consumers, you didn't say. But even with a variable number of worker threads there's probably a maximum you have configured or could configure (beyond which additional workers just add contention and no progress) and if max#consumers << typical # objects a per-consumer mutex could be good. Add a few details to your question to pin things down somewhat ...
    – davidbak
    Dec 16, 2021 at 23:56
  • Just to make it more concrete - I receive a notification from Outlook calendar. The callback fires a separate thread (async) and returns immediately to avoid dropped notifications under heavy loads. The async process check if the calendar item is recurring and has changed occurrences. In that case, it extracts changed occurrences and makes them into standalone appointments and converts the changed occurrence into a deleted occurrence. If two notifications for the same recurring appointment fire within a short interval the splitting code can run twice on separate threads resulting in duplicates
    – Dmitry Streblechenko
    Dec 16, 2021 at 23:57
  • good to know. but you haven't actually described a queue here. is that something you're considering adding to handle this situation? If so, then instead of firing off an arbitrary # of worker threads by firing off one per notification you'd (probably) be running them from a thread pool with a maximum (reasonable) size. Or is it not your intent to introduce an actual queue? (And if not ... maybe the title of this question is misleading ...)
    – davidbak
    Dec 17, 2021 at 0:10

1 Answer 1

Reset to default
5

[As discussed in comments: OP is moving towards an actual queue implementation - a genuine single-producer multiple-consumer pattern.]

Here's one approach based on a queue:

Have a map from underlying item to a queue of tasks to be performed.

Also there is a critical section that will be used to protect two data structures: 1) the task queue (at least removing stuff from it), 2) the map item->pending-task-queue.

All operations described below grab the single critical section as described.

Worker loop start: When pulling a task off the main queue the worker grabs the critical section, gets the task from the queue, and looks in the task and finds the underlying item. From the underlying item he looks in the map. There are two possibilities: it's there, or it is missing.

  1. If it is missing then nobody is working on it. He inserts it in the map where it is the only element of the task queue now associated with this underlying item. He releases the critical section and proceeds to work on the task. When he is done working on it he grabs the critical section again and looks his item up in the map again. Getting the pending task queue from the map he deletes his own task from it. Now, there are two possibilities: The task queue (for that item) is now empty, or it is not.

    1. If it is empty he removes the item from map, releases the crit sect, and goes to find more work to do (goes to "worker loop start").

    2. If it is not empty he's got more work to do on this item. So he gets the next entry of its pending task queue, releases the crit sec, and goes to work as above.

  2. Otherwise, here he has the crit sec and has found a map entry for the underlying item. Since it is in the map that means another worker is working on it. So he enqueues his task (that he got off the main task queue) onto the queue for this underlying item (which is in the map), releases the crit sec, and goes to find more work to do ("worker loop start").

(This approach might not be correct, but what the hell, I just thought of it just now. I'm fairly sure it can be made to work correctly in all concurrent scenarios. You might argue it is complex, you might argue it keeps several data structure operations together under a single critical section which might lead to longer latency than you'd like, you might argue that I didn't supply code. Whatever. Consider this a concept proposal. You can modify it until it works, you can weigh it against any other answer provided here, or you can weigh it against any other algorithm you propose. It's a start!)

Improve this answer
1

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.