python - 在asyncio中使用run_in_executor时，事件循环是否在主线程中执行？

Question

我正在尝试使用 asyncio 功能进行多线程执行。
我的理解是循环管理线程并在线程上执行函数。
如果是这种情况，那么在主线程仍在处理时应该不可能启动新函数。
但是，当我运行以下代码时，它会在我睡着的时候开始执行一个新函数。能告诉我原因吗？

引用:https://stackoverflow.com/a/60747799

import asyncio
from concurrent.futures.thread import ThreadPoolExecutor
from time import sleep
import logging

logging.basicConfig(
    level=logging.DEBUG, format="%(asctime)s %(thread)s %(funcName)s %(message)s"
)

def long_task(t):
    """Simulate long IO bound task."""
    logging.info("2. t: %s", t)
    sleep(t)
    logging.info("5. t: %s", t)
    return t ** 2

async def main():
    loop = asyncio.get_running_loop()
    executor = ThreadPoolExecutor(max_workers=2)
    inputs = range(1, 5)
    logging.info("1.")
    futures = [loop.run_in_executor(executor, long_task, i) for i in inputs]
    logging.info("3.")
    sleep(3)
    logging.info("4.")
    results = await asyncio.gather(*futures)
    logging.info("6.")

if __name__ == "__main__":
    asyncio.run(main())

预期输出

2022-02-08 22:59:08,896 139673219430208 __init__ Using selector: EpollSelector
2022-02-08 22:59:08,896 139673219430208 main 1.
2022-02-08 22:59:08,897 139673194632960 long_task 2. t: 1
2022-02-08 22:59:08,897 139673186240256 long_task 2. t: 2
2022-02-08 22:59:08,897 139673219430208 main 3.
2022-02-08 22:59:09,898 139673194632960 long_task 5. t: 1
2022-02-08 22:59:10,898 139673186240256 long_task 5. t: 2
2022-02-08 22:59:13,400 139673219430208 main 4.
2022-02-08 22:59:09,898 139673194632960 long_task 2. t: 3
2022-02-08 22:59:10,899 139673186240256 long_task 2. t: 4
2022-02-08 22:59:12,902 139673194632960 long_task 5. t: 3
2022-02-08 22:59:14,903 139673186240256 long_task 5. t: 4
2022-02-08 22:59:14,903 139673219430208 main 6.

实际输出

2022-02-08 22:59:08,896 139673219430208 __init__ Using selector: EpollSelector
2022-02-08 22:59:08,896 139673219430208 main 1.
2022-02-08 22:59:08,897 139673194632960 long_task 2. t: 1
2022-02-08 22:59:08,897 139673186240256 long_task 2. t: 2
2022-02-08 22:59:08,897 139673219430208 main 3.
2022-02-08 22:59:09,898 139673194632960 long_task 5. t: 1
2022-02-08 22:59:09,898 139673194632960 long_task 2. t: 3
2022-02-08 22:59:10,898 139673186240256 long_task 5. t: 2
2022-02-08 22:59:10,899 139673186240256 long_task 2. t: 4
2022-02-08 22:59:12,902 139673194632960 long_task 5. t: 3
2022-02-08 22:59:13,400 139673219430208 main 4.
2022-02-08 22:59:14,903 139673186240256 long_task 5. t: 4
2022-02-08 22:59:14,903 139673219430208 main 6.

在 3 和 4 之间，sleep(3) 正在主线程中执行。我明白Threadpool中早先运行的longtask(1)和longtask(2)的结束是在这段时间打印出来的，但是为什么下一个任务在这段时间运行呢？如果 event_loop 在主线程中，则 sleep(3) 不应允许执行新函数。

Answer 1

When using run_in_executor in asyncio, is the event loop executed in the main thread?

是的，它是 - 但是 run_in_executor 将可调用对象提交给执行器，允许它们在没有事件循环帮助的情况下运行。

Between 3 and 4, sleep(3) is being executed in the main thread. I understand that the end of longtask(1) and longtask(2) running earlier in Threadpool is printed during this time, but why is the next task running during this time? If event_loop is in the main thread, then sleep(3) should not allow the execution of the new function.

ThreadPoolExecutor(max_workers=2) 创建一个线程池，最多可扩展到两个工作线程。 run_in_executor 是 Executor.submit 的包装器确保最终结果传播到 asyncio。它的实现可能看起来像这样(actual code 有点 more complex 因为它处理取消和其他问题，但这是要点):

class EventLoop:
    # ...
    def run_in_executor(self, executor, f, *args):
        async_future = self.create_future()
        handle = executor.submit(f, *args)
        def when_done(_):
            self.call_soon_threadsafe(async_future.set_result, handle.result())
        handle.add_done_callback(when_done)
        return async_future

调用 submit 将可调用对象及其参数推送到多线程队列中。池中的工作人员在消耗该队列的无限循环中运行，只有当执行者被告知要执行 shut down 时才会退出。 .

如果您提交的任务多于池中工作人员的数量，额外的任务仍将放置在队列中，等待轮到它们进行处理。 (队列是一个无界 channel ，所以 Executor.submit() 永远不会阻塞。)一旦 worker 完成了一个任务，它就会从队列中请求下一个任务，这就是为什么你需要额外的任务被处决。主线程此时停留在 time.sleep() 中并不重要 - 函数在此之前已提交给执行程序，并且位于队列中，因此工作人员可以找到他们就好了。

---

最后，在普通的异步代码中，异步函数绝不能调用 time.sleep()，它必须改为 await asyncio.sleep()。 (我知道你是故意阻塞运行事件循环的线程，但初学者往往没有意识到这一点，所以需要指出。)