當(dāng)運(yùn)行 CPU 密集型的并行程序時(shí)，通常希望將線程或進(jìn)程池的大小設(shè)置為計(jì)算機(jī)上的 CPU 核數(shù)量，但有沒有考慮過是否真的是核數(shù)用的越多并行程序越快？

理論上線程過少，無法充分利用所有核心，線程過多，程序會(huì)因?yàn)槎鄠€(gè)線程爭(zhēng)奪同一核心而變得運(yùn)行緩慢。

事實(shí)上，確定要運(yùn)行多少個(gè)線程沒那么容易：

• Python 標(biāo)準(zhǔn)庫(kù)提供了多個(gè)獲取此信息的 API，但沒有一個(gè)是恰當(dāng)?shù)模ㄉ院髸?huì)舉例）
• 由于 CPU 具有指令級(jí)并行性和同時(shí)多線程等功能（在英特爾 CPU 上稱為超線程），可以有效使用的核心數(shù)量取決于編寫的代碼

從 Python 獲取 CPU 內(nèi)核數(shù)

前述提到在Python中獲取內(nèi)核數(shù)的API是不準(zhǔn)確的，為啥這么說，我們看個(gè)例子

Python提供 os.cpu_count() 函數(shù)，可以返回 "系統(tǒng)中邏輯 CPU 的數(shù)量"，文檔說明 "len(os.sched_getaffinity(0))可以獲取當(dāng)前進(jìn)程調(diào)用線程受限的邏輯 CPU 數(shù)量"，調(diào)度器親和性是一種限制進(jìn)程使用特定內(nèi)核的方法。

遺憾的是，這個(gè) API 也不夠恰當(dāng)，例如使用Docker在創(chuàng)建容器時(shí)人為限制CPU數(shù)量，比如將 CPU 限制為2.25 個(gè)內(nèi)核：

$ docker run -i -t --cpus=2.25 python:3.12-slim
Python 3.12.1 (main, Dec  9 2023, 00:21:37) [GCC 12.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import os
>>> os.cpu_count()
20
>>> len(os.sched_getaffinity(0))
20

在Docker中只提供了2.25個(gè)內(nèi)核資源，但顯然調(diào)用Python API時(shí)返回的數(shù)量仍不對(duì)。

說完這個(gè)問題，還需要先了解物理和邏輯 CPU 內(nèi)核是什么再進(jìn)入正題。

物理與邏輯 CPU內(nèi)核

以英特爾 i7-12700K 處理器為例，它具有：

• 12 個(gè)物理內(nèi)核（8 個(gè)高性能內(nèi)核和 4 個(gè)性能較弱的內(nèi)核）
• 20 個(gè)邏輯內(nèi)核

現(xiàn)代 CPU 內(nèi)核可以并行執(zhí)行多條指令，但如果 CPU 在等待從 RAM 中加載某些數(shù)據(jù)時(shí)卡住了，會(huì)發(fā)生什么情況？在此之前，它可能無法執(zhí)行任何工作。

為了充分利用這些可能被浪費(fèi)的資源，CPU 物理內(nèi)核的計(jì)算資源可以作為多個(gè)內(nèi)核向操作系統(tǒng)公開。在這臺(tái)電腦上，8 個(gè)高性能內(nèi)核中的每一個(gè)都可以作為兩個(gè)內(nèi)核公開，總共有 16 個(gè)邏輯內(nèi)核。成對(duì)的邏輯內(nèi)核將共享單個(gè)物理內(nèi)核的計(jì)算資源，例如，如果一個(gè)邏輯內(nèi)核沒有充分利用所有內(nèi)部算術(shù)邏輯單元，比如因?yàn)樗诘却齼?nèi)存加載，那么通過配對(duì)邏輯內(nèi)核運(yùn)行的代碼仍然可以使用這些閑置資源。

這種技術(shù)被稱為同步多線程技術(shù)，英特爾稱之為超線程技術(shù)。如果你有一臺(tái)電腦，通?？梢栽?BIOS 中禁用它。

這種解釋非常不準(zhǔn)確，而且不同型號(hào)的 CPU，即使是同一制造商生產(chǎn)的 CPU，實(shí)際執(zhí)行情況也不盡相同。不過，邏輯內(nèi)核與物理內(nèi)核并不完全相同的一般意義足以滿足這篇文章要表達(dá)的目的。

現(xiàn)在又有了一個(gè)新問題，拋開調(diào)度器親和性等因素不談，我們應(yīng)該使用物理內(nèi)核數(shù)還是邏輯內(nèi)核數(shù)作為線程池大?。?/p>

示例

在該例中，用 Numba 將兩個(gè)函數(shù)編譯成機(jī)器代碼，確保釋放 GIL 以實(shí)現(xiàn)并行。

