Throttling asynchronous tasks

by

As suggested, use TPL Dataflow.

A TransformBlock<TInput, TOutput> may be what you’re looking for.

You define a MaxDegreeOfParallelism to limit how many strings can be transformed (i.e., how many urls can be downloaded) in parallel. You then post urls to the block, and when you’re done you tell the block you’re done adding items and you fetch the responses.

var downloader = new TransformBlock<string, HttpResponse>(
        url => Download(url),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 50 }
    );

var buffer = new BufferBlock<HttpResponse>();
downloader.LinkTo(buffer);

foreach(var url in urls)
    downloader.Post(url);
    //or await downloader.SendAsync(url);

downloader.Complete();
await downloader.Completion;

IList<HttpResponse> responses;
if (buffer.TryReceiveAll(out responses))
{
    //process responses
}

Note: The TransformBlock buffers both its input and output. Why, then, do we need to link it to a BufferBlock?

Because the TransformBlock won’t complete until all items (HttpResponse) have been consumed, and await downloader.Completion would hang. Instead, we let the downloader forward all its output to a dedicated buffer block – then we wait for the downloader to complete, and inspect the buffer block.

More Related Contents:

  1. Does SemaphoreSlim (.NET) prevent same thread from entering block?
  2. Is it possible to await an event instead of another async method?
  3. Should I avoid ‘async void’ event handlers?
  4. Is Task.Result the same as .GetAwaiter.GetResult()?
  5. TAP global exception handler
  6. At the end of an async method, should I return or await?
  7. Fire and Forget approach
  8. How can I use Async with ForEach?
  9. Does the use of the “Async” suffix in a method name depend on whether the ‘async’ modifier is used?
  10. Async lock not allowed
  11. Using async without await
  12. async Task vs async void
  13. A reusable pattern to convert event into task
  14. Why would I bother to use Task.ConfigureAwait(continueOnCapturedContext: false);
  15. ConfigureAwait pushes the continuation to a pool thread
  16. How can I prevent synchronous continuations on a Task?
  17. C# GUI refresh and async serial port communication
  18. Correct way to get the CoreDispatcher in a Windows Store app
  19. Async action filter in MVC 4
  20. Async/Await – is it *concurrent*?
  21. Program exits upon calling await
  22. Converting a WebClient method to async / await
  23. How to throttle multiple asynchronous tasks?
  24. How do you implement an async action delegate method?
  25. Non-Generic TaskCompletionSource or alternative
  26. Do I need to Dispose a SemaphoreSlim?
  27. In C#7, how can I “roll my own” Task-like type to use with async?
  28. Intercept async method that returns generic Task via DynamicProxy
  29. How to implement an efficient WhenEach that streams an IAsyncEnumerable of task results?
  30. Wrapping ManualResetEvent as awaitable task

Leave a Comment