使用 asyncio ProactorEventLoop 时如何分配线程池

Posted 2023-02-21

【中文标题】使用 asyncio ProactorEventLoop 时如何分配线程池

【英文标题】：How to allocated thread pool while using asyncio ProactorEventLoop

【发布时间】：2020-06-14 08:03:59

【问题描述】：

我目前在 Python 3.7 中使用 asyncio，并使用 asyncio.start_server() 函数编写 TCP 服务器 参考这个例子：https://docs.python.org/3/library/asyncio-stream.html

还可以尝试使用“I/O 完成端口”(IOCP) 的 asyncio.ProactorEventLoop

根据这个微软官方文档https://docs.microsoft.com/en-ca/windows/win32/fileio/i-o-completion-ports，它提到使用带有预分配线程池的 I/O 完成端口，但我找不到可以分配线程数的地方

在哪里可以分配线程池中的线程数？ 任何人都可以在这里帮助我吗？非常感谢！

【问题讨论】：

为什么要关心 OS 提供的线程池中的线程数？我怀疑您将无法从 Python 代码访问这些线程。

因为我希望模型线程处理更多请求。比如在一秒钟内处理数千个请求因为c++可以在CreateIoCompletionPort函数中设置NumberOfConcurrentThreads参数

你知道 asyncio 是单线程的吗？增加完成端口使用的线程数不会改变这一点。

我认为 asyncio.start_server() 只使用单线程，但 asyncio 库使用协程，我们几乎拥有多线程的所有优点，而无需实际使用多线程

没错，这也是完成端口使用的并发线程数不太可能影响性能的原因。我强烈怀疑用户代码需要使用这些线程才能真正发挥作用，Python+asyncio 将执行运行事件循环的线程中的所有内容。您可能需要找到其他方法来提高程序的性能。

【参考方案1】：

首先是关于 I/O 完成端口 (iocp) 和线程池的一般信息。我们这里有 2 个选项：

---

自己创造一切：

通过CreateIoCompletionPort（或NtCreateIoCompletion）自行创建iocp。

通过自行创建线程，将调用GetQueuedCompletionStatus（或NtRemoveIoCompletion）。

您需要通过NtSetInformationFile 与FileCompletionInformation 和FILE_COMPLETION_INFORMATION 或通过CreateIoCompletionPort 再次绑定到您的iocp 的每个文件（此win32 api 结合了NtCreateIoCompletion 和@ 的功能987654349@)。

---

使用系统iocp(s)和线程池。

系统（ntdll.dll）在进程启动时创建默认线程池（现在它命名为TppPoolpGlobalPool）。你可以控制这个池的周数。你不能得到它直接指针PTP_POOL。存在未记录的TpSetDefaultPoolMaxThreads（用于设置此池中的最大线程数）但没有最小。

如果需要 - 您可以通过 CreateThreadpool 函数创建额外的线程池。

创建新线程池后，您可以（但不应该！）调用SetThreadpoolThreadMaximum 指定池可以分配的最大线程数，并调用SetThreadpoolThreadMinimum 指定池中可用的最小线程数。

线程池维护一个 I/O 完成端口。 iocp 在调用 CreateThreadpool 内部创建 - 我们无法直接访问它。

所以最初在进程中存在一个全局/默认线程池 (TppPoolpGlobalPool) 和 iocp (windows 10 用于并行加载程序创建一个线程池 LdrpThreadPool em> 但这当然只供内部使用 - 在加载 DDL 时）

最后你通过调用CreateThreadpoolIo将self文件绑定到iocp

请注意这里的 msdn 文档是错误的 -

创建一个新的 I/O 完成对象。

真的是CreateThreadpoolIo 函数不 创建新的 I/O 完成对象 - 它仅在调用 CreateThreadpool 中创建。此 api 将文件（不是句柄而是文件！）绑定到与池关联的 I/O 完成对象。去哪个池子？查找最后一个参数 - 指向 TP_CALLBACK_ENVIRON 的可选指针。

你可以通过下一种方式指定一个线程池——分配回调环境，为它调用InitializeThreadpoolEnvironment，然后调用SetThreadpoolCallbackPool。

如果你不指定线程池，全局线程池将在调用CreateThreadpoolIo 中使用 - 所以文件将绑定到默认/全局进程iocp

在这种情况下，您不需要自己调用GetQueuedCompletionStatus（或NtRemoveIoCompletion） - 系统会从池中为您执行此操作。然后调用你的IoCompletionCallback回调函数，你在CreateThreadpoolIo调用中传递给系统

---

我们也可以通过BindIoCompletionCallback使用系统全局线程池和iocp（ 或RtlSetIoCompletionCallback) - 它将全局 (TppPoolpGlobalPool) 线程池拥有的 I/O 完成端口与指定的文件句柄相关联。这是旧 api 和案例 2 的变体。这里我们不能使用自定义轮询 - 只能处理全局。

---

现在让我们回到具体的 Python 代码。它使用哪种情况？它是自己创建 iocp 和线程池吗？还是它使用系统线程池？如果使用系统 - 使用由 CreateThreadpool 分配的全局或自定义线程池？如果你不知道这一点 - 这里什么也做不了。即使知道..或者这个库有特殊的api/接口（或者在python中是如何调用的）来控制它（如果使用了self或custom pool），或者你只能按原样使用它。真的很难决定你在池中真正需要多少线程