Intercommunicator of MPIPoolExecutor

[DirkKuhn]

Hi,

I would like to use the futures interface to launch multiple jobs in parallel. Additionally, I would like to report intermediate values back to the master process which then can tell a job to stop prematurely, if the intermediate values are too bad (pruning with optuna).

Unfortunately, I am stuck on how best to establish the communication between the master and worker processes. Is there a way to get the intercommunicator of the MpiPoolExecutor or should I use Open_port, Introcomm.Accept and Intracomm.Connect.
Afterwards, I would like to merge the intercommunicator with the master process to get an intracommunicator.

Can I do this with mpi4py or should I rather use another library?

Thanks a lot!

Best regards,
Dirk

[dalcinl]

Prima facie, I would say that mpi4py.futures was not designed for the sort of requirements you have. Your "launch multiple jobs in parallel" is a bit vague, it all depends on what these "jobs" are. It they are sequential tasks independent of each other, then things could be manageable. Intermediate results is something that maybe can be implemented by simply running for some predefined number of iterations, return the result, check and resubmit, but perhaps at that point you can implement all these logics in the task itself. You cannot access the executor's intercommunicator, I designed things that way 100% on purpose. mpi4py.futures is a replacement for concurrent.futures, and so far I have no intention of growing the feature set at the expense of maintenance cost from my side. You should probably use another library that better fits your need. Or roll your own Python+MPI solution if you need to go large scale (other libraries may not work so robustly at scale).

PS: Not sure if this will be helpful to your use case, but look at my tip in #324 if your "jobs" have to be parallel.

[DirkKuhn]

Thank you for your answer!
I already have a working implementation using the ProcessPoolExecutor of concurrent.futures. There, I do the communication with Pipes and RawArray. If I can translate this to MPI I should be fine, at least theoretically.
Do you see a problem with the following idea?

from mpi4py.futures import MPIPoolExecutor
from mpi4py import MPI


def f(x: int) -> None:
    port_name = MPI.Lookup_name('ocean')
    intercomm = MPI.COMM_WORLD.Connect(port_name)
    comm = intercomm.Merge(high=True)
    
    # Here I would train a neural network and send the
    # validation metric at each epoch back to the master
    data = {'a': 7, 'b': 3.14}
    comm.send(data, dest=0, tag=0)
    
    comm.Disconnect()
    intercomm.Disconnect()


if __name__ == '__main__':
    port_name = MPI.Open_port()
    MPI.Publish_name('ocean', port_name)
    
    with MPIPoolExecutor(max_workers=2) as pool:
        for x in [0, 1]:
            pool.submit(f, x)

        intercomm = MPI.COMM_SELF.Accept(port_name)
        comm = intercomm.Merge()
        
        for _ in range(2):
            # Here I would listen for the validation metrics and if they are bad
            # I would send a signal to stop training and return from function f
            data = comm.recv()
            print(f'Recived {data}')
        
        comm.Disconnect()
        intercomm.Disconnect()
    
    MPI.Unpublish_name('ocean', port_name)
    MPI.Close_port(port_name)

Also, the setup using the port and creating the intracommunicator would be done in the initializer of the pool.
With wait from concurrent.futures I already ensure that the right amount of jobs are in the pool at all times.

Alternatively, I guess I could try to use dask or ray.

[dalcinl]

Well, your approach may definitely work, of course, as long as the executor is operating in spawn-mode rather than legacy MPI-1 mode.

I'm not sure why you are doing a Merge, do you really need that? IMHO using intercomms for the master-worker approach is slightly easier to reason about. But that's ultimately a matter of style, I guess.

I still think that rolling a custom Python+MPI may be more appropriate. After all, you are forcing all your tasks to synchronize and communicate. Of course, I can see that using mpi4py.futures will save you of many of the gory details of setting up your own master-worker framework. Just be aware that you are breaking the original design and intent of the framework, so you are a bit on your own.

Please take a quick look at #324, it is somewhat related to what you are doing (you need your tasks to synchronize).
Doing for x in range(executor._max_workers): ... is a trivial generalization to support the alternative ways of specifying the number of workers (universe size, environment variable).

[DirkKuhn]

I do not think that I understand what you want to tell me. I basically do

