1

I have a large number of pieces of data to process. The current code can be simplified as following:

public void ProcessData(string data) { string resultOfA = doCpuBoundWorkA(data); string resultOfS1 = sendToServiceS1(resultOfA); string resultOfB = doCpuBoundWorkB(resultOfS1); string resultOfS2 = sendToServiceS2(resultOfB); string resultOfC = doCpuBoundWorkC(resultOfS2); }

The ProcessData is invoked using Parallel.ForEach. This implementation is not optimal from at least two perspectives. First of all the calls to the services are blocking so we are blocking threads while waiting for the call to return. Secondly Parallel.ForEach creates Tasks that are scheduled for execution on the thread pool. The Thread Pool creates additional threads every 500ms (if I'm not wrong on that) and because 'ProcessData' takes longer than 500ms to complete, over time, we end up with hundreds of threads that spend most of the time waiting for the services to comeback.

My naive idea for 'improvement' was this:

public async Task ProcessData(string data) { string resultOfA = doCpuBoundWorkA(data); string resultOfS1 = await sendToServiceS1Async(resultOfA); string resultOfB = doCpuBoundWorkB(resultOfS1); string resultOfS2 = await sendToServiceS2Async(resultOfB); string resultOfC = doCpuBoundWorkC(resultOfS2); }

I'm new to async/await so I could be totally wrong in my understanding of what it is actually happening.

With the async/await keywords the compiler breaks the code of the ProcessData it into multiple Tasks.

  • Task-A: From the beginning of the ProcessData method up-to the point where call to The ServiceA "hits the wire".
  • Task-B: From the moment when we pick up the results of call to the ServiceA up-to the point where call to The ServiceB "hits the wire".
  • Task-C: From the moment when we pick up the results of call to the ServiceB up-to the end of the ProcessData method.

As a result instead of having a single "processing unit of work" we have three, where each piece is scheduled for execution based on its positions in the scheduler's queue.

The problem is that by the time Task-B (for the first piece of work) is put on the scheduler's queue I may have hundreds of Task-A, put there by Parallel.ForEach, and by the time Task-C (for the first piece of work) is placed on the scheduler's queue the situation will be even worse.

I would like the data pass through as fast as possible, so I need to be able to prioritize Task-C over Task-B over Task-A. What will be the best way to achieve this?

INotifyCompletion, SynchronizationContext come to mind, but it seems to be the "dark corners" of the async/await. ParallelExtensionsExtras has ReprioritizableTaskScheduler and QueuedTaskScheduler with priority queues, but how do I tell async/await to use the desired scheduler?

John Skeet talks about this problem in his blog: https://codeblog.jonskeet.uk/2010/11/02/configuring-waiting/

Improve this question

2 Answers 2

Reset to default
4

Throttling is probably a much easier approach than prioritization.

I think your problem is best addressed by the TPL Dataflow library. It combines parallel and async technologies.

You can create "blocks" and "link" them together to form a "mesh" (in your case, the mesh is a pipeline). TransformBlock can be used with both synchronous and asynchronous actions, and also supports parallelism and throttling built-in.

Alternatively, you could apply asynchronous throttling to your ProcessData method using SemaphoreSlim (calling WaitAsync at the beginning of the method and Release at the end). But do consider TPL Dataflow; I find that if people are doing something this complex, then they usually find they can use TPL Dataflow in other parts of their application, too.

Improve this answer
Sign up to request clarification or add additional context in comments.

6 Comments

I was thinking about taking a look at TPL Dataflow, but wouldn't it suffer from the same problem which is a lack of control over when the Tasks created by the use of async/await are executed? I'm not sure if throtteling will solve the problem. It is fine to to create and process many Task-A as long as I don't blow the memory. I just want jobs that are "in flight" to take priority.
As someone who's familiar with custom task schedulers and synchronization contexts, I strongly recommend you consider throttling. It's a whole ton of a lot easier.
Stephen, After spending some time trying to figure out how I could use throttling I came back empty handed. There is no fixed number that I can provide to the semaphore, as the number of Task-A which can be run, depends on the time it takes for ServiceA or ServiceB to come back. What I will have to do is to check if there is any Task-B or Task-C pending before executing the code of Task-A. Unfortunately this is no way for me to know that (it's all compiler magic). Isn't using semaphore in ProcessData is equivalent to putting MaxDegreeOfParallelism in the Paralle.ForEach?
@Nafas: Yes, both SemaphoreSlim and MaxDegreeOfParallelism in this scenario are doing the same thing: applying a simple throttling value. If that is insufficient, you may want to consider Reactive Extensions (Rx), which has more powerful throttling mechanisms. Rx is particularly good when you have to deal with time.
Thank you for your help. It looks to me that using async/await in this scenario doesn't make any sense. On one side I will be feeing threads by using the async/await on the other hand I won’t be using them as I’m throttling on some fixed number. I will take a look If Rx can be of any help here. I’m skeptical if it will be of any help. It is not about throttling by a fixed number or time span. The core issue remains the same: lack of control over when the continuation of await is scheduled for execution.
|
0

Your problem:

Thread Pool creates additional threads every 500ms (if I'm not wrong on that) and because 'ProcessData' takes longer than 500ms to complete, over time, we end up with hundreds of threads that spend most of the time waiting for the services to comeback.

can be "fixed" by await on ProcessData and only spawn new upon finishing. (or doing something like Task.WhenAll(...Task.Delay(500)..., ...ProcessData()).

All the calls inside ProcessData are data dependent,

string resultOfA = doCpuBoundWorkA(data); string resultOfS1 = await sendToServiceS1Async(resultOfA); string resultOfB = doCpuBoundWorkB(resultOfS1); string resultOfS2 = await sendToServiceS2Async(resultOfB); string resultOfC = doCpuBoundWorkC(resultOfS2);

IIRC, await passes execution only to outside the method "ProcessData". So it can only allow other async methods to run, but calls inside ProcessData are still back-to-back because of data dependency.

Improve this answer

Comments

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.