python - 在 Django 应用程序中启动 MPI 工作进程

Question

我们想在我们的 Django 应用程序中运行一些后台进程。看起来 Celery 是最常见的解决方案，但我们团队对 MPI 更熟悉，所以我正在尝试它。我想创建一个启动 MPI 工作人员池的 Django 管理命令，因此我阅读了Django 管理命令和 MPI4py 的动态进程管理。

我编写了一个管理命令来运行车队管理器和一个管理命令来运行一个工人。车队经理成功使用MPI.COMM_SELF.Spawn()启动工作人员，但他们无法相互通信。经理和第一个工人的等级都是 0，所以看起来他们正在使用单独的通信器。

我怎样才能让经理和工人使用同一个通信器？

Answer 1

诀窍是合并两个通信器，如this answer to a C question中所述。在MPI4py 文档的帮助下，我将其转换为 Python：

# myproject/myapp/management/commands/runfleet.py
from mpi4py import MPI
from optparse import make_option

from django.core.management.base import BaseCommand

import sys

class Command(BaseCommand):
    help = 'Launches the example manager and workers.'

    option_list = BaseCommand.option_list + (
        make_option('--workers', '-w', type='int', default=1), )

    def handle(self, *args, **options):
        self.stdout.write("Manager launched.")

        worker_count = options['workers']
        manage_script = sys.argv[0]
        comm = MPI.COMM_SELF.Spawn(sys.executable,
                                   args=[manage_script, 'fleetworker'],
                                   maxprocs=worker_count).Merge()
        self.stdout.write('Manager rank {}.'.format(comm.Get_rank()))

        start_data = [None] # First item is sent to manager and ignored
        for i in range(worker_count):
            start_data.append("Item {}".format(i))
        comm.scatter(start_data)

        end_data = comm.gather()
        self.stdout.write('Manager received data {!r}.'.format(end_data))

        comm.Disconnect()

worker 命令如下所示：

# myproject/myapp/management/commands/fleetworker.py
from mpi4py import MPI

from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = 'Example worker process.'

    def handle(self, *args, **options):
        self.stdout.write("Worker launched.")

        comm = MPI.Comm.Get_parent().Merge()
        rank = comm.Get_rank()
        self.stdout.write('Worker rank {}.'.format(rank))

        data = comm.scatter()
        result = "{!r}, {!r}".format(rank, data)

        comm.gather(result)
        self.stdout.write("Finished worker.")

        comm.Disconnect()