Task.WhenAll 用于 ValueTask

Question

是否有相当于Task.WhenAll 接受ValueTask？

我可以使用

Task.WhenAll(tasks.Select(t => t.AsTask()))

如果它们都包装了Task，这会很好，但它会强制为真正的ValueTask 分配一个无用的Task 对象。

Answer 1

按照设计，没有。来自the docs：

当方法的操作结果很可能同步可用并且方法被期望如此频繁地调用以致为每次调用分配新任务的成本将过高时，方法可能会返回此值类型的实例.

…

例如，考虑一个方法，它可以返回带有缓存任务的Task<TResult> 作为常见结果，也可以返回ValueTask<TResult>。如果结果的消费者想要将其用作Task<TResult>，例如在Task.WhenAll 和Task.WhenAny 等方法中使用，则首先需要使用@987654330 将ValueTask<TResult> 转换为Task<TResult> @，如果一开始就使用了缓存的Task<TResult>，这将导致本可以避免的分配。

Answer 2

除非我遗漏了什么，否则我们应该能够在一个循环中等待所有任务：

public static async ValueTask<T[]> WhenAll<T>(params ValueTask<T>[] tasks)
{
    // Argument validations omitted

    var results = new T[tasks.Length];
    for (var i = 0; i < tasks.Length; i++)
        results[i] = await tasks[i].ConfigureAwait(false);

    return results;
}

分配
等待同步完成的ValueTask 不应导致分配Task。所以这里发生的唯一“额外”分配是我们用于返回结果的数组。

订购
返回项目的顺序与产生它们的给定任务的顺序相同。

例外情况
当任务抛出异常时，上面的代码将停止等待其余的异常并直接抛出。如果这是不可取的，我们可以这样做：

public static async ValueTask<T[]> WhenAll<T>(params ValueTask<T>[] tasks)
{
    // We don't allocate the list if no task throws
    List<Exception>? exceptions = null;

    var results = new T[tasks.Length];
    for (var i = 0; i < tasks.Length; i++)
        try
        {
            results[i] = await tasks[i].ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            exceptions ??= new List<Exception>(tasks.Length);
            exceptions.Add(ex);
        }

    return exceptions is null
        ? results
        : throw new AggregateException(exceptions);
}

额外注意事项

• 我们可以将此作为扩展方法。
• 我们可以让重载接受IEnumerable<ValueTask<T>> 和IReadOnlyList<ValueTask<T>> 以获得更广泛的兼容性。

样本签名：

// There are some collections (e.g. hash-sets, queues/stacks,
// linked lists, etc) that only implement I*Collection interfaces
// and not I*List ones, but A) we're not likely to have our tasks
// in them and B) even if we do, IEnumerable accepting overload
// below should handle them. Allocation-wise; it's a ToList there
// vs GetEnumerator here.
public static async ValueTask<T[]> WhenAll<T>(
    IReadOnlyList<ValueTask<T>> tasks)
{
    // Our implementation above.
}

// ToList call below ensures that all tasks are initialized, so
// calling this with an iterator wouldn't cause the tasks to run
// sequentially (Thanks Sergey from comments to mention this
// possibility, which led me to add this Considerations section).
public static ValueTask<T[]> WhenAll<T>(
    IEnumerable<ValueTask<T>> tasks)
{
    return WhenAll(tasks?.ToList());
}

// Arrays already implement IReadOnlyList<T>, but this overload
// is still useful because as the `params` keyword allows callers 
// to pass individual tasks like they are different arguments.
public static ValueTask<T[]> WhenAll<T>(
    params ValueTask<T>[] tasks)
{
    return WhenAll(tasks as IReadOnlyList<ValueTask<T>>);
}

cmets 中的 Theodor 提到了将结果数组/列表作为参数传递的方法，因此 我们的 实现将没有所有额外的分配，但调用者仍然必须创建它，这可能如果他们批量等待任务是有道理的，但这听起来像是一个相当专业的场景，所以如果你发现自己需要，你可能不需要这个答案?

Answer 3

正如@stuartd 指出的那样，它不受设计支持，我不得不手动实现：

public static async Task<IReadOnlyCollection<T>> WhenAll<T>(this IEnumerable<ValueTask<T>> tasks)
{
    var results = new List<T>();
    var toAwait = new List<Task<T>>();

    foreach (var valueTask in tasks)
    {
        if (valueTask.IsCompletedSuccessfully)
            results.Add(valueTask.Result);
        else
            toAwait.Add(valueTask.AsTask());
    }

    results.AddRange(await Task.WhenAll(toAwait).ConfigureAwait(false));

    return results;
}

当然，这只会有助于高吞吐量和大量ValueTask，因为它会增加一些其他开销。

注意：正如@StephenCleary 指出的那样，这不会像Task.WhenAll 那样保持顺序，如果需要，可以轻松更改以实现它。

Answer 4

我正在使用这种扩展方法：

internal static class ValueTaskExtensions
{
    public static Task WhenAll(this IEnumerable<ValueTask> tasks)
    {
        return Task.WhenAll(tasks.Select(v => v.AsTask()));
    }
}

Answer 5

尝试做一些优化，结果返回正确的顺序和正确的异常处理。

public static ValueTask<T[]> WhenAll<T>(IEnumerable<ValueTask<T>> tasks)
    {
        var list = tasks.ToList();
        var length = list.Count;
        var result = new T[length];
        var i = 0;

        for (; i < length; i ++)
        {
            if (list[i].IsCompletedSuccessfully)
            {
                result[i] = list[i].Result;
            }
            else
            {
                return WhenAllAsync();
            }
        }

        return new ValueTask<T[]>(result);

        async ValueTask<T[]> WhenAllAsync()
        {
            for (; i < length; i ++)
            {
                try
                {
                    result[i] = await list[i];
                }
                catch
                {
                    for (i ++; i < length; i ++)
                    {
                        try
                        {
                            await list[i];
                        }
                        catch
                        {
                            // ignored
                        }
                    }

                    throw;
                }
            }

            return result;
        }
    }