一次使用 IEnumerable 多次

Question

是否可以编写一个高阶函数，使IEnumerable 被多次使用，但只在一次传递中，而不会将所有数据读入内存？ [请参阅下面的编辑以澄清我正在寻找的内容。]

例如，在下面的代码中，可枚举是mynums（我在上面标记了.Trace()，以便查看我们枚举它的次数）。目标是弄清楚它是否有任何大于 5 的数字，以及所有数字的总和。处理一个可枚举两次的函数是Both_TwoPass，但它会枚举它两次。相比之下，Both_NonStream 只枚举了一次，但代价是将其读入内存。原则上，可以像Any5Sum 所示以单次传递和流式传输方式执行这两项任务，但这是特定的解决方案。是否可以编写一个与Both_* 具有相同签名的函数，但这是两全其美？

（在我看来，这应该可以使用线程来实现。有没有更好的解决方案，比如使用async？）

编辑

以下是关于我正在寻找的内容的说明。我所做的是在方括号中对每个属性进行了非常实际的描述。

我正在寻找具有以下特征的函数Both：

1. 它有签名(S1, S2) Both<T, S1, S2>(this IEnumerable<T> tt, Func<IEnumerable<T>, S1>, Func<IEnumerable<T>, S2>)（并产生“正确”的输出！）
2. 它只迭代第一个参数tt 一次。 [我的意思是当通过mynums（定义如下）时，它只输出mynums: 0 1 2 ...once。这排除了函数Both_TwoPass。]
3. 它以流方式处理来自第一个参数tt 的数据。 [我的意思是，例如，没有足够的内存将tt中的所有项目同时存储在内存中，因此排除了函数Both_NonStream。]

using System;
using System.Collections.Generic;
using System.Linq;

namespace ConsoleApp
{
    static class Extensions
    {
        public static IEnumerable<T> Trace<T>(this IEnumerable<T> tt, string msg = "")
        {
            Console.Write(msg);
            try
            {
                foreach (T t in tt)
                {
                    Console.Write(" {0}", t);
                    yield return t;
                }
            }
            finally
            {
                Console.WriteLine('.');
            }
        }

        public static (S1, S2) Both_TwoPass<T, S1, S2>(this IEnumerable<T> tt, Func<IEnumerable<T>, S1> f1, Func<IEnumerable<T>, S2> f2)
        {
            return (f1(tt), f2(tt));
        }

        public static (S1, S2) Both_NonStream<T, S1, S2>(this IEnumerable<T> tt, Func<IEnumerable<T>, S1> f1, Func<IEnumerable<T>, S2> f2)
        {
            var tt2 = tt.ToList();
            return (f1(tt2), f2(tt2));
        }

        public static (bool, int) Any5Sum(this IEnumerable<int> ii)
        {
            int sum = 0;
            bool any5 = false;
            foreach (int i in ii)
            {
                sum += i;
                any5 |= i > 5; // or: if (!any5) any5 = i > 5;
            }
            return (any5, sum);
        }

    }
    class Program
    {
        static void Main()
        {
            var mynums = Enumerable.Range(0, 10).Trace("mynums:");
            Console.WriteLine("TwoPass: (any > 5, sum) = {0}", mynums.Both_TwoPass(tt => tt.Any(k => k > 5), tt => tt.Sum()));
            Console.WriteLine("NonStream: (any > 5, sum) = {0}", mynums.Both_NonStream(tt => tt.Any(k => k > 5), tt => tt.Sum()));
            Console.WriteLine("Manual: (any > 5, sum) = {0}", mynums.Any5Sum());
        }
    }
}

Answer 1

您编写计算模型的方式（即return (f1(tt), f2(tt))）无法避免枚举的多次迭代。你基本上是在说计算 Item1 然后计算 Item2。

您必须将模型从 (Func<IEnumerable<T>, S1>, Func<IEnumerable<T>, S2>) 更改为 (Func<T, S1>, Func<T, S2>) 或 Func<IEnumerable<T>, (S1, S2)> 才能并行运行计算。

Any5Sum 的实现基本上是第二种方法 (Func<IEnumerable<T>, (S1, S2)>)。但是已经有一个内置的方法。

