CODEWITHSUNDEEP
c#multithreadingconcurrencymessage-queue
Dabloons
Dabloons

Reputation: 1502

How to create custom message pump?

I am doing a small exercise where I need to create something akin to a message pump. What I have is a queue of work to do, and I want the work to be done entirely on one thread while any thread can add work to the queue to be done. 

Queue<WorkToDo> queue;

The threads use a wait handle to tell the pump that there is work to do.

WaitHandle signal;

The pump just loops as long as there is work to do and then waits for the signal to start again.

while(ApplicationIsRunning){
    while(queue.HasWork){
        DoWork(queue.NextWorkItem)
    }
    signal.Reset();
    signal.WaitOne();
}

Every other thread can add work to the queue and signal the wait handle...

public void AddWork(WorkToDo work){
    queue.Add(work);
    signal.Set();
}

The problem is, if work is being added fast enough, a condition can occur where work can be left in the queue because the between the queue check for work and the the WaitHandle reset, another thread can add work to the queue.

How would I go about mitigating that circumstance without putting a costly mutex around the the WaitHandle?

Upvotes: 4

Views: 2008

Answers (3)

Servy
Servy

Reputation: 203827

You can use BlockingCollection<T> to make implementing the queue much easier, as it will handle the synchronization for you:

public class MessagePump
{
    private BlockingCollection<Action> actions = new BlockingCollection<Action>();

    public void Run() //you may want to restrict this so that only one caller from one thread is running messages
    {
        foreach (var action in actions.GetConsumingEnumerable())
            action();
    }

    public void AddWork(Action action)
    {
        actions.Add(action);
    }

    public void Stop()
    {
        actions.CompleteAdding();
    }
}

Upvotes: 3

David Hope
David Hope

Reputation: 2256

You shouldn't have to do a full mutex, but can you put a lock statement

public void AddWork(WorkToDo work)
{
  queue.Add(work);
  lock(lockingObject)
  {
    signal.Set();
  }
}

use whatever you want for the locking object, most people would say that using the signal itself is a bad idea.

Responding to @500 - Server Internal Error's comment below, you could just reset the signal prior to doing the work. The following should protect things:

while(ApplicationIsRunning)
{
  while(queue.HasWork)
  {
    WorkItem wi;
    lock(lockingObject)
    {
       wi = queue.NextWorkItem;
       if(!queue.HasWork)
       {
          signal.Reset();
       }
    }
    DoWork(wi)
  }

  signal.WaitOne();
}

This way, if you have more work, the internal queue keeps going. If not, it falls out to the signal.WaitOne(), and we only reset if there isn't more work already queued.

The only downside here is that it's possible that we will reset multiple times in a row if work comes in while the DoWork is executing.

Upvotes: 2

NG.
NG.

Reputation: 22904

You could use the WaitOne(TimeSpan) method so that you have a hybrid signal/polling loop. Basically specify a timespan of 1sec of waiting at most. This would lead to a task getting caught in that race to be held for at most a second (or whatever poll time you specify) or until another task is added to your queue.

Upvotes: 0

Related Questions