缓存生成器

Question

最近的一个类似问题 (isinstance(foo, types.GeneratorType) or inspect.isgenerator(foo)?) 让我对如何通用地实现它感到好奇。

实际上，拥有一个生成器类型的对象似乎是一件普遍有用的事情，该对象将在第一次通过时进行缓存（如itertools.cycle），报告 StopIteration，然后在下一次通过时从缓存中返回项目，但如果对象不是生成器（即固有地支持 O(1) 查找的列表或字典），则不要缓存，并且具有相同的行为，但对于原始列表。

可能性：

1) 修改itertools.cycle。它看起来像这样：

def cycle(iterable):
    saved = []
    try: 
         saved.append(iterable.next())
         yield saved[-1]
         isiter = True
    except:
         saved = iterable
         isiter = False
    # cycle('ABCD') --> A B C D A B C D A B C D ...
    for element in iterable:
        yield element
        if isiter: 
            saved.append(element)

     # ??? What next?

如果我可以重新启动生成器，那就太完美了——我可以发回一个 StopIteration，然后在下一个 gen.next() 上，返回条目 0，即“ABCD StopIteration ABCD StopIteration”，但它看起来不像这实际上是可能的。

其次是一旦 StopIteration 被击中，然后保存有一个缓存。但看起来没有任何方法可以访问内部 saved[] 字段。也许这是一个班级版本？

2) 或者我可以直接传入列表：

def cycle(iterable, saved=[]):
    saved.clear()
    try: 
         saved.append(iterable.next())
         yield saved[-1]
         isiter = True
    except:
         saved = iterable
         isiter = False
    # cycle('ABCD') --> A B C D A B C D A B C D ...
    for element in iterable:
        yield element
        if isiter: 
            saved.append(element)

mysaved = []
myiter = cycle(someiter, mysaved)

但这看起来很糟糕。而在 C/++ 中，我可以传入一些引用，并将实际引用更改为已保存以指向可迭代 - 你实际上不能在 python 中这样做。所以这甚至行不通。

其他选择？

编辑：更多数据。 CachingIterable 方法似乎太慢而无法有效，但它确实将我推向了一个可行的方向。它比天真的方法（转换为列出我自己）稍慢，但如果它已经是可迭代的，它似乎不会受到打击。

一些代码和数据：

def cube_generator(max=100):
    i = 0
    while i < max:
        yield i*i*i
        i += 1

# Base case: use generator each time
%%timeit
cg = cube_generator(); [x for x in cg]
cg = cube_generator(); [x for x in cg]
cg = cube_generator(); [x for x in cg]
10000 loops, best of 3: 55.4 us per loop

# Fastest case: flatten to list, then iterate
%%timeit
cg = cube_generator()
cl = list(cg)
[x for x in cl]
[x for x in cl]
[x for x in cl]
10000 loops, best of 3: 27.4 us per loop

%%timeit
cg = cube_generator()
ci2 = CachingIterable(cg)
[x for x in ci2]
[x for x in ci2]
[x for x in ci2]
1000 loops, best of 3: 239 us per loop

# Another attempt, which is closer to the above
# Not exactly the original solution using next, but close enough i guess
class CacheGen(object):
    def __init__(self, iterable):
        if isinstance(iterable, (list, tuple, dict)):
            self._myiter = iterable
        else:
            self._myiter = list(iterable)
    def __iter__(self):
        return self._myiter.__iter__()
    def __contains__(self, key):
        return self._myiter.__contains__(key)
    def __getitem__(self, key):
        return self._myiter.__getitem__(key)

%%timeit
cg = cube_generator()
ci = CacheGen(cg)
[x for x in ci]
[x for x in ci]
[x for x in ci]
10000 loops, best of 3: 30.5 us per loop

# But if you start with a list, it is faster
cg = cube_generator()
cl = list(cg)
%%timeit
[x for x in cl]
[x for x in cl]
[x for x in cl]
100000 loops, best of 3: 11.6 us per loop

%%timeit
ci = CacheGen(cl)
[x for x in ci]
[x for x in ci]
[x for x in ci]
100000 loops, best of 3: 13.5 us per loop

任何更快的食谱可以更接近“纯”循环？

Answer 1

基于此评论：

我的意图是，只有当用户知道他想对“可迭代”进行多次迭代，但不知道输入是生成器还是可迭代时，才使用此方法。这可以让您忽略这种区别，同时不会损失（很多）性能。

这个简单的解决方案正是这样做的：

