Is it possible for the target to service MPI_Get's in parallel?

[frizwi]

Background information

What version of Open MPI are you using? (e.g., v3.0.5, v4.0.2, git branch name and hash, etc.)

Using MPI v4.1.1 on Linux

Describe how Open MPI was installed (e.g., from a source/distribution tarball, from a git clone, from an operating system distribution package, etc.)

Built from source tarball, no other configure flags other than --prefix

---

Details of the problem

Its more of a question: with point-to-point communication, either the non-blocking sends or blocking sends in a thread each can be used to parallelise transfers but is there any way to setup/build/configure the OpenMPI library to do this at the target?

I have a master process and many workers, each worker process performs and MPI_Get with the master process as the target via passive synchronisation. The issue is that as the number of workers increases so does the walltime for the transfers, so I'd like the target to service the Get's in parallel threads - is that possible? With point-to-point comms, I could either use non-blocking sends or blocking sends in a parallel OpenMP sections to achieve this but not clear if I can do this in any way via RMA? Is there some trick that can be done with active synchronisation? The only other thing I can think of is to do an MPI_Put from the master in parallel - will that work?

Thanks in advance for any suggestions, or please point me to some doc, if available. The MPI spec. doesn't really address my issue so I'm hoping to get an understanding of how OpenMPI actually implements RMA to be able to fine tune my application

[devreal]

Both main implementations in OMPI will eventually have parallel transfers (assuming they are large enough that they can actually overlap). The specifics depend on your platform. If your network supports RDMA (IB, Cray Aries) then the hardware will most likely handle transfers in parallel. If not, then there may be a serialization in the starting of transfers by a thread at the target. Either way, there is nothing in the MPI standard that gives you control over it and the design of RMA is such that puts and gets from different origins are independent and can thus happen in parallel.

Some questions regarding the problem at hand: what size are your transfers? What network are you using? Are you building with UCX support? What does the master process do while the worker are getting their data? Can you rule out that the slow-down is caused by limited bandwidth?

[frizwi]

The master array size is in the order of 150,000 doubles, which is divvyed up more or less evenly amongst the workers. Indices are not necessarily consecutive so I create MPI_Datatypes using the array mappings on each of the origins. Yes, good point, I'm not actually building with UCX support. For this test case, I'm only using a small amount of workers, 2/4/8 and only on a 4 x CPU single node so was assuming that it would all be handled via shared mem efficiently. But maybe I should still build with UCX and Knem too. Will try this but it's probably best to move to our HPC cluster (IB) as that's really the aim here - to scale across many nodes.

The master basically does file i/o while the workers are doing the computations.

Thanks for the info, its good to know that on suitable hardware and OpenMPI configurations, parallel transfers can be had by the library without explicit handling in the application

[devreal]

Another aspect that is of importance: how do you allocate the window? Using MPI_Win_allocate is typically the most efficient option as it allows the MPI implementation to manage the memory (and allocate shared memory through mmap or sysv if possible). You wouldn't need knem or xpmem for shared memory support. Things are more difficult with MPI_Win_create (OMPI might use xpmem if available, not sure though) and dynamic windows are the worst in my experience.

If you're running on a single node and you're using MPI_Win_allocate, OMPI should use shared memory to perform the gets without requiring the target to get involved in the transfer. You can force OMPI to use the shared memory-only implementation by passing --mca osc sm to mpirun. If that fails for some reason, something is off with your setup or you're not using allocated windows and might fall back to a less efficient implementation. If it works and scaling is still not what you expect it's likely a bandwidth issue.

When you move to an IB-based cluster, I highly recommend you compile OMPI against UCX. Let us know if you're seeing performance issues there, too. In that case, more information on the datatype layout and/or a small reproducer would be helpful.

[frizwi]

Using MPI_Win_create to create Windows, saves on modifying hundreds of arrays in existing code that uses malloc. Application needs to work in non MPI mode too but will add wrapper functions if it turns out to be a big performance gain

Also a follow up question, on the worker nodes I'm doing something like this call
err = MPI_Win_create(lpatm, szS*sizeof(double), sizeof(double), MPI_INFO_NULL, MPI_COMM_WORLD, &win);
where lpatm is the receive buffer for each worker and used in the MPI_Get call. But strictly speaking, I don't need to expose the worker's memory so can just pass in NULL. However, is it doing any harm? I.e. any issue with it being copied into a public area in memory even though its never referenced by any other process?

