Почему TPL Dataflow block.LinkTo не дает никакого вывода?

Question

Я довольно новичок в теме потока данных TPL. В книге Параллельность в C# Я проверил следующий пример. Я не могу понять, почему нет выхода, который должен быть 2*2-2=2;

static void Main(string[] args) 
{ 
    //Task tt = test(); 
    Task tt = test1(); 
    Console.ReadLine(); 
} 
static async Task test1() 
{ 
    try 
    { 
     var multiplyBlock = new TransformBlock<int, int>(item => 
     { 
      if (item == 1) 
       throw new InvalidOperationException("Blech."); 
      return item * 2; 
     }); 
     var subtractBlock = new TransformBlock<int, int>(item => item - 2); 
     multiplyBlock.LinkTo(subtractBlock, 
     new DataflowLinkOptions { PropagateCompletion = true }); 
     multiplyBlock.Post(2); 
     await subtractBlock.Completion; 
     int temp = subtractBlock.Receive(); 
     Console.WriteLine(temp); 
    } 
    catch (AggregateException e) 
    { 
     // The exception is caught here. 
     foreach (var v in e.InnerExceptions) 
     { 
      Console.WriteLine(v.Message); 
     } 

    } 

} 

Update1: Я попробовал другой пример. Тем не менее я не использовал Block.Complete(), но я подумал, что когда первый блок будет завершен, результат будет передан во второй блок автоматически.

private static async Task test3() 
{ 
    TransformManyBlock<int, int> tmb = new TransformManyBlock<int, int>((i) => { return new int[] {i, i + 1}; }); 
    ActionBlock<int> ab = new ActionBlock<int>((i) => Console.WriteLine(i)); 

    tmb.LinkTo(ab); 

    for (int i = 0; i < 4; i++) 
    { 
     tmb.Post(i); 
    } 
    //tmb.Complete(); 
    await ab.Completion; 
    Console.WriteLine("Finished post"); 
} 

Answer 1

Эта часть кода:

await subtractBlock.Completion; 
int temp = subtractBlock.Receive(); 

является первым (асинхронно) ждет блок вычитания для завершения, а затем пытается получить выходной сигнал из блока.

Есть две проблемы: исходный блок никогда не завершается, и код пытается получить выход из завершенного блока. Как только блок будет завершен, он не будет производить больше данных.

(Я предполагаю, что вы ссылаетесь на пример в рецепте 4.2, который выведет 1, вызывая исключение, которое завершает блок в неисправном состоянии).

Таким образом, вы можете исправить этот тест, заполнив блок источника (и завершение будет распространяться по ссылке на subtractBlock автоматически), и перед чтением вывода (асинхронно) ждет subtractBlock для завершения:

multiplyBlock.Complete(); 
int temp = subtractBlock.Receive(); 
await subtractBlock.Completion;