Python 多线程/多处理和限制 CPU 核心关联性

在Python中，您可以创建新的线程和进程来运行给定的任务multiprocessing.Pool, multiprocessing.ThreadPool, concurrent.futures.ProcessPoolExecutor, and concurrent.futures.ThreadPoolExecutor.

默认情况下，这些线程/进程以其父进程相同的 CPU 核心关联性运行，即所有可用核心/线程。

在 Linux/Unix 系统上，可以使用以下命令更改 CPU 核心关联性os.sched_setaffinity(pid, mask)。问题是这仅限于某些 Linux/Unix 系统。

有的是psutilpython 库公开了设置 CPU 核心关联性的能力psutil.Process().cpu_affinity(CPUS) where CPUS是一个整数列表，标识进程应使用哪些 CPU 核心/线程，从 0 开始。

问题在于，通常操作系统 CPU 调度程序可以处理给定进程应使用哪个核心/线程的挑选和选择，而不是让最终用户决定使用哪些 CPU 核心/线程。

我的问题是是否可以创建线程/进程池并限制每个实例使用 X 个 CPU 核心/线程，但不限制它们的确切核心关联性？

例如，如果我有16核的PC并想创建4个进程，我可以创建一个multiprocessing.Pool(processes=4)目的。现在，如果我希望这 4 个子进程中的每一个都被限制为仅使用 2 个 CPU 核心，我将不得不使用psutil抢先选择 2 个 CPU 核心并将它们分配给该进程，从可用的 CPU 核心列表中删除这 2 个 CPU 核心，并对所有 4 个进程重复该过程。

但这并不理想，如果我给一个进程提供系统中两个最弱的核心会怎么样？或者，如果这 2 个核心在物理上距离较远（例如现代多芯片 AMD Ryzen CPU 或双 CPU 插槽系统的情况）。

我想让操作系统自动为每个进程调度 2 个核心，并根据需要调整它们，而不必为每个进程手动设置和取消设置 CPU 核心。

有没有办法在 Python 中完成这个任务？

前段时间我有类似的需求，所以我编写了一个 CPUResourceManager 类来跟踪我分配给进程的核心。在这里你可以调用get_processors方法来获取您将使用的核心列表。您可以使用 PSUTIL 设置核心关联性，就像您已经在做的那样。当您的过程完成后，使用以下命令返回核心free_processors method.

from typing import NamedTuple
from enum import Enum


class CPUResponse(NamedTuple):
    """
    This is the response when the CPUResourceManager is asked for some cores.
    """
    success: bool     # whether or not there are enough cores available
    processors: list  # the list of processors to be used by the process


class ProcState(Enum):
    """
    Represents the state of a processor core.  This is only 
    represents what we are having the cores do.  Not what other unrelated
    processes on the machine are doing
    """
    idle = 0
    busy = 1


class CPUResourceManager:
    def __init__(self, cpu_count=max(cpu_count() - 2, 1)) -> None:

        self.cpu_count = cpu_count

        self.processors = {i: ProcState.idle for i in range(self.cpu_count)}

    def cpu_avalaible_count(self):
        available = [
            p for p, state in self.processors.items() if state == ProcState.idle
        ]
        return len(available)

    def get_processors(self, count=1):
        """Get some available cores"""
        available = [
            p for p, state in self.processors.items() if state == ProcState.idle
        ]

        if len(available) >= count:
            cpus = available[:count]
            for p in cpus:
                self.processors[p] = ProcState.busy
            return CPUResponse(True, available[:count])
        else:
            return CPUResponse(False, [])

    def free_processors(self, processors: list):
        """return the cores when you are done"""
        for p in processors:
            if p in self.processors:
                self.processors[p] = ProcState.idle
            else:
                # manager was likely resized and this processor
                # should no longer be considered available
                pass