c# - 涉及递归未完成的 TPL 数据流

Question

在测试 WPF 项目中，我尝试使用 TPL 数据流来枚举给定父目录的所有子目录，并创建具有特定文件扩展名的文件列表，例如“.xlsx”。我使用 2 个 block ，第一个 dirToFilesBlock，最后一个 fileActionBlock。

为了创建遍历所有子目录的递归效果，第一个 block 有一个返回自身的链接，链接谓词测试以查看输出项是否为目录。这是我在一本关于异步编程的书中找到的方法。第二个链接指向 fileActionBlock，然后根据链接谓词测试将文件添加到列表，以查看文件是否具有正确的扩展名。

我遇到的问题是在使用 btnStart_Click 启动之后，它永远不会完成。也就是说，我们永远不会到达事件处理程序中的等待下方以显示“已完成”消息。我知道我可能需要调用 dirToFilesBlock.Complete()，但我不知道它应该在代码中的什么位置以及在什么条件下调用？我不能在最初的帖子之后调用它，因为它会停止返回子目录的链接。我试过使用 InputCount 和 OutputCount 属性做一些事情，但没有取得太大进展。如果可能的话，我希望保持数据流的结构不变，因为这意味着我还可以使用每个新目录更新 UI，通过返回链接向用户提供一些进度反馈。

我是 TPL 数据流的新手，非常感谢收到任何帮助。

这是代码隐藏文件中的代码:

public partial class MainWindow : Window
{
    TransformManyBlock<string, string> dirToFilesBlock;
    ActionBlock<string> fileActionBlock;
    ObservableCollection<string> files;
    CancellationTokenSource cts;
    CancellationToken ct;
    public MainWindow()
    {
        InitializeComponent();

        files = new ObservableCollection<string>();

        lst.DataContext = files;

        cts = new CancellationTokenSource();
        ct = cts.Token;
    }

    private Task Start(string path)
    {
        var uiScheduler = TaskScheduler.FromCurrentSynchronizationContext();

        dirToFilesBlock = new TransformManyBlock<string, string>((Func<string, IEnumerable<string>>)(GetFileSystemItems), new ExecutionDataflowBlockOptions() { CancellationToken = ct });
        fileActionBlock = new ActionBlock<string>((Action<string>)ProcessFile, new ExecutionDataflowBlockOptions() {CancellationToken = ct, TaskScheduler = uiScheduler});

        // Order of LinkTo's important here!
        dirToFilesBlock.LinkTo(dirToFilesBlock, new DataflowLinkOptions() { PropagateCompletion = true }, IsDirectory);
        dirToFilesBlock.LinkTo(fileActionBlock, new DataflowLinkOptions() { PropagateCompletion = true }, IsRequiredDocType);

        // Kick off the recursion.
        dirToFilesBlock.Post(path);

        return Task.WhenAll(dirToFilesBlock.Completion, fileActionBlock.Completion);
    }

    private bool IsDirectory(string path)
    {

        return Directory.Exists(path);
    }


    private bool IsRequiredDocType(string fileName)
    {
        return System.IO.Path.GetExtension(fileName) == ".xlsx";
    }

    private IEnumerable<string> GetFilesInDirectory(string path)
    {
        // Check for cancellation with each new dir.
        ct.ThrowIfCancellationRequested();

        // Check in case of Dir access problems
        try
        {
            return Directory.EnumerateFileSystemEntries(path);
        }
        catch (Exception)
        {
            return Enumerable.Empty<string>();
        }
    }

    private IEnumerable<string> GetFileSystemItems(string dir)
    {
        return GetFilesInDirectory(dir);
    }

    private void ProcessFile(string fileName)
    {
        ct.ThrowIfCancellationRequested();

       files.Add(fileName);
    }

