Handling events in a multi threaded environment

Question

I have a service that raises multiple events, some of them can be raised at the same time. I need to process those events and run a potentially long running method based on the event arguments.
What I did is to create a BlockingCollection<T> that will store the events and a Task that will keep taking one event at a time until it will be signaled to stop using a CancellationTokenSource.
This is my code:

public class EventsTest
{
 //private fields
 private BlockingCollection<int> _queue; //a concurrent collection to hold incoming events
 private CancellationTokenSource _tokenSource;
 private IoService _ioService; //a service that raises multiple events
 private static EventWaitHandle _eventWaiter; // used for waiting inside a method
 public EventsTest()
 {
 _queue = new BlockingCollection<int>();
 _tokenSource = new CancellationTokenSource();
 _eventWaiter = new EventWaitHandle(false, EventResetMode.AutoReset);
 _ioService = new IoService();
 _ioService.IoEvent += _ioService_IoEvent;
 //Start Listening to IO events
 var t = Task.Factory.StartNew(StartListening, _tokenSource, TaskCreationOptions.LongRunning);
 }
 //IO events listener
 private void _ioService_IoEvent(byte[] desc, int portNum)
 {
 //add events to a blocking collection
 _queue.Add(portNum);
 }
 //a while loop running on a separate task 
 //and taking the event one by one
 private void StartListening(object dummy)
 {
 while (!_tokenSource.IsCancellationRequested)
 {
 try
 {
 var eve = _queue.Take(_tokenSource.Token);
 switch (eve)
 {
 case 0:
 //run the required method on a separate task so it won't block the loop
 Task.Factory.StartNew(LongRunningMethod);
 break;
 case 1:
 //invoke a potential blocking method
 break;
 default:
 break;
 }
 }
 catch (OperationCanceledException)
 {
 // _tokenSource was signaled to stop. do nothing
 }
 }
 }
 static readonly object _locker = new object();
 //sample long running method
 private void LongRunningMethod()
 {
 lock (_locker)
 {
 Thread.Sleep(5000);
 }
 }
}

Am I on the right track here? what can I do in order to make it more robust?
UPDATE
The process of this class is as follows:

• Listen to events raised by a remote service
• Handle each event based on it's event arguments in a predefined method
• A method execution time can be long
• Different methods can run simultaneously (That's why I start each method with the Task.Factory.StartNew(LongRunningMethod);)
• If an event handling method needs to run again when the previous one wasn't finished it should wait for it (that's why I placed the lock statement in the method implementation)

please note that I originally posted this question here but I guess that it's more suitable for this site

• Martin: Could you please clarify what the requirements on the order of execution of events are? Your code currently executes short running tasks synchronously by the order they were added to the BlockingCollection. Long running tasks are being executed concurrently with short running tasks, but one at a time and not necessarily in the order they were added to the BlockingCollection. Is this a requirement, or is this an implementation detail?
• Yoav: @Martin please see my update
• Martin: So, when _ioService.IoEvent raises the event twice with two different int portNum the event handlers can be called simultaneously but when it raises the event twice with the same int portNum the event handlers should be executed one by one?
• Yoav: @Martin correct
• Martin: Is there a different method for each int portNum that needs to be called (one method per a value) or is there only one method that needs to be called with a parameter?

Answer 1

1. The method StartListening takes one argument object dummy. It is not used in the method and from its name I conclude that it is not intended to be used. It is possible to declare the method with no arguments and call a variant of Task.Factory.StartNew that takes a parameter of type Action and TaskCreationOptions to start the new task.

2. The field private static EventWaitHandle _eventWaiter is only initialized in the constructor and never used. Its purpose is unclear from the code.

Answer 2

IoService event

I would suggest having an event handler for that iOservice event, and in that event handler you would have a dicitionary collection that will hold for example : "Port80" : WebHandler "Port21" : FtpHandler

this way you would initialize this collection or load it from a configuration file with all the required port numbers to handle. Of course you can handle them inside those classes with a task that receives your cancelation token from the main class calling them.