任務無法運行

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處理整個文件，並呼籲batcherComplete()。
2. batcher處理最後一批，發送到searcher並完成。
3. 繼續執行，在searcher和uiResult上調用Complete()。
4. 由於uiResult沒有工作要做，它完成。
5. searcher處理最後一批，嘗試將每個結果發送到uiResult。但uiResult已經完成，所以它拒絕了一切。這意味着Post()返回false，但你沒有檢查。

所以問題是，你正試圖發送一些東西到已經完成的塊，這是行不通的。

解決方法是在塊完成之前（即其Completion完成）僅在塊之後調用Complete()。可能最簡單的方法是使用PropagateCompletion和LinkTo()。