How to get with Tasks the same result achieved with Threads?

Question

Hello everybody.

I have a code in my app that opens a queue in MQSeries messages to me, reads a message from the queue and makes a series of internal processing.

Assuming I have only one message in the queue and be fired 5 Tasks for reading from the queue, the first Task reads the message in the queue (the message is automatically hidden in the queue by MQSeries). The second task arrives and finds no messages in the queue. What is expected is that it performs the cancellation of it and all the other three Tasks.

Well, using Tasks that was easy to do (using CancellationTokenSource) BUT, by the application logs, I realized the same task (not different Tasks), read the same message more than once. This should not happen and I honestly do not understand why this happened.

I changed all the code to use threads. The result was perfect, but I cant cancel the other threads when there are no more messages in the queue to be read. Besides, would not want to explicitly use threads.

How to get with Tasks, ensure that one and only one task at a time will execute my method using a single thread? It's possible?

thank you

Answer 1

There really should be no functional difference when switching from Tasks to Threads for processing.  Can you show some of your basic code flow?  It might help us determine what the actual problem here would be...

The main difference would be that the Task (by default) will get scheduled using a ThreadPool thread to execute, where a thread will use a standard Thread.  This shouldnt' change the behavior at all.  In your case, it's likely that something was written incorrectly in the port, but without more details, its impossible to know what the problem would actually be...

That being said, if your goal is to process a queue, you might want to consider using something like BlockingCollection<T> and a producer/consumer scenario (not sure - depends a lot on what your work is doing).  This can simplify your code, and allow you to keep any number of "consumer" tasks around to process, without having to worry about cancelation (when you call CompleteAdding on the BC, the tasks will just complete and exit on their own).

---

Reed Copsey, Jr. - http://reedcopsey.com
If a post answers your question, please click "Mark As Answer" on that post and "Mark as Helpful".

Answer 2

Hi Reed, tks for your answer. Here is a fragment of my code:

private void ReadAndProccessMessages()
        {
            int numMaxParallelProcess = 5; // (from Web.Config);
            List<Task> listTasks = new List<Task>();

            for (int i = 1; i <= numMaxParalellProcess; i++)
            {
                var myTask = Task.Factory.StartNew(() => new ReadMessageQueue());
                listTasks.Add(myTask);
            }

            Task.WaitAll(listTasks.ToArray());
        }

The "ReadMessageQueue()" method, open and get the message from MQSeries, process the message and log the result in SQLServer. I remove the extras lines from the code to let it more clean.

Answer 3

My sixthsense says that there might be some kind of synchronization problem lurking in your code. Is the queue thread-safe?

Answer 4

Like Reed mentioned should not be any difference
in this sample between threads and tasks. Can you try to use TaskCreationOptions.LongRunning
and see  how the scenario behaves? Using
this option the scenario will switch from ThreadPool threads to directly using
threads. The syntax is Task.Factory.StartNew(…, TaskCreationOptions.LongRunning)

However the behavior that you explain
is interesting, could you please add more detail why you think that the message
is read twice form the same task? Do you log the task ID?

Thanks,

Cristina