Python 3 asyncio 和 GIL（如何使用所有 cpu 内核 - 除 ProcessPoolExecutor 之外的任何其他选项）？

Question

如何使用所有 cpu 内核进行 asyncio - 除了 ProcessPoolExecutor 之外的任何其他选项？

我认为 asyncio 无法打破 GIL 限制（也许我错了），因此程序的执行速度将比踩踏版本快，但会在一个核心上执行。

我研究了一些例子，我发现一种方法是多处理和 ProcessPoolExecutor。

https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

# 3. Run in a custom process pool:
with concurrent.futures.ProcessPoolExecutor() as pool:
    result = await loop.run_in_executor(
        pool, cpu_bound)
    print('custom process pool', result)

这很好，但需要在进程之间进行“pickle”，因此需要一些开销并对传递的参数进行一些优化以减少“pickle”序列化。

使用上面这个简单的模式，我写了这样的测试代码（如果你不喜欢它，你可以跳过这段代码阅读，因为它和以前一样）。顺便说一句，这是我解析文件问题的最快解决方案。这部分代码不是整个程序。

def _match_general_and_specific_file_chunk(file_name):
    with codecs.open(file_name, encoding='utf8') as f:
        while True:
            lines = f.readlines(sizehint=10000)
            if not lines:
                break
            for line in lines:
                general_match = RE_RULES.match(line)
                if general_match:
                    specific_match = RULES[general_match.lastindex].match(line)
                    groups = list(specific_match.groups())
                    continue


async def _async_process_executor_match_general_and_specific_read_lines_with_limit_file_chunk():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        futures = []
        for file_name in get_file_names():
            future = loop.run_in_executor(pool, _match_general_and_specific_file_chunk, file_name)
            futures.append(future)
        await asyncio.gather(*futures)


def async_process_executor_match_general_and_specific_read_lines_with_limit_file_chunk():
    asyncio.run(_async_process_executor_match_general_and_specific_read_lines_with_limit_file_chunk())

Answer 1

如何使用所有 cpu 内核进行 asyncio - 除了 ProcessPoolExecutor 之外的任何其他选项？

Asyncio 是不适合这项工作的工具，因为它是专门为管理 IO 绑定程序的状态而设计的（您可以将其视为 Twisted 的继承者）。

要并行执行受 CPU 限制的代码，您将需要由线程或进程提供的操作系统级并发性。在 Python 中，最方便的方法是concurrent.futures模块，首先是类喜欢ThreadPoolExecutor和ProcessPoolExecutor来自的类。您只需要将submit()工作交给执行人并wait完成生成的期货。

如果你想避免酸洗的开销，有两种方法：

• 使用进程，并利用共享内存、映射内存或管理器对象在控制进程和工作进程之间共享状态。

• 使用线程，但在执行 CPU 密集型工作时调用内部释放 GIL 的代码。一些软件包已经这样做了，例如hashlib std 模块或numpy 的许多部分，但是如果您的需要没有被这些软件包覆盖，那对您没有帮助。在这种情况下，您可能需要编写一个新的 C 扩展，请参阅文档以了解有关如何临时发布 GIL 以及何时可以安全发布的详细信息。