Alex Rivera | Logout

C# queueing dependant tasks to be processed by a thread pool

Asked 2012-06-27T15:26:45.127
8

I want to queue dependant tasks across several flows that need to be processed in order (in each flow). The flows can be processed in parallel.

To be specific, let's say I need two queues and I want the tasks in each queue to be processed in order. Here is sample pseudocode to illustrate the desired behavior:

Queue1_WorkItem wi1a=...;

enqueue wi1a;

... time passes ...

Queue1_WorkItem wi1b=...;

enqueue wi1b; // This must be processed after processing of item wi1a is complete

... time passes ...

Queue2_WorkItem wi2a=...;

enqueue wi2a; // This can be processed concurrently with the wi1a/wi1b

... time passes ...

Queue1_WorkItem wi1c=...;

enqueue wi1c; // This must be processed after processing of item wi1b is complete

Here is a diagram with arrows illustrating dependencies between work items:

enter image description here

The question is how do I do this using C# 4.0/.NET 4.0? Right now I have two worker threads, one per queue and I use a BlockingCollection<> for each queue. I would like to instead leverage the .NET thread pool and have worker threads process items concurrently (across flows), but serially within a flow. In other words I would like to be able to indicate that for example wi1b depends on completion of wi1a, without having to track completion and remember wi1a, when wi1b arrives. In other words, I just want to say, "I want to submit a work item for queue1, which is to be processed serially with other items I have already submitted for queue1, but possibly in parallel with work items submitted to other queues".

I hope this description made sense. If not please feel free to ask questions in the comments and I will update this question accordingly.

Thanks for reading.

Update:

To summarize "flawed" solutions so far, here are the solutions from the answers section that I ca

c# .net c#-4.0 .net-4.0
Edit
Report

1 Answer

3

The best way is to use the Task Parallel Library (TPL) and Continuations. A continuation not only allows you to create a flow of tasks but also handles your exceptions. This is a great introduction to the TPL. But to give you some idea...

You can start a TPL task using 

Task task = Task.Factory.StartNew(() => 
{
    // Do some work here...
});

Now to start a second task when an antecedent task finishes (in error or successfully) you can use the ContinueWith method

Task task1 = Task.Factory.StartNew(() => Console.WriteLine("Antecedant Task"));
Task task2 = task1.ContinueWith(antTask => Console.WriteLine("Continuation..."));

So as soon as task1 completes, fails or is cancelled task2 'fires-up' and starts running. Note that if task1 had completed before reaching the second line of code task2 would be scheduled to execute immediately. The antTask argument passed to the second lambda is a reference to the antecedent task. See this link for more detailed examples...

You can also pass continuations results from the antecedent task

Task.Factory.StartNew<int>(() => 1)
    .ContinueWith(antTask => antTask.Result * 4)
    .ContinueWith(antTask => antTask.Result * 4)
    .ContinueWith(antTask =>Console.WriteLine(antTask.Result * 4)); // Prints 64.

Note. Be sure to read up on exception handling in the first link provided as this can lead a newcomer to TPL astray.

One last thing to look at in particular for what you want is child tasks. Child tasks are those which are created as AttachedToParent. In this case the continuation will not run until all c

answered 2012-06-27T15:32:47.187

Your Answer