python - asyncio 超时终止子进程

Question

我有一个必须执行一些 shell 命令的脚本。但是，如果命令花费太长时间才能完成，则必须强制终止它。考虑以下代码片段:

import asyncio, random

q = asyncio.Queue()

MAX_WAIT = 5

@asyncio.coroutine
def blocking_task(sec):
    print('This task will sleep {} sec.'.format(sec))
    create = asyncio.create_subprocess_shell(
       'sleep {s}; echo "Woke up after {s} sec." >> ./tst.log'.format(s=sec),
        stdout=asyncio.subprocess.PIPE)
    proc = yield from create
    yield from proc.wait()

@asyncio.coroutine
def produce():
    while True:
        q.put_nowait(random.randint(3,8))
        yield from asyncio.sleep(0.5 + random.random())

@asyncio.coroutine
def consume():
    while True:
        value = yield from q.get()
        try:
            yield from asyncio.wait_for(blocking_task(value), MAX_WAIT)
        except asyncio.TimeoutError:
            print('~/~ Job has been cancelled !!')
        else:
            print('=/= Job has been done :]')


loop = asyncio.get_event_loop()
asyncio.ensure_future(produce())
asyncio.ensure_future(consume())
loop.run_forever()

此代码产生以下输出:

This task will sleep 4 sec.
=/= Job has been done :]
This task will sleep 8 sec.
~/~ Job has been cancelled !!
This task will sleep 5 sec.
~/~ Job has been cancelled !!

看来它正在按预期工作，如果工作需要很长时间才能完成，工作就会停止。但是，如果我检查日志，我可以确认耗时的任务仍在继续运行，并且实际上并未停止/终止/中止:

Woke up after 4 sec.
Woke up after 8 sec.
Woke up after 5 sec.

我希望日志中应该只有一行，因为其他进程在有机会完成之前必须已中止:

Woke up after 4 sec.

有办法实现我想要的吗？

我什至不确定这里是否需要asyncio，也许也可以使用concurrent.futures。无论哪种方式，任务都是相同的 - 终止任务，这些任务需要太多时间才能完成。

Answer 1

您可以使用Process.terminate :

try:
    yield from proc.wait()
except asyncio.CancelledError:
    proc.terminate()
    raise

或者:

try:
    yield from proc.wait()
finally:
    if proc.returncode is None:
        proc.terminate()

<小时/>

编辑

Why I didn't see asyncio.CancelledError raised in my code?

当asyncio.wait_for(或其他任何内容)取消任务时，它会在相应的协程中抛出CancelledError。这允许协程在必要时执行一些清理(例如使用上下文管理器或 try/finally 子句)。不需要记录此错误，因为它是已取消任务的正常行为。但是在取消任务后尝试等待任务，则会引发 CancelledError。