有没有办法等待所有任务，直到特定结果为真，然后取消其余的？

Question

在我的 C# 控制台应用程序中，我尝试运行多个同时执行各种数据检查的任务。 如果其中一项任务返回 true，我应该停止其他任务，因为我有可操作的结果。也很有可能没有一个函数返回 true

我有代码可以一起运行这些任务（我想），但我无法到达终点线：

Task task1 = Task.Run(() => Task1(stoppingToken));
Task task2 = Task.Run(() => Task2(stoppingToken));
Task task3 = Task.Run(() => Task3(stoppingToken));
Task task4 = Task.Run(() => Task4(stoppingToken));
Task task5 = Task.Run(() => Task5(stoppingToken));
Task task6 = Task.Run(() => Task6(stoppingToken));

Task.WaitAll(task1, task2, task3, task4, task5, task6);

这与已知所需结果（超时值）的链接问题中的答案略有不同。我正在等待这些任务中的任何一个可能返回 true，然后如果它们仍在运行则取消剩余的任务

Task.WhenAny with cancellation of the non completed tasks and timeout

Answer 1

这是一个基于延续任务的解决方案。这个想法是将延续任务附加到每个原始（提供）任务，并在那里检查结果。如果匹配，完成源将设置一个结果（如果没有匹配，则根本不会设置结果）。

然后，代码将等待首先发生的任何事情：要么所有延续任务完成，要么设置任务完成结果。无论哪种方式，我们都准备好检查与任务完成源关联的任务的结果（这就是我们等待 继续任务 完成的原因，而不是原始任务），如果已设置，它就是几乎表明我们有一个匹配项（最后的额外检查有点偏执，但我猜比抱歉更安全......：D）

public static async Task<bool> WhenAnyHasResult<T>(Predicate<T> isExpectedResult, params Task<T>[] tasks)
{
    const TaskContinuationOptions continuationTaskFlags = TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.AttachedToParent;
         
    // Prepare TaskCompletionSource to be set only when one of the provided tasks
    // completes with expected result
    var tcs = new TaskCompletionSource<T>();

    // For every provided task, attach a continuation task that fires
    // once the original task was completed
    var taskContinuations = tasks.Select(task =>
    {
        return task.ContinueWith(x =>
        {
            var taskResult = x.Result;
            if (isExpectedResult(taskResult))
            {
                tcs.SetResult(taskResult);
            }
        },
        continuationTaskFlags);
    });

    // We either wait for all the continuation tasks to be completed 
    // (it's most likely an indication that none of the provided tasks completed with the expected result)
    // or for the TCS task to complete (which means a failure)
    await Task.WhenAny(Task.WhenAll(taskContinuations), tcs.Task);

    // If the task from TCS has run to completion, it means the result has been set from
    // the continuation task attached to one of the tasks provided in the arguments
    var completionTask = tcs.Task;
    if (completionTask.IsCompleted)
    {
        // We will check once more to make sure the result is set as expected 
        // and return this as our outcome
        var tcsResult = completionTask.Result;
        return isExpectedResult(tcsResult);
    }

    // TCS result was never set, which means we did not find a task matching the expected result.
    tcs.SetCanceled();
    return false;
}

现在，用法如下：

static async Task ExampleWithBooleans()
{
    Console.WriteLine("Example with booleans");

    var task1 = SampleTask(3000, true);
    var task2 = SampleTask(5000, false);

    var finalResult = await TaskUtils.WhenAnyHasResult(result => result == true, task1, task2);

    // go ahead and cancel your cancellation token here

    Console.WriteLine("Final result: " + finalResult);
    Debug.Assert(finalResult == true);
    Console.WriteLine();
}

将它放入泛型方法的好处在于，它适用于任何类型，而不仅仅是布尔值，这是原始任务的结果。

Answer 2

假设您的任务返回 bool，您可以执行以下操作：

CancellationTokenSource source = new CancellationTokenSource();
CancellationToken stoppingToken = source.Token;
Task<bool> task1 = Task.Run(() => Task1(stoppingToken));
....

var tasks = new List<Task<bool>>
{
    task1, task2, task3, ...
};

bool taskResult = false;
do
{
    var finished = await Task.WhenAny(tasks);
    taskResult = finished.Result;
    tasks.Remove(finished);
} while (tasks.Any() && !taskResult);

source.Cancel();

Answer 3

您可以使用将Task<bool> 包装到另一个Task<bool> 的异步方法，如果输入任务的结果是true，则取消CancellationTokenSource。在下面的例子中，这个方法是IfTrueCancel，它被实现为local function。这样capturesCancellationTokenSource，因此您不必在每次调用时都将其作为参数传递：

var cts = new CancellationTokenSource();
var stoppingToken = cts.Token;

var task1 = IfTrueCancel(Task.Run(() => Task1(stoppingToken)));
var task2 = IfTrueCancel(Task.Run(() => Task2(stoppingToken)));
var task3 = IfTrueCancel(Task.Run(() => Task3(stoppingToken)));
var task4 = IfTrueCancel(Task.Run(() => Task4(stoppingToken)));
var task5 = IfTrueCancel(Task.Run(() => Task5(stoppingToken)));
var task6 = IfTrueCancel(Task.Run(() => Task6(stoppingToken)));

Task.WaitAll(task1, task2, task3, task4, task5, task6);

async Task<bool> IfTrueCancel(Task<bool> task)
{
    bool result = await task.ConfigureAwait(false);
    if (result) cts.Cancel();
    return result;
}

另一个完全不同的解决方案是使用PLINQ 而不是显式创建的Tasks。 PLINQ 需要一个 IEnumerable 的东西才能对其进行并行工作，在您的情况下，这是您要调用的 Task1、Task2 等函数。您可以将它们放在Func<CancellationToken, bool> 的数组中，然后这样解决问题：

var functions = new Func<CancellationToken, bool>[]
{
    Task1, Task2, Task3, Task4, Task5, Task6
};

bool success = functions
    .AsParallel()
    .WithDegreeOfParallelism(4)
    .Select(function =>
    {
        try
        {
            bool result = function(stoppingToken);
            if (result) cts.Cancel();
            return result;
        }
        catch (OperationCanceledException)
        {
            return false;
        }
    })
    .Any(result => result);

这种方式的好处是可以配置并行度，不必依赖ThreadPool的可用性来限制整个操作的并发。缺点是所有函数都应该具有相同的签名。您可以通过将函数声明为 lambda 表达式来克服这个缺点，如下所示：

var functions = new Func<CancellationToken, bool>[]
{
    ct => Task1(arg1, ct),
    ct => Task2(arg1, arg2, ct),
    ct => Task3(ct),
    ct => Task4(arg1, arg2, arg3, ct),
    ct => Task5(arg1, ct),
    ct => Task6(ct)
};