Control when an ActionBlock starts processing

Question

I am using an ActionBlock to process files detected by FileSystemWatchers. I  first start the FileSystemWatchers to ensure we do not miss any new files. Then I process all files that already existed before the FSW started.

To enforce the files being processed according to their creation date I am looking for a way control when the ActionBlock starts processing. Currently I am using a ManualResetEvent for this. Is there a nicer way to do this?

Private _existingFilesProcessedEvent As New Threading.ManualResetEvent(False)
Private _newFilesActionBlock As New ActionBlock(Of String)(
    Async Function(filePath)
        _existingFilesProcessedEvent.WaitOne() 'Ensure we process new files after existing ones
        Await ProcessFile(filePath)
    End Function)
...
    
ProcessExistingFiles(existingFiles).Wait
_existingFilesProcessedEvent.Set()

Answer 1

Hi pmeinl-

I don't quite understand the scenario, but...

If your goal is to be able to queue items ahead of time but manually control when the ActionBlock starts processing, one solution is to use a BufferBlock and an ActionBlock.  You'd Post/Send to the BufferBlock, and then LinkTo from the BufferBlock to the ActionBlock when you want the processing to start.

The approach you outline will technically work, but it will result in a thread getting blocked.  If that's acceptable, ok.  If not, you could use a TaskCompletionSource instead of the ManualResetEvent, calling SetResult on the TCS to signal it and await'ing its Task in the ActionBlock's Async delegate.

Answer 2

Hi Stephen!

1.
Linking from a BufferBlock sounds like a good solution. I will try it.

2.
The scenario is to process files in the order they were created.

The FileSystemWatcher does not detect files that existed before it started watching, but these files must be processed first. I we start the FSW after processing existing files, we miss those that were created while we were processing. So we start the FSW first, and have to delay processing newly detected files until the existing ones were processed.

We could use a poll timer instead of FSWs but I like the reactivity of the FSW and that my code does nothing when idle.

Do you have better idea?

BTW: It is a pleasure to read your TPL articles.

Answer 3

I see.  Seems reasonable.

And I'm glad you've enjoyed the articles! :)  Thanks for letting me know.