diff --git a/src/include/mpir_comm.h b/src/include/mpir_comm.h index 6714b588104..d005f3944f8 100644 --- a/src/include/mpir_comm.h +++ b/src/include/mpir_comm.h @@ -230,6 +230,9 @@ struct MPIR_Comm { int topo_wave_tree_lat_diff_switches; int topo_wave_tree_lat_same_switches; MPIR_Treealgo_tree_t *topo_wave_tree; +#ifdef ENABLE_CNC + void *cnc_priv; +#endif } coll; #if defined HAVE_HCOLL diff --git a/src/mpi/coll/Makefile.mk b/src/mpi/coll/Makefile.mk index 567efafde43..14c8ab8436f 100644 --- a/src/mpi/coll/Makefile.mk +++ b/src/mpi/coll/Makefile.mk @@ -11,6 +11,7 @@ include $(top_srcdir)/src/mpi/coll/alltoallv/Makefile.mk include $(top_srcdir)/src/mpi/coll/alltoallw/Makefile.mk include $(top_srcdir)/src/mpi/coll/barrier/Makefile.mk include $(top_srcdir)/src/mpi/coll/bcast/Makefile.mk +include $(top_srcdir)/src/mpi/coll/cnc/Makefile.mk include $(top_srcdir)/src/mpi/coll/exscan/Makefile.mk include $(top_srcdir)/src/mpi/coll/gather/Makefile.mk include $(top_srcdir)/src/mpi/coll/gatherv/Makefile.mk diff --git a/src/mpi/coll/cnc/Makefile.mk b/src/mpi/coll/cnc/Makefile.mk new file mode 100644 index 00000000000..2ac14b0dcd2 --- /dev/null +++ b/src/mpi/coll/cnc/Makefile.mk @@ -0,0 +1,14 @@ +## +## Copyright (C) by Argonne National Laboratory +## See COPYRIGHT in top-level directory +## + +if BUILD_CNC + +mpi_core_sources += \ + src/mpi/coll/cnc/cnc_coll.c + +noinst_HEADERS += \ + src/mpi/coll/cnc/cnc_internal.h + +endif BUILD_CNC diff --git a/src/mpi/coll/cnc/cnc_coll.c b/src/mpi/coll/cnc/cnc_coll.c new file mode 100644 index 00000000000..5396c9916bf --- /dev/null +++ b/src/mpi/coll/cnc/cnc_coll.c @@ -0,0 +1,961 @@ +/* + * Copyright (C) by Argonne National Laboratory + * See COPYRIGHT in top-level directory + */ + +#include "mpiimpl.h" +#include +#include +#include "cnc_internal.h" + +/* Sentinel stored in comm->coll.cnc_priv on the private dup'd communicator + * while cnc_init() runs its internal MPI collectives, preventing recursive + * entry into the cnc algorithms. */ +#define CNC_PRIV_SKIP ((void *) -1) + +struct cnc_comm_priv { + MPIR_Comm *dup_comm; + cnc_oob_t oob; + cnc_transport_t *transport; +}; + +static int cnc_get_transport(MPIR_Comm * comm_ptr, cnc_transport_t ** t_out) +{ + int mpi_errno = MPI_SUCCESS; + struct cnc_comm_priv *priv = comm_ptr->coll.cnc_priv; + + if (priv == CNC_PRIV_SKIP) + return MPI_ERR_OTHER; + if (priv) { + *t_out = priv->transport; + return MPI_SUCCESS; + } + + priv = MPL_calloc(1, sizeof(*priv), MPL_MEM_COLL); + if (!priv) { + comm_ptr->coll.cnc_priv = CNC_PRIV_SKIP; + return MPI_ERR_OTHER; + } + + /* MPII_Comm_copy allocates a dedicated context ID without copying user + * communicator attributes or invoking attribute callbacks. */ + mpi_errno = MPII_Comm_copy(comm_ptr, comm_ptr->local_size, NULL, &priv->dup_comm); + if (mpi_errno != MPI_SUCCESS) { + MPL_free(priv); + comm_ptr->coll.cnc_priv = CNC_PRIV_SKIP; + return mpi_errno; + } + + priv->dup_comm->coll.cnc_priv = CNC_PRIV_SKIP; + + if (cnc_oob_mpi_create(&priv->oob, priv->dup_comm->handle) != CNC_OK) { + MPIR_Comm_release(priv->dup_comm); + MPL_free(priv); + comm_ptr->coll.cnc_priv = CNC_PRIV_SKIP; + return MPI_ERR_OTHER; + } + + if (cnc_init(&priv->transport, &priv->oob) != CNC_OK) { + cnc_oob_mpi_destroy(&priv->oob); + MPIR_Comm_release(priv->dup_comm); + MPL_free(priv); + comm_ptr->coll.cnc_priv = CNC_PRIV_SKIP; + return MPI_ERR_OTHER; + } + + comm_ptr->coll.cnc_priv = priv; + *t_out = priv->transport; + return MPI_SUCCESS; +} + +int MPII_Cnc_comm_cleanup(MPIR_Comm * comm_ptr) +{ + struct cnc_comm_priv *priv = comm_ptr->coll.cnc_priv; + + if (!priv || priv == CNC_PRIV_SKIP) { + comm_ptr->coll.cnc_priv = NULL; + return MPI_SUCCESS; + } + + comm_ptr->coll.cnc_priv = NULL; + cnc_fini(priv->transport); + cnc_oob_mpi_destroy(&priv->oob); + MPIR_Comm_release(priv->dup_comm); + MPL_free(priv); + return MPI_SUCCESS; +} + +static int cnc_can_run(MPIR_Comm * comm_ptr) +{ + return !MPIR_ThreadInfo.isThreaded && comm_ptr->coll.cnc_priv != CNC_PRIV_SKIP; +} + +static int cnc_contig_host(const void *buf, MPI_Datatype datatype) +{ + int is_contig; + + MPIR_Datatype_is_contig(datatype, &is_contig); + return is_contig && !MPIR_GPU_query_pointer_is_dev(buf); +} + +/* Stage a (buf, count, datatype) buffer of nbytes bytes for reading by + * libcncoll. Contiguous host buffers are used in place; non-contiguous or + * device buffers are packed into a temporary host buffer via MPIR_Localcopy + * so ranks with differing local datatype layouts or memory kinds never + * diverge on collective eligibility. */ +static int cnc_stage_in(const void *buf, MPI_Aint count, MPI_Datatype datatype, size_t nbytes, + const void **wire_buf_out, void **tmp_out) +{ + MPI_Aint true_lb; + + *tmp_out = NULL; + if (cnc_contig_host(buf, datatype)) { + MPIR_Datatype_get_true_lb(datatype, &true_lb); + *wire_buf_out = (const char *) buf + true_lb; + return MPI_SUCCESS; + } + + *tmp_out = MPL_malloc(nbytes ? nbytes : 1, MPL_MEM_COLL); + if (!*tmp_out) + return MPI_ERR_OTHER; + if (nbytes > 0) { + int mpi_errno = MPIR_Localcopy(buf, count, datatype, *tmp_out, (MPI_Aint) nbytes, + MPIR_BYTE_INTERNAL); + if (mpi_errno != MPI_SUCCESS) { + MPL_free(*tmp_out); + *tmp_out = NULL; + return mpi_errno; + } + } + *wire_buf_out = *tmp_out; + return MPI_SUCCESS; +} + +static int cnc_stage_out_alloc(void *buf, MPI_Datatype datatype, size_t nbytes, + void **wire_buf_out, void **tmp_out) +{ + MPI_Aint true_lb; + + *tmp_out = NULL; + if (cnc_contig_host(buf, datatype)) { + MPIR_Datatype_get_true_lb(datatype, &true_lb); + *wire_buf_out = (char *) buf + true_lb; + return MPI_SUCCESS; + } + + *tmp_out = MPL_malloc(nbytes ? nbytes : 1, MPL_MEM_COLL); + if (!*tmp_out) + return MPI_ERR_OTHER; + *wire_buf_out = *tmp_out; + return MPI_SUCCESS; +} + +static int cnc_stage_out_finish(void *user_buf, MPI_Aint count, MPI_Datatype datatype, + size_t nbytes, void *tmp_buf, int status_ok) +{ + int mpi_errno = MPI_SUCCESS; + + if (!tmp_buf) + return status_ok ? MPI_SUCCESS : MPI_ERR_OTHER; + if (status_ok && nbytes > 0) + mpi_errno = MPIR_Localcopy(tmp_buf, (MPI_Aint) nbytes, MPIR_BYTE_INTERNAL, user_buf, + count, datatype); + MPL_free(tmp_buf); + return status_ok ? mpi_errno : MPI_ERR_OTHER; +} + +/* Prepare byte counts, byte displacements, and a contiguous host buffer for a + * variable-count (*v) collective side. When the caller buffer is contiguous on + * host and all non-empty displacements are non-negative, the user buffer is + * used directly; otherwise segments are packed/unpacked into a contiguous + * temporary buffer. */ +static int cnc_v_stage_prep(const void *buf, const MPI_Aint * counts, const MPI_Aint * displs, + MPI_Datatype datatype, int n, int copy_in, int inplace_slot, + void **wire_buf_out, void **tmp_out, size_t * counts_bytes, + ptrdiff_t * displs_bytes) +{ + MPI_Aint type_size, type_extent, true_lb = 0; + int direct; + size_t total = 0; + + *tmp_out = NULL; + MPIR_Datatype_get_size_macro(datatype, type_size); + MPIR_Datatype_get_extent_macro(datatype, type_extent); + direct = cnc_contig_host(buf, datatype); + for (int i = 0; i < n; i++) { + size_t seg = counts[i] > 0 ? (size_t) (counts[i] * type_size) : 0; + counts_bytes[i] = seg; + if (counts[i] > 0 && displs[i] * type_extent < 0) + direct = 0; + } + + if (direct) { + MPIR_Datatype_get_true_lb(datatype, &true_lb); + for (int i = 0; i < n; i++) + displs_bytes[i] = counts[i] > 0 ? (ptrdiff_t) (displs[i] * type_extent) : 0; + *wire_buf_out = (char *) buf + true_lb; + return MPI_SUCCESS; + } + + for (int i = 0; i < n; i++) { + displs_bytes[i] = (ptrdiff_t) total; + total += counts_bytes[i]; + } + *tmp_out = MPL_malloc(total ? total : 1, MPL_MEM_COLL); + if (!*tmp_out) + return MPI_ERR_OTHER; + if (copy_in) { + for (int i = 0; i < n; i++) { + if (counts_bytes[i] > 0) { + int rc = MPIR_Localcopy((const char *) buf + displs[i] * type_extent, counts[i], + datatype, (char *) *tmp_out + displs_bytes[i], + (MPI_Aint) counts_bytes[i], MPIR_BYTE_INTERNAL); + if (rc != MPI_SUCCESS) { + MPL_free(*tmp_out); + *tmp_out = NULL; + return rc; + } + } + } + } else if (inplace_slot >= 0 && counts_bytes[inplace_slot] > 0) { + int rc = MPIR_Localcopy((const char *) buf + displs[inplace_slot] * type_extent, + counts[inplace_slot], datatype, + (char *) *tmp_out + displs_bytes[inplace_slot], + (MPI_Aint) counts_bytes[inplace_slot], MPIR_BYTE_INTERNAL); + if (rc != MPI_SUCCESS) { + MPL_free(*tmp_out); + *tmp_out = NULL; + return rc; + } + } + *wire_buf_out = *tmp_out; + return MPI_SUCCESS; +} + +static int cnc_v_stage_unpack(void *user_buf, const MPI_Aint * counts, const MPI_Aint * displs, + MPI_Datatype datatype, int n, const size_t * counts_bytes, + const ptrdiff_t * displs_bytes, void *tmp_buf, int status_ok) +{ + MPI_Aint type_extent; + int mpi_errno = MPI_SUCCESS; + + if (!tmp_buf) + return status_ok ? MPI_SUCCESS : MPI_ERR_OTHER; + if (status_ok) { + MPIR_Datatype_get_extent_macro(datatype, type_extent); + for (int i = 0; i < n; i++) { + if (counts_bytes[i] > 0) { + mpi_errno = MPIR_Localcopy((const char *) tmp_buf + displs_bytes[i], + (MPI_Aint) counts_bytes[i], MPIR_BYTE_INTERNAL, + (char *) user_buf + displs[i] * type_extent, + counts[i], datatype); + if (mpi_errno != MPI_SUCCESS) + break; + } + } + } + MPL_free(tmp_buf); + return status_ok ? mpi_errno : MPI_ERR_OTHER; +} + +static int cnc_map_dtype_op(MPI_Datatype datatype, MPI_Op op, cnc_datatype_t * dt_out, + cnc_reduce_op_t * op_out) +{ + switch (MPIR_DATATYPE_GET_RAW_INTERNAL(datatype)) { + case MPIR_FLOAT32: + *dt_out = CNC_DTYPE_F32; + break; + case MPIR_INT32: + *dt_out = CNC_DTYPE_I32; + break; + default: + return 0; + } + + switch (op) { + case MPI_SUM: + *op_out = CNC_REDUCE_SUM; + break; + case MPI_PROD: + *op_out = CNC_REDUCE_PROD; + break; + case MPI_MIN: + *op_out = CNC_REDUCE_MIN; + break; + case MPI_MAX: + *op_out = CNC_REDUCE_MAX; + break; + default: + return 0; + } + + return 1; +} + +int MPIR_Barrier_intra_cnc(MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Barrier_intra_dissemination(comm_ptr, coll_attr); + + cnc_barrier(comm_ptr->local_size, comm_ptr->rank, t); + return MPI_SUCCESS; +} + +int MPIR_Bcast_intra_cnc(void *buffer, MPI_Aint count, MPI_Datatype datatype, int root, + MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + MPI_Aint type_size; + size_t nbytes; + void *wire_buf = NULL, *tmp_buf = NULL; + int rank = comm_ptr->rank; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Bcast_intra_binomial(buffer, count, datatype, root, comm_ptr, coll_attr); + + MPIR_Datatype_get_size_macro(datatype, type_size); + nbytes = (size_t) (count * type_size); + + if (rank == root) { + const void *sbuf_wire = NULL; + rc = cnc_stage_in(buffer, count, datatype, nbytes, &sbuf_wire, &tmp_buf); + wire_buf = (void *) sbuf_wire; + } else { + rc = cnc_stage_out_alloc(buffer, datatype, nbytes, &wire_buf, &tmp_buf); + } + if (rc != MPI_SUCCESS) + return rc; + + st = cnc_bcast(CNC_IN_PLACE, wire_buf, nbytes, comm_ptr->local_size, rank, root, t, NULL); + if (rank == root) { + MPL_free(tmp_buf); + return st == CNC_OK ? MPI_SUCCESS : MPI_ERR_OTHER; + } + return cnc_stage_out_finish(buffer, count, datatype, nbytes, tmp_buf, st == CNC_OK); +} + +int MPIR_Gather_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + size_t nbytes; + const void *sbuf_adj = NULL; + void *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rank = comm_ptr->rank; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Gather_intra_binomial(sendbuf, sendcount, sendtype, recvbuf, recvcount, + recvtype, root, comm_ptr, coll_attr); + + if (rank == root) { + MPI_Aint recv_size, recv_extent; + MPIR_Datatype_get_size_macro(recvtype, recv_size); + nbytes = (size_t) (recvcount * recv_size); + rc = cnc_stage_out_alloc(recvbuf, recvtype, (size_t) n * nbytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) + return rc; + if (sendbuf == MPI_IN_PLACE) { + sbuf_adj = (const char *) rbuf_adj + rank * nbytes; + if (tmp_rbuf && nbytes > 0) { + MPIR_Datatype_get_extent_macro(recvtype, recv_extent); + rc = MPIR_Localcopy((const char *) recvbuf + rank * recvcount * recv_extent, + recvcount, recvtype, (void *) sbuf_adj, (MPI_Aint) nbytes, + MPIR_BYTE_INTERNAL); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_rbuf); + return rc; + } + } + } else { + rc = cnc_stage_in(sendbuf, sendcount, sendtype, nbytes, &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_rbuf); + return rc; + } + } + } else { + MPI_Aint send_size; + MPIR_Datatype_get_size_macro(sendtype, send_size); + nbytes = (size_t) (sendcount * send_size); + rc = cnc_stage_in(sendbuf, sendcount, sendtype, nbytes, &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + } + + st = cnc_gather(sbuf_adj, rbuf_adj, nbytes, n, rank, root, t, NULL); + MPL_free(tmp_sbuf); + if (rank == root) + return cnc_stage_out_finish(recvbuf, recvcount * (MPI_Aint) n, recvtype, + (size_t) n * nbytes, tmp_rbuf, st == CNC_OK); + return st == CNC_OK ? MPI_SUCCESS : MPI_ERR_OTHER; +} + +int MPIR_Gatherv_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, const MPI_Aint * displs, + MPI_Datatype recvtype, int root, MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + size_t send_nbytes; + const void *sbuf_adj = NULL; + void *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rank = comm_ptr->rank; + size_t *rcounts_bytes = NULL; + ptrdiff_t *rdispls_bytes = NULL; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Gatherv_allcomm_linear(sendbuf, sendcount, sendtype, recvbuf, recvcounts, + displs, recvtype, root, comm_ptr, coll_attr); + + if (rank == root) { + rcounts_bytes = MPL_malloc((size_t) n * sizeof(*rcounts_bytes), MPL_MEM_COLL); + rdispls_bytes = MPL_malloc((size_t) n * sizeof(*rdispls_bytes), MPL_MEM_COLL); + if (!rcounts_bytes || !rdispls_bytes) { + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return MPI_ERR_OTHER; + } + rc = cnc_v_stage_prep(recvbuf, recvcounts, displs, recvtype, n, 0, + sendbuf == MPI_IN_PLACE ? rank : -1, &rbuf_adj, &tmp_rbuf, + rcounts_bytes, rdispls_bytes); + if (rc != MPI_SUCCESS) { + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return rc; + } + send_nbytes = rcounts_bytes[rank]; + if (sendbuf == MPI_IN_PLACE) { + sbuf_adj = (const char *) rbuf_adj + rdispls_bytes[rank]; + } else { + rc = cnc_stage_in(sendbuf, sendcount, sendtype, send_nbytes, &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_rbuf); + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return rc; + } + } + } else { + MPI_Aint send_size; + MPIR_Datatype_get_size_macro(sendtype, send_size); + send_nbytes = sendcount > 0 ? (size_t) (sendcount * send_size) : 0; + rc = cnc_stage_in(sendbuf, sendcount, sendtype, send_nbytes, &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + rcounts_bytes = MPL_calloc((size_t) n, sizeof(*rcounts_bytes), MPL_MEM_COLL); + if (!rcounts_bytes) { + MPL_free(tmp_sbuf); + return MPI_ERR_OTHER; + } + rcounts_bytes[rank] = send_nbytes; + } + + st = cnc_gatherv_linear(sbuf_adj, send_nbytes, rbuf_adj, rcounts_bytes, rdispls_bytes, + n, rank, root, t); + MPL_free(tmp_sbuf); + if (rank == root) + rc = cnc_v_stage_unpack(recvbuf, recvcounts, displs, recvtype, n, rcounts_bytes, + rdispls_bytes, tmp_rbuf, st == CNC_OK); + else + rc = (st == CNC_OK) ? MPI_SUCCESS : MPI_ERR_OTHER; + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return rc; +} + +int MPIR_Scatter_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + size_t nbytes; + const void *sbuf_adj = NULL; + void *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rank = comm_ptr->rank; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Scatter_intra_binomial(sendbuf, sendcount, sendtype, recvbuf, recvcount, + recvtype, root, comm_ptr, coll_attr); + + if (rank == root) { + MPI_Aint send_size; + MPIR_Datatype_get_size_macro(sendtype, send_size); + nbytes = (size_t) (sendcount * send_size); + rc = cnc_stage_in(sendbuf, sendcount * (MPI_Aint) n, sendtype, (size_t) n * nbytes, + &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + if (recvbuf == MPI_IN_PLACE) { + rbuf_adj = CNC_IN_PLACE; + } else { + rc = cnc_stage_out_alloc(recvbuf, recvtype, nbytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + return rc; + } + } + } else { + MPI_Aint recv_size; + MPIR_Datatype_get_size_macro(recvtype, recv_size); + nbytes = (size_t) (recvcount * recv_size); + rc = cnc_stage_out_alloc(recvbuf, recvtype, nbytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) + return rc; + } + + st = cnc_scatter(sbuf_adj, rbuf_adj, nbytes, n, rank, root, t, NULL); + MPL_free(tmp_sbuf); + if (rank == root && recvbuf == MPI_IN_PLACE) + return st == CNC_OK ? MPI_SUCCESS : MPI_ERR_OTHER; + return cnc_stage_out_finish(recvbuf, recvcount, recvtype, nbytes, tmp_rbuf, st == CNC_OK); +} + +int MPIR_Scatterv_intra_cnc(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * displs, MPI_Datatype sendtype, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + size_t recv_nbytes; + void *sbuf_adj = NULL, *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rank = comm_ptr->rank; + size_t *scounts_bytes = NULL; + ptrdiff_t *sdispls_bytes = NULL; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Scatterv_allcomm_linear(sendbuf, sendcounts, displs, sendtype, recvbuf, + recvcount, recvtype, root, comm_ptr, coll_attr); + + if (rank == root) { + scounts_bytes = MPL_malloc((size_t) n * sizeof(*scounts_bytes), MPL_MEM_COLL); + sdispls_bytes = MPL_malloc((size_t) n * sizeof(*sdispls_bytes), MPL_MEM_COLL); + if (!scounts_bytes || !sdispls_bytes) { + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return MPI_ERR_OTHER; + } + rc = cnc_v_stage_prep(sendbuf, sendcounts, displs, sendtype, n, 1, -1, &sbuf_adj, + &tmp_sbuf, scounts_bytes, sdispls_bytes); + if (rc != MPI_SUCCESS) { + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return rc; + } + recv_nbytes = scounts_bytes[rank]; + if (recvbuf == MPI_IN_PLACE) { + rbuf_adj = (char *) sbuf_adj + sdispls_bytes[rank]; + } else { + rc = cnc_stage_out_alloc(recvbuf, recvtype, recv_nbytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return rc; + } + } + } else { + MPI_Aint recv_size; + MPIR_Datatype_get_size_macro(recvtype, recv_size); + recv_nbytes = recvcount > 0 ? (size_t) (recvcount * recv_size) : 0; + rc = cnc_stage_out_alloc(recvbuf, recvtype, recv_nbytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) + return rc; + scounts_bytes = MPL_calloc((size_t) n, sizeof(*scounts_bytes), MPL_MEM_COLL); + if (!scounts_bytes) { + MPL_free(tmp_rbuf); + return MPI_ERR_OTHER; + } + scounts_bytes[rank] = recv_nbytes; + } + + st = cnc_scatterv_linear(sbuf_adj, scounts_bytes, sdispls_bytes, rbuf_adj, recv_nbytes, + n, rank, root, t); + MPL_free(tmp_sbuf); + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + if (rank == root && recvbuf == MPI_IN_PLACE) + return st == CNC_OK ? MPI_SUCCESS : MPI_ERR_OTHER; + return cnc_stage_out_finish(recvbuf, recvcount, recvtype, recv_nbytes, tmp_rbuf, st == CNC_OK); +} + +int MPIR_Allgather_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + MPI_Aint recv_size, recv_extent; + size_t nbytes; + const void *sbuf_adj = NULL; + void *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rank = comm_ptr->rank; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Allgather_intra_ring(sendbuf, sendcount, sendtype, recvbuf, recvcount, + recvtype, comm_ptr, coll_attr); + + MPIR_Datatype_get_size_macro(recvtype, recv_size); + nbytes = (size_t) (recvcount * recv_size); + rc = cnc_stage_out_alloc(recvbuf, recvtype, (size_t) n * nbytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) + return rc; + + if (sendbuf == MPI_IN_PLACE) { + sbuf_adj = (const char *) rbuf_adj + rank * nbytes; + if (tmp_rbuf && nbytes > 0) { + MPIR_Datatype_get_extent_macro(recvtype, recv_extent); + rc = MPIR_Localcopy((const char *) recvbuf + rank * recvcount * recv_extent, + recvcount, recvtype, (void *) sbuf_adj, (MPI_Aint) nbytes, + MPIR_BYTE_INTERNAL); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_rbuf); + return rc; + } + } + } else { + rc = cnc_stage_in(sendbuf, sendcount, sendtype, nbytes, &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_rbuf); + return rc; + } + } + + st = cnc_allgather(sbuf_adj, rbuf_adj, nbytes, n, rank, t, NULL); + MPL_free(tmp_sbuf); + return cnc_stage_out_finish(recvbuf, recvcount * (MPI_Aint) n, recvtype, + (size_t) n * nbytes, tmp_rbuf, st == CNC_OK); +} + +int MPIR_Allgatherv_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, const MPI_Aint * displs, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + const void *sbuf_adj = NULL; + void *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rank = comm_ptr->rank; + size_t *rcounts_bytes = NULL; + ptrdiff_t *rdispls_bytes = NULL; + int rc; + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Allgatherv_intra_ring(sendbuf, sendcount, sendtype, recvbuf, recvcounts, + displs, recvtype, comm_ptr, coll_attr); + + rcounts_bytes = MPL_malloc((size_t) n * sizeof(*rcounts_bytes), MPL_MEM_COLL); + rdispls_bytes = MPL_malloc((size_t) n * sizeof(*rdispls_bytes), MPL_MEM_COLL); + if (!rcounts_bytes || !rdispls_bytes) { + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return MPI_ERR_OTHER; + } + + rc = cnc_v_stage_prep(recvbuf, recvcounts, displs, recvtype, n, 0, + sendbuf == MPI_IN_PLACE ? rank : -1, &rbuf_adj, &tmp_rbuf, + rcounts_bytes, rdispls_bytes); + if (rc != MPI_SUCCESS) { + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return rc; + } + + if (sendbuf == MPI_IN_PLACE) { + sbuf_adj = (const char *) rbuf_adj + rdispls_bytes[rank]; + } else { + rc = cnc_stage_in(sendbuf, sendcount, sendtype, rcounts_bytes[rank], &sbuf_adj, &tmp_sbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_rbuf); + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return rc; + } + } + + st = cnc_allgatherv_ring(sbuf_adj, rbuf_adj, rcounts_bytes, rdispls_bytes, n, rank, t); + MPL_free(tmp_sbuf); + rc = cnc_v_stage_unpack(recvbuf, recvcounts, displs, recvtype, n, rcounts_bytes, + rdispls_bytes, tmp_rbuf, st == CNC_OK); + MPL_free(rcounts_bytes); + MPL_free(rdispls_bytes); + return rc; +} + +int MPIR_Alltoall_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + MPI_Aint recv_size; + size_t nbytes, total_bytes; + const void *sbuf_adj = NULL; + void *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + int rc; + + if (sendbuf == MPI_IN_PLACE) + return MPIR_Alltoall_intra_pairwise_sendrecv_replace(sendbuf, sendcount, sendtype, + recvbuf, recvcount, recvtype, + comm_ptr, coll_attr); + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Alltoall_intra_pairwise(sendbuf, sendcount, sendtype, recvbuf, recvcount, + recvtype, comm_ptr, coll_attr); + + MPIR_Datatype_get_size_macro(recvtype, recv_size); + nbytes = (size_t) (recvcount * recv_size); + total_bytes = n * nbytes; + + rc = cnc_stage_in(sendbuf, sendcount * (MPI_Aint) n, sendtype, total_bytes, &sbuf_adj, + &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + rc = cnc_stage_out_alloc(recvbuf, recvtype, total_bytes, &rbuf_adj, &tmp_rbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + return rc; + } + + st = cnc_alltoall(sbuf_adj, rbuf_adj, nbytes, n, comm_ptr->rank, t, NULL); + MPL_free(tmp_sbuf); + return cnc_stage_out_finish(recvbuf, recvcount * (MPI_Aint) n, recvtype, total_bytes, + tmp_rbuf, st == CNC_OK); +} + +int MPIR_Alltoallv_intra_cnc(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * sdispls, MPI_Datatype sendtype, void *recvbuf, + const MPI_Aint * recvcounts, const MPI_Aint * rdispls, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + void *sbuf_adj = NULL, *rbuf_adj = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int n = comm_ptr->local_size; + size_t *scounts_bytes = NULL, *rcounts_bytes = NULL; + ptrdiff_t *sdispls_bytes = NULL, *rdispls_bytes = NULL; + int rc; + + if (sendbuf == MPI_IN_PLACE) + return MPIR_Alltoallv_intra_pairwise_sendrecv_replace(sendbuf, sendcounts, sdispls, + sendtype, recvbuf, recvcounts, + rdispls, recvtype, comm_ptr, + coll_attr); + + if (!cnc_can_run(comm_ptr) || cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + return MPIR_Alltoallv_intra_scattered(sendbuf, sendcounts, sdispls, sendtype, recvbuf, + recvcounts, rdispls, recvtype, comm_ptr, coll_attr); + + scounts_bytes = MPL_malloc(2 * (size_t) n * sizeof(*scounts_bytes), MPL_MEM_COLL); + sdispls_bytes = MPL_malloc(2 * (size_t) n * sizeof(*sdispls_bytes), MPL_MEM_COLL); + if (!scounts_bytes || !sdispls_bytes) { + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return MPI_ERR_OTHER; + } + rcounts_bytes = scounts_bytes + n; + rdispls_bytes = sdispls_bytes + n; + + rc = cnc_v_stage_prep(sendbuf, sendcounts, sdispls, sendtype, n, 1, -1, &sbuf_adj, &tmp_sbuf, + scounts_bytes, sdispls_bytes); + if (rc != MPI_SUCCESS) { + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return rc; + } + rc = cnc_v_stage_prep(recvbuf, recvcounts, rdispls, recvtype, n, 0, -1, &rbuf_adj, &tmp_rbuf, + rcounts_bytes, rdispls_bytes); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return rc; + } + + st = cnc_alltoallv_pairwise(sbuf_adj, scounts_bytes, sdispls_bytes, rbuf_adj, rcounts_bytes, + rdispls_bytes, n, comm_ptr->rank, t); + MPL_free(tmp_sbuf); + rc = cnc_v_stage_unpack(recvbuf, recvcounts, rdispls, recvtype, n, rcounts_bytes, + rdispls_bytes, tmp_rbuf, st == CNC_OK); + MPL_free(scounts_bytes); + MPL_free(sdispls_bytes); + return rc; +} + +int MPIR_Reduce_intra_cnc(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, int root, MPIR_Comm * comm_ptr, + int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + cnc_reduce_opts_t opts; + cnc_datatype_t dt; + MPI_Aint type_size; + size_t nbytes; + int rank = comm_ptr->rank; + const void *sbuf_use = NULL; + void *rbuf_use = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int rc; + + if (!cnc_can_run(comm_ptr)) + goto fallback; + + memset(&opts, 0, sizeof(opts)); + if (!cnc_map_dtype_op(datatype, op, &dt, &opts.op) || + cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + goto fallback; + + MPIR_Datatype_get_size_macro(datatype, type_size); + nbytes = (size_t) (count * type_size); + + if (rank == root) { + if (sendbuf == MPI_IN_PLACE) { + rc = cnc_stage_in(recvbuf, count, datatype, nbytes, &sbuf_use, &tmp_rbuf); + if (rc != MPI_SUCCESS) + return rc; + rbuf_use = (void *) sbuf_use; + } else { + rc = cnc_stage_in(sendbuf, count, datatype, nbytes, &sbuf_use, &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + rc = cnc_stage_out_alloc(recvbuf, datatype, nbytes, &rbuf_use, &tmp_rbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + return rc; + } + } + } else { + rc = cnc_stage_in(sendbuf, count, datatype, nbytes, &sbuf_use, &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + } + + st = cnc_reduce(sbuf_use, rbuf_use, count, dt, comm_ptr->local_size, rank, root, t, + opts.op == CNC_REDUCE_SUM ? NULL : &opts); + MPL_free(tmp_sbuf); + if (rank == root) + return cnc_stage_out_finish(recvbuf, count, datatype, nbytes, tmp_rbuf, st == CNC_OK); + return st == CNC_OK ? MPI_SUCCESS : MPI_ERR_OTHER; + + fallback: + return MPIR_Reduce_intra_binomial(sendbuf, recvbuf, count, datatype, op, root, comm_ptr, + coll_attr); +} + +int MPIR_Allreduce_intra_cnc(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, MPIR_Comm * comm_ptr, int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + cnc_allreduce_opts_t opts; + cnc_datatype_t dt; + MPI_Aint type_size; + size_t nbytes; + const void *sbuf_use = NULL; + void *rbuf_use = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int rc; + + if (!cnc_can_run(comm_ptr)) + goto fallback; + + memset(&opts, 0, sizeof(opts)); + if (!cnc_map_dtype_op(datatype, op, &dt, &opts.op) || + cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + goto fallback; + + MPIR_Datatype_get_size_macro(datatype, type_size); + nbytes = (size_t) (count * type_size); + + if (sendbuf == MPI_IN_PLACE) { + const void *in_wire = NULL; + rc = cnc_stage_in(recvbuf, count, datatype, nbytes, &in_wire, &tmp_rbuf); + if (rc != MPI_SUCCESS) + return rc; + rbuf_use = (void *) in_wire; + sbuf_use = CNC_IN_PLACE; + } else { + rc = cnc_stage_in(sendbuf, count, datatype, nbytes, &sbuf_use, &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + rc = cnc_stage_out_alloc(recvbuf, datatype, nbytes, &rbuf_use, &tmp_rbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + return rc; + } + } + + st = cnc_allreduce(sbuf_use, rbuf_use, count, dt, comm_ptr->local_size, comm_ptr->rank, t, + opts.op == CNC_REDUCE_SUM ? NULL : &opts); + MPL_free(tmp_sbuf); + return cnc_stage_out_finish(recvbuf, count, datatype, nbytes, tmp_rbuf, st == CNC_OK); + + fallback: + return MPIR_Allreduce_intra_recursive_doubling(sendbuf, recvbuf, count, datatype, op, + comm_ptr, coll_attr); +} + +int MPIR_Reduce_scatter_block_intra_cnc(const void *sendbuf, void *recvbuf, MPI_Aint recvcount, + MPI_Datatype datatype, MPI_Op op, MPIR_Comm * comm_ptr, + int coll_attr) +{ + cnc_transport_t *t; + cnc_status_t st; + cnc_reduce_scatter_opts_t opts; + cnc_datatype_t dt; + MPI_Aint type_size; + size_t recv_nbytes, total_nbytes; + int n = comm_ptr->local_size; + const void *sbuf_use = NULL; + void *rbuf_use = NULL, *tmp_sbuf = NULL, *tmp_rbuf = NULL; + int rc; + + if (!cnc_can_run(comm_ptr) || sendbuf == MPI_IN_PLACE) + goto fallback; + + memset(&opts, 0, sizeof(opts)); + if (!cnc_map_dtype_op(datatype, op, &dt, &opts.op) || + cnc_get_transport(comm_ptr, &t) != MPI_SUCCESS) + goto fallback; + + MPIR_Datatype_get_size_macro(datatype, type_size); + recv_nbytes = (size_t) (recvcount * type_size); + total_nbytes = n * recv_nbytes; + + rc = cnc_stage_in(sendbuf, recvcount * (MPI_Aint) n, datatype, total_nbytes, &sbuf_use, + &tmp_sbuf); + if (rc != MPI_SUCCESS) + return rc; + rc = cnc_stage_out_alloc(recvbuf, datatype, recv_nbytes, &rbuf_use, &tmp_rbuf); + if (rc != MPI_SUCCESS) { + MPL_free(tmp_sbuf); + return rc; + } + + st = cnc_reduce_scatter(sbuf_use, rbuf_use, (size_t) n * (size_t) recvcount, dt, n, + comm_ptr->rank, t, opts.op == CNC_REDUCE_SUM ? NULL : &opts); + MPL_free(tmp_sbuf); + return cnc_stage_out_finish(recvbuf, recvcount, datatype, recv_nbytes, tmp_rbuf, st == CNC_OK); + + fallback: + return MPIR_Reduce_scatter_block_intra_recursive_doubling(sendbuf, recvbuf, recvcount, + datatype, op, comm_ptr, coll_attr); +} diff --git a/src/mpi/coll/cnc/cnc_internal.h b/src/mpi/coll/cnc/cnc_internal.h new file mode 100644 index 00000000000..6971250e9ca --- /dev/null +++ b/src/mpi/coll/cnc/cnc_internal.h @@ -0,0 +1,11 @@ +/* + * Copyright (C) by Argonne National Laboratory + * See COPYRIGHT in top-level directory + */ + +#ifndef CNC_INTERNAL_H_INCLUDED +#define CNC_INTERNAL_H_INCLUDED + +int MPII_Cnc_comm_cleanup(MPIR_Comm * comm_ptr); + +#endif /* CNC_INTERNAL_H_INCLUDED */ diff --git a/src/mpi/coll/cnc/subconfigure.m4 b/src/mpi/coll/cnc/subconfigure.m4 new file mode 100644 index 00000000000..909e3245d0d --- /dev/null +++ b/src/mpi/coll/cnc/subconfigure.m4 @@ -0,0 +1,21 @@ +[#] start of __file__ + +dnl Optional offload of selected collectives to an external libcncoll. +AC_DEFUN([PAC_SUBCFG_PREREQ_]PAC_SUBCFG_AUTO_SUFFIX,[ + PAC_SET_HEADER_LIB_PATH(cnc) + pac_have_cnc=no + if test -n "${with_cnc}" -a "${with_cnc}" != "no" ; then + AC_CHECK_HEADER([cnc/collective.h],[pac_have_cnc=yes], + [AC_MSG_ERROR([--with-cnc is given but cnc/collective.h was not found])]) + fi + if test "$pac_have_cnc" = "yes" ; then + PAC_LIBS_ADD(-lcncoll) + AC_DEFINE([ENABLE_CNC],1,[Define if building with libcncoll]) + fi + AM_CONDITIONAL([BUILD_CNC],[test "$pac_have_cnc" = "yes"]) +])dnl end PREREQ + +AC_DEFUN([PAC_SUBCFG_BODY_]PAC_SUBCFG_AUTO_SUFFIX,[ +])dnl end BODY + +[#] end of __file__ diff --git a/src/mpi/coll/coll_algorithms.txt b/src/mpi/coll/coll_algorithms.txt index 637dbd615c5..3a64656cdb5 100644 --- a/src/mpi/coll/coll_algorithms.txt +++ b/src/mpi/coll/coll_algorithms.txt @@ -173,6 +173,8 @@ barrier-intra: inline: 1 macro_guard: defined(MPIDI_CH4_SHM_POSIX) restrictions: MPIDI_CH4_release_gather + cnc + macro_guard: defined(ENABLE_CNC) barrier-inter: bcast inner: barrier-intra@local_comm, bcast-inter@comm @@ -210,6 +212,8 @@ bcast-intra: inline: 1 macro_guard: defined(MPIDI_CH4_SHM_POSIX) restrictions: MPIDI_CH4_release_gather + cnc + macro_guard: defined(ENABLE_CNC) ibcast-intra: sched_binomial sched_smp @@ -243,6 +247,8 @@ ibcast-inter: gather-intra: binomial + cnc + macro_guard: defined(ENABLE_CNC) gather-inter: linear local_gather_remote_send @@ -260,6 +266,8 @@ igather-inter: gatherv-intra: linear allcomm: 1 + cnc + macro_guard: defined(ENABLE_CNC) gatherv-inter: linear allcomm: 1 @@ -276,6 +284,8 @@ igatherv-inter: scatter-intra: binomial + cnc + macro_guard: defined(ENABLE_CNC) scatter-inter: linear remote_send_local_scatter @@ -293,6 +303,8 @@ iscatter-inter: scatterv-intra: linear allcomm: 1 + cnc + macro_guard: defined(ENABLE_CNC) scatterv-inter: linear allcomm: 1 @@ -322,6 +334,8 @@ allgather-intra: extra_params: k, single_phase_recv cvar_params: RECEXCH_KVAL, RECEXCH_SINGLE_PHASE_RECV circ_graph + cnc + macro_guard: defined(ENABLE_CNC) allgather-inter: local_gather_remote_bcast inner: gather-intra@local_comm, bcast-inter@comm @@ -349,6 +363,8 @@ allgatherv-intra: recursive_doubling restrictions: pof2 ring + cnc + macro_guard: defined(ENABLE_CNC) allgatherv-inter: remote_gather_local_bcast inner: gatherv-inter@comm @@ -385,6 +401,8 @@ alltoall-intra: restrictions: inplace scattered inplace_copy: 1 + cnc + macro_guard: defined(ENABLE_CNC) alltoall-inter: pairwise_exchange ialltoall-intra: @@ -411,6 +429,8 @@ alltoallv-intra: restrictions: inplace scattered inplace_copy: 1 + cnc + macro_guard: defined(ENABLE_CNC) alltoallv-inter: pairwise_exchange ialltoallv-intra: @@ -466,6 +486,8 @@ reduce-intra: inline: 1 macro_guard: defined(MPIDI_CH4_SHM_POSIX) restrictions: MPIDI_CH4_release_gather + cnc + macro_guard: defined(ENABLE_CNC) reduce-inter: local_reduce_remote_send inner: reduce-intra@local_comm @@ -518,6 +540,8 @@ allreduce-intra: inline: 1 macro_guard: defined(MPIDI_CH4_SHM_POSIX) restrictions: MPIDI_CH4_release_gather + cnc + macro_guard: defined(ENABLE_CNC) allreduce-inter: reduce_exchange_bcast inner: reduce-intra@local_comm, bcast-intra@local_comm @@ -587,6 +611,8 @@ reduce_scatter_block-intra: restrictions: commutative recursive_halving restrictions: commutative + cnc + macro_guard: defined(ENABLE_CNC) reduce_scatter_block-inter: remote_reduce_local_scatter inner: reduce-inter@comm, scatter-intra@local_comm diff --git a/src/mpi/coll/cvars.txt b/src/mpi/coll/cvars.txt index d0387327a51..380e506715f 100644 --- a/src/mpi/coll/cvars.txt +++ b/src/mpi/coll/cvars.txt @@ -22,6 +22,7 @@ cvars: k_dissemination - Force high radix dissemination algorithm recexch - Force recursive exchange algorithm release_gather - Force CH4 POSIX release_gather algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_BARRIER_INTER_ALGORITHM category : COLLECTIVE @@ -133,6 +134,7 @@ cvars: pipelined_tree - Force tree-based pipelined algorithm tree - Force tree-based algorithm release_gather - Force CH4 POSIX release_gather algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_BCAST_TREE_KVAL category : COLLECTIVE @@ -364,6 +366,7 @@ cvars: auto - Internal algorithm selection (can be overridden with MPIR_CVAR_COLL_SELECTION_JSON_FILE) binomial - Force binomial algorithm nb - Force nonblocking algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_GATHER_INTER_ALGORITHM category : COLLECTIVE @@ -427,6 +430,7 @@ cvars: auto - Internal algorithm selection (can be overridden with MPIR_CVAR_COLL_SELECTION_JSON_FILE) linear - Force linear algorithm nb - Force nonblocking algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_GATHERV_INTER_ALGORITHM category : COLLECTIVE @@ -479,6 +483,7 @@ cvars: auto - Internal algorithm selection (can be overridden with MPIR_CVAR_COLL_SELECTION_JSON_FILE) binomial - Force binomial algorithm nb - Force nonblocking algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_SCATTER_INTER_ALGORITHM category : COLLECTIVE @@ -542,6 +547,7 @@ cvars: auto - Internal algorithm selection (can be overridden with MPIR_CVAR_COLL_SELECTION_JSON_FILE) linear - Force linear algorithm nb - Force nonblocking algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_SCATTERV_INTER_ALGORITHM category : COLLECTIVE @@ -600,6 +606,7 @@ cvars: recexch_doubling - Force recexch distance doubling algorithm recexch_halving - Force recexch distance halving algorithm circ_graph - Force queued circulant graph algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_ALLGATHER_BRUCKS_KVAL category : COLLECTIVE @@ -720,6 +727,7 @@ cvars: nb - Force nonblocking algorithm recursive_doubling - Force recursive doubling algorithm ring - Force ring algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_ALLGATHERV_INTER_ALGORITHM category : COLLECTIVE @@ -812,6 +820,7 @@ cvars: pairwise - Force pairwise algorithm pairwise_sendrecv_replace - Force pairwise sendrecv replace algorithm scattered - Force scattered algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_ALLTOALL_BRUCKS_KVAL category : COLLECTIVE @@ -879,6 +888,7 @@ cvars: nb - Force nonblocking algorithm pairwise_sendrecv_replace - Force pairwise_sendrecv_replace algorithm scattered - Force scattered algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_ALLTOALLV_INTER_ALGORITHM category : COLLECTIVE @@ -1012,6 +1022,7 @@ cvars: smp - Force smp algorithm reduce_scatter_gather - Force reduce scatter gather algorithm release_gather - Force CH4 POSIX release_gather algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_REDUCE_INTER_ALGORITHM category : COLLECTIVE @@ -1189,6 +1200,7 @@ cvars: k_reduce_scatter_allgather - Force reduce scatter allgather algorithm ccl - Force CCL algorithm release_gather - Force CH4 POSIX release_gather algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_ALLREDUCE_RECURSIVE_MULTIPLYING_KVAL category : COLLECTIVE @@ -1525,6 +1537,7 @@ cvars: pairwise - Force pairwise algorithm recursive_halving - Force recursive halving algorithm nb - Force nonblocking algorithm + cnc - Force CN collectives (libcncoll) algorithm - name : MPIR_CVAR_REDUCE_SCATTER_BLOCK_INTER_ALGORITHM category : COLLECTIVE diff --git a/src/mpi/coll/include/coll_impl.h b/src/mpi/coll/include/coll_impl.h index 04de88fbdac..d23f1f8f71c 100644 --- a/src/mpi/coll/include/coll_impl.h +++ b/src/mpi/coll/include/coll_impl.h @@ -13,6 +13,9 @@ #include "../algorithms/stubalgo/stubalgo.h" #include "../algorithms/treealgo/treealgo.h" #include "../algorithms/recexchalgo/recexchalgo.h" +#ifdef ENABLE_CNC +#include "../cnc/cnc_internal.h" +#endif #define MPII_COLLECTIVE_FALLBACK_CHECK(check, mpi_errno) \ do { \ diff --git a/src/mpi/coll/src/coll_impl.c b/src/mpi/coll/src/coll_impl.c index 99ba0c4f21b..844642a30f4 100644 --- a/src/mpi/coll/src/coll_impl.c +++ b/src/mpi/coll/src/coll_impl.c @@ -274,6 +274,9 @@ int MPIR_Coll_comm_init(MPIR_Comm * comm) mpi_errno = MPII_Recexchalgo_comm_init(comm); if (mpi_errno) MPIR_ERR_POP(mpi_errno); +#ifdef ENABLE_CNC + comm->coll.cnc_priv = NULL; +#endif fn_exit: return mpi_errno; @@ -300,6 +303,10 @@ int MPII_Coll_comm_cleanup(MPIR_Comm * comm) mpi_errno = MPII_Recexchalgo_comm_cleanup(comm); if (mpi_errno) MPIR_ERR_POP(mpi_errno); +#ifdef ENABLE_CNC + mpi_errno = MPII_Cnc_comm_cleanup(comm); + MPIR_ERR_CHECK(mpi_errno); +#endif fn_exit: return mpi_errno; diff --git a/test/units/cnc_coll.unit b/test/units/cnc_coll.unit new file mode 100644 index 00000000000..de148471acd --- /dev/null +++ b/test/units/cnc_coll.unit @@ -0,0 +1,607 @@ +sources: + src/mpi/coll/cnc/cnc_coll.c + +include: + src/mpi/coll/cnc + +shim: + core + cnc + +template: + #include "mpiimpl.h" + #include + #include + #include "cnc_internal.h" + + MPIR_Thread_info_t MPIR_ThreadInfo = { 0 }; + + struct cnc_transport { + MPI_Comm comm; + }; + + static int next_handle = 100; + static int live_comms = 0; + static int live_oob = 0; + static int live_transports = 0; + static int is_dev_ptr = 0; + static MPI_Aint true_lb_val = 0; + static int fail_comm_copy = 0; + static int fail_oob_create = 0; + static int fail_cnc_init = 0; + static MPIR_Comm *init_dup_comm = NULL; + + int MPII_Comm_copy(MPIR_Comm * comm_ptr, int size, MPIR_Info * info, MPIR_Comm ** outcomm_ptr) + { + if (fail_comm_copy) { + return MPI_ERR_OTHER; + } + MPIR_Comm *c = calloc(1, sizeof(*c)); + c->handle = ++next_handle; + c->rank = comm_ptr->rank; + c->local_size = size; + live_comms++; + init_dup_comm = c; + *outcomm_ptr = c; + return MPI_SUCCESS; + } + + int MPIR_Comm_release(MPIR_Comm * comm_ptr) + { + live_comms--; + free(comm_ptr); + return MPI_SUCCESS; + } + + void MPIR_Datatype_is_contig(MPI_Datatype datatype, int *flag) + { + *flag = (datatype != 99); + } + + int MPIR_GPU_query_pointer_is_dev(const void *ptr) + { + return is_dev_ptr; + } + + MPI_Aint unit_datatype_size(MPI_Datatype datatype) + { + return (datatype == MPI_DOUBLE) ? 8 : 4; + } + + void MPIR_Datatype_get_true_lb(MPI_Datatype datatype, MPI_Aint * true_lb) + { + *true_lb = true_lb_val; + } + + int MPIR_Localcopy(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype) + { + printf("localcopy sc=%ld st=%x rc=%ld rt=%x\n", + (long) sendcount, (unsigned) sendtype, (long) recvcount, (unsigned) recvtype); + return MPI_SUCCESS; + } + + int MPIR_Barrier_intra_dissemination(MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_barrier\n"); + return MPI_SUCCESS; + } + + int MPIR_Bcast_intra_binomial(void *buffer, MPI_Aint count, MPI_Datatype datatype, int root, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_bcast count=%ld root=%d\n", (long) count, root); + return MPI_SUCCESS; + } + + int MPIR_Gather_intra_binomial(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + int root, MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_gather sc=%ld rc=%ld root=%d\n", (long) sendcount, (long) recvcount, root); + return MPI_SUCCESS; + } + + int MPIR_Gatherv_allcomm_linear(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, + const MPI_Aint * displs, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_gatherv sc=%ld root=%d\n", (long) sendcount, root); + return MPI_SUCCESS; + } + + int MPIR_Scatter_intra_binomial(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + int root, MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_scatter sc=%ld rc=%ld root=%d\n", (long) sendcount, (long) recvcount, root); + return MPI_SUCCESS; + } + + int MPIR_Scatterv_allcomm_linear(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * displs, MPI_Datatype sendtype, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_scatterv rc=%ld root=%d\n", (long) recvcount, root); + return MPI_SUCCESS; + } + + int MPIR_Allgather_intra_ring(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_allgather sc=%ld rc=%ld\n", (long) sendcount, (long) recvcount); + return MPI_SUCCESS; + } + + int MPIR_Allgatherv_intra_ring(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, + const MPI_Aint * displs, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_allgatherv sc=%ld\n", (long) sendcount); + return MPI_SUCCESS; + } + + int MPIR_Alltoall_intra_pairwise(const void *sendbuf, MPI_Aint sendcount, + MPI_Datatype sendtype, void *recvbuf, MPI_Aint recvcount, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_alltoall sc=%ld rc=%ld\n", (long) sendcount, (long) recvcount); + return MPI_SUCCESS; + } + + int MPIR_Alltoall_intra_pairwise_sendrecv_replace(const void *sendbuf, MPI_Aint sendcount, + MPI_Datatype sendtype, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_alltoall_inplace rc=%ld\n", (long) recvcount); + return MPI_SUCCESS; + } + + int MPIR_Alltoallv_intra_scattered(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * sdispls, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, + const MPI_Aint * rdispls, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_alltoallv\n"); + return MPI_SUCCESS; + } + + int MPIR_Alltoallv_intra_pairwise_sendrecv_replace(const void *sendbuf, + const MPI_Aint * sendcounts, + const MPI_Aint * sdispls, + MPI_Datatype sendtype, void *recvbuf, + const MPI_Aint * recvcounts, + const MPI_Aint * rdispls, + MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_alltoallv_inplace\n"); + return MPI_SUCCESS; + } + + int MPIR_Reduce_intra_binomial(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, int root, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_reduce count=%ld dt=%x op=%d root=%d\n", + (long) count, (unsigned) datatype, op, root); + return MPI_SUCCESS; + } + + int MPIR_Allreduce_intra_recursive_doubling(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_allreduce count=%ld dt=%x op=%d\n", + (long) count, (unsigned) datatype, op); + return MPI_SUCCESS; + } + + int MPIR_Reduce_scatter_block_intra_recursive_doubling(const void *sendbuf, void *recvbuf, + MPI_Aint recvcount, + MPI_Datatype datatype, MPI_Op op, + MPIR_Comm * comm_ptr, int coll_attr) + { + printf("fallback_reduce_scatter_block rc=%ld dt=%x op=%d\n", + (long) recvcount, (unsigned) datatype, op); + return MPI_SUCCESS; + } + + cnc_status_t cnc_oob_mpi_create(cnc_oob_t * oob, MPI_Comm comm) + { + if (fail_oob_create) { + return CNC_ERR; + } + memset(oob, 0, sizeof(*oob)); + oob->context = malloc(sizeof(MPI_Comm)); + *(MPI_Comm *) oob->context = comm; + live_oob++; + printf("oob_create\n"); + return CNC_OK; + } + + void cnc_oob_mpi_destroy(cnc_oob_t * oob) + { + free(oob->context); + oob->context = NULL; + live_oob--; + printf("oob_destroy\n"); + } + + cnc_status_t cnc_init(cnc_transport_t ** out, const cnc_oob_t * oob) + { + if (fail_cnc_init) { + return CNC_ERR; + } + /* Simulate mpi_backend.c calling MPI_Allreduce on dup_comm during cnc_init: + * CNC_PRIV_SKIP must route this to the fallback, not recurse into cnc_init. */ + if (init_dup_comm) { + int dummy = 0; + MPIR_Allreduce_intra_cnc(&dummy, &dummy, 1, MPI_INT, MPI_MIN, init_dup_comm, 0); + } + cnc_transport_t *t = calloc(1, sizeof(*t)); + t->comm = *(MPI_Comm *) oob->context; + live_transports++; + *out = t; + printf("cnc_init\n"); + return CNC_OK; + } + + void cnc_fini(cnc_transport_t * t) + { + free(t); + live_transports--; + printf("cnc_fini\n"); + } + + void cnc_barrier(int n, int rank, cnc_transport_t * t) + { + printf("cnc_barrier n=%d rank=%d\n", n, rank); + } + + cnc_status_t cnc_bcast(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, int root, cnc_transport_t * t, + const cnc_bcast_opts_t * opts) + { + printf("cnc_bcast inplace=%d nbytes=%zu n=%d rank=%d root=%d\n", + sendbuf == CNC_IN_PLACE, nbytes, n, rank, root); + return CNC_OK; + } + + cnc_status_t cnc_gather(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, int root, cnc_transport_t * t, + const cnc_gather_opts_t * opts) + { + printf("cnc_gather alias_slot=%d nbytes=%zu n=%d rank=%d root=%d\n", + recvbuf && sendbuf == (const char *) recvbuf + (size_t) rank * nbytes, + nbytes, n, rank, root); + return CNC_OK; + } + + cnc_status_t cnc_gatherv_linear(const void *sendbuf, size_t sendcount, void *recvbuf, + const size_t * recvcounts, const ptrdiff_t * displs, + int n, int rank, int root, cnc_transport_t * t) + { + printf("cnc_gatherv alias_slot=%d sc=%zu rc_rank=%zu disp_null=%d n=%d rank=%d root=%d\n", + recvbuf && displs && sendbuf == (const char *) recvbuf + displs[rank], + sendcount, recvcounts[rank], displs == NULL, n, rank, root); + return CNC_OK; + } + + cnc_status_t cnc_scatter(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, int root, cnc_transport_t * t, + const cnc_scatter_opts_t * opts) + { + printf("cnc_scatter snull=%d rinplace=%d nbytes=%zu n=%d rank=%d root=%d\n", + sendbuf == NULL, recvbuf == CNC_IN_PLACE, nbytes, n, rank, root); + return CNC_OK; + } + + cnc_status_t cnc_scatterv_linear(const void *sendbuf, const size_t * sendcounts, + const ptrdiff_t * displs, void *recvbuf, size_t recvcount, + int n, int rank, int root, cnc_transport_t * t) + { + printf("cnc_scatterv snull=%d alias_slot=%d rc=%zu sc_rank=%zu disp_null=%d n=%d rank=%d root=%d\n", + sendbuf == NULL, + sendbuf && displs && recvbuf == (const char *) sendbuf + displs[rank], + recvcount, sendcounts[rank], displs == NULL, n, rank, root); + return CNC_OK; + } + + cnc_status_t cnc_allgather(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, cnc_transport_t * t, + const cnc_allgather_opts_t * opts) + { + printf("cnc_allgather alias_slot=%d nbytes=%zu n=%d rank=%d\n", + sendbuf == (const char *) recvbuf + (size_t) rank * nbytes, + nbytes, n, rank); + return CNC_OK; + } + + cnc_status_t cnc_allgatherv_ring(const void *sendbuf, void *recvbuf, + const size_t * recvcounts, const ptrdiff_t * displs, + int n, int rank, cnc_transport_t * t) + { + printf("cnc_allgatherv alias_slot=%d rc_rank=%zu disp_rank=%td n=%d rank=%d\n", + sendbuf == (const char *) recvbuf + displs[rank], + recvcounts[rank], displs[rank], n, rank); + return CNC_OK; + } + + cnc_status_t cnc_alltoall(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, cnc_transport_t * t, + const cnc_alltoall_opts_t * opts) + { + printf("cnc_alltoall nbytes=%zu n=%d rank=%d\n", nbytes, n, rank); + return CNC_OK; + } + + cnc_status_t cnc_alltoallv_pairwise(const void *sendbuf, const size_t * sendcounts, + const ptrdiff_t * sdispls, void *recvbuf, + const size_t * recvcounts, const ptrdiff_t * rdispls, + int n, int rank, cnc_transport_t * t) + { + printf("cnc_alltoallv sc0=%zu sd1=%td rc0=%zu rd1=%td n=%d rank=%d\n", + sendcounts[0], sdispls[1], recvcounts[0], rdispls[1], n, rank); + return CNC_OK; + } + + cnc_status_t cnc_reduce(const void *sendbuf, void *recvbuf, size_t count, + cnc_datatype_t datatype, int n, int rank, int root, + cnc_transport_t * t, const cnc_reduce_opts_t * opts) + { + printf("cnc_reduce s_eq_r=%d count=%zu dt=%d op=%d n=%d rank=%d root=%d\n", + sendbuf == recvbuf, count, datatype, opts ? (int) opts->op : 0, n, rank, root); + return CNC_OK; + } + + cnc_status_t cnc_allreduce(const void *sendbuf, void *recvbuf, size_t count, + cnc_datatype_t datatype, int n, int rank, + cnc_transport_t * t, const cnc_allreduce_opts_t * opts) + { + printf("cnc_allreduce inplace=%d count=%zu dt=%d op=%d n=%d rank=%d\n", + sendbuf == CNC_IN_PLACE, count, datatype, opts ? (int) opts->op : 0, n, rank); + return CNC_OK; + } + + cnc_status_t cnc_reduce_scatter(const void *sendbuf, void *recvbuf, size_t count, + cnc_datatype_t datatype, int n, int rank, + cnc_transport_t * t, const cnc_reduce_scatter_opts_t * opts) + { + printf("cnc_reduce_scatter count=%zu dt=%d op=%d n=%d rank=%d\n", + count, datatype, opts ? (int) opts->op : 0, n, rank); + return CNC_OK; + } + + int main(int argc, char **argv) + { + MPIR_Comm comm = { .handle = 1, .rank = 1, .local_size = 4 }; + char sbuf[128] = { 0 }, rbuf[128] = { 0 }; + const char *mode = argc > 1 ? argv[1] : ""; + + if (strcmp(mode, "bcast") == 0) { + true_lb_val = 8; + MPIR_Bcast_intra_cnc(rbuf, 5, MPI_INT, 0, &comm, 0); + MPIR_Bcast_intra_cnc(rbuf, 3, MPI_FLOAT, 2, &comm, 0); + MPII_Cnc_comm_cleanup(&comm); + } else if (strcmp(mode, "allreduce") == 0) { + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_FLOAT, MPI_SUM, &comm, 0); + MPIR_Allreduce_intra_cnc(MPI_IN_PLACE, rbuf, 6, MPI_INT, MPI_MAX, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 2, MPI_INT32_T, MPI_PROD, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 2, MPIR_FLOAT32, MPI_MIN, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 5, MPIR_INT32, MPI_SUM, &comm, 0); + MPII_Cnc_comm_cleanup(&comm); + } else if (strcmp(mode, "other_colls") == 0) { + MPIR_Barrier_intra_cnc(&comm, 0); + /* gather: non-root then root (out-of-place and in-place) */ + MPIR_Gather_intra_cnc(sbuf, 3, MPI_INT, NULL, 0, MPI_INT, 0, &comm, 0); + MPIR_Gather_intra_cnc(sbuf, 3, MPI_INT, rbuf, 3, MPI_INT, 1, &comm, 0); + MPIR_Gather_intra_cnc(MPI_IN_PLACE, 0, MPI_INT, rbuf, 3, MPI_INT, 1, &comm, 0); + /* gatherv: non-root (NULL counts/displs) then root (out-of-place and in-place) */ + { + MPI_Aint rc[4] = { 1, 2, 3, 4 }; + MPI_Aint rd[4] = { 0, 1, 3, 6 }; + MPIR_Gatherv_intra_cnc(sbuf, 2, MPI_INT, NULL, NULL, NULL, MPI_INT, 0, &comm, 0); + MPIR_Gatherv_intra_cnc(sbuf, 2, MPI_INT, rbuf, rc, rd, MPI_INT, 1, &comm, 0); + MPIR_Gatherv_intra_cnc(MPI_IN_PLACE, 0, MPI_INT, rbuf, rc, rd, MPI_INT, 1, &comm, 0); + } + /* scatter: non-root then root (out-of-place and in-place) */ + MPIR_Scatter_intra_cnc(NULL, 0, MPI_INT, rbuf, 2, MPI_INT, 0, &comm, 0); + MPIR_Scatter_intra_cnc(sbuf, 2, MPI_INT, rbuf, 2, MPI_INT, 1, &comm, 0); + MPIR_Scatter_intra_cnc(sbuf, 2, MPI_INT, MPI_IN_PLACE, 0, MPI_INT, 1, &comm, 0); + /* scatterv: non-root (NULL counts/displs) then root (out-of-place and in-place) */ + { + MPI_Aint sc[4] = { 1, 2, 3, 4 }; + MPI_Aint sd[4] = { 0, 1, 3, 6 }; + MPIR_Scatterv_intra_cnc(NULL, NULL, NULL, MPI_INT, rbuf, 2, MPI_INT, 0, &comm, 0); + MPIR_Scatterv_intra_cnc(sbuf, sc, sd, MPI_INT, rbuf, 2, MPI_INT, 1, &comm, 0); + MPIR_Scatterv_intra_cnc(sbuf, sc, sd, MPI_INT, MPI_IN_PLACE, 0, MPI_INT, 1, &comm, + 0); + } + /* allgather: out-of-place and in-place */ + MPIR_Allgather_intra_cnc(sbuf, 4, MPI_INT, rbuf, 4, MPI_INT, &comm, 0); + MPIR_Allgather_intra_cnc(MPI_IN_PLACE, 0, MPI_INT, rbuf, 4, MPI_INT, &comm, 0); + /* allgatherv: out-of-place and in-place */ + { + MPI_Aint rc[4] = { 1, 2, 3, 4 }; + MPI_Aint rd[4] = { 0, 1, 3, 6 }; + MPIR_Allgatherv_intra_cnc(sbuf, 2, MPI_INT, rbuf, rc, rd, MPI_INT, &comm, 0); + MPIR_Allgatherv_intra_cnc(MPI_IN_PLACE, 0, MPI_INT, rbuf, rc, rd, MPI_INT, &comm, + 0); + } + /* alltoall and alltoallv */ + MPIR_Alltoall_intra_cnc(sbuf, 2, MPI_INT, rbuf, 2, MPI_INT, &comm, 0); + { + MPI_Aint sc[4] = { 1, 2, 3, 4 }; + MPI_Aint sd[4] = { 0, 1, 3, 6 }; + MPIR_Alltoallv_intra_cnc(sbuf, sc, sd, MPI_INT, rbuf, sc, sd, MPI_INT, &comm, 0); + } + /* reduce: non-root, root out-of-place, root in-place */ + MPIR_Reduce_intra_cnc(sbuf, NULL, 5, MPI_FLOAT, MPI_SUM, 0, &comm, 0); + MPIR_Reduce_intra_cnc(sbuf, rbuf, 5, MPI_INT, MPI_MAX, 1, &comm, 0); + MPIR_Reduce_intra_cnc(MPI_IN_PLACE, rbuf, 5, MPI_INT, MPI_MIN, 1, &comm, 0); + /* reduce_scatter_block */ + MPIR_Reduce_scatter_block_intra_cnc(sbuf, rbuf, 3, MPI_FLOAT, MPI_SUM, &comm, 0); + MPII_Cnc_comm_cleanup(&comm); + } else if (strcmp(mode, "fallback") == 0) { + /* unsupported datatype and op (uniform across ranks) */ + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_DOUBLE, MPI_SUM, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_INT, MPI_BAND, &comm, 0); + MPIR_Reduce_intra_cnc(sbuf, rbuf, 4, MPI_DOUBLE, MPI_SUM, 0, &comm, 0); + MPIR_Reduce_scatter_block_intra_cnc(sbuf, rbuf, 4, MPI_DOUBLE, MPI_SUM, &comm, 0); + MPIR_Reduce_scatter_block_intra_cnc(MPI_IN_PLACE, rbuf, 4, MPI_INT, MPI_SUM, &comm, 0); + MPIR_Alltoall_intra_cnc(MPI_IN_PLACE, 0, MPI_INT, rbuf, 4, MPI_INT, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, 99, MPI_SUM, &comm, 0); + /* non-contiguous datatype (99) stages locally via MPIR_Localcopy + * instead of diverging per rank */ + MPIR_Bcast_intra_cnc(rbuf, 4, 99, 0, &comm, 0); + MPIR_Gather_intra_cnc(sbuf, 4, 99, rbuf, 4, MPI_INT, 0, &comm, 0); + MPIR_Scatter_intra_cnc(sbuf, 4, MPI_INT, rbuf, 4, 99, 0, &comm, 0); + MPIR_Allgather_intra_cnc(sbuf, 4, MPI_INT, rbuf, 4, 99, &comm, 0); + /* GPU buffer stages locally via MPIR_Localcopy */ + is_dev_ptr = 1; + MPIR_Bcast_intra_cnc(rbuf, 4, MPI_INT, 0, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_INT, MPI_SUM, &comm, 0); + is_dev_ptr = 0; + /* multi-threaded */ + MPIR_ThreadInfo.isThreaded = 1; + MPIR_Barrier_intra_cnc(&comm, 0); + MPIR_Bcast_intra_cnc(rbuf, 4, MPI_INT, 0, &comm, 0); + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_INT, MPI_SUM, &comm, 0); + MPIR_ThreadInfo.isThreaded = 0; + MPII_Cnc_comm_cleanup(&comm); + } else if (strcmp(mode, "errors") == 0) { + fail_comm_copy = 1; + printf("rc_copy=%d\n", MPIR_Bcast_intra_cnc(rbuf, 4, MPI_INT, 0, &comm, 0) == MPI_SUCCESS); + fail_comm_copy = 0; + /* Cached failure skips retrying setup until cleanup resets it. */ + MPIR_Bcast_intra_cnc(rbuf, 4, MPI_INT, 0, &comm, 0); + MPII_Cnc_comm_cleanup(&comm); + + fail_oob_create = 1; + printf("rc_oob=%d\n", MPIR_Bcast_intra_cnc(rbuf, 4, MPI_INT, 0, &comm, 0) == MPI_SUCCESS); + fail_oob_create = 0; + MPIR_Bcast_intra_cnc(rbuf, 4, MPI_INT, 0, &comm, 0); + MPII_Cnc_comm_cleanup(&comm); + + fail_cnc_init = 1; + printf("rc_init=%d\n", MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_INT, MPI_SUM, &comm, 0) == MPI_SUCCESS); + fail_cnc_init = 0; + MPIR_Allreduce_intra_cnc(sbuf, rbuf, 4, MPI_INT, MPI_SUM, &comm, 0); + MPII_Cnc_comm_cleanup(&comm); + } + + printf("leaks=%d\n", live_comms + live_oob + live_transports); + return 0; + } + +TESTS: + cmd: ./t bcast + expect: + oob_create + fallback_allreduce count=1 dt=4c810405 op=3 + cnc_init + cnc_bcast inplace=1 nbytes=20 n=4 rank=1 root=0 + cnc_bcast inplace=1 nbytes=12 n=4 rank=1 root=2 + cnc_fini + oob_destroy + leaks=0 + + cmd: ./t allreduce + expect: + oob_create + fallback_allreduce count=1 dt=4c810405 op=3 + cnc_init + cnc_allreduce inplace=0 count=4 dt=0 op=0 n=4 rank=1 + cnc_allreduce inplace=1 count=6 dt=1 op=3 n=4 rank=1 + cnc_allreduce inplace=0 count=2 dt=1 op=1 n=4 rank=1 + cnc_allreduce inplace=0 count=2 dt=0 op=2 n=4 rank=1 + cnc_allreduce inplace=0 count=5 dt=1 op=0 n=4 rank=1 + cnc_fini + oob_destroy + leaks=0 + + cmd: ./t other_colls + expect: + oob_create + fallback_allreduce count=1 dt=4c810405 op=3 + cnc_init + cnc_barrier n=4 rank=1 + cnc_gather alias_slot=0 nbytes=12 n=4 rank=1 root=0 + cnc_gather alias_slot=0 nbytes=12 n=4 rank=1 root=1 + cnc_gather alias_slot=1 nbytes=12 n=4 rank=1 root=1 + cnc_gatherv alias_slot=0 sc=8 rc_rank=8 disp_null=1 n=4 rank=1 root=0 + cnc_gatherv alias_slot=0 sc=8 rc_rank=8 disp_null=0 n=4 rank=1 root=1 + cnc_gatherv alias_slot=1 sc=8 rc_rank=8 disp_null=0 n=4 rank=1 root=1 + cnc_scatter snull=1 rinplace=0 nbytes=8 n=4 rank=1 root=0 + cnc_scatter snull=0 rinplace=0 nbytes=8 n=4 rank=1 root=1 + cnc_scatter snull=0 rinplace=1 nbytes=8 n=4 rank=1 root=1 + cnc_scatterv snull=1 alias_slot=0 rc=8 sc_rank=8 disp_null=1 n=4 rank=1 root=0 + cnc_scatterv snull=0 alias_slot=0 rc=8 sc_rank=8 disp_null=0 n=4 rank=1 root=1 + cnc_scatterv snull=0 alias_slot=1 rc=8 sc_rank=8 disp_null=0 n=4 rank=1 root=1 + cnc_allgather alias_slot=0 nbytes=16 n=4 rank=1 + cnc_allgather alias_slot=1 nbytes=16 n=4 rank=1 + cnc_allgatherv alias_slot=0 rc_rank=8 disp_rank=4 n=4 rank=1 + cnc_allgatherv alias_slot=1 rc_rank=8 disp_rank=4 n=4 rank=1 + cnc_alltoall nbytes=8 n=4 rank=1 + cnc_alltoallv sc0=4 sd1=4 rc0=4 rd1=4 n=4 rank=1 + cnc_reduce s_eq_r=0 count=5 dt=0 op=0 n=4 rank=1 root=0 + cnc_reduce s_eq_r=0 count=5 dt=1 op=3 n=4 rank=1 root=1 + cnc_reduce s_eq_r=1 count=5 dt=1 op=2 n=4 rank=1 root=1 + cnc_reduce_scatter count=12 dt=0 op=0 n=4 rank=1 + cnc_fini + oob_destroy + leaks=0 + + cmd: ./t fallback + expect: + fallback_allreduce count=4 dt=4c83080b op=1 + fallback_allreduce count=4 dt=4c810405 op=5 + fallback_reduce count=4 dt=4c83080b op=1 root=0 + fallback_reduce_scatter_block rc=4 dt=4c83080b op=1 + fallback_reduce_scatter_block rc=4 dt=4c810405 op=1 + fallback_alltoall_inplace rc=4 + fallback_allreduce count=4 dt=63 op=1 + oob_create + fallback_allreduce count=1 dt=4c810405 op=3 + cnc_init + cnc_bcast inplace=1 nbytes=16 n=4 rank=1 root=0 + localcopy sc=16 st=4c820100 rc=4 rt=63 + localcopy sc=4 st=63 rc=16 rt=4c820100 + cnc_gather alias_slot=0 nbytes=16 n=4 rank=1 root=0 + cnc_scatter snull=1 rinplace=0 nbytes=16 n=4 rank=1 root=0 + localcopy sc=16 st=4c820100 rc=4 rt=63 + cnc_allgather alias_slot=0 nbytes=16 n=4 rank=1 + localcopy sc=64 st=4c820100 rc=16 rt=63 + cnc_bcast inplace=1 nbytes=16 n=4 rank=1 root=0 + localcopy sc=16 st=4c820100 rc=4 rt=4c810405 + localcopy sc=4 st=4c810405 rc=16 rt=4c820100 + cnc_allreduce inplace=0 count=4 dt=1 op=0 n=4 rank=1 + localcopy sc=16 st=4c820100 rc=4 rt=4c810405 + fallback_barrier + fallback_bcast count=4 root=0 + fallback_allreduce count=4 dt=4c810405 op=1 + cnc_fini + oob_destroy + leaks=0 + + cmd: ./t errors + expect: + fallback_bcast count=4 root=0 + rc_copy=1 + fallback_bcast count=4 root=0 + fallback_bcast count=4 root=0 + rc_oob=1 + fallback_bcast count=4 root=0 + oob_create + oob_destroy + fallback_allreduce count=4 dt=4c810405 op=1 + rc_init=1 + fallback_allreduce count=4 dt=4c810405 op=1 + leaks=0 diff --git a/test/units/shim/cnc/cnc/cnc_oob_mpi.h b/test/units/shim/cnc/cnc/cnc_oob_mpi.h new file mode 100644 index 00000000000..f5094ac43d1 --- /dev/null +++ b/test/units/shim/cnc/cnc/cnc_oob_mpi.h @@ -0,0 +1,16 @@ +/* + * Copyright (C) by Argonne National Laboratory + * See COPYRIGHT in top-level directory + */ + +#ifndef CNC_OOB_MPI_SHIM_H_INCLUDED +#define CNC_OOB_MPI_SHIM_H_INCLUDED +#define CNC_OOB_MPI_H + +#include "mpiimpl.h" +#include + +cnc_status_t cnc_oob_mpi_create(cnc_oob_t * oob, MPI_Comm comm); +void cnc_oob_mpi_destroy(cnc_oob_t * oob); + +#endif /* CNC_OOB_MPI_SHIM_H_INCLUDED */ diff --git a/test/units/shim/cnc/cnc/collective.h b/test/units/shim/cnc/cnc/collective.h new file mode 100644 index 00000000000..a745a3120d0 --- /dev/null +++ b/test/units/shim/cnc/cnc/collective.h @@ -0,0 +1,196 @@ +/* + * Copyright (C) by Argonne National Laboratory + * See COPYRIGHT in top-level directory + */ + +#ifndef CNC_COLLECTIVE_SHIM_H_INCLUDED +#define CNC_COLLECTIVE_SHIM_H_INCLUDED + +#include +#include + +#ifndef CNC_TRANSPORT_H +typedef struct cnc_transport cnc_transport_t; + +typedef enum cnc_status { + CNC_OK = 0, + CNC_ERR = 1, + CNC_ERR_INVAL = 2, + CNC_ERR_NOTSUP = 3, + CNC_INPROGRESS = 4 +} cnc_status_t; + +typedef struct cnc_oob { + cnc_status_t(*allgather) (const void *src, void *recv, size_t size, + void *context, void **request); + cnc_status_t(*req_test) (void *request); + cnc_status_t(*req_free) (void *request); + void *context; + uint32_t n_eps; + uint32_t ep; +} cnc_oob_t; +#endif /* CNC_TRANSPORT_H */ + +#ifndef CNC_COLLECTIVE_H +#define CNC_IN_PLACE ((void *) -1) + +typedef enum cnc_bcast_algo { + CNC_BCAST_BINOMIAL = 0, + CNC_BCAST_SCATTER_ALLGATHER = 1, + CNC_BCAST_CIRCULANT = 2, + CNC_BCAST_BINE = 3 +} cnc_bcast_algo_t; + +typedef enum cnc_gather_algo { + CNC_GATHER_BINOMIAL = 0, + CNC_GATHER_BINE = 1 +} cnc_gather_algo_t; + +typedef enum cnc_allgather_algo { + CNC_ALLGATHER_RING = 0, + CNC_ALLGATHER_RECURSIVE_DOUBLING = 1, + CNC_ALLGATHER_BRUCK = 2, + CNC_ALLGATHER_BINE = 3 +} cnc_allgather_algo_t; + +typedef enum cnc_allreduce_algo { + CNC_ALLREDUCE_RECURSIVE_DOUBLING = 0, + CNC_ALLREDUCE_RABENSEIFNER = 1, + CNC_ALLREDUCE_RING = 2, + CNC_ALLREDUCE_BINE = 3, + CNC_ALLREDUCE_BINE_BANDWIDTH = 4 +} cnc_allreduce_algo_t; + +typedef enum cnc_alltoall_algo { + CNC_ALLTOALL_PAIRWISE = 0, + CNC_ALLTOALL_BRUCK = 1 +} cnc_alltoall_algo_t; + +typedef enum cnc_reduce_scatter_algo { + CNC_REDUCE_SCATTER_RECURSIVE_HALVING = 0, + CNC_REDUCE_SCATTER_BINE = 1 +} cnc_reduce_scatter_algo_t; + +typedef enum cnc_scatter_algo { + CNC_SCATTER_BINOMIAL = 0, + CNC_SCATTER_BINE = 1 +} cnc_scatter_algo_t; + +typedef enum cnc_reduce_algo { + CNC_REDUCE_BINOMIAL = 0, + CNC_REDUCE_RABENSEIFNER = 1, + CNC_REDUCE_CIRCULANT = 2, + CNC_REDUCE_BINE = 3 +} cnc_reduce_algo_t; + +typedef enum cnc_reduce_op { + CNC_REDUCE_SUM = 0, + CNC_REDUCE_PROD = 1, + CNC_REDUCE_MIN = 2, + CNC_REDUCE_MAX = 3 +} cnc_reduce_op_t; + +typedef enum cnc_datatype { + CNC_DTYPE_F32 = 0, + CNC_DTYPE_I32 = 1 +} cnc_datatype_t; + +typedef struct cnc_bcast_opts { + cnc_bcast_algo_t algo; + int chunk; + int topology_remap; +} cnc_bcast_opts_t; + +typedef struct cnc_gather_opts { + int chunk; + cnc_gather_algo_t algo; +} cnc_gather_opts_t; + +typedef struct cnc_scatter_opts { + int chunk; + cnc_scatter_algo_t algo; +} cnc_scatter_opts_t; + +typedef struct cnc_allgather_opts { + int chunk; + cnc_allgather_algo_t algo; +} cnc_allgather_opts_t; + +typedef struct cnc_alltoall_opts { + int chunk; + cnc_alltoall_algo_t algo; +} cnc_alltoall_opts_t; + +typedef struct cnc_reduce_opts { + cnc_reduce_op_t op; + int chunk; + cnc_reduce_algo_t algo; + int topology_remap; +} cnc_reduce_opts_t; + +typedef struct cnc_allreduce_opts { + cnc_reduce_op_t op; + int chunk; + cnc_allreduce_algo_t algo; +} cnc_allreduce_opts_t; + +typedef struct cnc_reduce_scatter_opts { + cnc_reduce_op_t op; + int chunk; + cnc_reduce_scatter_algo_t algo; +} cnc_reduce_scatter_opts_t; +#endif /* CNC_COLLECTIVE_H */ + +cnc_status_t cnc_init(cnc_transport_t ** out, const cnc_oob_t * oob); +void cnc_fini(cnc_transport_t * t); + +void cnc_barrier(int n, int rank, cnc_transport_t * t); + +cnc_status_t cnc_bcast(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, int root, cnc_transport_t * t, + const cnc_bcast_opts_t * opts); + +cnc_status_t cnc_gather(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, int root, cnc_transport_t * t, + const cnc_gather_opts_t * opts); + +cnc_status_t cnc_gatherv_linear(const void *sendbuf, size_t sendcount, void *recvbuf, + const size_t * recvcounts, const ptrdiff_t * displs, + int n, int rank, int root, cnc_transport_t * t); + +cnc_status_t cnc_scatter(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, int root, cnc_transport_t * t, + const cnc_scatter_opts_t * opts); + +cnc_status_t cnc_scatterv_linear(const void *sendbuf, const size_t * sendcounts, + const ptrdiff_t * displs, void *recvbuf, size_t recvcount, + int n, int rank, int root, cnc_transport_t * t); + +cnc_status_t cnc_allgather(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, cnc_transport_t * t, const cnc_allgather_opts_t * opts); + +cnc_status_t cnc_allgatherv_ring(const void *sendbuf, void *recvbuf, + const size_t * recvcounts, const ptrdiff_t * displs, + int n, int rank, cnc_transport_t * t); + +cnc_status_t cnc_alltoall(const void *sendbuf, void *recvbuf, size_t nbytes, + int n, int rank, cnc_transport_t * t, const cnc_alltoall_opts_t * opts); + +cnc_status_t cnc_alltoallv_pairwise(const void *sendbuf, const size_t * sendcounts, + const ptrdiff_t * sdispls, void *recvbuf, + const size_t * recvcounts, const ptrdiff_t * rdispls, + int n, int rank, cnc_transport_t * t); + +cnc_status_t cnc_reduce(const void *sendbuf, void *recvbuf, size_t count, + cnc_datatype_t datatype, int n, int rank, int root, + cnc_transport_t * t, const cnc_reduce_opts_t * opts); + +cnc_status_t cnc_allreduce(const void *sendbuf, void *recvbuf, size_t count, + cnc_datatype_t datatype, int n, int rank, + cnc_transport_t * t, const cnc_allreduce_opts_t * opts); + +cnc_status_t cnc_reduce_scatter(const void *sendbuf, void *recvbuf, size_t count, + cnc_datatype_t datatype, int n, int rank, + cnc_transport_t * t, const cnc_reduce_scatter_opts_t * opts); + +#endif /* CNC_COLLECTIVE_SHIM_H_INCLUDED */ diff --git a/test/units/shim/core/mpiimpl.h b/test/units/shim/core/mpiimpl.h index ed08cd40014..767d4c727f9 100644 --- a/test/units/shim/core/mpiimpl.h +++ b/test/units/shim/core/mpiimpl.h @@ -19,16 +19,153 @@ typedef intptr_t MPI_Aint; typedef int MPI_Datatype; typedef int MPI_Op; +typedef int MPI_Comm; +typedef struct MPIR_Info MPIR_Info; + +#define MPI_IN_PLACE ((void *) -1) + +#define MPIR_INT32 ((MPI_Datatype) 0x4c810400) +#define MPIR_UINT8 ((MPI_Datatype) 0x4c820100) +#define MPIR_FLOAT32 ((MPI_Datatype) 0x4c830400) +#define MPIR_FLOAT64 ((MPI_Datatype) 0x4c830800) +#define MPIR_BYTE_INTERNAL MPIR_UINT8 +#define MPIR_DATATYPE_GET_RAW_INTERNAL(type) ((type) & 0xffffff00) + +#define MPI_FLOAT (MPIR_FLOAT32 | 0x0a) +#define MPI_INT (MPIR_INT32 | 0x05) +#define MPI_INT32_T (MPIR_INT32 | 0x33) +#define MPI_DOUBLE (MPIR_FLOAT64 | 0x0b) + +#define MPI_SUM 1 +#define MPI_PROD 2 +#define MPI_MIN 3 +#define MPI_MAX 4 +#define MPI_BAND 5 /* Types that headers under test only mention. Pointers to them are fine as * incomplete types; the ones used by value are placeholders whose layout is * NOT the real one, so units must not depend on their contents. */ +struct MPIR_Comm { + MPI_Comm handle; + int rank; + int local_size; + struct { + void *cnc_priv; + } coll; +}; typedef struct MPIR_Comm MPIR_Comm; typedef struct MPIR_Request MPIR_Request; typedef struct { int placeholder; } MPIR_gpu_req; +typedef struct { + int isThreaded; +} MPIR_Thread_info_t; +extern MPIR_Thread_info_t MPIR_ThreadInfo; + +int MPII_Comm_copy(MPIR_Comm * comm_ptr, int size, MPIR_Info * info, MPIR_Comm ** outcomm_ptr); +int MPIR_Comm_release(MPIR_Comm * comm_ptr); +int MPIR_Localcopy(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype); +void MPIR_Datatype_is_contig(MPI_Datatype datatype, int *flag); +int MPIR_GPU_query_pointer_is_dev(const void *ptr); +void MPIR_Datatype_get_size_macro(MPI_Datatype datatype, MPI_Aint size); +void MPIR_Datatype_get_extent_macro(MPI_Datatype datatype, MPI_Aint extent); +void MPIR_Datatype_get_true_lb(MPI_Datatype datatype, MPI_Aint * true_lb); +#define MPIR_Datatype_get_size_macro(dt_, sz_) ((sz_) = unit_datatype_size(dt_)) +#define MPIR_Datatype_get_extent_macro(dt_, ext_) ((ext_) = unit_datatype_size(dt_)) +MPI_Aint unit_datatype_size(MPI_Datatype datatype); + +int MPIR_Barrier_intra_dissemination(MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Bcast_intra_binomial(void *buffer, MPI_Aint count, MPI_Datatype datatype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Gather_intra_binomial(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Gatherv_allcomm_linear(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, const MPI_Aint * displs, + MPI_Datatype recvtype, int root, MPIR_Comm * comm_ptr, + int coll_attr); +int MPIR_Scatter_intra_binomial(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Scatterv_allcomm_linear(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * displs, MPI_Datatype sendtype, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Allgather_intra_ring(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Allgatherv_intra_ring(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, const MPI_Aint * displs, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Alltoall_intra_pairwise(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Alltoall_intra_pairwise_sendrecv_replace(const void *sendbuf, MPI_Aint sendcount, + MPI_Datatype sendtype, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Alltoallv_intra_scattered(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * sdispls, MPI_Datatype sendtype, void *recvbuf, + const MPI_Aint * recvcounts, const MPI_Aint * rdispls, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Alltoallv_intra_pairwise_sendrecv_replace(const void *sendbuf, + const MPI_Aint * sendcounts, + const MPI_Aint * sdispls, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, + const MPI_Aint * rdispls, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Reduce_intra_binomial(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, int root, MPIR_Comm * comm_ptr, + int coll_attr); +int MPIR_Allreduce_intra_recursive_doubling(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Reduce_scatter_block_intra_recursive_doubling(const void *sendbuf, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype datatype, + MPI_Op op, MPIR_Comm * comm_ptr, + int coll_attr); + +int MPIR_Barrier_intra_cnc(MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Bcast_intra_cnc(void *buffer, MPI_Aint count, MPI_Datatype datatype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Gather_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Gatherv_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, const MPI_Aint * displs, + MPI_Datatype recvtype, int root, MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Scatter_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Scatterv_intra_cnc(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * displs, MPI_Datatype sendtype, void *recvbuf, + MPI_Aint recvcount, MPI_Datatype recvtype, int root, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Allgather_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Allgatherv_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, const MPI_Aint * recvcounts, const MPI_Aint * displs, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Alltoall_intra_cnc(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype, + void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype, + MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Alltoallv_intra_cnc(const void *sendbuf, const MPI_Aint * sendcounts, + const MPI_Aint * sdispls, MPI_Datatype sendtype, void *recvbuf, + const MPI_Aint * recvcounts, const MPI_Aint * rdispls, + MPI_Datatype recvtype, MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Reduce_intra_cnc(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, int root, MPIR_Comm * comm_ptr, + int coll_attr); +int MPIR_Allreduce_intra_cnc(const void *sendbuf, void *recvbuf, MPI_Aint count, + MPI_Datatype datatype, MPI_Op op, MPIR_Comm * comm_ptr, int coll_attr); +int MPIR_Reduce_scatter_block_intra_cnc(const void *sendbuf, void *recvbuf, MPI_Aint recvcount, + MPI_Datatype datatype, MPI_Op op, MPIR_Comm * comm_ptr, + int coll_attr); + /* Errors: instead of creating an error code, the MPIR_ERR_ macros format the * generic message, or the specific message with its arguments, into * unit_last_error, so a driver can print it. */ diff --git a/test/units/shim/mpl/mpl.h b/test/units/shim/mpl/mpl.h index d3817d7ef33..d26378b42ba 100644 --- a/test/units/shim/mpl/mpl.h +++ b/test/units/shim/mpl/mpl.h @@ -22,6 +22,7 @@ #include "../../../../src/mpl/include/mpl_err.h" /* memory allocation, ignoring the memory class */ +#define MPL_MEM_COLL 0 #define MPL_malloc(size, class) malloc(size) #define MPL_calloc(n, size, class) calloc(n, size) #define MPL_realloc(p, size, class) realloc(p, size)