我正在学习使用 Python 的 Multiprocessing 包解决令人尴尬的并行问题，因此我编写了串行和并行版本来确定小于或等于自然数 n 的素数的数量。基于我从 blog post 中读到的内容和一个 Stack Overflow question ，我想出了以下代码:
系列
import math
import time
def is_prime(start, end):
"""determine how many primes within given range"""
numPrime = 0
for n in range(start, end+1):
isPrime = True
for i in range(2, math.floor(math.sqrt(n))+1):
if n % i == 0:
isPrime = False
break
if isPrime:
numPrime += 1
if start == 1:
numPrime -= 1 # since 1 is not prime
return numPrime
if __name__ == "__main__":
natNum = 0
while natNum < 2:
natNum = int(input('Enter a natural number greater than 1: '))
startTime = time.time()
finalResult = is_prime(1, natNum)
print('Elapsed time:', time.time()-startTime, 'seconds')
print('The number of primes <=', natNum, 'is', finalResult)
平行
import math
import multiprocessing as mp
import numpy
import time
def is_prime(vec, output):
"""determine how many primes in vector"""
numPrime = 0
for n in vec:
isPrime = True
for i in range(2, math.floor(math.sqrt(n))+1):
if n % i == 0:
isPrime = False
break
if isPrime:
numPrime += 1
if vec[0] == 1:
numPrime -= 1 # since 1 is not prime
output.put(numPrime)
def chunks(vec, n):
"""evenly divide list into n chunks"""
for i in range(0, len(vec), n):
yield vec[i:i+n]
if __name__ == "__main__":
natNum = 0
while natNum < 2:
natNum = int(input('Enter a natural number greater than 1: '))
numProc = 0
while numProc < 1:
numProc = int(input('Enter the number of desired parallel processes: '))
startTime = time.time()
numSplits = math.ceil(natNum/numProc)
splitList = list(chunks(tuple(range(1, natNum+1)), numSplits))
output = mp.Queue()
processes = [mp.Process(target=is_prime, args=(splitList[jobID], output))
for jobID in range(numProc)]
for p in processes:
p.start()
for p in processes:
p.join()
print('Elapsed time:', time.time()-startTime, 'seconds')
procResults = [output.get() for p in processes]
finalResult = numpy.sum(numpy.array(procResults))
print('Results from each process:\n', procResults)
print('The number of primes <=', natNum, 'is', finalResult)
这是我得到的 n=10000000(对于并行我请求 8 个进程):
$ python serial_prime_test.py
Enter a natural number greater than 1: 10000000
Elapsed time: 162.1960825920105 seconds
The number of primes <= 10000000 is 664579
$ python parallel_prime_test.py
Enter a natural number greater than 1: 10000000
Enter the number of desired parallel processes: 8
Elapsed time: 49.41204643249512 seconds
Results from each process:
[96469, 86603, 83645, 80303, 81796, 79445, 78589, 77729]
The number of primes <= 10000000 is 664579
所以看起来我可以获得超过 3 倍的加速。这是我的 问题 :
任何帮助表示赞赏。
编辑:有 4 个物理内核，能够进行超线程。
最佳答案
我认为你想以不同的方式划分工作。
尽管您的程序将候选整数的范围均匀地划分到内核之间，但每个范围内的工作不可能是均匀的。这意味着一些内核提前完成，无事可做，而其他内核仍在运行。这会很快失去并行效率。
只是为了说明这一点，假设您有 1000 个内核。第一个核心看到非常小的候选数字，不需要很长时间就可以分解它们，然后就闲置了。最后一个(千分之一)核心只看到非常大的候选数字，并且需要更长的时间来分解它们。所以它运行，而第一个核心闲置。浪费的周期。 4核也是如此。
当传递给内核的工作量未知时，您想要做的是将许多中等大小的块交给所有内核，比内核数量多得多。然后内核可能会以不均匀的速度完成，每个内核都会回去寻找更多的工作要做。这本质上是一种工作列表算法。你最终会得到不均匀的结果，但它只是在小块上，所以不会浪费太多。
我不是 Python 程序员，所以我用 Parlanse 编写了一个解决方案。
(includeunique `Console.par')
(includeunique `Timer.par')
(define upper_limit 10000000)
(define candidates_per_segment 10)
(define candidates_per_segment2 (constant (* candidates_per_segment 2)))
(= [prime_count natural] 0)
[prime_finding_team team]
(define primes_in_segment
(action (procedure [lower natural] [upper natural])
(;;
(do [candidate natural] lower upper 2
(block test_primality
(local (= [divisor natural] 3)
(;;
(while (< (* divisor divisor) candidate)
(ifthenelse (== (modulo candidate divisor) 0)
(exitblock test_primality)
(+= divisor 2)
)ifthenelse
)while
(ifthen (~= (* divisor divisor) candidate)
(consume (atomic+= prime_count))
)ifthen
);;
)local
)block
)do
);;
)action
)define
(define main
(action (procedure void)
(local [timer Timer:Timer]
(;;
(Console:Put (. `Number of primes found: '))
(Timer:Reset (. timer))
(do [i natural] 1 upper_limit candidates_per_segment2
(consume (draft prime_finding_team primes_in_segment
`lower':i
`upper':(minimum upper_limit (- (+ i candidates_per_segment2) 2))))
)do
(consume (wait (@ (event prime_finding_team))))
(Timer:Stop (. timer))
(Console:PutNatural prime_count)
(Console:PutNewline)
(Timer:PrintElapsedTime (. timer) (. `Parallel computed in '))
(Console:PutNewline)
);;
)local
)action
)define
Parlanse 看起来像 LISP，但工作和编译更像 C。
worker 是 primes_in_segment；它采用由参数下限和上限定义的一系列候选值。它尝试该范围内的每个候选者，如果该候选者是素数，则(原子地)增加总 prime_count。
整个范围被 do 分成小范围的数据包(奇数序列)
在主循环。并行发生在draft 命令上，它创建一个并行执行的计算粒度(不是Windows 线程)并将其添加到prime_finding_team，这是一个代表所有素数分解的聚合工作集。 (团队的目的是让所有这些工作作为一个单元进行管理，例如，必要时销毁，本程序不需要)。 Draft 的参数是要由 fork 的grain 运行的函数及其参数。这项工作是由 Parlanse 管理的一组 (Windows) 线程使用工作窃取算法完成的。如果工作过多，Parlanse 会限制产生工作的颗粒，并将其能量用于执行纯计算的颗粒。
一个人只能将一个候选值传递给每个粒度，但是每个候选者会有更多的 fork 开销，总运行时间相应地变得更糟。我们根据经验选择了 10 个，以确保每个候选范围的 fork 开销很小；将每个段的候选设置为 1000 并不会带来太多额外的加速。
do 循环只是尽可能快地制造工作。当有足够的并行性可用时，Parlanse 会限制草稿步骤。等待团队事件，导致主程序等待所有团队成员完成。
我们在 HP 六核 AMD Phenom II X6 1090T 3.2 GHz 上运行它。
执行运行如下；首先是 1 个 CPU:
>run -p1 -v ..\teamprimes
PARLANSE RTS: Version 19.1.53
# Processors = 1
Number of primes found: 664579
Parallel computed in 13.443294 seconds
---- PARLANSE RTS: Performance Statistics
Duration = 13.527557 seconds.
CPU Time Statistics:
Kernel CPU Time: 0.031s
User CPU Time: 13.432s
Memory Statistics:
Peak Memory Usage : 4.02 MBytes
Steals: 0 Attempts: 0 Success rate: 0.0% Work Rediscovered: 0
Exiting with final status 0.
然后对于 6 个 CPU(很好地扩展):
>run -p6 -v ..\teamprimes
PARLANSE RTS: Version 19.1.53
# Processors = 6
Number of primes found: 664579
Parallel computed in 2.443123 seconds
---- PARLANSE RTS: Performance Statistics
Duration = 2.538972 seconds.
CPU Time Statistics:
Kernel CPU Time: 0.000s
User CPU Time: 14.102s
Total CPU Time: 14.102s
Memory Statistics:
Peak Memory Usage : 4.28 MBytes
Steals: 459817 Attempts: 487334 Success rate: 94.4% Work Rediscovered: 153
您注意到并行版本的总 CPU 时间与串行版本大致相同；这是因为他们在做同样的工作。
考虑到 Python 的“fork”和“join”操作，我确信有一个 Python 等价物可以轻松编写代码。由于可能同时出现太多 fork ，它可能会耗尽空间或线程。 (使用
candidates_per_segment 在 10 处，Parlanse 下运行的活 Cereal 多达 100 万个)。这就是自动限制工作生成的地方。作为替代，您可以设置 candidates_per_segment到更大的数字，例如 10000，这意味着您只能在最坏的情况下获得 1000 个线程。 (由于 Python 的解释性，我认为您仍然会付出高昂的代价)。当您将每个段的候选设置越来越接近 1e7/4 时，您将接近使用当前 Python 代码的确切行为。
关于python - 使用 Python Multiprocessing 的令人尴尬的并行任务的预期加速，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/26555745/