Poll for new files, avoid duplicates while processing files

Question

I want to poll for new files in a remote directory (pdf of few MB) from a windows service.

Each files has to be processed by an intensive CPU job (image recognition within the pdf files). After the process is done, the file has to be moved elsewhere or deleted.

I'd like to trigger my work as quickly as possible while benefiting multi processor capabilities to parallalise works.

However, I'm facing an issue: while it's easy to enumerate files in a directory, how to avoid duplicate entries in my job queue? Actually, each time I enumerate my files, it is possible that some files are yet queued or are yet being processed.

My first approach was to look into System.Collections.Concurrent.* but no class seems to provides a contains method to test before adding.

I also looked at HashSet, but I fear some issues with concurrent access.

My current skeleton is:

    private async void GetNewFiles(CancellationToken cancellationToken)
    {
        if (!cancellationToken.IsCancellationRequested)
        {
            var newfiles = Directory.GetFileSystemEntries(inputDirectory, "*.pdf", SearchOption.AllDirectories);
            logger.Trace($"{newfiles.Length} new files detected in {inputDirectory}");

            foreach (var file in newfiles)
            {
                Task.Factory.StartNew(()=>ProcessFile(file), cancellationToken);
            }

            await Task.Delay(frequency, cancellationToken);
            if (!cancellationToken.IsCancellationRequested)
            {
                GetNewFiles(cancellationToken);
           }
        }
    }

However, this code does not avoid having a file queued twice.

If I removed the Task.Delay call and wait for all files to be processed, It would work, but it can leads to have only one running tasks, even if new files are added (each iteration of processing new files must be completely processed before checking for new files).

user_5311735 · Answer

The easiest way requiring least amount of modifications to your current code is using ConcurrentDictionary I think:

private readonly ConcurrentDictionary _filesInProgress = new ConcurrentDictionary();
private async Task GetNewFiles(CancellationToken cancellationToken) {            
    if (!cancellationToken.IsCancellationRequested) {
        var newfiles = Directory.GetFileSystemEntries(inputDirectory, "*.pdf", SearchOption.AllDirectories);
        foreach (var file in newfiles) {
            // TryAdd returns true if key was not already in dictionary
            if (_filesInProgress.TryAdd(file, 0) && File.Exists(file)) {
                Task.Factory.StartNew(() => {
                    ProcessFile(file);                         
                    _filesInProgress.TryRemove(file, out _);
                }, cancellationToken);
            }
        }
        await Task.Delay(frequency, cancellationToken);
        if (!cancellationToken.IsCancellationRequested) {
            GetNewFiles(cancellationToken);
        }
    }
}

Beware that ideally you'd want to process items with limited number of threads (equal to number of your cores\virtual cores). As of now if you find 100 files in directory you might potentially spawn 100 threads which are all CPU-heavy and so will contend with each other for resources for no good reason.

Poll for new files, avoid duplicates while processing files

1 Answers