python - 异步任务意外延迟

Question

我一直在努力学习一些关于 asyncio 的知识，但我遇到了一些意想不到的行为。我设置了一个简单的斐波那契服务器，它支持使用流的多个连接。 fib 计算是递归编写的，因此我可以通过输入大量数字来模拟长时间运行的计算。正如预期的那样，长时间运行的计算会阻塞 I/O，直到长时间运行的计算完成。

不过问题来了。我将斐波那契函数重写为协程。我预计通过从每次递归中产生，控制将回退到事件循环，并且等待 I/O 任务将有机会执行，并且您甚至可以同时运行多个 fib 计算。然而，情况似乎并非如此。

代码如下:

import asyncio

@asyncio.coroutine
def fib(n):
    if n < 1:
        return 1
    a = yield from fib(n-1)
    b = yield from fib(n-2)
    return a + b


@asyncio.coroutine
def fib_handler(reader, writer):
    print('Connection from : {}'.format(writer.transport.get_extra_info('peername')))
    while True:
        req = yield from reader.readline()
        if not req:
            break
        print(req)
        n = int(req)
        result = yield from fib(n)
        writer.write('{}\n'.format(result).encode('ascii'))
        yield from writer.drain()
    writer.close()
    print("Closed")


def server(address):
    loop = asyncio.get_event_loop()
    fib_server = asyncio.start_server(fib_handler, *address, loop=loop)
    fib_server = loop.run_until_complete(fib_server)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print('closing...')
        fib_server.close()
        loop.run_until_complete(fib_server.wait_closed())
        loop.close()


server(('', 25000))

如果您 netcat 到端口 25000 并开始输入数字，则该服务器运行良好。但是，如果您开始长时间运行的计算(例如 35)，则在第一个计算完成之前不会运行其他计算。事实上，甚至不会处理额外的连接。

我知道事件循环正在反馈递归 fib 调用的 yield ，因此控制必须一路下降。但我认为循环会在“蹦床”返回 fib 函数之前处理 I/O 队列中的其他调用(例如生成第二个 fib_handler)。

我确定我一定是误会了什么，或者我忽略了某种错误，但我一辈子都找不到它。

我们将不胜感激您提供的任何见解。

Answer 1

第一个问题是您正在调用 yield from fib(n) fib_handler 内部.包括yield from意味着 fib_handler将阻塞直到调用 fib(n)已完成，这意味着它无法处理您在 fib 期间提供的任何输入在跑。即使您所做的只是 fib 内部的 I/O，您也会遇到这个问题。 .要解决此问题，您应该使用 asyncio.async(fib(n)) (或者最好是 asyncio.ensure_future(fib(n)) ，如果你有足够新的 Python 版本)安排 fib使用事件循环，实际上没有阻塞 fib_handler .从那里，您可以使用 Future.add_done_callback准备就绪后将结果写入客户端:

import asyncio
from functools import partial
from concurrent.futures import ProcessPoolExecutor

@asyncio.coroutine
def fib(n):
    if n < 1:
        return 1
    a = yield from fib(n-1)
    b = yield from fib(n-2)
    return a + b

def do_it(writer, result):
    writer.write('{}\n'.format(result.result()).encode('ascii'))
    asyncio.async(writer.drain())

@asyncio.coroutine
def fib_handler(reader, writer):
    print('Connection from : {}'.format(writer.transport.get_extra_info('peername')))
    executor = ProcessPoolExecutor(4)
    loop = asyncio.get_event_loop()
    while True:
        req = yield from reader.readline()
        if not req:
            break
        print(req)
        n = int(req)
        result = asyncio.async(fib(n))
        # Write the result to the client when fib(n) is done.
        result.add_done_callback(partial(do_it, writer))
    writer.close()
    print("Closed")

也就是说，仅此更改仍不能完全解决问题；虽然它将允许多个客户端同时连接和发出命令，但单个客户端仍将获得同步行为。发生这种情况是因为当您调用 yield from coro()直接在协程函数上，控制权直到 coro() 才交还给事件循环(或由 coro 调用的另一个协程)实际上执行一些非阻塞 I/O。否则，Python 只会执行 coro没有屈服的控制。这是一个有用的性能优化，因为当您的协程实际上不会执行阻塞 I/O 时将控制权交给事件循环是浪费时间，尤其是考虑到 Python 的高函数调用开销。

在你的例子中，fib从不做任何 I/O，所以一旦你调用 yield from fib(n-1) fib 内部本身，事件循环在完成递归之前永远不会再次运行，这将阻止 fib_handler从读取客户端的任何后续输入直到调用 fib已经完成了。将您的所有 调用包装到fib在 asyncio.async保证每次你创建 yield from asyncio.async(fib(...)) 时都会将控制权交给事件循环称呼。当我进行此更改时，除了使用 asyncio.async(fib(n))在 fib_handler ，我能够同时处理来自单个客户端的多个输入。这是完整的示例代码:

import asyncio
from functools import partial
from concurrent.futures import ProcessPoolExecutor

@asyncio.coroutine
def fib(n):
    if n < 1:
        return 1
    a = yield from fib(n-1)
    b = yield from fib(n-2)
    return a + b

def do_it(writer, result):
    writer.write('{}\n'.format(result.result()).encode('ascii'))
    asyncio.async(writer.drain())

@asyncio.coroutine
def fib_handler(reader, writer):
    print('Connection from : {}'.format(writer.transport.get_extra_info('peername')))
    executor = ProcessPoolExecutor(4)
    loop = asyncio.get_event_loop()
    while True:
        req = yield from reader.readline()
        if not req:
            break
        print(req)
        n = int(req)
        result = asyncio.async(fib(n))
        result.add_done_callback(partial(do_it, writer))
    writer.close()
    print("Closed")

客户端的输入/输出:

dan@dandesk:~$ netcat localhost 25000
35 # This was input
4  # This was input
8  # output
24157817 # output

现在，即使这可行，我也不会使用此实现，因为它在单线程程序中执行大量 CPU 绑定(bind)工作，该程序还希望在同一线程中提供 I/O 服务。这不会很好地扩展，也不会有理想的性能。相反，我建议使用 loop.run_in_executor运行对 fib 的调用在后台进程中，它允许 asyncio 线程满负荷运行，还允许我们将调用扩展到 fib跨多个核心:

import asyncio
from functools import partial
from concurrent.futures import ProcessPoolExecutor

def fib(n):
    if n < 1:
        return 1
    a = fib(n-1)
    b = fib(n-2)
    return a + b

def do_it(writer, result):
    writer.write('{}\n'.format(result.result()).encode('ascii'))
    asyncio.async(writer.drain())

@asyncio.coroutine
def fib_handler(reader, writer):
    print('Connection from : {}'.format(writer.transport.get_extra_info('peername')))
    executor = ProcessPoolExecutor(8)  # 8 Processes in the pool
    loop = asyncio.get_event_loop()
    while True:
        req = yield from reader.readline()
        if not req:
            break
        print(req)
        n = int(req)
        result = loop.run_in_executor(executor, fib, n)
        result.add_done_callback(partial(do_it, writer))
    writer.close()
    print("Closed")