缓冲LINQ查询

Question

最终编辑:

我选择了Timothy的答案,但如果你想要一个利用C#yield语句的可行实现,请检查Eamon的答案:https://stackoverflow.com/a/19825659/145757

---

默认情况下,LINQ查询是延迟流式传输的.

ToArray/ ToList给予完全缓冲,但首先他们渴望,其次可能需要相当长的时间来完成无限序列.

有没有办法将这两种行为结合起来:生成时动态流式传输和缓冲值,以便下一次查询不会触发已经查询过的元素的生成.

这是一个基本用例:

static IEnumerable<int> Numbers
{
    get
    {
        int i = -1;

        while (true)
        {
            Console.WriteLine("Generating {0}.", i + 1);
            yield return ++i;
        }
    }
}

static void Main(string[] args)
{
    IEnumerable<int> evenNumbers = Numbers.Where(i => i % 2 == 0);

    foreach (int n in evenNumbers)
    {
        Console.WriteLine("Reading {0}.", n);
        if (n == 10) break;
    }

    Console.WriteLine("==========");

    foreach (int n in evenNumbers)
    {
        Console.WriteLine("Reading {0}.", n);
        if (n == 10) break;
    }
}

这是输出:

Generating 0.
Reading 0.
Generating 1.
Generating 2.
Reading 2.
Generating 3.
Generating 4.
Reading 4.
Generating 5.
Generating 6.
Reading 6.
Generating 7.
Generating 8.
Reading 8.
Generating 9.
Generating 10.
Reading 10.
==========
Generating 0.
Reading 0.
Generating 1.
Generating 2.
Reading 2.
Generating 3.
Generating 4.
Reading 4.
Generating 5.
Generating 6.
Reading 6.
Generating 7.
Generating 8.
Reading 8.
Generating 9.
Generating 10.
Reading 10.

生成代码被触发22次.

我希望它被触发11次,这是第一次迭代枚举.

然后第二次迭代将受益于已生成的值.

它会是这样的:

IEnumerable<int> evenNumbers = Numbers.Where(i => i % 2 == 0).Buffer();

对于那些熟悉Rx的人来说,这是一种类似于a的行为ReplaySubject.

Answer 1

IEnumerable<T>.Buffer() 扩展方法

public static EnumerableExtensions
{
    public static BufferEnumerable<T> Buffer(this IEnumerable<T> source)
    {
        return new BufferEnumerable<T>(source);
    }
}

public class BufferEnumerable<T> : IEnumerable<T>, IDisposable
{
    IEnumerator<T> source;
    List<T> buffer;
    public BufferEnumerable(IEnumerable<T> source)
    {
        this.source = source.GetEnumerator();
        this.buffer = new List<T>();
    }
    public IEnumerator<T> GetEnumerator()
    {
        return new BufferEnumerator<T>(source, buffer);
    }
    public void Dispose()
    {
        source.Dispose()
    }
}

public class BufferEnumerator<T> : IEnumerator<T>
{
    IEnumerator<T> source;
    List<T> buffer;
    int i = -1;
    public BufferEnumerator(IEnumerator<T> source, List<T> buffer)
    {
        this.source = source;
        this.buffer = buffer;
    }
    public T Current
    {
        get { return buffer[i]; }
    }
    public bool MoveNext()
    {
        i++;
        if (i < buffer.Count)
            return true;
        if (!source.MoveNext())
            return false;
        buffer.Add(source.Current);
        return true;
    }
    public void Reset()
    {
        i = -1;
    }
    public void Dispose()
    {
    }
}

用法

using (var evenNumbers = Numbers.Where(i => i % 2 == 0).Buffer())
{
    ...
}

评论

这里的关键点是,IEnumerable<T> source给定的Buffer方法输入只GetEnumerator调用一次,而不管Buffer枚举结果的次数.所有枚举器的结果都Buffer共享相同的源枚举器和内部列表.

Answer 2

您可以使用Microsoft.FSharp.Collections.LazyList<>F#power pack中的类型(是的,来自C#,没有安装F# - 没问题!).它在Nuget包中FSPowerPack.Core.Community.

特别是,您希望调用LazyListModule.ofSeq(...)哪个返回一个LazyList<T>实现IEnumerable<T>并且是惰性和缓存的.

