我正在用mpi 4py做一个并行化练习,两个骰子被掷出一个定义的次数(除以进程,即npp
),点数被计算。结果被存储在一个字典中,平均偏差被计算出来,直到mean_dev
小于0.001,掷出的次数加倍。
所有这些都按预期工作,问题是代码没有退出。条件满足,没有更多的输出,但代码挂起。
from ctypes.wintypes import SIZE
from dice import * #This is just a class that creates the dictionaries
from random import randint
import matplotlib.pyplot as plt
import numpy as np
from mpi4py import MPI
from math import sqrt
def simulation(f_events, f_sides, f_n_dice):
f_X = dice(sides, n_dice).myDice() #Nested dictionary composed of dices (last dict stores the sum)
for j in range(f_events): #for loop to handle all the dice throwings aka events
n = [] #List to store index respective to number on each dice
for i in range(1, f_n_dice+1): #for cycle for each dice
k = randint(1, f_sides) #Random number
n.append(k)
f_X[i][k] += 1 #The index (k) related to each throw is increased for the dice (i)
sum_throw = sum(n) #Sum of the last throw
f_X[f_n_dice+1][sum_throw] += 1 #Sum dictionary "increases" the index respective to the sum of the last throw
return f_X
npp = int(4)//4 #Number of events divided by the number of processes
sides = 6 #Number of sides per dice
n_dice = 2 #Number of dices
comm = MPI.COMM_WORLD #Communicator to handle point-to-point communication
rank = comm.Get_rank() #Hierarchy of processes
size = comm.Get_size() #Number of processes
#-------------------- Parallelization portion of the code --------------------#
seq = (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12)
AUX = dict.fromkeys(seq, 0)
mean_dev = 1
while True:
msg = comm.bcast(npp, root = 0)
print("---> msg: ", msg, " for rank ", rank)
print("The mean dev for %d" %rank + " is: ", mean_dev)
D = simulation(npp, sides, n_dice)
Dp = comm.gather(D, root = 0)
print("This is Dp: ", Dp)
summ = 0
prob = [1/36, 2/36, 3/36, 4/36, 5/36, 6/36, 5/36, 4/36, 3/36, 2/36, 1/36]
if rank==0:
for p in range(0, size):
for n in range(dice().min, dice().max+1): #Range from minimum sum possible to the maximum sum possible depending on the number of dices used
AUX[n] += Dp[p][n_dice+1][n] #Adds the new data to the final sum dictionary
#of the previously initiated nested dictionary
print(Dp[p][n_dice+1])
print("The final dictionary is: ", AUX, sum(AUX[j] for j in AUX))
for i in range(dice().min, dice().max+1):
exp = (prob[i-2])*(sum(AUX[j] for j in AUX))
x = (AUX[i]-exp)/exp
summ = summ + pow(x, 2)
mean_dev = (1/11)*sqrt(summ)
print("The deviation for {} is {}.".format(sum(AUX[j] for j in AUX), mean_dev))
if mean_dev > 0.001:
npp = 2*npp
# new_msg = comm.bcast(npp, root = 0)
# print("---> new_msg: ", new_msg, " for rank ", rank)
else:
break
我被这个难住了。提前感谢你的任何意见!
新代码采用了@victor-eijkhout提出的解决方案:
from ctypes.wintypes import SIZE
from dice import *
from random import randint
import matplotlib.pyplot as plt
import numpy as np
from mpi4py import MPI
from math import sqrt
def simulation(f_events, f_sides, f_n_dice):
f_X = dice(sides, n_dice).myDice() #Nested dictionary composed of dices (last dict stores the sum)
for j in range(f_events): #for loop to handle all the dice throwings aka events
n = [] #List to store index respective to number on each dice
for i in range(1, f_n_dice+1): #for cycle for each dice
k = randint(1, f_sides) #Random number
n.append(k)
f_X[i][k] += 1 #The index (k) related to each throw is increased for the dice (i)
sum_throw = sum(n) #Sum of the last throw
f_X[f_n_dice+1][sum_throw] += 1 #Sum dictionary "increases" the index respective to the sum of the last throw
return f_X
npp = int(4)//4 #Number of events divided by the number of processes
sides = 6 #Number of sides per dice
n_dice = 2 #Number of dices
comm = MPI.COMM_WORLD #Communicator to handle point-to-point communication
rank = comm.Get_rank() #Hierarchy of processes
size = comm.Get_size() #Number of processes
#-------------------- Parallelization portion of the code --------------------#
seq = (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12)
AUX = dict.fromkeys(seq, 0)
mean_dev = 1
while True:
msg = comm.bcast(npp, root = 0)
#print("---> msg: ", msg, " for rank ", rank)
D = simulation(npp, sides, n_dice)
Dp = comm.gather(D, root = 0)
#if Dp != None: print("This is Dp: ", Dp)
#print("The mean dev for %d" %rank + " is: ", mean_dev)
if rank==0:
summ = 0
prob = [1/36, 2/36, 3/36, 4/36, 5/36, 6/36, 5/36, 4/36, 3/36, 2/36, 1/36]
for p in range(0, size):
for n in range(dice().min, dice().max+1): #Range from minimum sum possible to the maximum sum possible depending on the number of dices used
AUX[n] += Dp[p][n_dice+1][n] #Adds the new data to the final sum dictionary
#of the previously initiated nested dictionary
print(Dp[p][n_dice+1])
print("The final dictionary is: ", AUX, sum(AUX[j] for j in AUX))
for i in range(dice().min, dice().max+1):
exp = (prob[i-2])*(sum(AUX[j] for j in AUX))
x = (AUX[i]-exp)/exp
summ = summ + pow(x, 2)
mean_dev = (1/11)*sqrt(summ)
print("The deviation for {} is {}.".format(sum(AUX[j] for j in AUX), mean_dev))
#new_mean_dev = comm.gather(mean_dev, root = 0)
new_mean_dev = comm.bcast(mean_dev, root = 0)
print("---> msg2: ", new_mean_dev, " for rank ", rank)
if new_mean_dev < 0.001:
break
# new_msg = comm.bcast(npp, root = 0)
# print("---> new_msg: ", new_msg, " for rank ", rank)
else:
npp = 2*npp
print("The new npp is: ", npp)
2条答案
按热度按时间eagi6jfj1#
您只在进程0上计算平均偏差,因此该进程将退出。但是,其他进程不会获得该值,因此它们永远不会退出。您应该在计算完该值后广播它。
njthzxwz2#
你正在打破if语句。只要用
while mean_dev > 0.001:
替换whileTrue:
就可以了。你也可以在结尾做一个赋值,而不是把它 Package 在if
中。如果这不起作用,那就意味着
mean_dev
总是大于0.001。您可以将mean_dev
计算为(1/11)*sqrt(sum …)
。在整个算法中,如果2个骰子的最小和为2,那么mean_dev
不会降到0.14以下。试着放入一个print语句,每次通过循环都打印出mean_dev
,看看它是否会降到0.14以下。是不是每次都要用mean_dev
除以npp
之类的东西?一般来说,当估计值的变化非常小时,迭代求解的问题就会终止。当mean_dev的变化小于0.001时,是否应该停止迭代?您需要执行abs(last_mean_dev-mean_dev)〈0.001这样的操作。