How to mark a TPL dataflow cycle to complete?

by

If the purpose of your code is to traverse the directory structure using some sort of parallelism then I would suggest not using TPL Dataflow and use Microsoft’s Reactive Framework instead. I think it becomes much simpler.

Here’s how I would do it.

First define a recursive function to build the list of directories:

Func<DirectoryInfo, IObservable<DirectoryInfo>> recurse = null;
recurse = di =>
    Observable
        .Return(di)
        .Concat(di.GetDirectories()
            .ToObservable()
            .SelectMany(di2 => recurse(di2)))
        .ObserveOn(Scheduler.Default);

This performs the recurse of the directories and uses the default Rx scheduler which causes the observable to run in parallel.

So by calling recurse with an input DirectoryInfo I get an observable list of the input directory and all of its descendants.

Now I can build a fairly straight-forward query to get the results I want:

var query =
    from di in recurse(new DirectoryInfo(@"C:\dev\kortforsyningen_dsm\tiles"))
    from fi in di.GetFiles().ToObservable()
    let zxy =
        fi
            .FullName
            .Split('\\')
            .Reverse()
            .Take(3)
            .Reverse()
            .Select(s => int.Parse(Path.GetFileNameWithoutExtension(s)))
            .ToArray()
    let suffix = String.Format("{0}/{1}/{2}.png", zxy[0], zxy[1], zxy[2])
    select new FileInfo(Path.Combine(di.FullName, suffix));

Now I can action the query like this:

query
    .Subscribe(s =>
    {
        Trace.TraceInformation("Done combining : {0}", s.Name);
    });

Now I may have missed a little bit in your custom code but if this is an approach you want to take I’m sure you can fix any logical issues quite easily.

This code automatically handles completion when it runs out of child directories and files.

To add Rx to your project look for “Rx-Main” in NuGet.

More Related Contents:

  1. Asynchronously wait for Task to complete with timeout
  2. How to cancel Task await after a timeout period
  3. Running multiple async tasks and waiting for them all to complete
  4. StaTaskScheduler and STA thread message pumping
  5. Why should I prefer single ‘await Task.WhenAll’ over multiple awaits?
  6. Regarding usage of Task.Start() , Task.Run() and Task.Factory.StartNew()
  7. Is it considered acceptable to not call Dispose() on a TPL Task object?
  8. No ConcurrentList in .Net 4.0?
  9. Task sequencing and re-entracy
  10. Synchronously waiting for an async operation, and why does Wait() freeze the program here
  11. How to use non-thread-safe async/await APIs and patterns with ASP.NET Web API?
  12. Multiple Parallel.ForEach loops in .Net
  13. How to put a task to sleep (or delay) in C# 4.0?
  14. Simplest way to do a fire and forget method in c# 4.0
  15. Create a completed Task
  16. Waiting for async/await inside a task
  17. Revisiting Task.ConfigureAwait(continueOnCapturedContext: false)
  18. multiple awaits vs Task.WaitAll – equivalent?
  19. Can .NET Task instances go out of scope during run?
  20. Executing tasks in parallel
  21. Difference between returning and awaiting a Task in an async method [duplicate]
  22. Use Task.Run() in synchronous method to avoid deadlock waiting on async method?
  23. A pattern for self-cancelling and restarting task
  24. How to correctly write Parallel.For with async methods
  25. For a TPL Dataflow: How do I get my hands on all the output produced by a TransformBlock while blocking until all inputs have been processed?
  26. BroadcastBlock with guaranteed delivery in TPL Dataflow
  27. TPL Dataflow exception in transform block with bounded capacity
  28. Converting async/await to Task.ContinueWith
  29. Task chaining (wait for the previous task to completed)
  30. Wrapping ManualResetEvent as awaitable task

Leave a Comment