在您的情况下,使用只是...的问题

var evenNumbers = LazyListModule.ofSeq(Numbers.Where(i => i % 2 == 0));
var cachedEvenNumbers = LazyListModule.ofSeq(evenNumbers);

虽然我个人更喜欢var所有这些情况,但请注意,这确实意味着编译时类型将更具体而不仅仅是IEnumerable<>- 而不是这可能是一个缺点.F#非接口类型的另一个优点是它们暴露了一些有效的操作,你无法通过简单的IEnumerables来提高效率,例如LazyListModule.skip.

我不确定是否LazyList是线程安全的,但我怀疑它是.

下面的评论中指出的另一个替代方案(如果你安装了F#)是SeqModule.Cache(命名空间Microsoft.FSharp.Collections,它将在GACed程序集FSharp.Core.dll中),它具有相同的有效行为.像其他.NET枚举一样,Seq.cache没有尾部(或跳过)运算符,您可以有效地链接.

线程安全:与此问题的其他解决方案不同,Seq.cache是线程安全的,因为您可以并行运行多个枚举器(每个枚举器不是线程安全的).

性能我做了一个快速的基准测试,并且LazyList可枚举的开销至少是SeqModule.Cache变体的4倍,后者的开销至少是自定义实现答案的三倍.因此,虽然F#变体可行,但它们并不是那么快.请注意,与可执行的(例如)I/O或任何非平凡的计算相比,3-12倍的速度仍然不是很慢,因此这在大多数情况下可能无关紧要,但保留在心神.

TL; DR如果您需要一个高效的,线程安全的缓存可枚举,请使用SeqModule.Cache.

Answer 3

在Eamon上面的答案的基础上,这是另一个功能解决方案(没有新的类型),也适用于同时评估.这表明一般模式(具有共享状态的迭代)是此问题的基础.

首先我们定义一个非常通用的帮助器方法,这意味着我们可以在C#中模拟匿名迭代器的缺失特性:

public static IEnumerable<T> Generate<T>(Func<Func<Tuple<T>>> generator)
{
    var tryGetNext = generator();
    while (true)
    {
        var result = tryGetNext();
        if (null == result)
        {
            yield break;
        }
        yield return result.Item1;
    }
}

生成就像一个有状态的聚合器.它接受一个返回初始状态的函数,以及一个在其中具有匿名性的生成器函数yield return(如果在C#中允许它).返回的状态initialize意味着是枚举,而更全局的状态(在所有枚举之间共享)可以由调用者维护,例如在闭包变量中生成,如下所示.

现在我们可以将它用于"缓冲的可枚举"问题:

public static IEnumerable<T> Cached<T>(IEnumerable<T> enumerable)
{
    var cache = new List<T>();
    var enumerator = enumerable.GetEnumerator();

    return Generate<T>(() =>
    {
        int pos = -1;
        return () => {
            pos += 1;
            if (pos < cache.Count())
            {
                return new Tuple<T>(cache[pos]);
            }
            if (enumerator.MoveNext())
            {
                cache.Add(enumerator.Current);
                return new Tuple<T>(enumerator.Current);
            }
            return null;
        };
    });
}

Answer 4

我希望这个答案结合了sinelaw答案的简洁和清晰,以及对Timothy答案的多个枚举的支持:

public static IEnumerable<T> Cached<T>(this IEnumerable<T> enumerable) {
    return CachedImpl(enumerable.GetEnumerator(), new List<T>());
}

static IEnumerable<T> CachedImpl<T>(IEnumerator<T> source, List<T> buffer) {
    int pos=0;
    while(true) {
        if(pos == buffer.Count) 
            if (source.MoveNext()) 
                buffer.Add(source.Current); 
            else 
                yield break;
        yield return buffer[pos++];
    }
}

关键思想是使用yield return语法来实现一个简短的可枚举实现,但是您仍然需要一个状态机来决定是否可以从缓冲区获取下一个元素,或者是否需要检查基础枚举器.

限制: 这不会尝试是线程安全的,也不会处理底层的枚举器(一般来说,这是非常棘手的,因为只要仍然可以使用任何缓存的enumerabl,底层的未缓存枚举器必须保持不存在).