Tell if Concurrent Queue updated without looping

Question

thanks for the assistance. I've got a triple-threaded process, linked by a concurrent queue. Thread one processes information, returns to the second thread, which places data into a concurrent queue. The third thread is just looping like so:

while (true) {
    if(queue.TryDequeue(out info)) {
        doStuff(info);
    } else {
        Thread.Sleep(1);
    }
}

Is there a better way to handle it such that I'm not iterating over the loop so much? The application is extremely performance sensitive, and currently just the TryDequeue is taking ~8-9% of the application runtime. Looking to decrease that as much as possible, but not really sure what my options are.

score 2 · Accepted Answer · answered Aug 07 '15 at 19:37

2

You should consider using System.Collections.Concurrent.BlockingCollection and its Add() / Take() methods. With Take() your third thread will be just suspended while waiting for new item. Add() is thread safe and can be used by second thread.

With that approach you should be able to simplify your code into something like that:

while (true) {
   var info = collection.Take();
   doStuff(info);
}

answered Aug 07 '15 at 19:37

pkmiec

2,594
18
16

Or use `GetConsumingEnumerable` https://msdn.microsoft.com/en-us/library/dd287186(v=vs.110).aspx – tia Aug 07 '15 at 19:39
1

I had been using a BlockingCollection previously, but found I was spending even more time in the *.take() than with the while(true) loop. Not sure on the programmatic differences there but swapping to a ConcurrentQueue seemed to show a performance boost. I'll swap back and see if I can confirm/deny that again – user2864757 Aug 07 '15 at 19:50
How are you measuring that time? – pkmiec Aug 07 '15 at 20:04
before the action I save off Environment.TickCount, afterwards I subtract that from the saved TickCount and add it to a counter – user2864757 Aug 07 '15 at 20:17
Seems ok.. One more possibility you can try is ConcurrentBag, if you don't care about order of items.. – pkmiec Aug 08 '15 at 04:27

score 0 · Answer 2 · edited May 23 '17 at 10:30

0

You can increase the sleep time. I would also use await Task.Delay instead of sleep. This way you can wait longer without the extra cpu cycles that Thread.Sleep uses and still be able to cancel the delay by making use of the CancellationTokenSource.

On another note, there are better ways of queuing up jobs. Taking into consideration that it appears you want to run these jobs synchronously, an example would be to have a singleton class that takes your work items and queues them up. So if there are no items in the queue when you add one, it should detect that and then start your job process. At the end of your job process, check for more work, use recursion to do that work or if no more jobs then exit the job process, which will run again when you add an item to the empty queue. If my assumption is wrong and you can run these jobs in parallel, why use a queue?

You may like to use a thread safe implementation of ObservableCollection. Check out this SO question ObservableCollection and threading

edited May 23 '17 at 10:30

Community

1
1

answered Aug 07 '15 at 19:36

Jacob Roberts

1,725
3
16
24

Same note as the above answer, I need these to all run asynchronously for performance reasons. App is handling > 100k rabbitmq messages per second right now (peak ~138k/s), running synchronously destroys the speed. I'm worried that if I wait until I put data into the queue to spawn a processing thread that I'll get backed up, as even a few ms delay between the task creation and task actual start would be devastating. That's why I have a dedicated task running right now full time – user2864757 Aug 07 '15 at 19:57
So why even use the queue at all and just have your second task call directly into your third task? – Jacob Roberts Aug 07 '15 at 20:04
I've got multiple of the first task running asynchronously, and preservation of order is mandatory so I can't have them fire off into the next task whenever they finish. Current setup is a queue of tasks, such that they run asynchronously and I just pop the top task, wait for it to finish, grab the result and pass it to the next task. This is all based on RabbitMQ messages, one of the first tasks is spawned on receiving a message and begins to process it. I've got up to 3 of those process tasks running simultaneously as that's the bulk of the actual workload done in the process – user2864757 Aug 07 '15 at 20:20
Seems you should consider re-writing your workflow. You are worried about the order of when they start but if you are adding tasks to the thread pool, the thread pool is responsible for determining which task should run next. This doesn't guarantee any type of order. Without knowing your entire workflow, I'm afraid I can't be of much help as to how your execution should be laid out. Maybe consider raising events to trigger the next step... again this is just me speculating a particular order in chaos. – Jacob Roberts Aug 07 '15 at 20:55
So the way I ensure order is the queue of tasks. As tasks are spawned (so, as messages come in and cause a task to be spawned), the task is placed into a queue. The tasks themselves don't affect outside data, so the order in which they actually finish is of no consequence. All I need to do to ensure proper order is pop the top task off of the queue, call task.wait() if necessary, and then grab that result. Setting it up in this way allows parallelization of the processing, while maintaining order in the output – user2864757 Aug 07 '15 at 21:40
@user2864757 what was your final solution? – pogorman Apr 12 '17 at 18:56

score -1 · Answer 3 · answered Aug 07 '15 at 19:41

-1

I don't have a recommendation that avoids looping, however I would recommend you move away from

while (true)

and consider this instead:

MyThing thing;
while (queue.TryDequeue(out thing))
{
    doWork(thing);
}

Put this in a method that gets called each time the queue is modified, this ensures it is running when needed, but ends when not needed.

answered Aug 07 '15 at 19:41

Colorado Matt

349
3
8

This seems to take away from what the OP wants. This code will exit if `TryDequeue` is false so there is no way to keep checking for new jobs. – Jacob Roberts Aug 07 '15 at 19:46
I'm a little reluctant to do this, as I necessarily must ensure this doWork() never gets backed up. I'm concerned spawning a new task when there's work to do could result in a delay between the task creation and activation, which is completely unacceptable for the process. – user2864757 Aug 07 '15 at 19:52
Jacob, If TryDequeue is false there is nothing left to do, which is why I state the looping method must be called again any time something is added. Once this process starts, assuming multiple items queued, then it will continue 'till finished. – Colorado Matt Aug 07 '15 at 21:51

Tell if Concurrent Queue updated without looping

3 Answers3