Removing this odd while(true) for checking the queue

Question

I was reviewing some WPF code I've written in C# last year.

The scenario is the following:

• I've got a usercontrol that show usermessages and it's docked in the bottom of the application's workspace
• I've got a ConcurrentQueue that receives message via a IMessageMediator (from Catel, but's irrilevant how data are pushed)

• I've a workerprocess that checks if there's some item in the queue and add them to a list that's bound to a grid

  private readonly ConcurrentQueue<UserMessage> queue;
  private readonly List<UserMessage> dataItems;
  private void worker_DoWork(object sender, DoWorkEventArgs e)
  {
   while (true)
   {
   if (!queue.IsEmpty)
   {
   UserMessage message;
   while (queue.TryDequeue(out message))
   {
   dataItems.Add(message);
   }
   RaisePropertyChanged(() => FilteredDataItems);
   RaisePropertyChanged(() => ErrorCountMessage);
   RaisePropertyChanged(() => WarningsCountMessage);
   }
   Thread.Sleep(500);
   }
  }
  

  //That's used to show only filtered items (Error/Warning....)

  public IEnumerable<UserMessage> FilteredDataItems
  {
   get
   {
   if (!enabledFilters.Any())
   return dataItems;
   return dataItems.Where(x => enabledFilters.Contains(x.LogMessageType));
   }
  }
  

How can I remove this while(true) and the ugly/odd Thread.Sleep(500)?

• Der Kommissar: Use a Timer.
• moarboilerplate: RaisePropertyChanged--are you mixing MVVM in here?

Answer 1

It sounds like you are sleeping only to avoiding busy waiting on the queue too frequently.

How about following the suggestion in this SO answer:

https://stackoverflow.com/a/5014271/2268996

so that you use the blocking Take method in your worker with code like this:

while (true)
{
 UserMessage m = blockingQueue.Take();
 dataItems.add(m);
 RaisePropertyChanged(() => FilteredDataItems);
 ... (no need to sleep here)...
}

You will still have the while (true) but not the Thread.sleep(...) call. Will that work for your situation?

• dcastro: If you're going to use a blocking collection, you might as well get rid of Take, and replace the while loop with foreach(var msg in queue.GetConsumingEnumerable()), which is how it's meant to be used.
• ErikR: @dcastro - thanks for the tip! I'm not a C# programmer. :-)

Answer 2

You can easily use a System.Timers.Timer here. This would be the best-practice in my opinion.

Assuming worker_DoWork only gets called once (otherwise why would you need a while (true) loop in it?), then a Timer can be initiated in this method.

This isn't guaranteed valid, compile-able code (I wrote it far outside an IDE):

private Timer queueManagementTimer;
private void worker_DoWork(object sender, DoWorkEventArgs e)
{
 queueManagementTimer = new Timer(500); /* timeout, 500ms = 0.5s*/
 queueManagementTimer.Elapsed += OnQueueManagementTimerElapsed;
 queueManagementTimer.Enabled = true;
 queueManagementTimer.AutoReset = false;
}
private static void OnQueueManagementTimerElapsed(Object source, ElapsedEventArgs e)
{
 if (!queue.IsEmpty)
 {
 UserMessage message;
 while (queue.TryDequeue(out message))
 {
 dataItems.Add(message);
 }
 RaisePropertyChanged(() => FilteredDataItems);
 RaisePropertyChanged(() => ErrorCountMessage);
 RaisePropertyChanged(() => WarningsCountMessage);
 }
 queueManagementTimer.Enabled = true;
}

This makes it much more idiomatic as to what is going on. It's precisely clear what the intention is, and Windows will manage all the triggers for you. :) It also allows you to keep control and not have an additional thread just looping.

By setting queueManagementTimer.AutoReset = false in the worker_DoWork method, you guarantee that the timer will not continue counting down after it fires. The queueManagementTimer.Enabled = true line in OnQueueManagementTimerElapsed will re-enable the timer once the method is done. This means the timer can never fire twice if the method takes an absurdly long time to run.

• Laurent Trentin: I've adopted for your solution since it can happen that I've a lot of "traffic" and so the solution propsed by @ErikR was performing worste... how can I block the OnQueueManagementTimerElapsed so that if it tooks more then 500ms I don't have event fired twice?
• Der Kommissar: @LaurentTrentin Simply set queueManagementTimer.AutoReset = false; in the end of the worker_DoWork method, then add queueManagementTimer.Enabled = true; at the end of the OnQueueManagementTimerElapsed method.
• Nikita B: I would not recommend this solution. Any approach which uses time intervals (be it Thread.Sleep or Timer) to check the input introduces unnecessary lag. You should use model based on signals instead. It can be BlockingCollection as suggested above (if you are using .Net 4.5+), or it can be simple ManualResetEvent. The idea is the same either way: you shoud wake up your thread only when (and as soon as) you receive new data.
• Der Kommissar: @NikitaBrizhak Until there are other things you need to do with the queue in that time interval. (Log the fact that no records have been updated, etc.) As it stands, using either of those two methods will not allow you to care for certain time-sensitive issues that can occur.

Answer 3

What you want is to have multiple producers - and a single consumer.

You can use TPL Dataflow's AsyncBlock<T> for this.

Here, AsyncBlock<T> acts as a single-threaded thread-safe consumer.

var opts = new ExecutionDataflowBlockOptions { BoundedCapacity = 1, };
var consumer = new ActionBlock<UserMessage>(msg =>
{
 dataItems.Add(msg);
 RaisePropertyChanged(() => FilteredDataItems);
 RaisePropertyChanged(() => ErrorCountMessage);
 RaisePropertyChanged(() => WarningsCountMessage);
}, opts);

And then, use Post to process new messages

consumer.Post(msg1);
consumer.Post(msg2);

When you're done sending messages (if ever), use this:

consumer.Complete(); //signal you're done sending messages
await consumer.Completion; //wait for all messages to be dequeued and processed.

See Stephen Cleary's Async Producer/Consumer Queue using Dataflow