If it helps, I'll try and simplify my code into a standalone example and post it here with some timing stats

[devreal]

Using MPI_Win_create to create Windows, saves on modifying hundreds of arrays in existing code that uses malloc. Application needs to work in non MPI mode too but will add wrapper functions if it turns out to be a big performance gain

That is a general problem with MPI RMA, it's not trivial to integrate with existing allocation structures. MPI_Win_create will certainly work with IB networks but shared memory is a little more challenging. I glanced at the rdma one-sided communication implementation in OMPI (enable with --mca osc rdma) and see traces of xpmem so you might want to try to build with xpmem support (requires the xpmem kernel module to work).

Also a follow up question, on the worker nodes I'm doing something like this call
err = MPI_Win_create(lpatm, szS*sizeof(double), sizeof(double), MPI_INFO_NULL, MPI_COMM_WORLD, &win);
where lpatm is the receive buffer for each worker and used in the MPI_Get call.

It shouldn't do any harm but the standard is not clear on whether passing NULL is permitted or not and it certainly does not mandate that the origin buffer is part of a window. By using a window location as origin buffer of a get or put you might save on the registration of memory with the NIC because it is likely already registered. Whether or not that is significant is hard to predict.

[frizwi]

Okay, thanks very much, this is certainly helpful info.

Here is some example code:

#define NSTEPS 5

/*
 * Entry function
 */
