为啥增加工人数量（超过核心数量）仍然会减少执行时间？

Posted 2023-02-19

【中文标题】为啥增加工人数量（超过核心数量）仍然会减少执行时间？

【英文标题】：Why increasing number of workers (more than number of cores) still decrease execution time?为什么增加工人数量（超过核心数量）仍然会减少执行时间？

【发布时间】：2017-12-07 19:24:15

【问题描述】：

我始终确信，拥有比 CPU 内核更多的线程/进程是没有意义的（从性能角度来看）。但是，我的 python 示例显示了不同的结果。

import concurrent.futures
import random
import time


def doSomething(task_num):
    print("executing...", task_num)
    time.sleep(1)  # simulate heavy operation that takes ~ 1 second    
    return random.randint(1, 10) * random.randint(1, 500)  # real operation, used random to avoid caches and so on...


def main():
    # This part is not taken in consideration because I don't want to
    # measure the worker creation time
    executor = concurrent.futures.ProcessPoolExecutor(max_workers=60)

    start_time = time.time()

    for i in range(1, 100): # execute 100 tasks
        executor.map(doSomething, [i, ])
    executor.shutdown(wait=True)

    print("--- %s seconds ---" % (time.time() - start_time))


if __name__ == '__main__':
    main()

节目结果：

1 工人 --- 100.28233647346497 秒 --- 2 名工人 --- 50.26122164726257 秒 --- 3 名工人 --- 33.32741022109985 秒 --- 4 名工人 --- 25.399883031845093 秒 --- 5 名工人 --- 20.434186220169067 秒 --- 10 名工人--- 10.903695344924927 秒 --- 50 名工人--- 6.363946914672852 秒 --- 60 名工人--- 4.819359302520752 秒 ---

如何在只有 4 个逻辑处理器的情况下更快地工作？

这是我的电脑规格（在 Windows 8 和 Ubuntu 14 上测试）：

CPU Intel(R) Core(TM) i5-3210M CPU @ 2.50GHz 插座：1 核心：2 逻辑处理器：4 个

【问题讨论】：

time.sleep(1) 不会阻塞 CPU。在一名工作人员睡觉的时候，其他工作人员可以使用 CPU。我怀疑如果您使用主动等待，您会看到预期的结果。

确实，睡眠引起的意外行为。非常感谢。

【参考方案1】：

原因是因为sleep() 只使用了微不足道的 CPU。在这种情况下，它对线程执行的实际工作的模拟效果不佳。

sleep() 所做的只是暂停线程直到计时器到期。当线程挂起时，它不使用任何 CPU 周期。

【参考方案2】：

我通过更密集的计算（例如矩阵求逆）扩展了您的示例。您将看到您所期望的：计算时间将减少到核心数量，然后增加（因为上下文切换的成本）。

import concurrent.futures
import random
import time
import numpy as np
import matplotlib.pyplot as plt


def doSomething(task_num):
    print("executing...", task_num)
    for i in range(100000):
        A = np.random.normal(0,1,(1000,1000))
        B = np.inv(A)

    return random.randint(1, 10) * random.randint(1, 500)  # real operation, used random to avoid caches and so on...

def measureTime(nWorkers: int):
    executor = concurrent.futures.ProcessPoolExecutor(max_workers=nWorkers)
    start_time = time.time()
    for i in range(1, 40):  # execute 100 tasks
        executor.map(doSomething, [i, ])
    executor.shutdown(wait=True)
    return (time.time() - start_time)

def main():
    # This part is not taken in consideration because I don't want to
    # measure the worker creation time
    maxWorkers = 20
    dT = np.zeros(maxWorkers)
    for i in range(maxWorkers):
        dT[i] = measureTime(i+1)
        print("--- %s seconds ---" % dT[i])
    plt.plot(np.linspace(1,maxWorkers, maxWorkers), dT)
    plt.show()

if __name__ == '__main__':
    main()