如何同时压缩两个 IAsyncEnumerables？

Question

我有两个要成对“压缩”的异步序列，为此我使用了 Zip operator from the System.Linq.Async 包。不过，该运算符的行为不尽如人意，至少对于我而言是这样。它不是同时枚举两个序列，而是按顺序枚举它们，结果是延迟相加。我的每个序列平均每一秒发出一个元素，我预计组合序列也会每隔一秒发出压缩对，但实际上我每 2 秒得到一对。下面是演示此行为的最小示例：

static async IAsyncEnumerable<int> First()
{
    for (int i = 1; i <= 5; i++) { await Task.Delay(1000); yield return i; }
}

static async IAsyncEnumerable<int> Second()
{
    for (int i = 1; i <= 5; i++) { await Task.Delay(1000); yield return i; }
}

var stopwatch = Stopwatch.StartNew();
await foreach (var pair in First().Zip(Second()))
    Console.WriteLine(pair);
Console.WriteLine($"Duration: {stopwatch.ElapsedMilliseconds:#,0} msec");

输出：

(1, 1)
(2, 2)
(3, 3)
(4, 4)
(5, 5)
Duration: 10,155 msec

Try it on Fiddle.

有什么方法可以让程序在 5 秒而不是 10 秒内完成 Zip 这两个序列？我对具有理想行为的自定义运算符或官方包中的运算符组合感兴趣。

Answer 1

类似这样的方法似乎有效：

public static async IAsyncEnumerable<(TFirst, TSecond)> Zip<TFirst, TSecond>(this IAsyncEnumerable<TFirst> first, IAsyncEnumerable<TSecond> second)
{
    await using var e1 = first.GetAsyncEnumerator();    
    await using var e2 = second.GetAsyncEnumerator();
    
    while (true)
    {
        var t1 = e1.MoveNextAsync().AsTask();
        var t2 = e2.MoveNextAsync().AsTask();
        await Task.WhenAll(t1, t2);
        
        if (!t1.Result || !t2.Result)
            yield break;
        
        yield return (e1.Current, e2.Current);
    }
}

在 dotnetfiddle.net 上查看。

当然，这遗漏了诸如空值检查之类的东西，因此可以做一些改进：留作 reader.

的练习

我也不相信 Task.WhenAll 比这里的 bool r1 = await t1; bool r2 = await t2; if (!r1 || !r2) yield break; 好。

如何同时压缩两个 IAsyncEnumerables？

How to Zip concurrently two IAsyncEnumerables?

c#

linq

concurrency

iasyncenumerable

ix.net