试试这个：

Console.WriteLine("Aggregate: (any > 5, sum) = {0}",
    mynums
        .Aggregate<int, (bool any5, int sum)>(
            (false, 0),
            (a, x) => (a.any5 | x > 5, a.sum + x)));

Answer 2

我认为 you 和 I 在 cmets 中描述的是同一件事。不过，没有必要创建这样的“专用IEnumerable”，因为BlockingCollection<> class 已经存在于此类生产者-消费者场景中。您可以按如下方式使用它...

• 为每个消费函数创建一个BlockingCollection<>（即tt1和tt2）。
  • 默认情况下，BlockingCollection<> 包装 ConcurrentQueue<>，因此元素将按 FIFO 顺序到达。
  • 为满足您一次只在内存中保存一个元素的要求，将为bounded capacity 指定1。请注意，此容量是每个集合的容量，因此对于两个集合，在任何给定时刻最多有两个排队的元素。
  • 每个集合都将保存该消费者的输入元素。
• 为每个消费函数创建一个线程/任务。
  • 线程/任务将简单地为其输入集合调用GetConsumingEnumerator()，将生成的IEnumerable<> 传递给其消费函数，然后返回该结果。
    • GetConsumingEnumerable() 就像它的名字所暗示的那样：它创建了一个IEnumerable<>，它使用（删除）集合中的元素。如果集合为空，则枚举将阻塞，直到添加元素。 CompleteAdding() 在生产者完成后被调用，这允许消费枚举器在集合清空时退出。
• 生产者枚举IEnumerable<>、tt，并将每个元素添加到两个集合中。这是唯一一次枚举tt。
  • 如果集合已达到其容量，BlockingCollection<>.Add() 将阻塞，从而防止整个 tt 在内存中缓冲。
• tt 被完全枚举后，每个集合都会调用 CompleteAdding()。
• 一旦每个消费者线程/任务完成，就会返回它们的结果。

这是代码中的样子...

public static (S1, S2) Both<T, S1, S2>(this IEnumerable<T> tt, Func<IEnumerable<T>, S1> tt1, Func<IEnumerable<T>, S2> tt2)
{
    const int MaxQueuedElementsPerCollection = 1;

    using (BlockingCollection<T> collection1 = new BlockingCollection<T>(MaxQueuedElementsPerCollection))
    using (Task<S1> task1 = StartConsumerTask(collection1, tt1))
    using (BlockingCollection<T> collection2 = new BlockingCollection<T>(MaxQueuedElementsPerCollection))
    using (Task<S2> task2 = StartConsumerTask(collection2, tt2))
    {
        foreach (T element in tt)
        {
            collection1.Add(element);
            collection2.Add(element);
        }

        // Inform any enumerators created by .GetConsumingEnumerable()
        // that there will be no more elements added.
        collection1.CompleteAdding();
        collection2.CompleteAdding();

        // Accessing the Result property blocks until the Task<> is complete.
        return (task1.Result, task2.Result);
    }

    Task<S> StartConsumerTask<S>(BlockingCollection<T> collection, Func<IEnumerable<T>, S> func)
    {
        return Task.Run(() => func(collection.GetConsumingEnumerable()));
    }
}

请注意，为了提高效率，您可以将MaxQueuedElementsPerCollection 增加到10 或100，这样消费者就不必彼此同步运行。

不过，此代码存在一个问题。当集合为空时，消费者必须等待生产者生产元素，而当集合已满时，生产者必须等待消费者消费元素。考虑一下在执行 tt => tt.Any(k => k > 5) lambda 的过程中会发生什么...

1. 生产者等待集合未满并添加5。
2. 消费者等待集合非空并删除5。
   • 5 > 5 返回 false 并继续枚举。
3. 生产者等待集合未满并添加6。
4. 消费者等待集合非空并删除6。
   • 6 > 5 返回 true 并且枚举停止。 Any()、lambda 和消费者任务都返回。
5. 生产者等待集合未满并添加7。
6. 生产者等待集合未满，然后……这永远不会发生！
   • 消费者已经放弃了枚举，所以它不会消​​耗任何元素来为新的元素腾出空间。 Add() 永远不会回来。

