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

Question

我在 .NET 4.7.1 中使用 DataFlow 库出于某种原因，当我在 ActionBlock 构造函数的 async lambda 中有 await Console.Out.WriteLineAsync(DateTime.Now.TimeOfDay.ToString()); 时，我的程序永远不会终止。它只会输出 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]))
                {
                    string line;
                    // Read and display lines from the file until the end of the file is reached.
                    while ((line = await sr.ReadLineAsync()) != null)
                    {
                        await queBuffer.SendAsync(line);
                    }
                }
                await Console.Out.WriteLineAsync("Finished Reading the file");
            }
            queBuffer.Complete();
            processingBlock.Complete();
            await Task.WhenAll(queBuffer.Completion, processingBlock.Completion);
            watch.Stop();
            Console.WriteLine("Time elapsed:" + watch.Elapsed);
            Console.WriteLine("Number of lines read:" + numberOfLines.ToString());           
        }
    }

但是，如果我删除导致问题的行，它会工作并从文本文件中读取所有行。 W:\test>.\CompressAMS.exe token2-small.txt Number of cores used:24 Processing started at:12/17/2018 6:32:50 PM Finished Reading the file Time elapsed:00:00:00.3569824 Number 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);

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

Program never finishes when I use await Console.Out.WriteLineAsync

.net

c#

task-parallel-library

async-await

tpl-dataflow