数据流 Task.WhenAll 导致任务被取消异常

Question

我是 Dataflow 的新手，我按照此演练 How to: Cancel a Dataflow Block。
我先单击添加按钮，然后单击取消，但单击取消按钮后出现“A task was cancelled Exception”异常。我找不到解决此错误的任何方法。
任何帮助将不胜感激。
更新： 演示代码：

    public partial class Form1 : Form
{
    CancellationTokenSource cancellationTokenSource;
    TransformBlock<WorkItem, WorkItem> startWork;
    ActionBlock<WorkItem> completeWork;
    ActionBlock<ToolStripProgressBar> incProgress;
    ActionBlock<ToolStripProgressBar> decProgress;
    TaskScheduler uiTaskScheduler;
    public Form1()
    {
        InitializeComponent();
        uiTaskScheduler = TaskScheduler.FromCurrentSynchronizationContext();
        Cancel.Enabled = false;
    }

    private void Add_Click(object sender, EventArgs e)
    {
        if (!Cancel.Enabled)
        {
            CreatePipeline();
            Cancel.Enabled = true;
        }

        for (int i = 0; i < 20; i++)
        {
            toolStripProgressBar1.Value++;
            startWork.Post(new WorkItem());
        }
    }

    private async void Cancel_Click(object sender, EventArgs e)
    {
        Add.Enabled = false;
        Cancel.Enabled = false;

        cancellationTokenSource.Cancel();

        try
        {
             await Task.WhenAll(
                completeWork.Completion,
                incProgress.Completion,
                decProgress.Completion);
        }
        catch (OperationCanceledException)
        {

            throw;
        }
        toolStripProgressBar4.Value += toolStripProgressBar1.Value;
        toolStripProgressBar4.Value += toolStripProgressBar2.Value;

        // Reset the progress bars that track the number of active work items.
        toolStripProgressBar1.Value = 0;
        toolStripProgressBar2.Value = 0;

        // Enable the Add Work Items button.      
        Add.Enabled = true;
    }
    private void CreatePipeline()
    {
        cancellationTokenSource = new CancellationTokenSource();

        startWork = new TransformBlock<WorkItem, WorkItem>(workItem =>
        {
            workItem.DoWork(250, cancellationTokenSource.Token);
            decProgress.Post(toolStripProgressBar1);
            incProgress.Post(toolStripProgressBar2);
            return workItem;
        },
        new ExecutionDataflowBlockOptions
        {
            CancellationToken = cancellationTokenSource.Token
        });

        completeWork = new ActionBlock<WorkItem>(workItem =>
        {
            workItem.DoWork(1000, cancellationTokenSource.Token);
            decProgress.Post(toolStripProgressBar2);
            incProgress.Post(toolStripProgressBar3);
        },
        new ExecutionDataflowBlockOptions
        {
            CancellationToken = cancellationTokenSource.Token,
            MaxDegreeOfParallelism = 2
        });

        startWork.LinkTo(completeWork);

        startWork.Completion.ContinueWith(delegate { completeWork.Complete(); },cancellationTokenSource.Token);
        incProgress = new ActionBlock<ToolStripProgressBar>(progress =>
        {
            progress.Value++;
        },
        new ExecutionDataflowBlockOptions
        {
            CancellationToken = cancellationTokenSource.Token,
            TaskScheduler = uiTaskScheduler
        });

        decProgress = new ActionBlock<ToolStripProgressBar>(progress => progress.Value--,
            new ExecutionDataflowBlockOptions
            {
                CancellationToken = cancellationTokenSource.Token,
                TaskScheduler = uiTaskScheduler
            });

    }

    class WorkItem
    {
        public void DoWork(int milliseconds, CancellationToken cancellationToken)
        {
            if (cancellationToken.IsCancellationRequested == false)
            {
                Thread.Sleep(milliseconds);
            }
        }
    }
}

Answer 1

正如@SirRufo 指出的那样，您的问题的解决方案就是在捕获异常后不要重新抛出异常。但是为了突出您可以在 cmets 中讨论的数据流中使用的其他一些技术，我整理了一个小示例。我试图保持原始代码的精神和意图不变。为此；原始代码没有显示流程如何正常完成，而不是取消，所以我也把它留在这里了。