int main(int argc, char **argv)
{
  int infid, wpfid, acfid;
  char *infname, *wpfname;
  int isMaster;
  int err;
  int mpi_rank = 0;
  int mpi_size = 0;
  MPI_Win win = NULL;
  int gsz = 0;
  
  /* Initialise MPI */
  MPI_Init(&argc, &argv);  // xxx threaded, how?
  MPI_Comm_size(MPI_COMM_WORLD, &mpi_size);
  MPI_Comm_rank(MPI_COMM_WORLD, &mpi_rank);

  isMaster = (mpi_rank == 0);
  
  /* Check input args */
  if (argc != 3) {
    printf("\n");
    printf("Usage %s <input file> <win_map>\n", argv[0]);
    printf("\n");
    exit(1);
  }

  /* Cache input file name */
  infname = argv[1];
  wpfname = argv[2];

  if (isMaster) {
    geometry_t geom;
    double *patm;
    double *lons, *lats;
    size_t sz;
    int patm_id, lons_id, lats_id;
    int n, s;
    double *wtimes = d_alloc_1d(mpi_size);
    double *ltimes = d_alloc_1d(mpi_size);
    
    printf("\n");
    printf("========\n");
    printf("EMS sim2\n");
    printf("========\n");
    printf("\n");

    /* Initialise */
    memset(&geom, 0, sizeof(geometry_t));
    
    /* Open input file */
    err = nc_open(infname, NC_NOWRITE, &infid);
    NC_ERR(err);

    /* This is the whole sparse array */
    geom.b2_t = getDimLen(infid, "nMesh2_face");

    /* Global arrays */
    sz = geom.b2_t;
    patm = d_alloc_1d(sz+1);
    lons = d_alloc_1d(sz+1);
    lats = d_alloc_1d(sz+1);

    gsz = (int)sz;
    printf("Master gsz = %d\n", gsz);
    MPI_Bcast(&gsz, 1, MPI_INT, 0, MPI_COMM_WORLD);
    
    /* Get the spatial info */
    err = nc_get_var_double(infid, getVarid(infid, "Mesh2_face_x"), lons);
    NC_ERR(err);
    err = nc_get_var_double(infid, getVarid(infid, "Mesh2_face_y"), lats);
    NC_ERR(err);
    
    /* Initialise RMA window */
    err = MPI_Win_create(patm, sz*sizeof(double), sizeof(double),
			 MPI_INFO_NULL, MPI_COMM_WORLD, &win);
    MPI_ERR(err);

    /* 
     * Simulate steps
     * 
     * In the real application, data would be read out of a netCDF file
     */
    for (s=0; s<NSTEPS; s++) {
      double dt = 1.0;
      
      for (n=1; n<=sz; n++)
	patm[n] = n + s*0.12;

      printf("Master step %d\n", s);

      /* Sync with slaves */
      MPI_Barrier(MPI_COMM_WORLD);

      /* Get timings */
      MPI_Gather(&dt, 1, MPI_DOUBLE, ltimes, 1, MPI_DOUBLE, 0, MPI_COMM_WORLD);
      for (n=1; n<mpi_size; n++)
	wtimes[n] += ltimes[n];
    }
    fflush(stdout);
    sleep(1);

    printf("\nTiming stats:\n");
    /* Report timing - xxx better to do reduce? */
    for (n=1; n<mpi_size; n++)
      printf("%03d time = %f\n", n, wtimes[n]);
      
  } else {
    /*
     * Slave code
     */
    int nw = mpi_rank;
    geometry_t window;
    size_t start[2], count[2];
    int szS;
    double *lpatm = NULL;
    MPI_Datatype dtype;
    int s, n;
    size_t indexp = nw;
    FILE *f;
    char buf[256];
    
    /* Initialise */
    memset(&window, 0, sizeof(geometry_t));

    /* Open win_map file */
    err = nc_open(wpfname, NC_NOWRITE, &wpfid);
    NC_ERR(err);

    /* get b2_t, wsa info */
    err = nc_get_var1_int(wpfid, getVarid(wpfid, "b2_t_w"),
			    &indexp, &window.b2_t);
    NC_ERR(err);
    
    /* Allocate local array for this window */
    szS = window.b2_t;
    lpatm = d_alloc_1d(szS+1);
    window.wsa = i_alloc_1d(szS+1);

    /* Get window local to global array mapping */
    start[0] = nw; start[1] = 1;
    count[0] = 1;  count[1] = szS;
    err = nc_get_vara_int(wpfid, getVarid(wpfid, "wsa"),
			    start, count, window.wsa);
    NC_ERR(err);

    MPI_Bcast(&gsz, 1, MPI_INT, 0, MPI_COMM_WORLD);
    printf("%d:Slave szS = %d\n", mpi_rank, szS);
    /* Reset aux/ghost cells */
    for (n=0; n<szS; n++) {
      if (window.wsa[n] >= gsz) {
	// printf("Ooops! wn=%d, window.wsa[%d]=%d\n", nw, n, window.wsa[n]);
	window.wsa[n] = 1;
      }
    }

    /* Initialise RMA window */
    err = MPI_Win_create(lpatm, szS*sizeof(double), sizeof(double),
			 MPI_INFO_NULL, MPI_COMM_WORLD, &win);
    MPI_ERR(err);
    
    /* Set up mapping */
    err = MPI_Type_create_indexed_block(szS, 1, window.wsa, MPI_DOUBLE, &dtype);
    MPI_ERR(err);
    err = MPI_Type_commit(&dtype);
    MPI_ERR(err);

    // sleep(5);
    //printf("Slave(%d) ready\n", nw);
    
    /* sleep to simulate loop and barrier */
    for (s=0; s<NSTEPS; s++) {
      double t0, t1, dt;
      
      /* Sync with master */
      MPI_Barrier(MPI_COMM_WORLD);

      //printf("(%d) before = %f\n", nw, lpatm[10]);

      t0 = MPI_Wtime();
      err = MPI_Win_lock(MPI_LOCK_SHARED, 0, 0, win);
      MPI_ERR(err);
      
      err = MPI_Get(lpatm, szS, MPI_DOUBLE, 0, 0, 1, dtype, win);
      MPI_ERR(err);
      
      err = MPI_Win_unlock(0, win);
      MPI_ERR(err);
      t1 = MPI_Wtime();
      dt = t1-t0;
      
      MPI_Gather(&dt, 1, MPI_DOUBLE, NULL, 1, MPI_DOUBLE, 0, MPI_COMM_WORLD);
      
      /* test */
      //      for (n=0; n<szS; n++)
      //printf("(%d) after  = %f\n", nw, lpatm[10]);
    }

  }

  /* Handy for debugging */
  //  sleep(10);
  // MPI_Barrier(MPI_COMM_WORLD);

  MPI_Finalize();

  return(0);
}

And if I run it with 2 workers:

% mpirun -np 3 ./modelrun tile0b.nc ../compas-mpi/model/tests/setile/tile0_win_map_2W.nc.bak
========
EMS sim2
========
Master gsz = 145230
1:Slave szS = 72606
2:Slave szS = 72624
Master step 0
Master step 1
Master step 2
Master step 3
Master step 4

Timing stats:
001 time = 0.323801
002 time = 0.324102

But with 4 workers, it takes longer!

% mpirun -np 5 ./modelrun tile0b.nc ../compas-mpi/model/tests/setile/tile0_win_map_4W.nc

