Início / coding / Perguntas / 79411372
Accepted
Jeppe Stig Nielsen
Asked: 2025-02-04 18:40:52 +0800 CST

Obtendo um 2º `IAsyncEnumerator<>` do mesmo `IAsyncEnumerable<>` com base no método `Task.WhenEach<>`

  • 772

Dado um IAsyncEnumerable<> , em geral funciona bem (mesmo que possa ser um desempenho não ótimo) chamar GetAsyncEnumeratormais de uma vez, obtendo um novo IAsyncEnumerator<> a cada vez. Para dar um exemplo explícito (todo o código nesta questão está em C#), suponha que eu tenha este método:

static async IAsyncEnumerable<string> GetStr() {
    var item = await Task.FromResult("Bravo");
    yield return "Alfa";
    yield return item;
    yield return "Charlie";
}

então funciona bem fazer isso:

IAsyncEnumerable<string> oneEnumerable = GetStr();
await foreach (var s in oneEnumerable) {
    Console.WriteLine(s);
}
Console.WriteLine("Between");
await foreach (var t in oneEnumerable) {
    Console.WriteLine(t);
}
Console.WriteLine("End");

Se você quiser avançar, você pode até mesmo ter os dois enumeradores (do mesmo enumerável) vivendo juntos, sem descartar o primeiro antes de adquirir o segundo, digamos:

IAsyncEnumerable<string> oneEnumerable = GetStr();
IAsyncEnumerator<string> e = oneEnumerable.GetAsyncEnumerator();
IAsyncEnumerator<string> f = oneEnumerable.GetAsyncEnumerator();
bool c = await f.MoveNextAsync();
bool b = await e.MoveNextAsync();

Console.WriteLine($"b {b} with {e.Current}");
Console.WriteLine($"c {c} with {f.Current}");
await e.DisposeAsync();
await f.DisposeAsync();

O código acima funciona da maneira que vocês esperam, os dois enumeradores são independentes e não misturam seus estados.

Então, isso foi apenas uma introdução; quero perguntar sobre um caso em que adquirir dois enumeradores do mesmo enumerável leva a um resultado estranho. Então, agora considere este método baseado em Task.WhenEach :

static IAsyncEnumerable<Task<int>> GetTasks() {
    IEnumerable<Task<int>> source = [Task.FromResult(7), Task.FromResult(9), Task.FromResult(13)];
    return Task.WhenEach(source);
}

e use este código:

IAsyncEnumerable<Task<int>> oneEnumerable = GetTasks();
await foreach (var t in oneEnumerable) {
    Console.WriteLine(t);
}
Console.WriteLine("Between");
await foreach (var u in oneEnumerable) {
    Console.WriteLine(u);
}
Console.WriteLine("End");

Isso é executado sem exceção. MAS: A segunda enumeração fornece zero itens (corpo de foreach com uexecução zero vezes)!

Minhas perguntas:

  • Esse é o comportamento esperado de Task.WhenEach?
  • Esse comportamento está documentado?

Acho isso muito propenso a erros. Se, por alguma razão técnica, for impossível obter mais de uma enumeração de um enumerável, seria muito melhor se a segunda chamada para GetAsyncEnumeratorlançasse uma exceção (ou, alternativamente, a primeira chamada para MoveNextAsyncno segundo enumerador lançasse, ou, aguardando o ValueTask<bool>produzido por MoveNextAsynclançaria).

(Foi-me mostrado outro tópico onde um IAsyncEnumerable<>produzido por um GetRecordsAsync<>de a CsvReaderteve um problema semelhante .)

c#

2 respostas

  1. Best Answer

    Parece que ainda não está documentado, mas os comentários do código mencionam isso explicitamente:

    // The enumerable could have GetAsyncEnumerator called on it multiple times.
// As we're dealing with Tasks that only ever transition from non-completed
// to completed, re-enumeration doesn't have much benefit, so we take advantage
// of the optimizations possible by not supporting that and simply have the
// semantics that, no matter how many times the enumerable is enumerated, every
// task is yielded only once. The original GetAsyncEnumerator call will give back
// all the tasks, and all subsequent iterations will be empty.
if (waiter?.TryStart() is not true)
{
    yield break;
}

    Também:

    /// <summary>0 if this has never been used in an iteration; 1 if it has.</summary>
/// <remarks>This is used to ensure we only ever iterate through the tasks once.</remarks>
private int _enumerated;

/// <summary>Called at the beginning of the iterator to assume ownership of the state.</summary>
/// <returns>true if the caller owns the state; false if the caller should end immediately.</returns>
public bool TryStart() => Interlocked.Exchange(ref _enumerated, 1) == 0;

    Concordo que isso parece não óbvio e deve ser documentado. De preferência, deve permitir múltiplas enumerações. Sugiro que você crie um feature-request no GitHub para isso.

    Você pode ser capaz de escrever seu próprio iterador para fazer isso. Exemplo:

    public async static IAsyncEnumerable<T> WhenEach<T>(
    IEnumerable<T> tasks,
    [EnumeratorCancellation] CancellationToken cancellationToken = default) where T : Task
{
    ArgumentNullException.ThrowIfNull(tasks);
    await foreach (T task in Task.WhenEach(tasks)
        .WithCancellation(cancellationToken).ConfigureAwait(false))
    {
        yield return task;
    }
}

    Os métodos iteradores do C# criam enumeráveis ​​que invocam o corpo do método toda vez que uma nova enumeração é iniciada.

    • 4

  2. A resposta de Charlieface explica o comportamento de Task.WhenEach, e inclui uma solução especializada. Esta resposta é sobre uma solução geral para todos IAsyncEnumerable<T>os s que têm a mesma peculiaridade com o Task.WhenEach. O pacote System.Interactive.AsyncAsyncEnumerableEx.Defer inclui o método, com esta assinatura:

    /// <summary>
/// Returns an async-enumerable sequence that invokes the specified factory function
/// whenever a new observer subscribes.
/// </summary>
public static IAsyncEnumerable<TSource> Defer<TSource>(
    Func<IAsyncEnumerable<TSource>> factory);

    Com a ajuda deste método, você pode criar uma sequência enumerável adiada que cria uma nova IAsyncEnumerable<TSource>cada vez que é enumerada.

    Exemplo de uso:

    IAsyncEnumerable<Task<int>> oneEnumerable = AsyncEnumerableEx.Defer(() => GetTasks());

    Aqui está uma demonstração online. Ela produz esta saída:

    7
9
13
Between
7
9
13
End
    • 1

relate perguntas