Process files concurrently and asynchronously

by

For this kind of job a powerful tool you could use is the TPL Dataflow library. With this tool you can create a processing pipeline consisting of many linked blocks, with the data flowing from the first block to the last (circles and meshes are also possible).

The advantages of this approach are:

  1. You get data-parallelism on top of task-parallelism. All blocks are working concurrently and independently from each other.
  2. You can configure optimally the level of concurrency (a.k.a. degree of parallelism) of each heterogeneous operation. For example doing API calls may be highly parallelizable, while reading from the hard disk may be not parallelizable at all.
  3. You get advanced options out of the box (BoundedCapacity, CancellationToken and others).
  4. You get built-in support for both synchronous and asynchronous delegates.

Below is how you could rewrite your original code in TPL Dataflow terms. Three blocks are used, two TransformManyBlocks and one ActionBlock.

var directoryBlock = new TransformManyBlock<DirectoryConfig, string>(config =>
{
    return Directory.GetFiles(config.DirectoryPath, config.Token);
});

var fileBlock = new TransformManyBlock<string, string>(filePath =>
{
    return File.ReadLines(filePath);
});

var lineBlock = new ActionBlock<string>(async line =>
{
    await Task.Delay(1000);
}, new ExecutionDataflowBlockOptions()
{
    MaxDegreeOfParallelism = 4
});

directoryBlock.LinkTo(fileBlock, new DataflowLinkOptions { PropagateCompletion = true });
fileBlock.LinkTo(lineBlock, new DataflowLinkOptions { PropagateCompletion = true });

foreach (DirectoryConfig config in GetDirectoryConfigs())
    await directoryBlock.SendAsync(config);

directoryBlock.Complete();
await lineBlock.Completion;

This example is not very good since all the work is done by the last block (the lineBlock), and the first two blocks are doing essentially nothing. It is also not memory-efficient since all lines of all files of all directories will soon become queued in the input buffer of the ActionBlock, unless processing the lines happens to be faster than reading them from the disk. You’ll need to configure the blocks with BoundedCapacity to solve this problem.

This example also fails to demonstrate how you could have different blocks for different types of files, and link the directoryBlock to all of them using a different filtering predicate for each link:

directoryBlock.LinkTo(csvBlock, filePath => Path.GetExtension(filePath) == "csv");
directoryBlock.LinkTo(xlsBlock, filePath => Path.GetExtension(filePath) == "xls");
directoryBlock.LinkTo(generalFileBlock); // Anything that is neither csv nor xls

There are also other types of blocks you could use, like the TransformBlock and the BatchBlock. The TPL Dataflow is based on the Task Parallel Library (TPL), and it is essentially a high level task-generator that creates and controls the lifecycle of the tasks needed in order to process a workload of given type, based on declarative configuration. It is built-in the .NET Core, and available as a package for .NET Framework.

More Related Contents:

  1. How to limit the amount of concurrent async I/O operations?
  2. What is the difference between asynchronous programming and multithreading?
  3. Async/Await – is it *concurrent*?
  4. How and when to use ‘async’ and ‘await’
  5. How would I run an async Task method synchronously?
  6. Should I avoid ‘async void’ event handlers?
  7. await Task.Run vs await
  8. How do yield and await implement flow of control in .NET?
  9. How can I call an async method in Main?
  10. Best way in .NET to manage queue of tasks on a separate (single) thread
  11. Enforce an async method to be called once
  12. Why use Async/await all the way down
  13. async/await different thread ID
  14. awaitable Task based queue
  15. Different forms of the WCF service contract interface
  16. Named Mutex with await
  17. Is async HttpClient from .Net 4.5 a bad choice for intensive load applications?
  18. How to make BackgroundWorker return an object
  19. How could the new async feature in c# 5.0 be implemented with call/cc?
  20. What is the correct way to cancel an async operation that doesn’t accept a CancellationToken?
  21. ForEachAsync with Result
  22. Async ShowDialog
  23. Async thread body loop, It just works, but how?
  24. How to throttle multiple asynchronous tasks?
  25. Await vs Task.Result in an Async Method [duplicate]
  26. call async method without await #2
  27. Async and Await with For Loop [duplicate]
  28. When is the best place to use Task.Result instead of awaiting Task
  29. ConcurrentDictionary GetOrAdd async
  30. Calling TaskCompletionSource.SetResult in a non blocking manner

Leave a Comment