任务运行失败

Question

我编写了一个小实用程序来读取大型文本文件并搜索包含搜索词的行。我借此机会学习 TPL 数据流。

代码运行良好，除非搜索词接近文件末尾。在这种情况下，uiResult 操作块不会被调用除非其中有断点。

我的理解是数据从searcher 发布到uiResult，之后 searcher 变得完整（它已经处理了最后一个数据块）。由于数据已发布到uiResult，因此在处理完该数据之前它不应该变得完整。

问题

为什么uiResult 已经完成，即使数据已经发送到它（除非在uiResult 中设置了断点）？

代码

以下是相关代码，尽可能精简：

ActionBlock<LineInfo> uiResult = new ActionBlock<LineInfo>(li =>
    {
        // If match found near end of file, the following line only runs
        // if a breakpoint is set on it:
        if (results != null) results.Add(li);
    },
    new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = 1,
        CancellationToken = cancelSource.Token,
        TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
    });

BatchBlock<LineInfo> batcher = new BatchBlock<LineInfo>(5000); 

ActionBlock<LineInfo[]> searcher = new ActionBlock<LineInfo[]>(lines =>
    {
        foreach (LineInfo li in lines)
        {
            if (li.TextOfLine.Contains(searchTerm))
            {
                uiResult.Post(li);
            }
        }
    },
    new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = 1,
        CancellationToken = cancelSource.Token
    });

batcher.LinkTo(searcher);

batcher.Completion.ContinueWith(t =>
{
    if (t.IsFaulted) ((IDataflowBlock)searcher).Fault(t.Exception);
    else searcher.Complete();

    if (t.IsFaulted) ((IDataflowBlock)uiResult).Fault(t.Exception);
    else uiResult.Complete();
});

Task.Run(() =>
    {
        using (FileStream fs = File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
        using (BufferedStream bs = new BufferedStream(fs))
        using (StreamReader sr = new StreamReader(bs))
        {
            string line;
            while ((line = sr.ReadLine()) != null && cancelSource.IsCancellationRequested == false)
            {
                batcher.Post(new LineInfo() { LineNumber = lineNumber, OffsetOfLine = offset, TextOfLine = line });
            }

            batcher.Complete();
            try
            {
                searcher.Completion.Wait();
                uiResult.Completion.Wait();
            }
            catch (AggregateException ae)
            {
                TaskCanceledException taskCancelled = ae.InnerException as TaskCanceledException;
                if (taskCancelled != null)
                {
                    // Swallow the Exception if is just a user cancellation
                    throw;
                }
            }
            finally
            {
                signalDone();
            }
        }
    });

Answer 1

由于您处理完成的方式，您的代码是不确定的。可能的事件顺序是这样的：

1. Task 处理整个文件并在batcher 上调用Complete()。
2. batcher 处理最后一批，将其发送到 searcher 并完成。
3. 继续执行，它在searcher 和uiResult 上调用Complete()。
4. 由于uiResult没有工作要做，它完成了。
5. searcher 处理最后一批，尝试将每个结果发送到 uiResult。但是uiResult 已经完成，所以它拒绝一切。这意味着Post() 返回false，但您并没有对此进行检查。

所以问题是你试图向一个已经完成的块发送一些东西，这是行不通的。

解决方案是仅在块完成之前的块之后调用Complete()（即其Completion 完成）。可能最简单的方法是使用PropagateCompletion 和LinkTo()。