這兩個(gè)函數(shù)做的事情一毛一樣，但slow_threshold特意寫成比較慢的方式而fast_threshold則更快（感興趣的可以對(duì)比學(xué)習(xí)下為何另一個(gè)更快，很簡(jiǎn)單）?，F(xiàn)在可以在多個(gè)線程上并行運(yùn)行這些函數(shù)，在大多數(shù)人眼里，只需并行處理更多圖像，就能線性提高吞吐量，直到內(nèi)核耗盡，先從單核上進(jìn)行測(cè)試：

from numba import njit
import numpy as np

@njit(nogil=True)
def slow_threshold(img, noise_threshold):
    noise_threshold = img.dtype.type(noise_threshold)
    result = np.empty(img.shape, dtype=np.uint8)
    for i in range(result.shape[0]):
        for j in range(result.shape[1]):
            result[i, j] = img[i, j] // 256
    for i in range(result.shape[0]):
        for j in range(result.shape[1]):
            if result[i, j] < noise_threshold // 256:
                result[i, j] = 0
    return result

@njit(nogil=True)
def fast_threshold(img, noise_threshold):
    noise_threshold = np.uint8(noise_threshold // 256)
    result = np.empty(img.shape, dtype=np.uint8)
    for i in range(result.shape[0]):
        for j in range(result.shape[1]):
            value = img[i, j] >> 8
            value = (
                0 if value < noise_threshold else value
            )
            result[i, j] = value
    return result

rng = np.random.default_rng(12345)

def make_image(size=256):
    noise = rng.integers(0, high=1000, size=(size, size), dtype=np.uint16)
    signal = rng.integers(0, high=5000, size=(size, size), dtype=np.uint16)
    # A noisy, hard to predict image:
    return noise | signal

NOISY_IMAGE = make_image()
assert np.array_equal(
    slow_threshold(NOISY_IMAGE, 1000),
    fast_threshold(NOISY_IMAGE, 1000)
)

借助timeit測(cè)試單核上運(yùn)行每個(gè)功能的性能，結(jié)果如下：

%timeit slow_threshold(NOISY_IMAGE, 1000)

# 90.6 μs ± 77.7 ns per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

%timeit fast_threshold(NOISY_IMAGE, 1000)

# 24.6 μs ± 10.8 ns per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

結(jié)果如前所述，確實(shí)fast_threshold表現(xiàn)更好。

并行化示例

現(xiàn)在我們使用線程池處理上述函數(shù)：

from multiprocessing.dummy import Pool as ThreadPool

def apply_in_thread_pool(
    num_threads, function, images
):
    with ThreadPool(num_threads) as pool:
        result = pool.map(
            lambda img: function(img, 1000),
            images,
            chunksize=10
        )
        assert len(result) == len(images)

借助benchit繪制不同線程數(shù)運(yùn)行不同函數(shù)所需的時(shí)間圖：

import benchit
benchit.setparams(rep=1)

# 4000 images to run through the pool:
IMAGES = [make_image() for _ in range(4000)]

def slow_threshold_in_pool(num_threads):
    apply_in_thread_pool(num_threads, slow_threshold, IMAGES)

def fast_threshold_in_pool(num_threads):
    apply_in_thread_pool(num_threads, fast_threshold, IMAGES)

# Measure the two functions with 1 to 24 threads:
timings = benchit.timings(
    [slow_threshold_in_pool, fast_threshold_in_pool],
    range(1, 25),
    input_name="Number of threads"
)
timings.plot(logy=True, logx=False)

繪制的圖片如下：

可以注意到隨著線程數(shù)變多，運(yùn)行時(shí)間先是有明顯下降，但到一定程度后無明顯改進(jìn)，且另一個(gè)發(fā)現(xiàn)是每個(gè)函數(shù)的最佳線程數(shù)不同：

timings.to_dataframe().idxmin(axis="rows")

slow_threshold函數(shù)基本上可以利用所有邏輯內(nèi)核，單線程可能無法充分利用特定物理內(nèi)核的所有可用處理能力，因此邏輯內(nèi)核允許更多并行性。

相比之下，fast_threshold函數(shù)使用超過 9 個(gè)內(nèi)核后，速度就開始減慢。可能遇到計(jì)算以外的瓶頸，比如內(nèi)存帶寬。

總結(jié)

• 考慮到操作系統(tǒng)限制 CPU 使用的所有不同方式，很難獲得準(zhǔn)確的內(nèi)核數(shù)量
• 最佳并行程度（如線程數(shù)）取決于工作量
• 內(nèi)核數(shù)量并不是唯一的瓶頸

如果有一個(gè)長(zhǎng)期運(yùn)行的數(shù)據(jù)處理任務(wù)，需要在多個(gè)線程中運(yùn)行相同的代碼一段時(shí)間，通常也值得這樣做，花一點(diǎn)時(shí)間根據(jù)經(jīng)驗(yàn)測(cè)算出最佳線程數(shù)。