如何强制 ActionBlock 快速完成

Question

根据documentation：

当数据流块当前未处理消息并且已保证不再处理任何消息时，数据流块被视为已完成。

在我的情况下，这种行为并不理想。我希望能够随时取消作业，但是处理每个单独的操作需要很长时间。因此，当我取消令牌时，效果不会立即生效。我必须等待当前处理的项目完成。我无法直接取消操作，因为我使用的 API 是不可取消的。我可以做任何事情使该块忽略当前正在运行的操作并立即完成吗？

这是一个演示我的问题的示例。令牌在 500 毫秒后被取消，每个动作的持续时间为 1000 毫秒：

static async Task Main()
{
    var cts = new CancellationTokenSource(500);
    var block = new ActionBlock<int>(async x =>
    {
        await Task.Delay(1000);
    }, new ExecutionDataflowBlockOptions() { CancellationToken = cts.Token });
    block.Post(1); // I must wait for this one to complete
    block.Post(2); // This one is ignored
    block.Complete();
    var stopwatch = Stopwatch.StartNew();
    try
    {
        await block.Completion;
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine($"Canceled after {stopwatch.ElapsedMilliseconds} msec");
    }
}

输出：

1035 毫秒后取消

所需的输出将是约 500 毫秒后的取消。

Answer 1

根据您的评论摘录...：

如果出现取消请求，我希望忽略当前正在运行的工作项。我已经不在乎了，为什么还要等呢？

...假设您真的可以让任务运行，您可以简单地将您希望调用的作业 包装在另一个 Task 中，这将 不断轮询取消或完成，然后取消 那个 Task。看一下下面的“概念验证”代码，它将“长时间运行”的任务包装在另一个“任务”任务中，不断轮询包装的任务以完成，并使用 CancellationToken 取消（完全“刺激” the-moment" 状态，你当然会想重新调整一下）：

public class LongRunningTaskSource
{
    public Task LongRunning(int milliseconds)
    {
        return Task.Run(() =>
        {
            Console.WriteLine("Starting long running task");
            Thread.Sleep(3000);
            Console.WriteLine("Finished long running task");
        });
    }

    public Task LongRunningTaskWrapper(int milliseconds, CancellationToken token)
    {
        Task task = LongRunning(milliseconds);

        Task wrapperTask = Task.Run(() =>
        {
            while (true)
            {
                //Check for completion (you could, of course, do different things
                //depending on whether it is faulted or completed).
                if (!(task.Status == TaskStatus.Running))
                    break;

                //Check for cancellation.
                if (token.IsCancellationRequested)
                {
                    Console.WriteLine("Aborting Task.");

                    token.ThrowIfCancellationRequested();
                }
            }

        }, token);

        return wrapperTask;
    }
}

使用以下代码：

static void Main()
{
    LongRunningTaskSource longRunning = new LongRunningTaskSource();

    CancellationTokenSource cts = new CancellationTokenSource(1500);

    Task task = longRunning.LongRunningTaskWrapper(3000, cts.Token);

    //Sleep long enough to let things roll on their own.
    Thread.Sleep(5000);

    Console.WriteLine("Ended Main");
}

...产生以下输出：

Starting long running task
Aborting Task.
Exception thrown: 'System.OperationCanceledException' in mscorlib.dll
Finished long running task
Ended Main

被包装的任务显然是在它自己的好时机完成的。如果您对此没有问题（通常情况下不是），希望这能满足您的需求。

作为补充示例，运行以下代码（让包装的任务在超时之前完成）：

static void Main()
{
    LongRunningTaskSource longRunning = new LongRunningTaskSource();

    CancellationTokenSource cts = new CancellationTokenSource(3000);

    Task task = longRunning.LongRunningTaskWrapper(1500, cts.Token);

    //Sleep long enough to let things roll on their own.
    Thread.Sleep(5000);

    Console.WriteLine("Ended Main");
}

...产生以下输出：

Starting long running task
Finished long running task
Ended Main

因此任务在超时之前开始并完成，无需取消任何内容。当然，等待时没有任何阻塞。当然，您可能已经知道，如果您知道长期运行的代码在幕后使用了什么，那么在必要时最好进行清理。

希望您可以修改此示例以将类似的内容传递给您的 ActionBlock。

免责声明和注意事项

我不熟悉 TPL Dataflow 库，所以这当然只是一种解决方法。此外，如果您只有一个同步方法调用，而您根本没有任何影响，那么您显然需要 两个 任务。一个包装器任务用于包装同步调用，另一个用于包装包装器任务以包括连续状态轮询和取消检查。