我能想出的防止这种死锁的最干净的方法是确保枚举整个集合，即使func 没有这样做。这只需要对StartConsumerTask<>()local method进行简单的更改...

Task<S> StartConsumerTask<S>(BlockingCollection<T> collection, Func<IEnumerable<T>, S> func)
{
    return Task.Run(
        () => {
            try
            {
                return func(collection.GetConsumingEnumerable());
            }
            finally
            {
                // Prevent BlockingCollection<>.Add() calls from
                // deadlocking by ensuring the entire collection gets
                // consumed even if func abandoned its enumeration early.
                foreach (T element in collection.GetConsumingEnumerable())
                {
                    // Do nothing...
                }
            }
        }
    );
}

这样做的缺点是 tt 将始终被枚举到完成，即使 两者 tt1 和 tt2 都提前放弃了它们的枚举器。

解决了这个问题，这...

static void Main()
{
    IEnumerable<int> mynums = Enumerable.Range(0, 10).Trace("mynums:");

    Console.WriteLine("Both: (any > 5, sum) = {0}", mynums.Both(tt => tt.Any(k => k > 5), tt => tt.Sum()));
}

...输出这个...

mynums: 0 1 2 3 4 5 6 7 8 9.
Both: (any > 5, sum) = (True, 45)

Answer 3

这里的核心问题是谁负责调用Enumeration.MoveNext()（例如通过使用foreach 循环）。跨线程同步多个 foreach 循环将是缓慢而繁琐的。

实现IAsyncEnumerable<T>，这样多个await foreach 循环可以轮流处理项目会更容易。但还是很傻。

所以更简单的解决方案是改变问题。与其尝试调用多个都尝试枚举项目的方法，不如将接口更改为简单地访问每个项目。

Answer 4

我相信有可能满足问题的所有要求，还有一个（非常自然的）要求，即如果两个Func<IEnumerable<T>, S> 中的每一个都部分消耗原始可枚举，则仅部分枚举原始可枚举。 （@BACON 对此进行了讨论）。该方法在我的GitHub repo 'CoEnumerable' 中有更详细的讨论。这个想法是Barrier 类提供了一种相当简单的方法来实现代理IEnumerable，每个Func<IEnumerable<T>, S>s 都可以使用它，而代理只使用一次真正的IEnumerable。特别是，实现只消耗原始可枚举中绝对必要的部分（即，它满足上面提到的额外要求）。

代理是：

class BarrierEnumerable<T> : IEnumerable<T>
{
    private Barrier barrier;
    private bool moveNext;
    private readonly Func<T> src;

    public BarrierEnumerable(IEnumerator<T> enumerator)
    {
        src = () => enumerator.Current;
    }

    public Barrier Barrier
    {
        set => barrier = value;
    }

    public bool MoveNext
    {
        set => moveNext = value;
    }

    public IEnumerator<T> GetEnumerator()
    {
        try
        {
            while (true)
            {
                barrier.SignalAndWait();
                if (moveNext)
                {
                    yield return src();
                }
                else
                {
                    yield break;
                }
            }
        }
        finally
        {
            barrier.RemoveParticipant();
        }
    }

    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}

我们可以结合两个消费者

public static T Combine<S, T1, T2, T>(this IEnumerable<S> source,
    Func<IEnumerable<S>, T1> coenumerable1,
    Func<IEnumerable<S>, T2> coenumerable2,
    Func<T1, T2, T> resultSelector)
{
    using var ss = source.GetEnumerator();
    var enumerable1 = new BarrierEnumerable<S>(ss);
    var enumerable2 = new BarrierEnumerable<S>(ss);
    using var barrier = new Barrier(2, _ => enumerable1.MoveNext = enumerable2.MoveNext = ss.MoveNext());
    enumerable2.Barrier = enumerable1.Barrier = barrier;
    using var t1 = Task.Run(() => coenumerable1(enumerable1));
    using var t2 = Task.Run(() => coenumerable2(enumerable2));
    return resultSelector(t1.Result, t2.Result);
}

GitHub 存储库有几个使用上述代码的示例，以及一些简短的设计讨论（包括限制）。