用于测试发电机是否"正在运行"和/或已经用尽的标志

Question

摘要

我所追求的是一种确定发电机是否"正在运行"的方法.

细节

我对"当前正在运行"的定义是:

1. 生成器代码已经开始执行(即,next或者.send(None)某些其他迭代操作,例如for循环,先前已在生成器上执行,AND
2. 发电机先前没有耗尽/关闭(再次注意,对于上述发电机,.throw(some_error)由于一些新的回路启动而被捕获并且不会耗尽发电机;没有回路或yield from处理错误通常会耗尽发电机).

我知道没有办法(或者至少是一种简单易行的方法)在生成器中"窥视"并查明它是否会StopIteration在下一次迭代中升起.这与发电机的性质是对立的,而不是我所追求的.

例

假设我有一些生成器,g由一些生成器函数创建,f:

def f():
    '''Lots of 1s.'''
    print('I am now running.')
    while True:
        try:
            signal = yield 1
            if signal:
                break
        except GeneratorExit:
            raise #  important so that closing the generator doesn't cause a RuntimeError
        except Exception:
            print('caught exception!')

我们可以这样"初始化"生成器:

>>> g = f()
>>> next(g)
I am now running.
1

# OR:

>>> g.send(None) #  equivalent to next(g)
I am now running.
1

如果我将任何真相发送到发生器中,StopIteration则会引发错误(例如g.send('foo')).或者,如果我打电话g.close(),然后做next(g),我也会得到StopIteration.如果我抛出任何错误g.throw(e),发电机将继续.这一切都符合预期.

我希望能够做的是以下内容,它显示此生成器是否正在运行:

>>> g=f()
>>> g.running
False
>>> g.send(None)
I am now running.
1
>>> g.running
True
>>> g.close()
>>> g.running
False

我在答案中提供了一种可能的方法.但是,我认为必须有更好的方法.

Answer 1

inspect.getgeneratorstate 告诉你发电机的状态:

>>> import inspect
... 
... def gen():
...     yield 1
...     
... g = gen()
...
>>> inspect.getgeneratorstate(g)
'GEN_CREATED'
>>> next(g)
1
>>> inspect.getgeneratorstate(g)
'GEN_SUSPENDED'
>>> next(g)
>>> inspect.getgeneratorstate(g)
'GEN_CLOSED'