========
EMS sim2
========

Master gsz = 145230
2:Slave szS = 36320
4:Slave szS = 36307
1:Slave szS = 36291
3:Slave szS = 36312
Master step 0
Master step 1
Master step 2
Master step 3
Master step 4

Timing stats:
001 time = 0.593371
002 time = 0.556528
003 time = 0.592080
004 time = 0.602115

Note: This is still OpenMPI built without UCX or xpmem, so are these timings what you might expect?

[devreal]

Mhh, when breaking it down to (3 operations (lock, get, unlock) x 5 iterations x N worker) I end up with ~10ms per operation in both cases, which seems quite slow. I wanted to test your example both with and without xpmem but it doesn't seem to be self-contained so I fail to compile it, unfortunately. Have you had a chance to run with UCX and/or xpmem?

[frizwi]

You mean the above code is not self-contained? I've attached all the missing source and a Makefile (may need to tweak MPI paths)

ems-sim2-master@0a10f9c0f0b.zip

I tried UCX but the numbers are pretty much the same, probably because UCX wanted KNEM/XPMEM which I haven't installed yet. Do I get xpmem from Google Code?

Also, what about the mpirun command, do I need processor binding or something? Does OpenMPI use [p]threads or OpenMP - if the latter, should I be setting OMP_NUM_THREADS or some other affinity env?

[yosefe]

Do I get xpmem from Google Code?

The original xpmem was written by @hjelmn at https://github.com/hjelmn/xpmem
In addition, UCX community maintains a fork of xpmem at https://github.com/openucx/xpmem
I think they have almost same code now (except maybe adaptions for new kernels)

[frizwi]

Here are the data files need to run my code. The 3rd argument is bogus, not used
data.tar.gz

Thanks @yosefe, I can't get autoconf to work but looks like the OpenMPI on our HPC cluster is built with KNEM, so will try that

[frizwi]

Okay, have tried running the code on a single node on our HPC cluster using OpenMPI 4.1.1 which is built with KNEM.

Here are the results:

[ffr599@gadi-cpu-clx-0971 ems-sim2]$ mpirun -np 3 ./modelrun in.nc win_map_2W.nc 
Master gsz = 145230
1:Slave szS = 72606
2:Slave szS = 72624
Master step 0
Master step 1
Master step 2
Master step 3
Master step 4

Timing stats:
001 time = 0.092372
002 time = 0.092501

[ffr599@gadi-cpu-clx-0971 ems-sim2]$ mpirun -np 5 ./modelrun in.nc win_map_4W.nc 
Master gsz = 145230
4:Slave szS = 36307
1:Slave szS = 36291
2:Slave szS = 36320
3:Slave szS = 36312
Master step 0
Master step 1
Master step 2
Master step 3
Master step 4

Timing stats:
001 time = 0.132498
002 time = 0.123287
003 time = 0.131582
004 time = 0.133267

[ffr599@gadi-cpu-clx-0971 ems-sim2]$ mpirun -np 9 ./modelrun in.nc win_map_8W.nc 
Master gsz = 145230
2:Slave szS = 18157
3:Slave szS = 18163
8:Slave szS = 18137
1:Slave szS = 18159
4:Slave szS = 18154
5:Slave szS = 18150
6:Slave szS = 18158
7:Slave szS = 18152
Master step 0
Master step 1
Master step 2
Master step 3
Master step 4

Timing stats:
001 time = 0.177935
002 time = 0.177614
003 time = 0.179040
004 time = 0.178401
005 time = 0.174531
006 time = 0.178017
007 time = 0.173686
008 time = 0.167491

The absolute timing's not as important as the lack of scaling. As the number of processes increases, the size of each MPI_Get decreases (by factors of 2,4 & 8), so I would expect the timing for transfers to also decrease but that is clearly not the case. The computation scales well but the overall walltime for the application doesn't improve due to the increase in transfer times. Am I not timing properly or have some other conceptual issue?

[bosilca]

Let me try to put in words what your application is doing. A set of worker processes will access the memory of the master process to retrieve some value. Each one of these processes will obtain a lock on the window prior to it's access, get the private value it needs and then unlock the window to pass it to some other process. Thus, you have a totally sequential access pattern, not because of the gets, but because the gets are protected by a lock.

Think about the same pattern between threads, where each thread access a memory location protected by a shared mutex. Can't really make a more sequential process. Thus, your code as is has 0 parallelism in the part you are provided here so far.