def ensure_list(it):
    if isinstance(it, (list, tuple, dict)):
        return it
    else:
        return list(it)

现在 ensure_list(a_list) 实际上是一个无操作 - 两个函数调用 - 而 ensure_list(a_generator) 会将其转换为一个列表并返回它，结果证明这比任何其他方法都快。

Answer 2

你想要的不是一个迭代器，而是一个可迭代对象。迭代器只能遍历其内容一次。你想要一个带有迭代器的东西，然后你可以在上面迭代多次，从迭代器中产生相同的值，即使迭代器不记得它们，比如生成器。然后，只需对那些不需要缓存的输入进行特殊处理。这是一个非线程安全的示例（编辑：为提高效率而更新）：

import itertools
class AsYouGoCachingIterable(object):
    def __init__(self, iterable):
        self.iterable = iterable
        self.iter = iter(iterable)
        self.done = False
        self.vals = []

    def __iter__(self):
        if self.done:
            return iter(self.vals)
        #chain vals so far & then gen the rest
        return itertools.chain(self.vals, self._gen_iter())

    def _gen_iter(self):
        #gen new vals, appending as it goes
        for new_val in self.iter:
            self.vals.append(new_val)
            yield new_val
        self.done = True

还有一些时间安排：

class ListCachingIterable(object):
    def __init__(self, obj):
        self.vals = list(obj)

    def __iter__(self):
        return iter(self.vals)

def cube_generator(max=1000):
    i = 0
    while i < max:
        yield i*i*i
        i += 1

def runit(iterable_factory):
    for i in xrange(5):
        for what in iterable_factory():
            pass

def puregen():
    runit(lambda: cube_generator())
def listtheniter():
    res = list(cube_generator())
    runit(lambda: res)
def listcachingiterable():
    res = ListCachingIterable(cube_generator())
    runit(lambda: res)
def asyougocachingiterable():
    res = AsYouGoCachingIterable(cube_generator())
    runit(lambda: res)

结果是：

In [59]: %timeit puregen()
1000 loops, best of 3: 774 us per loop

In [60]: %timeit listtheniter()
1000 loops, best of 3: 345 us per loop

In [61]: %timeit listcachingiterable()
1000 loops, best of 3: 348 us per loop

In [62]: %timeit asyougocachingiterable()
1000 loops, best of 3: 630 us per loop

因此，就类而言，最简单的方法ListCachingIterable 与手动执行list 一样有效。 “as-you-go”变体的速度几乎是原来的两倍，但如果您不使用整个列表，则具有优势，例如假设您只寻找第一个超过 100 的立方体：

def first_cube_past_100(cubes):
    for cube in cubes:
        if cube > 100:
            return cube
    raise Error("No cube > 100 in this iterable")

然后：

In [76]: %timeit first_cube_past_100(cube_generator())
100000 loops, best of 3: 2.92 us per loop

In [77]: %timeit first_cube_past_100(ListCachingIterable(cube_generator()))
1000 loops, best of 3: 255 us per loop

In [78]: %timeit first_cube_past_100(AsYouGoCachingIterable(cube_generator()))
100000 loops, best of 3: 10.2 us per loop

Answer 3

刚刚创建了一个library，它正好解决了这个问题——支持对返回迭代器的函数进行缓存：

from typing import *
from cacheable_iter import iter_cache

@iter_cache
def iterator_function(n: int) -> Iterator[int]:
    yield from range(n)

使用示例：

from typing import *
from cacheable_iter import iter_cache

@iter_cache
def my_iter(n: int) -> Iterator[int]:
    print(" * my_iter called")
    for i in range(n):
        print(f" * my_iter step {i}")
        yield i

gen1 = my_iter(4)
print("Creating an iterator...")
print(f"The first value of gen1 is {next(gen1)}")
print(f"The second value of gen1 is {next(gen1)}")

gen2 = my_iter(4)
print("Creating an iterator...")
print(f"The first value of gen2 is {next(gen2)}")
print(f"The second value of gen2 is {next(gen2)}")
print(f"The third value of gen2 is {next(gen2)}")

哪个会打印：

Creating an iterator...
 * my_iter called
 * my_iter step 0
The first value of gen1 is 0
 * my_iter step 1
The second value of gen1 is 1
Creating an iterator...
The first value of gen2 is 0
The second value of gen2 is 1
 * my_iter step 2
The third value of gen2 is 2

还支持缓存等待迭代器和异步迭代器