[starpu-devel] MPI cache bug when switching communicators

[Atte Torri <atte.torri@universite-paris-saclay.fr>]

Dear all,

I have a bug in the cache system of StarPU in my code; I'm trying to run tasks asynchronously on two MPI communicators as follows:
1. Run async tasks on data handles using communicator MPI_COMM_WORLD

2. Run async tasks on the same data handles using a new communicator that renames the ranks in MPI_COMM_WORLD

3. Run async tasks using MPI_COMM_WORLD again on the same data handles.

I have attached a simplified example file where this bug occurs (with synchronous tasks).

I compile it with `mpicc $(pkg-config --cflags starpumpi-1.4) $(pkg-config --libs starpumpi-1.4) starpu-bug.c -o starpu-bug`

Then run it with `mpirun -np 3 --bind-to numa --tag-output starpu-bug`

When the cache is not flushed manually (line 63), it does not finish as rank 1 does not send data to rank 0 (it believes that rank 0 already has it). When the cache is flushed (or disabled), everything works as expected.

Best,

TORRI Atte

#include <starpu.h>
#include <starpu_mpi.h>

#define GRID_SIZE 3

void task(void *buffers[], void *cl_arg) {
	int *a = (int*)STARPU_VARIABLE_GET_PTR(buffers[0]);
	int *b = (int*)STARPU_VARIABLE_GET_PTR(buffers[1]);
	*a = *b;
}

struct starpu_codelet codelet = {
  .cpu_funcs = { task },
	.nbuffers = 2,
	.modes = { STARPU_RW, STARPU_R }
};

int main(int argc, char** argv) {
	starpu_mpi_init_conf(&argc, &argv, 1, MPI_COMM_WORLD, NULL);

	int rank, size;
	starpu_mpi_comm_rank(MPI_COMM_WORLD, &rank);
  starpu_mpi_comm_size(MPI_COMM_WORLD, &size);

	MPI_Comm comm_inverse;
	MPI_Comm_split(MPI_COMM_WORLD, 0, size - 1 - rank, &comm_inverse);
	starpu_mpi_comm_register(comm_inverse);

	int local_var = rank;
	starpu_data_handle_t handles[GRID_SIZE];
	for(int i = 0; i < GRID_SIZE; i++) {
		if(i == rank) {
			starpu_variable_data_register(&handles[i], STARPU_MAIN_RAM, (uintptr_t)&local_var, sizeof(int));
		} else {
			starpu_variable_data_register(&handles[i], -1, NULL, sizeof(int));
		}
		starpu_mpi_data_register(handles[i], i, i);
	}

	for(int i = 0; i < GRID_SIZE; i++) {
		starpu_mpi_task_insert(MPI_COMM_WORLD, &codelet,
													 STARPU_RW, handles[i],
													 STARPU_R, handles[(i+1) % GRID_SIZE],
													 0);
	}

	starpu_mpi_wait_for_all(MPI_COMM_WORLD);
	starpu_mpi_barrier(MPI_COMM_WORLD);

	for(int i = 0; i < GRID_SIZE; i++) {
		starpu_mpi_data_set_rank_comm(handles[i], size - 1 - i, comm_inverse);
	}

	for(int i = 0; i < GRID_SIZE; i++) {
		starpu_mpi_task_insert(comm_inverse, &codelet,
													 STARPU_RW, handles[(i+1) % GRID_SIZE],
													 STARPU_R, handles[i],
													 0);
	}
	
	starpu_mpi_wait_for_all(comm_inverse);
	starpu_mpi_barrier(MPI_COMM_WORLD);
	// starpu_mpi_cache_flush_all_data(comm_inverse);

	for(int i = 0; i < GRID_SIZE; i++) {
		starpu_mpi_data_set_rank_comm(handles[i], i, MPI_COMM_WORLD);
	}

	for(int i = 0; i < GRID_SIZE; i++) {
		starpu_mpi_task_insert(MPI_COMM_WORLD, &codelet,
													 STARPU_RW, handles[i],
													 STARPU_R, handles[(i+1) % GRID_SIZE],
													 0);
	}
	
	starpu_mpi_shutdown();
	return 0;
}