当我使用 await Console.Out.WriteLineAsync 时，程序永远不会完成

Question

我在 .NET 4.7.1 中使用 DataFlow 库由于某种原因，当我await Console.Out.WriteLineAsync(DateTime.Now.TimeOfDay.ToString());在构造函数的asynclambda 中时，我的程序永远不会终止。ActionBlock它只会输出行的流DateTime.Now.TimeOfDay.ToString()并随机停止，永远不会到达Console.WriteLine("Time elapsed:" + watch.Elapsed);，尽管在某些情况下我观察到控制台输出"Finished Reading the file"。 class Program    {        public static async Task Main(string[] args)        {            int numberOfLines = 0;            Console.WriteLine("Number of cores used:" + Convert.ToInt32(Math.Ceiling((Environment.ProcessorCount * 0.75) * 2.0)));            BufferBlock<string> queBuffer = new BufferBlock<string>(new DataflowBlockOptions { BoundedCapacity = 100000 });            var processingBlock = new ActionBlock<string>(async inputline =>            {                Interlocked.Increment(ref numberOfLines);                //Line that causes issue                //await Console.Out.WriteLineAsync(DateTime.Now.TimeOfDay.ToString());            }            , new ExecutionDataflowBlockOptions()            {                MaxDegreeOfParallelism = 48,                SingleProducerConstrained = true,                BoundedCapacity = 500            });            queBuffer.LinkTo(processingBlock);            //Start            var watch = System.Diagnostics.Stopwatch.StartNew();            Console.WriteLine("Processing started at:" + DateTime.Now);            if (File.Exists(args[0]))            {                using (StreamReader sr = new StreamReader(args[0]))                {        }    }但是，如果我取出导致问题的行，它会工作并从文本文件中读取所有行。 W:\test>.\CompressAMS.exe token2-small.txtNumber of cores used:24Processing started at:12/17/2018 6:32:50 PMFinished Reading the fileTime elapsed:00:00:00.3569824Number of lines read:100000

Answer 1

当你完成时，你所拥有的是一个竞争条件。您正在调用Complete()两个块，强制处理块停止接收消息，此时缓冲区可能仍有数据要传递。然后，当您等待两个块完成时，如果缓冲区尚未发送所有消息，它将永远不会完成并且执行将挂起Finished Reading File。

您可以安全地等待两个块，但只调用Complete()缓冲区并允许 TDF 将完成传播到下游处理块：

queBuffer.LinkTo(processingBlock, new DataflowLinkOptions() { PropagateCompletion = true });

/******/

queBuffer.Complete();

await Task.WhenAll(queBuffer.Completion, processingBlock.Completion);