OpenMDAO：ParallelGroup 的意外行为
OpenMDAO: unexpected behaviour of ParallelGroup
在下面的例子中，两个 ParallelGroups 被设置为需要不同数量的 procs 来计算它们的串行子组件，似乎一个不应该填满全局通信的组实际上确实如此, 并执行 redundant/repeated 个案例。在以前的版本中，未分配的过程不会执行它们的子组件。虽然它运行没有错误。
from mpi4py import MPI
import openmdao.api as om
class Exec(om.ExplicitComponent):
def __init__(self, val=-10):
super().__init__()
self.val = val
def setup(self):
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec1, %s, global rank %i, val=%f' % (self.name, MPI.COMM_WORLD.rank, self.val))
outputs['y'] = self.val
class Exec2(om.ExplicitComponent):
def __init__(self, val):
super().__init__()
self.val = val
def setup(self):
self.add_input('x', 0.)
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec2, %s, global rank %i, val=%f' % (self.name, MPI.COMM_WORLD.rank, self.val))
outputs['y'] = self.val * inputs['x']
class Summer(om.ExplicitComponent):
def __init__(self, ncase):
super().__init__()
self.ncase = ncase
def setup(self):
for i in range(self.ncase):
self.add_input('y%i' % i, 0.)
self.add_output('sum', 0.)
def compute(self, inputs, outputs):
for i in range(self.ncase):
outputs['sum'] += inputs['y%i' % i]
p = om.Problem()
ncase = 3
par1 = p.model.add_subsystem('par1', om.ParallelGroup())
p.model.add_subsystem('summer1', Summer(ncase))
for i in range(ncase):
par1.add_subsystem('ex_%i' % i, Exec(val=float(i)), min_procs=1)
p.model.connect('par1.ex_%i.y' % i, 'summer1.y%i' % i)
ncase2 = 4
par2 = p.model.add_subsystem('par2', om.ParallelGroup())
p.model.add_subsystem('summer2', Summer(ncase2))
for i in range(ncase2):
par2.add_subsystem('ex_%i' % i, Exec2(float(i)), min_procs=1)
p.model.connect('summer1.sum', 'par2.ex_%i.x' % i)
p.model.connect('par2.ex_%i.y' % i, 'summer2.y%i' % i)
p.setup()
p.run_model()
根据 docs，处理器分配主要基于 proc_weight 使用循环分配方式完成。
The algorithm used for the allocation starts, assuming that the number of processes is greater than or equal to the number of subsystems, by assigning the min_procs for each subsystem. It then adds any remaining processes to subsystems based on their weights, being careful not to exceed their specified max_procs, if any.
所以行为符合预期。 OpenMDAO 寻求分配所有给定的处理器。然后由组件作者明智地使用它们。
如果你真的想要，你可以修改组件，让除 Rank 0 之外的任何东西都不起作用，但我不建议这样做。
顺便说一句，您不应该那样引用 COMM_WORLD。每个组件都有一个您应该使用的本地通信。这是您示例的修改版本：
from mpi4py import MPI
import openmdao.api as om
class Exec(om.ExplicitComponent):
def __init__(self, val=-10):
super().__init__()
self.val = val
def setup(self):
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec1, %s, global rank %i, comm size %i, val=%f' % (self.pathname, self.comm.rank, self.comm.size, self.val))
outputs['y'] = self.val
class Exec2(om.ExplicitComponent):
def __init__(self, val):
super().__init__()
self.val = val
def setup(self):
self.add_input('x', 0.)
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec2, %s, global rank %i, comm size %i, val=%f' % (self.pathname, self.comm.rank, self.comm.size, self.val))
outputs['y'] = self.val * inputs['x']
class Summer(om.ExplicitComponent):
def __init__(self, ncase):
super().__init__()
self.ncase = ncase
def setup(self):
for i in range(self.ncase):
self.add_input('y%i' % i, 0.)
self.add_output('sum', 0.)
def compute(self, inputs, outputs):
for i in range(self.ncase):
outputs['sum'] += inputs['y%i' % i]
p = om.Problem()
ncase = 2
par1 = p.model.add_subsystem('par1', om.ParallelGroup())
p.model.add_subsystem('summer1', Summer(ncase))
for i in range(ncase):
par1.add_subsystem('ex_%i' % i, Exec(val=float(i)), min_procs=1)
p.model.connect('par1.ex_%i.y' % i, 'summer1.y%i' % i)
ncase2 = 4
par2 = p.model.add_subsystem('par2', om.ParallelGroup())
p.model.add_subsystem('summer2', Summer(ncase2))
for i in range(ncase2):
par2.add_subsystem('ex_%i' % i, Exec2(float(i)), min_procs=1, max_procs=1)
p.model.connect('summer1.sum', 'par2.ex_%i.x' % i)
p.model.connect('par2.ex_%i.y' % i, 'summer2.y%i' % i)
p.setup()
p.run_model()
运行 在 4 个处理器上给出：
HELLO from Exec1, par1.ex_0, global rank 0, comm size 2, val=0.000000
HELLO from Exec1, par1.ex_0, global rank 1, comm size 2, val=0.000000
HELLO from Exec1, par1.ex_1, global rank 0, comm size 2, val=1.000000
HELLO from Exec1, par1.ex_1, global rank 1, comm size 2, val=1.000000
HELLO from Exec2, par2.ex_3, global rank 0, comm size 1, val=3.000000
HELLO from Exec2, par2.ex_0, global rank 0, comm size 1, val=0.000000
HELLO from Exec2, par2.ex_1, global rank 0, comm size 1, val=1.000000
HELLO from Exec2, par2.ex_2, global rank 0, comm size 1, val=2.000000
所以你看到在 par1 中每个组件都有一个大小为 2 的通信。这是你称之为浪费的重复。不过，我认为这并不浪费，因为您已经了解了如何设置组件的详细信息。 Exec 和 Exec2 都是串行组件（即它们没有 self.options['distributed'] = True）。 OpenMDAO 总是在拥有该组件的本地组通信中的所有级别复制任何串行组件。这种重复的价值是降低 MPI 通信开销。由于该值是在本地计算的，因此您可以在本地传输到该进程上的任何其他串行组件（而不是必须从根广播）。
如果您不想使用本地值，您可以选择自己在连接语句中设置 src_indices=[0]。然后你会强制 OpenMDAO 从那个 comm 的根广播。重复计算仍然会发生。它不应该浪费任何时间，因为在根过程进行计算时重复的过程会一直闲置。您可以争辩说，由于额外的计算，它会浪费一些电能。在大多数情况下，这个成本会很小，但如果您担心它，您可以更改要分配的组件，并将除根以外的所有等级的变量大小设置为 0。然后你可以设置不重复计算。
我们的经验是，大多数时候，通信开销是您想要避免的。这就是我们将其设计为复制的原因，但如果您愿意，您可以自由地解决它。
在下面的例子中，两个 ParallelGroups 被设置为需要不同数量的 procs 来计算它们的串行子组件，似乎一个不应该填满全局通信的组实际上确实如此, 并执行 redundant/repeated 个案例。在以前的版本中，未分配的过程不会执行它们的子组件。虽然它运行没有错误。
from mpi4py import MPI
import openmdao.api as om
class Exec(om.ExplicitComponent):
def __init__(self, val=-10):
super().__init__()
self.val = val
def setup(self):
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec1, %s, global rank %i, val=%f' % (self.name, MPI.COMM_WORLD.rank, self.val))
outputs['y'] = self.val
class Exec2(om.ExplicitComponent):
def __init__(self, val):
super().__init__()
self.val = val
def setup(self):
self.add_input('x', 0.)
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec2, %s, global rank %i, val=%f' % (self.name, MPI.COMM_WORLD.rank, self.val))
outputs['y'] = self.val * inputs['x']
class Summer(om.ExplicitComponent):
def __init__(self, ncase):
super().__init__()
self.ncase = ncase
def setup(self):
for i in range(self.ncase):
self.add_input('y%i' % i, 0.)
self.add_output('sum', 0.)
def compute(self, inputs, outputs):
for i in range(self.ncase):
outputs['sum'] += inputs['y%i' % i]
p = om.Problem()
ncase = 3
par1 = p.model.add_subsystem('par1', om.ParallelGroup())
p.model.add_subsystem('summer1', Summer(ncase))
for i in range(ncase):
par1.add_subsystem('ex_%i' % i, Exec(val=float(i)), min_procs=1)
p.model.connect('par1.ex_%i.y' % i, 'summer1.y%i' % i)
ncase2 = 4
par2 = p.model.add_subsystem('par2', om.ParallelGroup())
p.model.add_subsystem('summer2', Summer(ncase2))
for i in range(ncase2):
par2.add_subsystem('ex_%i' % i, Exec2(float(i)), min_procs=1)
p.model.connect('summer1.sum', 'par2.ex_%i.x' % i)
p.model.connect('par2.ex_%i.y' % i, 'summer2.y%i' % i)
p.setup()
p.run_model()
根据 docs，处理器分配主要基于 proc_weight 使用循环分配方式完成。
The algorithm used for the allocation starts, assuming that the number of processes is greater than or equal to the number of subsystems, by assigning the min_procs for each subsystem. It then adds any remaining processes to subsystems based on their weights, being careful not to exceed their specified max_procs, if any.
所以行为符合预期。 OpenMDAO 寻求分配所有给定的处理器。然后由组件作者明智地使用它们。 如果你真的想要，你可以修改组件，让除 Rank 0 之外的任何东西都不起作用，但我不建议这样做。
顺便说一句，您不应该那样引用 COMM_WORLD。每个组件都有一个您应该使用的本地通信。这是您示例的修改版本：
from mpi4py import MPI
import openmdao.api as om
class Exec(om.ExplicitComponent):
def __init__(self, val=-10):
super().__init__()
self.val = val
def setup(self):
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec1, %s, global rank %i, comm size %i, val=%f' % (self.pathname, self.comm.rank, self.comm.size, self.val))
outputs['y'] = self.val
class Exec2(om.ExplicitComponent):
def __init__(self, val):
super().__init__()
self.val = val
def setup(self):
self.add_input('x', 0.)
self.add_output('y', 0.)
def compute(self, inputs, outputs):
print('HELLO from Exec2, %s, global rank %i, comm size %i, val=%f' % (self.pathname, self.comm.rank, self.comm.size, self.val))
outputs['y'] = self.val * inputs['x']
class Summer(om.ExplicitComponent):
def __init__(self, ncase):
super().__init__()
self.ncase = ncase
def setup(self):
for i in range(self.ncase):
self.add_input('y%i' % i, 0.)
self.add_output('sum', 0.)
def compute(self, inputs, outputs):
for i in range(self.ncase):
outputs['sum'] += inputs['y%i' % i]
p = om.Problem()
ncase = 2
par1 = p.model.add_subsystem('par1', om.ParallelGroup())
p.model.add_subsystem('summer1', Summer(ncase))
for i in range(ncase):
par1.add_subsystem('ex_%i' % i, Exec(val=float(i)), min_procs=1)
p.model.connect('par1.ex_%i.y' % i, 'summer1.y%i' % i)
ncase2 = 4
par2 = p.model.add_subsystem('par2', om.ParallelGroup())
p.model.add_subsystem('summer2', Summer(ncase2))
for i in range(ncase2):
par2.add_subsystem('ex_%i' % i, Exec2(float(i)), min_procs=1, max_procs=1)
p.model.connect('summer1.sum', 'par2.ex_%i.x' % i)
p.model.connect('par2.ex_%i.y' % i, 'summer2.y%i' % i)
p.setup()
p.run_model()
运行 在 4 个处理器上给出：
HELLO from Exec1, par1.ex_0, global rank 0, comm size 2, val=0.000000
HELLO from Exec1, par1.ex_0, global rank 1, comm size 2, val=0.000000
HELLO from Exec1, par1.ex_1, global rank 0, comm size 2, val=1.000000
HELLO from Exec1, par1.ex_1, global rank 1, comm size 2, val=1.000000
HELLO from Exec2, par2.ex_3, global rank 0, comm size 1, val=3.000000
HELLO from Exec2, par2.ex_0, global rank 0, comm size 1, val=0.000000
HELLO from Exec2, par2.ex_1, global rank 0, comm size 1, val=1.000000
HELLO from Exec2, par2.ex_2, global rank 0, comm size 1, val=2.000000
所以你看到在 par1 中每个组件都有一个大小为 2 的通信。这是你称之为浪费的重复。不过，我认为这并不浪费，因为您已经了解了如何设置组件的详细信息。 Exec 和 Exec2 都是串行组件（即它们没有 self.options['distributed'] = True）。 OpenMDAO 总是在拥有该组件的本地组通信中的所有级别复制任何串行组件。这种重复的价值是降低 MPI 通信开销。由于该值是在本地计算的，因此您可以在本地传输到该进程上的任何其他串行组件（而不是必须从根广播）。
如果您不想使用本地值，您可以选择自己在连接语句中设置 src_indices=[0]。然后你会强制 OpenMDAO 从那个 comm 的根广播。重复计算仍然会发生。它不应该浪费任何时间，因为在根过程进行计算时重复的过程会一直闲置。您可以争辩说，由于额外的计算，它会浪费一些电能。在大多数情况下，这个成本会很小，但如果您担心它，您可以更改要分配的组件，并将除根以外的所有等级的变量大小设置为 0。然后你可以设置不重复计算。
我们的经验是，大多数时候，通信开销是您想要避免的。这就是我们将其设计为复制的原因，但如果您愿意，您可以自由地解决它。