futures = set()
with MPIPoolExecutor(max_workers=num_workers) as pool:
    for _ in range(num_workers):
        f = pool.submit(...)
        futures.add(f)

    while ...:
        next_result = wait(futures, return_when=FIRST_COMPLETE)
        futures.remove(next_result)
        f = pool.submit(...)
        futures.add(f)

i.e., at the beginning I fill all workers and whenever a job is completed I submit the next job. Logically, I do not think that I need more synchronization. Could you please tell me if I miss something?
I would also like to know whether the actual number of workers can be lower as max_workers. If so can I get the actual number of workers?

Independent of that I want to report intermediate values to a backgroup thread running in the master process which can tell jobs to stop prematurely. I tought I could send intermediate values to the master process. And if a job should be stopped tell that to the worker by setting a flag using one-sided communication with passive target synchronization (to avoid waiting on the worker side and since another job on another worker might be affected aswell).

datatype = MPI.BOOL
np_dtype = dtlib.to_numpy_dtype(datatype)
itemsize = datatype.Get_size()

## master side (rank=0)

true_value = np.ones(1, dtype=np_dtype)

while True:
    # Block until next intermediate value is recieved
    trial_num, step_num, intermediate_value = comm.recv()
    should_prune = report(trial_num, step_num, intermediate_value)
    
    if should_prune:  # Happens rarely
        # Tell each worker that job `trial_num` should be stopped using one-sided communication
        for worker_rank in range(1, comm.Get_size()):
            win.Lock(rank=worker_rank)
            win.Put(true_value, target_rank=worker_rank, target=(trial_num, 1, datatype))
            win.Unlock(rank=worker_rank)

## worker side (rank>0)

should_stop = np.zeros(1, dtype=np_dtype)

training = True
while training:  # Many iterations if not stopped
    intermediate_value, step_num = train_step()
    
    # Report intermediate_value
    comm.send((trial_num, step_num, intermediate_value), dest=0)
    
    # Check whether should stop
    win.Lock(rank=rank)
    win.Get(should_stop, target_rank=rank, target=(trial_num, 1, datatype))
    win.Unlock(rank=rank)
    training = not should_stop[0]

However, I am not so sure whether I am using one-sided communication correctly.

[dalcinl]

i.e., at the beginning I fill all workers and whenever a job is completed I submit the next job.

Your "outer-loop" jobs are parallel tasks (you are creating a communicator in there), and that's not the model mpi4py.futures was designed for. Nonetheless, if you make sure, by any mechanism, that all workers agree on executing concurrently the parallel parts of the job, then you are fine.

Logically, I do not think that I need more synchronization. Could you please tell me if I miss something?

You know the logics of your application better than I. I was just throwing out general warnings about things that could go wrong.

I would also like to know whether the actual number of workers can be lower as max_workers.

mpi4py.futures spawns workers with mpi_info = {'soft': f'1:{max_workers}'} (look for it in the sources). When you pass the soft key to MPI, the implementation is free to spawn any number of processes in the specified soft range (Fortran 90 style triplet)

If so can I get the actual number of workers?

Use executor._max_workers. Despite the name, it is the actual number of worker processes, see #94. I'm considering adding a public property executor.num_workers in the future.

by setting a flag using one-sided communication with passive target synchronization

OK, in that case, you definitely need an intracomm to create the Window object, so disregard my comments about intercomms and intracomms.

[DirkKuhn]

I thought I could set a flag in a worker process asynchronously using one-sided communciation. Unfortunately, the following toy problem does not seem to confirm this. First, I initialize each buffer with zeros. Then, I set in rank=0 the first and third flag of rank=1. In rank=1 I wait some time so that rank=0 has time to complete the transfer and then print the buffer a couple of times.

import time
import numpy as np
from mpi4py import MPI
from mpi4py.util import dtlib

comm = MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

datatype = MPI.BOOL
np_dtype = dtlib.to_numpy_dtype(datatype)
itemsize = datatype.Get_size()

N = 3
win_size = N * itemsize
win = MPI.Win.Allocate(win_size, comm=comm)

# Fill buffer with 0s
buf = np.zeros(N, dtype=np_dtype)
buf.fill(0)
win.Lock(rank=rank)
win.Put(buf, target_rank=rank)
win.Unlock(rank=rank)
comm.Barrier()

