multiprocessing 模組的注意事項

Python 雖然是跨平台的程式語言, 不過實際上特定的功能還是會因為平台的差異, 產生不同的結果, 如果沒有特別注意, 就會對執行結果感到訝異, multiprocessing 就是一個明顯的例子。請看以下這個簡單的程式：

from multiprocessing import Process
from multiprocessing import cpu_count
from multiprocessing import get_start_method
from multiprocessing import current_process

def info(name):
    print(f'{name}[pid:{current_process().pid}] in {__name__}')

info("P1")

p2 = Process(target=info, args=('P2',))
p2.start()
p2.join()

其中的 info() 會顯示傳入的名稱以及目前行程的識別號碼與模組名稱, 主程式會傳入 "P1" 叫用 info(), 然後建立一個新的行程, 並在這個新行程中傳入 "P2" 叫用 info()。這個程式在 Windows 下執行會列印兩行資料後噴出一大堆錯誤 (以 ... 省略中間一大段)：

# python .\mp_win.py
P1[pid:4040] in __main__
P1[pid:14180] in __mp_main__
...
  File "D:\Program Files\Python311\Lib\multiprocessing\spawn.py", line 158, in get_preparation_data
    _check_not_importing_main()
  File "D:\Program Files\Python311\Lib\multiprocessing\spawn.py", line 138, in _check_not_importing_main
    raise RuntimeError('''
RuntimeError:
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

        The "freeze_support()" line can be omitted if the program
        is not going to be frozen to produce an executable.

但是相同的程式如果在 Linux 下執行, 卻非常正常：

$ python /mnt/d/code/python/mp_win.py
P1[pid:81] in __main__
P2[pid:82] in __main__

到底發生什麼事, 我們就來探究看看。

行程的起始方式

上述程式遇到的問題, 主要是因為在 multiprocessing 模組中, 起始新行程有 3 種方式：

起始方式	說明	可套用平台	預設平台
spawn	建立一個新的 Python 直譯器行程, 由它重新匯入程式檔後執行指定的函式。	Windows/Linux/macOS	Windows/macOS
fork	從當前行程以 os.fork()分叉出新行程來執行指定的函式。	Unix/macOS	Unix
forkserver	程式會先建立一個伺服端行程, 之後所有起始新行程的工作就轉交由這個行程先分叉出新行程, 並在這個新行程中以 spawn 方式運作。	可以透過資料通道 (pipe) 傳送檔案描述器 (file descriptor) 的 UNIX 系統/macOS	無

現在就可以回頭來看看剛剛發生什麼事？由於在 Windows 平台下預設的行程起始方式是 spawn, 因此會執行一個新的 Python 直譯器來載入程式檔, 這個動作相當於 import, 所以會重新執行一遍程式, 這樣一來, 在新行程中又會再建立一個新的行程, 沒完沒了。Python 會將這樣的情況視為錯誤, 如果查看錯誤訊息的最後面：

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

它說錯誤的肇因是新行程並不是以 fork 方式起始, 而且主模組又沒有依照慣例使用 if __name__ == '__main__': 來區分模組與獨立程式。我們就來嘗試加上看看會不會讓錯誤消失：

from multiprocessing import Process
from multiprocessing import cpu_count
from multiprocessing import get_start_method
from multiprocessing import current_process
from multiprocessing import set_start_method

def info(name):
    print(f'{name}[pid:{current_process().pid}] in {__name__}')

info("P1")
if __name__ == "__main__":
    p2 = Process(target=info, args=('P2',))
    p2.start()
    p2.join()

執行結果如下：

> python mp_main.py
P1[pid:18636] in __main__
P1[pid:11476] in __mp_main__
P2[pid:11476] in __mp_main__

你可以看到不會產生錯誤, 從輸出結果也可以看到, 由 multiprocessing 建立的新行程的 __name__ 不是 "__main__", 而是 "__mp_main__", 因此我們加上的這一行：

if __name__ == "__main__":

在新行程中不成立, 不會進入 if 執行建立新行程的程式碼了。你也可以看到新行程會重新匯入程式檔, 所以會再執行一次 info("P1"), 因此會看到有兩筆 P1 開頭但模組名稱不同的訊息, 第一筆是一開始執行程式時顯示、第二筆則是新行程匯入程式檔時執行的。

如果你看 Linux 下的執行結果：

$ python mp_main.py 
P1[pid:220] in __main__
P2[pid:221] in __main__

就會發現新行程的 __name__ 仍然是 "__main__", 這是因為新的行程是從原行程分叉出來, 並沒有重新匯入程式檔, 因此 info("P1") 只會執行一次, 而且在新行程中, 模組名稱維持和原始的行程一樣不變。

由於 fork 是由本的行程分叉出來, 不需要重新匯入程式檔, 因此是三種方式中效能最高的作法。

更改新行程的起始方式

你可以透過 multiprocessing.get_all_start_methods() 取得目前平台可用的新行程起始方式, 也可以透過 multiprocessing.get_start_method() 取得目前採用的新行程起始方式, 例如在 Windows 上：

>>> import multiprocessing
>>> multiprocessing.get_all_start_methods()
['spawn']
>>> multiprocessing.get_start_method()
'spawn'

若是 Linux, 則是：

>>> import multiprocessing
>>> multiprocessing.get_all_start_methods()
['fork', 'spawn', 'forkserver']
>>> multiprocessing.get_start_method()
'fork'
>>>

如果想要設定新行程的起始方式, 可以使用 multiprocessing.set_start_method(method, force=False), 如果 force 是 False, 而且系統已經設定過起始方式, 就會引發 RuntimeError 錯誤, 例如在已經有預設起始方式的 Windows 上：

>>> import multiprocessing
>>> multiprocessing.set_start_method("spawn")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "D:\apps\PortableApps\thonny\lib\multiprocessing\context.py", line 247, in set_start_method
    raise RuntimeError('context has already been set')
RuntimeError: context has already been set

如果要強制設定起始方式, 就要指定 force 為 True, 以下就將前面的範例程式改為強制採用 spawn 起始方式：

from multiprocessing import Process
from multiprocessing import cpu_count
from multiprocessing import get_start_method
from multiprocessing import current_process
from multiprocessing import set_start_method

def info(name):
    print(f'{name}[pid:{current_process().pid}] in {__name__}')

info("P1")
if __name__ == "__main__":
    set_start_method('spawn', True)    
    p2 = Process(target=info, args=('P2',))
    p2.start()
    p2.join()

在 Windows 下執行結果不變, 但是在 Linux 下因為改變了起始方式, 所以執行結果也會和 Windows 一樣了：

$ python mp_spawn.py 
P1[pid:255] in __main__
P1[pid:257] in __mp_main__
P2[pid:257] in __mp_main__

特別的 forkserver 方式

如果是 Linux, 還有一種 forkserver 起始方式, 等於是會先建立一個專門負責建立新行程的代理行程, 以下是修改版的範例：

from multiprocessing import Process
from multiprocessing import cpu_count
from multiprocessing import get_start_method
from multiprocessing import current_process
from multiprocessing import set_start_method

def info(name):
    print(f'{name}[pid:{current_process().pid}] in {__name__}')

info("P1")
if __name__ == "__main__":
    set_start_method('forkserver', True)    
    p2 = Process(target=info, args=('P2',))
    p3 = Process(target=info, args=('P3',))
    p2.start()
    p3.start()
    p2.join()
    p3.join()

我們特意建立兩個新行程, 觀察執行結果：

$ python mp_forkserver.py 
P1[pid:291] in __main__
P1[pid:294] in __mp_main__
P2[pid:294] in __mp_main__
P1[pid:295] in __mp_main__
P3[pid:295] in __mp_main__

由於伺服端程式分叉出的新行程會以 spawn 方式起始新直譯器行程, 所以也一樣會重新匯入程式檔執行 info("P1") 一次, 因此輸出結果中就會有兩行 P1 開頭但模組名稱是 '__mp_main__' 的訊息。

彈性變化新行程起始方式

理論上你可以叫用 multiprocessing.set_start_method(method, force=False) 多次來變更新行程起始方式, 像是這樣：

from multiprocessing import Process
from multiprocessing import cpu_count
from multiprocessing import get_start_method
from multiprocessing import current_process
from multiprocessing import set_start_method

def info(name):
    print(f'{name}[pid:{current_process().pid}] in {__name__}')

info("P1")
if __name__ == "__main__":
    set_start_method('spawn', True)    
    p2 = Process(target=info, args=('P2',))
    p3 = Process(target=info, args=('P3',))
    p2.start()
    set_start_method('fork', True)    
    p3.start()
    p2.join()
    p3.join()

或者你也可以透過 multiprocessing.get_context(method=None) 取得不同起始方式的執行環境物件, 例如：

from multiprocessing import Process
from multiprocessing import cpu_count
from multiprocessing import get_start_method
from multiprocessing import current_process
from multiprocessing import set_start_method
from multiprocessing import get_context 

def info(name):
    print(f'{name}[pid:{current_process().pid}] in {__name__}')

info("P1")
if __name__ == "__main__":
    ctx2 = get_context('spawn')
    ctx3 = get_context('fork')
    p2 = ctx2.Process(target=info, args=('P2',))
    p3 = ctx3.Process(target=info, args=('P3',))
    p2.start()
    p3.start()
    p2.join()
    p3.join()

分別取得以 spawn 和 fork 方式起始新行程的執行環境, 然後利用他們個別建立新的行程。上述兩種版本的執行結果都一樣如下：

$ python mp_ctx.py
P1[pid:64] in __main__
P3[pid:67] in __main__
P1[pid:66] in __mp_main__
P2[pid:66] in __mp_main__

可以看到兩個行程起始的方式的確不同, p2 是以 spawn 方式起始, 所以會匯入程式檔, 多執行了一次 info("P1")。

取得執行環境物件的方式尤其適合暫時修改起始方式, 或是撰寫模組時, 不想變更使用此模組的程式原本起始模式時。

模擬 multiprocessing 新行程的起始方式

對於 spawn 和 fork 兩種模式的差異, 以下分別提供簡化的模擬程式碼, 實際上的程式碼要複雜許多。首先是模擬 fork 的方式：

import os 

def info():
    print(f'{os.getpid()}:{__name__}')

def fork_process(target):
    pid = os.fork()
    if pid == 0:
        target()
        exit(0)
    else:
        return pid

info()
if __name__ == '__main__':
    fork_process(info)

在 fork_process() 中利用 os.fork() 分叉出新的行程叫用目標函式, 要特別留意的是叫用目標函式後必須結束不返回, 否則如果在叫用 fork_process(info) 後有其他程式碼, 就變成會在原本的行程和新行程中都個別執行一次。以下是執行結果：

$ python mp_fork.py
747:__main__
748:__main__

spawn 的方式比較複雜, 如以下模擬程式：

import runpy
import subprocess
import os 
import sys

def info():
    print(f'{os.getpid()}:{__name__}')

def spqwn_process(target):
    subprocess.run([
        'python',
        '-c',
        'import runpy\n' +
        f'd = runpy.run_path("{sys.argv[0]}", run_name="__mp_main__")\n' +
        f'd["{target.__name__}"]()'
])

info()
if __name__ == "__main__":
    spqwn_process(info)

首先利用 subprocess.run() 執行新的直譯器, 並透過 Python 直譯器的命令列 -c 選項執行一組 Python 敘述, 其中 runpy.run_path() 會重新匯入程式檔, 並取得匯入模組的全域名稱字典物件, 再透過這個字典物件叫用目標函式。以下是執行結果：

$ python /mnt/d/code/python/mp_sp.py
751:__main__
752:__mp_main__
752:__mp_main__

由上述兩段模擬程式即可看出, spawn 的方式要比 spawn 繁瑣, 效能一定比不上 fork。

結語

對於需要在不同平台執行同一 Python 程式的人來說, 如果沒有好好閱讀文件, 很可能就會遇到像是本文提到的狀況, 尤其許多網路上的教學可能只是針對單一平台測試, 希望這篇文章可以讓大家避免掉入陷阱。