Pool.apply_async() 按顺序创建和运行工作线程

Question

我从python多处理教程中复制了一个扩展代码并对其进行了轻微修改：

import multiprocessing as mp
import os
import time

start = time.time()
# Step 1: Redefine, to accept `i`, the iteration number
def howmany_within_range2(i, row, minimum, maximum):
"""Returns how many numbers lie within `maximum` and `minimum` in a given `row`"""
print(os.getpid(),' going to sleep')
time.sleep(3)
print(os.getpid(),' waking up')
count = 0
for n in row:
if minimum <= n <= maximum:
count = count + 1
return (i, count)

# Step 2: Define callback function to collect the output in `results`
def collect_result(result):
global results
results.append(result)
data = []
for i in range(8):
arr_row = []
for j in range(5):
arr_row.append(random.randint(0,10))
data.append(arr_row)
k = 6
print('CPU-count: ',mp.cpu_count())
print('using this many CPUs: ',k)
pool = mp.Pool(k)
results = []
i=0
for row in data:
pool.apply_async(howmany_within_range2, args=(i, row, 4, 8),callback=collect_result)
i+=1
# Step 4: Close Pool and let all the processes complete
pool.close()
pool.join()  # postpones the execution of next line of code until all processes in the queue are done.
# Step 5: Sort results [OPTIONAL]
results.sort(key=lambda x: x[0])
results_final = [r for i, r in results]
print('final result: ',results_final[:10])
now = time.time()
print('script total time taken: ',now-start)

我得到了这个意想不到的输出：

CPU-count:  24
using this many CPUs:  6
143259  going to sleep
143259  waking up
143258  going to sleep
143258  waking up
143260  going to sleep
143260  waking up
143256  going to sleep
143256  waking up
143255  going to sleep
143255  waking up
143255  going to sleep
143255  waking up
143257  going to sleep
143257  waking up
143257  going to sleep
143257  waking up
final result:  [2, 1, 1, 1, 1, 1, 1, 4]
script total time taken:  6.07246470451355

似乎由于某些原因，程序在启动下一个进程之前等待howmany_within_range2()完成，我不明白为什么会这样，从我在文档和 Web 上读到的内容来看，工作进程应该异步并行运行，而不是显示这种顺序行为。

编辑：

这种错觉确实是按照罗兰的解释是基于操作系统的。辅助角色确实异步运行。这是我的输出，添加了进程进入睡眠状态和唤醒的时间：

CPU-count:  24
using this many CPUs:  6
1579047423.7 96321  going to sleep
1579047426.7 96321  waking up
1579047423.7 96319  going to sleep
1579047426.7 96319  waking up
1579047423.7 96320  going to sleep
1579047426.7 96320  waking up
1579047423.7 96322  going to sleep
1579047426.7 96322  waking up
1579047423.7 96317  going to sleep
1579047426.7 96317  waking up
1579047426.7 96317  going to sleep
1579047429.7 96317  waking up
1579047423.7 96318  going to sleep
1579047426.7 96318  waking up
1579047426.7 96318  going to sleep
1579047429.7 96318  waking up
final result:  [4, 1, 3, 3, 2, 1, 2, 2]
script total time taken:  6.050582647323608

Answer 1

每个子进程休眠三秒，父进程中的总运行时间约为 6 秒。 这证明，实际上这些进程不是按顺序运行的。如果是，总时间为 8 × 3 = 24 秒。

您的问题带有一个隐含的假设，即您看到行出现在终端上的顺序指示它们的发送顺序，即使它们来自不同的进程。在具有单核处理器的机器上，这甚至可能是真的。但是在现代多核机器上，我认为不能保证这一点。

标准输出一般是缓冲的;sys.stdout是一个io.TextIOWrapper实例：

In [1]: import sys                                                                                       
In [2]: type(sys.stdout)                                                                                 
Out[2]: _io.TextIOWrapper

因此，如果使用print，则输出不会立即显示，除非您在print调用中使用flush=True，或者在启动子进程之前reconfiguresys.stdout对象。即便如此，操作系统级别也可能有缓冲。

AFAIK，操作系统文件对象将按照接收它的顺序生成输出，但由于输入是从不同的进程接收的，这些进程都继承了相同的文件描述符sys.stdout我认为不能保证这是相同的发送顺序！由于我们在这里谈论不同的进程，因此操作系统调度程序也开始发挥作用。

为了规避所有这些，您可以包括调用 print 的时间：

def howmany_within_range2(i, row, minimum, maximum):
"""Returns how many numbers lie within `maximum` and `minimum` in a given `row`"""
print(str(round(time.time(), 2)), os.getpid(), ' going to sleep')
time.sleep(3)
print(str(round(time.time(), 2)), os.getpid(), ' waking up')
count = 0
for n in row:
if minimum <= n <= maximum:
count = count + 1
return (i, count)

就我而言，这给出了：

> python3 cputest.py
CPU-count:  4
using this many CPUs:  6
1578218933.69 89875  going to sleep
1578218933.69 90129  going to sleep
1578218933.69 90509  going to sleep
1578218933.69 90760  going to sleep
1578218933.69 91187  going to sleep
1578218933.69 91209  going to sleep
1578218936.71 90509  waking up
1578218936.71 89875  waking up
1578218936.71 90760  waking up
1578218936.71 91209  waking up
1578218936.71 90129  waking up
1578218936.71 91187  waking up
1578218936.71 89875  going to sleep
1578218936.71 90509  going to sleep
1578218939.75 90509  waking up
1578218939.75 89875  waking up
final result:  [1, 4, 4, 2, 1, 2, 4, 2]
script total time taken:  6.099705934524536

这清楚地表明，至少在 FreeBSD 上，所有六个 worker 函数都在 1/100 秒内启动，并在 1/100 秒内完成。