if rank == 0:
    true_value = np.ones(1, dtype=np_dtype)
    for pos in [0, 2]:
        win.Lock(rank=1, lock_type=MPI.LOCK_EXCLUSIVE)
        win.Put(true_value, target_rank=1, target=(pos, 1, datatype))
        win.Unlock(rank=1)
        print(f'{rank} {size} End put {pos}', flush=True)
elif rank == 1:
    time.sleep(1)
    for _ in range(5):
        win.Lock(rank=1, lock_type=MPI.LOCK_EXCLUSIVE)
        win.Get(buf, target_rank=1)
        win.Unlock(rank=1)
        print(f'{rank} {size} {list(buf)}', flush=True)

I would expect the output

End put 0
End put 2
[True, False, True]
[True, False, True]

but what I am getting is

[False, False, False]
End put 0
[True, False, False]
End put 2

It seems to me like rank=0 is waiting for rank=1. What is going on here? Does Win behave more like a queue than memory, is this a result of the separate memory model or am I doing some kind of mistake?
Sorry, for the long question. And thank you for your help.

[dalcinl]

Alternatively, search for the installed mpi4py/__init__.py file and change this line to False.

[DirkKuhn]

Sorry that question was redundant as it has already been answered in an issue.
Unfortunately, I do not seem to be able to establish an intercommunicator between the master and worker processes when using the MPIPoolExecutor with Intel MPI.

from mpi4py.futures import MPIPoolExecutor
from mpi4py import MPI

def initialize() -> None:
    port_name = MPI.Lookup_name('ocean')
    print('Connect worker', flush=True)
    intercomm = MPI.COMM_WORLD.Connect(port_name)

def f(x: int) -> None:  
    print(f'Got {x}', flush=True)

if __name__ == '__main__':
    port_name = MPI.Open_port()
    MPI.Publish_name('ocean', port_name)
    
    with MPIPoolExecutor(max_workers=1, initializer=initialize) as pool:
        pool.submit(f, 0)
        print('Connect master', flush=True)
        intercomm = MPI.COMM_SELF.Accept(port_name)
    
    MPI.Unpublish_name('ocean', port_name)
    MPI.Close_port(port_name)

prints "Connect master" and blocks. initialize and f do not seem to be executed on the worker. If I remove the Accept and Connect calls everything works fine.
Do you have any idea what I could try?
If you don't have an idea immediately, please don't bother.

[dalcinl]

I don't really know what's going on. Just a quick tip: Instead of using name publishing/lookup, why don't you pass port_name with MPIPoolExecutor(..., initializer=initialize, initargs=(port_name,)) and then you write def initialize(port_name): ...? Supporting publish/lookup may be tricky for implementations, passing the port name directly may ease things.

FWIW, I reviewed my CI logs, and publish/lookup and accept/connect seem to be working just fine with Intel MPI on Windows.

from mpi4py.futures import MPIPoolExecutor
from mpi4py import MPI

def initialize(port_name) -> None:
    print('Connect worker', flush=True)
    intercomm = MPI.COMM_WORLD.Connect(port_name)

def f(x: int) -> None:
    print(f'Got {x}', flush=True)

if __name__ == '__main__':

    port_name = MPI.Open_port()

    with MPIPoolExecutor(
        max_workers=1,
        initializer=initialize,
        initargs=(port_name,)
    ) as pool:
        pool.submit(f, 0)
        print('Connect master', flush=True)
        intercomm = MPI.COMM_SELF.Accept(port_name)
    MPI.Close_port(port_name)

    print("done")

[DirkKuhn]

Using initargs is a great idea!
I wonder what I do differently. I mean I already install the newest wheel. And I guess you also set MPI4PY_RC_RECV_MPROBE=false.
If I follow the steps at https://github.com/mpi4py/mpi4py/actions/runs/5102155021/jobs/9171538354 will I have to same setup?

[dalcinl]

If I follow the steps at https://github.com/mpi4py/mpi4py/actions/runs/5102155021/jobs/9171538354 will I have to same setup?

Well, I guess so. I'm not an Windows expert, haven't used it in 20 years. The most important step would be to run setvars.bat once in your console before running. Beyond that, I don't know what else suggest.