予期しない動作 - TPL DataFlow BatchBlockは、TriggerBatchの実行中にアイテムを拒否します。

Question

バインドされた容量のバッチブロックを作成し、triggerBatchを呼び出して新しいアイテムをポストすると、トリガーのバッチ実行中に新しいアイテムが失敗します。

呼び出しトリガーバッチ（X回ごと）は、受信データストリームが一時停止または遅くなった場合に、データがブロック内で長すぎないようにするために行われます。

次のコードは、「失敗後」イベントを出力します。たとえば ：

public static void Main(string[] args) 
    { 
     var batchBlock = new BatchBlock<int>(10, new GroupingDataflowBlockOptions() { BoundedCapacity = 10000000 }); 
     var actionBlock = new ActionBlock<int[]>(x => ProcessBatch(x), new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 }); 
     batchBlock.LinkTo(actionBlock); 

     var producerTask = Task.Factory.StartNew(() => 
     { 
      //Post 10K Items 
      for (int i = 0; i < 10000; i++) 
      { 
       var postResult = batchBlock.Post(i); 
       if (!postResult) 
        Console.WriteLine("Failed to Post"); 
      } 
     }); 

     var triggerBatchTask = Task.Factory.StartNew(() => 
      {      
       //Trigger Batch.. 
       for (int i = 0; i < 1000000; i++) 
        batchBlock.TriggerBatch(); 
      }); 

     producerTask.Wait(); 
     triggerBatchTask.Wait(); 
    } 

    public static void ProcessBatch(int[] batch) 
    { 
     Console.WriteLine("{0} - {1}", batch.First(), batch.Last()); 
    } 

*このシナリオはbatchBlockが制限された場合にのみ再現可能であることに注意してください。

は、私が何かをしないのですか、それはbatchBlockの問題ですか？ BatchBlockが実際にアイテムを拒否していない

Answer 1

、それを延期しようとします。 Post()の場合を除いて、延期はオプションではありません。これを修正する簡単な方法は、batchBlock.Post(i)の代わりにawait batchBlock.SendAsync(i)を使用することです（これはTask.Factory.StartNew(() =>をTask.Run(async() =>に変更する必要があることを意味します）。

どうしてですか？ the source codeによれば、BatchBlockがバインドされている場合、TriggerBatch()は非同期に処理され、処理されている間は新しい項目は受け入れられません。いずれの場合においても

、あなたはPost()は常にブロックがいっぱいの場合、Post()もfalseを返します、有界ブロックにtrueを返すことを期待すべきではありません。