    private async void btnStart_Click(object sender, RoutedEventArgs e)
    {
        try
        {
            await Start(@"C:\");
            // Never gets here!!!
            MessageBox.Show("Completed");

        }
        catch (OperationCanceledException)
        {
            MessageBox.Show("Cancelled");

        }
        catch (Exception)
        {
            MessageBox.Show("Unknown err");
        }
        finally
        {
        }
    }

    private void btnCancel_Click(object sender, RoutedEventArgs e)
    {
        cts.Cancel();
    }
}

Answer 1

即使这是一个老问题，在数据流循环中处理完成仍然是一个问题。

在您的情况下，您可以让 TransfomBlock 对仍在飞行中的项目进行计数。这表明该 block 正忙于处理任意数量的项目。然后，您只会在 block 不忙且两个缓冲区都为空时才调用 Complete()。您可以在我写的帖子中找到有关处理完成的更多信息 Finding Completion in a Complex Flow: Feedback Loops

public partial class MainWindow : Window {

        TransformManyBlock<string, string> dirToFilesBlock;
        ActionBlock<string> fileActionBlock;
        ObservableCollection<string> files;
        CancellationTokenSource cts;
        CancellationToken ct;
        public MainWindow() {
            InitializeComponent();

            files = new ObservableCollection<string>();

            lst.DataContext = files;

            cts = new CancellationTokenSource();
            ct = cts.Token;
        }

        private async Task Start(string path) {
            var uiScheduler = TaskScheduler.FromCurrentSynchronizationContext();

            dirToFilesBlock = new TransformManyBlock<string, string>((Func<string, IEnumerable<string>>)(GetFileSystemItems), new ExecutionDataflowBlockOptions() { CancellationToken = ct });
            fileActionBlock = new ActionBlock<string>((Action<string>)ProcessFile, new ExecutionDataflowBlockOptions() { CancellationToken = ct, TaskScheduler = uiScheduler });

            // Order of LinkTo's important here!
            dirToFilesBlock.LinkTo(dirToFilesBlock, new DataflowLinkOptions() { PropagateCompletion = true }, IsDirectory);
            dirToFilesBlock.LinkTo(fileActionBlock, new DataflowLinkOptions() { PropagateCompletion = true }, IsRequiredDocType);

            // Kick off the recursion.
            dirToFilesBlock.Post(path);

            await ProcessingIsComplete();
            dirToFilesBlock.Complete();
            await Task.WhenAll(dirToFilesBlock.Completion, fileActionBlock.Completion);
        }

        private async Task ProcessingIsComplete() {
            while (!ct.IsCancellationRequested && DirectoryToFilesBlockIsIdle()) {
                await Task.Delay(500);
            }
        }

        private bool DirectoryToFilesBlockIsIdle() {
            return dirToFilesBlock.InputCount == 0 &&
                dirToFilesBlock.OutputCount == 0 &&
                directoriesBeingProcessed <= 0;
        }

        private bool IsDirectory(string path) {
            return Directory.Exists(path);
        }


        private bool IsRequiredDocType(string fileName) {
            return System.IO.Path.GetExtension(fileName) == ".xlsx";
        }

        private int directoriesBeingProcessed = 0;

        private IEnumerable<string> GetFilesInDirectory(string path) {
            Interlocked.Increment(ref directoriesBeingProcessed)
            // Check for cancellation with each new dir.
            ct.ThrowIfCancellationRequested();

            // Check in case of Dir access problems
            try {
                return Directory.EnumerateFileSystemEntries(path);
            } catch (Exception) {
                return Enumerable.Empty<string>();
            } finally {
                Interlocked.Decrement(ref directoriesBeingProcessed);
            }
        }

        private IEnumerable<string> GetFileSystemItems(string dir) {
            return GetFilesInDirectory(dir);
        }

        private void ProcessFile(string fileName) {
            ct.ThrowIfCancellationRequested();

            files.Add(fileName);
        }

        private async void btnStart_Click(object sender, RoutedEventArgs e) {
            try {
                await Start(@"C:\");
                // Never gets here!!!
                MessageBox.Show("Completed");

            } catch (OperationCanceledException) {
                MessageBox.Show("Cancelled");

            } catch (Exception) {
                MessageBox.Show("Unknown err");
            } finally {
            }
        }

        private void btnCancel_Click(object sender, RoutedEventArgs e) {
            cts.Cancel();
        }
    }