python - 通过超时取消异步迭代器

Question

我有一个使用 asyncio 运行的进程，它应该永远运行。

我可以使用 ProcessIterator 与该进程交互，ProcessIterator 可以(此处省略)将数据发送到标准输入并从标准输出获取。

我可以使用 async for fd, data in ProcessIterator(...): 访问数据。

现在的问题是这个异步迭代器的执行必须有时间限制。如果时间用完，则调用 timeout() 函数，但异常并非源自 __anext__ 函数以通知超时。

如何在异步迭代器中引发此异常？我找不到调用 awaitable.throw(something) 或类似方法的方法。

class ProcessIterator:
    def __init__(self, process, loop, run_timeout):
        self.process = process
        self.loop = loop

        self.run_timeout = run_timeout

        # set the global timer
        self.overall_timer = self.loop.call_later(
            self.run_timeout, self.timeout)

    def timeout(self):
        # XXX: how do i pass this exception into the iterator?
        raise ProcTimeoutError(
            self.process.args,
            self.run_timeout,
            was_global,
        )

    async def __aiter__(self):
        return self

    async def __anext__(self):    
        if self.process.exited:
            raise StopAsyncIteration()

        else:
            # fetch output from the process asyncio.Queue()
            entry = await self.process.output_queue.get()
            if entry == StopIteration:
                raise StopAsyncIteration()

            return entry

现在异步迭代器的用法大致是:

async def test_coro(loop):
    code = 'print("rofl"); time.sleep(5); print("lol")'

    proc = Process([sys.executable, '-u', '-c', code])

    await proc.create()

    try:
        async for fd, line in ProcessIterator(proc, loop, run_timeout=1):
            print("%d: %s" % (fd, line))

    except ProcessTimeoutError as exc:
        # XXX This is the exception I'd like to get here! How can i throw it?
        print("timeout: %s" % exc)

    await proc.wait()

tl;dr:我怎样才能抛出一个定时异常，使其源自异步迭代器？

Answer 1

编辑:添加解决方案 2

解决方案一:

timeout() 回调是否可以将 ProcTimeoutError 异常存储在实例变量中？然后 __anext__() 可以检查实例变量并在设置时引发异常。

class ProcessIterator:
    def __init__(self, process, loop, run_timeout):
        self.process = process
        self.loop = loop
        self.error = None

        self.run_timeout = run_timeout

        # set the global timer
        self.overall_timer = self.loop.call_later(
            self.run_timeout, self.timeout)

    def timeout(self):
        # XXX: set instance variable
        self.error = ProcTimeoutError(
                         self.process.args,
                         self.run_timeout,
                         was_global
                     )

    async def __aiter__(self):
        return self

    async def __anext__(self): 
        # XXX: if error is set, then raise the exception
        if self.error:
            raise self.error

        elif self.process.exited:
            raise StopAsyncIteration()

        else:
            # fetch output from the process asyncio.Queue()
            entry = await self.process.output_queue.get()
            if entry == StopIteration:
                raise StopAsyncIteration()

            return entry

解决方案 2:

将异常放入 process.output_queue。

....
def timeout(self):
    # XXX: set instance variable
    self.process.ouput_queue.put(ProcTimeoutError(
                                     self.process.args,
                                     self.run_timeout,
                                     was_global
                                 ))

....

# fetch output from the process asyncio.Queue()
entry = await self.process.output_queue.get()
if entry == StopIteration:
    raise StopAsyncIteration()

elif entry = ProcTimeoutError:
    raise entry
....

如果队列中可能有条目，请使用优先队列。为 ProcTimeoutError 分配比其他条目更高的优先级，例如，(0, ProcTimeoutError) 与 (1, other_entry)。