using System;
using System.Data;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Windows.Forms;

namespace WindowsFormsApp1 {
    public partial class Form1 : Form {

        private CancellationTokenSource cancellationTokenSource;
        private TransformBlock<WorkItem, WorkItem> startWork;
        private ActionBlock<WorkItem> completeWork;
        private IProgress<int> progressBar1Value;
        private IProgress<int> progressBar2Value;

        public Form1() {
            InitializeComponent();
            btnCancel.Enabled = false;
        }

        private async void btnAdd_Click(object sender, EventArgs e) {
            if(!btnCancel.Enabled) {
                CreatePipeline();
                btnCancel.Enabled = true;
            }
            var data = Enumerable.Range(0, 20).Select(_ => new WorkItem());
            foreach(var workItem in data) {
                await startWork.SendAsync(workItem);
                progressBar1.Value++;                
            }
        }

        private async void btnCancel_Click(object sender, EventArgs e) {
            btnAdd.Enabled = false;
            btnCancel.Enabled = false;

            cancellationTokenSource.Cancel();

            await completeWork.Completion.ContinueWith(tsk => this.Invoke(new Action(() => this.Text = "Flow Cancelled")), 
                                                       TaskContinuationOptions.OnlyOnCanceled);

            progressBar4.Value += progressBar1.Value;
            progressBar4.Value += progressBar2.Value;

            // Reset the progress bars that track the number of active work items.
            progressBar1.Value = 0;
            progressBar2.Value = 0;

            // Enable the Add Work Items button.      
            btnAdd.Enabled = true;
        }

        private void CreatePipeline() {
            cancellationTokenSource = new CancellationTokenSource();
            progressBar1Value = new Progress<int>(_ => progressBar1.Value++);
            progressBar2Value = new Progress<int>(_ => progressBar2.Value++);

            startWork = new TransformBlock<WorkItem, WorkItem>(async workItem => {
                await workItem.DoWork(250, cancellationTokenSource.Token);
                progressBar1Value.Report(0); //Value is ignored since the progressbar value is simply incremented
                progressBar2Value.Report(0); //Value is ignored since the progressbar value is simply incremented
                return workItem;
            },
            new ExecutionDataflowBlockOptions {
                CancellationToken = cancellationTokenSource.Token
            });

            completeWork = new ActionBlock<WorkItem>(async workItem => {
                await workItem.DoWork(1000, cancellationTokenSource.Token);
                progressBar1Value.Report(0); //Value is ignored since the progressbar value is simply incremented
                progressBar2Value.Report(0); //Value is ignored since the progressbar value is simply incremented
            },
            new ExecutionDataflowBlockOptions {
                CancellationToken = cancellationTokenSource.Token,
                MaxDegreeOfParallelism = 2
            });

            startWork.LinkTo(completeWork, new DataflowLinkOptions() { PropagateCompletion = true });
        }
    }

    public class WorkItem {
        public async Task DoWork(int milliseconds, CancellationToken cancellationToken) {
            if(cancellationToken.IsCancellationRequested == false) {
                await Task.Delay(milliseconds);
            }
        }
    }
}

Answer 2

查看代码后，我发布了如果点击取消，任务将被取消。

await Task.WhenAll(
                completeWork.Completion,
                incProgress.Completion,
                decProgress.Completion);   

但是，上面的代码 Task.WhenAll 需要所有的任务都返回完成状态，那么“任务被取消异常”如果返回取消而不是完成，则按预期抛出。
对于解决此问题的可能方法，如果我们取消任务，我们应该返回 Task completed，并且下面的代码适用于我。

       await Task.WhenAll(
           completeWork.Completion.ContinueWith(task => cancelWork(task, "completeWork"), TaskContinuationOptions.OnlyOnCanceled),
           incProgress.Completion.ContinueWith(task => cancelWork(task, "incProgress"), TaskContinuationOptions.OnlyOnCanceled),
           decProgress.Completion.ContinueWith(task => cancelWork(task, "decProgress"), TaskContinuationOptions.OnlyOnCanceled));

合理吗？