如何知道多处理中剩余输入项的数量？答案

【问题标题】：How to know the number of remaining input items in multiprocessing?如何知道多处理中剩余输入项的数量？
【发布时间】：2021-10-06 17:54:25
【问题描述】：

我正在使用multiprocessing.Pool 来并行化任务。考虑下面的示例代码，

rangeval = range(1000)
pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
pool.map(f, iterable=rangeval)

在这里，我希望能够为每个函数f 执行打印len(rangeval)。我试图用计数器来了解每个任务何时完成以及任务的数量是多少。但是打印是在没有排序的情况下发生的，因此无法知道有多少 rangeval 项目剩余或完成。我正在执行以下操作。

import multiprocessing

counter = multiprocessing.Value('i', 0)
def f(x):
    global counter
    counter.value += 1
    print("Done with task: ", counter.value)
    return x


if __name__ == '__main__':
    rangeval = range(1000)
    pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
    pool.map(f, iterable=rangeval)

打印出来，

Done with task: Done with task:  Done with task: 30
 163Done with task: 
 31
Done with task:  Done with task: 164

但我希望能够说出rangeval 的长度，以便我可以得到输入中还剩下多少项。或者，如果此计数器值打印以排序方式发生，而不引入睡眠或任何额外计算，那也可以。

如何做到这一点？

【问题讨论】：

您可能希望（或需要）实现自己的生产者/消费者模型（池为您做什么），以便您可以直接检查工作队列的大小。
也许你应该创建其他变量 liek counter 但其值为 len(rangeval) 并且每个进程将在 return 之前运行 variable.value -= 1 这样你就会知道有多少进程仍在运行。
您应该将其打印为单个字符串print( f"Done with task: {counter.value}\n", end='')，并且您的任务编号将与Done with task:在同一行中

标签： python python-3.x multithreading multiprocessing

【解决方案1】：

编辑：

似乎它适用于Linux，它使用方法fork 创建进程，并在进程之间共享一些元素 - 但Windows 使用方法spawn（并且它没有fork）并且它可能需要完全不同的解决方案。也许它需要将数据作为参数发送到进程，或者使用Queue 与必须更改running 的主进程通信。我不使用Windows，所以无法测试。

我会用len(rangeval)创建变量

running = multiprocessing.Value('i', 0)

running.value = len(rangeval)

每个函数都应该在return之前减少它

要获得正确的值，它可能还需要Lock

    with lock:    
        running.value -= 1

if __name__ == '__main__':
    lock = multiprocessing.Lock()

要在一行中获取带有数字的文本，我会将其打印为单个字符串

print(f"Still running: {running.value}\n", end='', flush=True)

编辑：

据我所知，multiprocessing 在 Linux 和 Windows 上的工作方式可能略有不同。其中一些可能需要if __name__ == '__main__': 内部或外部的变量。

import multiprocessing

running = multiprocessing.Value('i', 0)
lock = multiprocessing.Lock()

def f(x):
    global running
    global lock

    # ... code ...
    
    with lock:    
        running.value -= 1
        print(f"Still running: {running.value}\n", end='', flush=True)
   
    return x

if __name__ == '__main__':
    #lock = multiprocessing.Lock()
    
    rangeval = range(100)
    running.value = len(rangeval)
    
    pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
    result = pool.map(f, iterable=rangeval)

结果：

Still running: 99
Still running: 98
Still running: 97
Still running: 96
Still running: 95
Still running: 94
Still running: 93
Still running: 92
Still running: 91
Still running: 90
Still running: 89
Still running: 88
Still running: 87
Still running: 86
Still running: 85
Still running: 84
Still running: 83
Still running: 82
Still running: 81
Still running: 80
Still running: 79
Still running: 78
Still running: 77
Still running: 76
Still running: 75
Still running: 74
Still running: 73
Still running: 72
Still running: 71
Still running: 70
Still running: 69
Still running: 68
Still running: 67
Still running: 66
Still running: 65
Still running: 64
Still running: 63
Still running: 62
Still running: 61
Still running: 60
Still running: 59
Still running: 58
Still running: 57
Still running: 56
Still running: 55
Still running: 54
Still running: 53
Still running: 52
Still running: 51
Still running: 50
Still running: 49
Still running: 48
Still running: 47
Still running: 46
Still running: 45
Still running: 44
Still running: 43
Still running: 42
Still running: 41
Still running: 40
Still running: 39
Still running: 38
Still running: 37
Still running: 36
Still running: 35
Still running: 34
Still running: 33
Still running: 32
Still running: 31
Still running: 30
Still running: 29
Still running: 28
Still running: 27
Still running: 26
Still running: 25
Still running: 24
Still running: 23
Still running: 22
Still running: 21
Still running: 20
Still running: 19
Still running: 18
Still running: 17
Still running: 16
Still running: 15
Still running: 14
Still running: 13
Still running: 12
Still running: 11
Still running: 10
Still running: 9
Still running: 8
Still running: 7
Still running: 6
Still running: 5
Still running: 4
Still running: 3
Still running: 2
Still running: 1
Still running: 0

【讨论】：

我收到一个错误`第 11 行，在带锁的 f 中：NameError: name 'lock' is not defined` 但lock 被定义为全局。我不知道发生了什么！
据我所知multiprocessing 在 Linux 或 Windows 上可能需要一些差异。它适用于 Linux，但您的系统可能需要在 if __name__ == '__main__': 之外的 lock，例如 running。
是的，我正在使用 Windows。我尝试将lock = multiprocessing.Lock() 放在if __name__ == '__main__' 之外，但它会输出一些奇怪的数字，例如still running: -1 ... -15，并且从不显示100。
也许它还需要在 if __name__ == '__main__'. OR maybe it will need to send running` 之前将值 running.value = len(rangeval) 设置为 pool.map() 中的所有函数作为参数 - 类似于 pool.map(..., args=[running for _ in range(rangeval)]