os.sched_getaffinity(0) 与 os.cpu_count()

Question

所以，我知道标题中两种方法的区别，但不知道实际含义。

据我了解：如果您使用的 NUM_WORKERS 数量多于实际可用的核心数量，您将面临性能大幅下降，因为您的操作系统不断地来回切换，试图保持并行。不知道这有多真实，但我在某处从比我聪明的人那里读到了它。

在它的文档中os.cpu_count()说：

返回系统中 CPU 的数量。如果未确定则返回 None。该数字不等于当前进程可以使用的 CPU 数量。可用CPU的数量可以通过len(os.sched_getaffinity(0))获得

因此，我试图弄清楚“系统”指的是一个进程可使用的 CPU 数量是否多于“系统”中的 CPU 数量。

我只是想安全有效地实现multiprocessing.pool功能。所以这是我的问题总结：

以下内容有何实际意义：

NUM_WORKERS = os.cpu_count() - 1
# vs.
NUM_WORKERS = len(os.sched_getaffinity(0)) - 1

这-1是因为我发现，如果我尝试在处理数据时工作，我的系统的延迟会少很多。

Answer 1

这两个函数非常不同，并且NUM_WORKERS = os.sched_getaffinity(0) - 1会立即失败，TypeError因为您尝试从集合中减去整数。虽然os.cpu_count()告诉您系统有多少个核心，但告诉您允许os.sched_getaffinity(pid)某个线程/进程在哪些核心上运行。

os.cpu_count()

os.cpu_count()显示操作系统已知的可用核心数量（虚拟核心）。您很可能拥有这个数量的物理核心的一半。使用比物理核心更多的进程，甚至比虚拟核心更多的进程是否有意义，很大程度上取决于您正在做什么。计算循环越紧密（指令多样性很少，缓存未命中次数很少，...），您就越有可能无法从更多使用的核心（通过使用更多工作进程）中受益，甚至会遇到性能下降。

显然，这还取决于您的系统正在运行什么，因为您的系统试图为系统中的每个线程（作为进程的实际执行单元）提供可用内核上公平的运行时间份额。因此，就应该使用多少工作人员而言，不可能一概而论。但是，例如，如果您有一个紧密的循环并且您的系统处于空闲状态，那么优化的一个很好的起点是

os.cpu_count() // 2 # same as mp.cpu_count() // 2 

...并从那里开始增加。

@Frank Yellin 已经提到过，multiprocessing.Pool使用os.cpu_count()工人数量作为默认值。

os.sched_getaffinity(pid)

os.sched_getaffinity(pid)

返回具有 PID pid 的进程（如果为零，则返回当前进程）所限制的 CPU 集合。

现在 core/cpu/processor/-affinity 是关于允许您的线程（在工作进程内）运行的具体（虚拟）核心。您的操作系统为每个核心提供一个 id，从 0 到（核心数 - 1），并且更改关联性可以限制（“固定”）某个线程允许在哪些实际核心上运行。

至少在 Linux 上，我发现这意味着如果当前没有允许的核心可用，则子进程的线程将不会运行，即使其他不允许的核心处于空闲状态。所以“亲和力”在这里有点误导。

摆弄关联性时的目标是最大限度地减少上下文切换和核心迁移造成的缓存失效。您的操作系统通常具有更好的洞察力，并且已经尝试通过其调度策略来保持缓存“热”，因此除非您知道自己在做什么，否则您不能指望从干扰中轻松获得收益。

默认情况下，亲和力设置为所有核心，对于multiprocessing.Pool，更改它没有太大意义，至少在您的系统处于空闲状态时是这样。

请注意，尽管这里的文档谈到了“进程”，但设置亲和力实际上是每个线程的事情。因此，例如，在“子”线程中为“当前进程（如果为零）”设置亲和力，不会更改主线程或进程内其他线程的亲和力。但是，子线程从主线程继承其亲和力，子进程（通过其主线程）从父进程的主线程继承亲和力。这会影响所有可能的启动方法（“spawn”、“fork”、“forkserver”）。下面的示例演示了这一点以及如何使用 修改亲和力multiprocessing.Pool。

import multiprocessing as mp
import threading
import os


def _location():
    return f"{mp.current_process().name} {threading.current_thread().name}"


def thread_foo():
    print(f"{_location()}, affinity before change: {os.sched_getaffinity(0)}")
    os.sched_setaffinity(0, {4})
    print(f"{_location()}, affinity after change: {os.sched_getaffinity(0)}")


def foo(_, iterations=200e6):

    print(f"{_location()}, affinity before thread_foo:"
          f" {os.sched_getaffinity(0)}")

    for _ in range(int(iterations)):  # some dummy computation
        pass

    t = threading.Thread(target=thread_foo)
    t.start()
    t.join()

    print(f"{_location()}, affinity before exit is unchanged: "
          f"{os.sched_getaffinity(0)}")

    return _


if __name__ == '__main__':

    mp.set_start_method("spawn")  # alternatives on Unix: "fork", "forkserver"

    # for current process, exclude cores 0,1 from affinity-mask
    print(f"parent affinity before change: {os.sched_getaffinity(0)}")
    excluded_cores = {0, 1}
    os.sched_setaffinity(0, os.sched_getaffinity(0).difference(excluded_cores))
    print(f"parent affinity after change: {os.sched_getaffinity(0)}")

    with mp.Pool(2) as pool:
        pool.map(foo, range(5))

输出：

os.cpu_count() // 2 # same as mp.cpu_count() // 2 

Answer 2

如果您有一个纯粹 100% CPU 限制的任务，即除了计算之外什么都不做，那么显然，如果进程池大小大于计算机上可用的 CPU 数量，则不会获得任何好处。但是，如果存在混合 I/O，进程将放弃 CPU 等待 I/O 完成（或者，例如，从网站返回 URL，这需要相对较长的时间），该怎么办？ ？对我来说，不清楚在这种情况下，如果进程池大小超过 ，您是否无法提高吞吐量os.cpu_count()。

更新

这是演示这一点的代码。这段代码可能最好通过使用线程来实现，它正在使用进程。我的桌面上有 8 个核心。该程序只是同时（或在本例中并行）检索 54 个 URL。该程序传递一个参数，即要使用的池的大小。不幸的是，创建额外的进程会产生初始开销，因此如果创建太多进程，节省的成本就会开始下降。但如果任务运行时间较长并且有大量 I/O，那么创建进程的开销最终是值得的：

from concurrent.futures import ProcessPoolExecutor, as_completed
import requests
from timing import time_it

def get_url(url):
    resp = requests.get(url, headers={'user-agent': 'my-app/0.0.1'})
    return resp.text


@time_it
def main(poolsize):
    urls = [
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
        'https://ibm.com',
        'https://microsoft.com',
        'https://google.com',
    ]
    with ProcessPoolExecutor(poolsize) as executor:
        futures = {executor.submit(get_url, url): url for url in urls}
        for future in as_completed(futures):
            text = future.result()
            url = futures[future]
            print(url, text[0:80])
            print('-' * 100)

if __name__ == '__main__':
    import sys
    main(int(sys.argv[1]))

8个进程：（我拥有的核心数量）：

func: main args: [(8,), {}] took: 2.316840410232544 sec.

16个流程：

func: main args: [(16,), {}] took: 1.7964842319488525 sec.

24个进程：

func: main args: [(24,), {}] took: 2.2560818195343018 sec.