Merge pull request #7529 from hzhou/2507_pipeline

ch4/ofi: add new pipeline implementation

Approved-by: Ken Raffenetti

Author: hzhou

doc/mpich/tuning_parameters.md

@@ -1096,27 +1096,22 @@ GPU pipeline uses host buffer and pipelining technique to send internode
 messages instead of GPU RDMA. To enable this mode, use the following two
 CVARs:
 
-* `MPIR_CVAR_CH4_OFI_ENABLE_GPU_PIPELINE`: This CVAR enables GPU pipeline
-for inter-node pt2pt messages
-* `MPIR_CVAR_CH4_OFI_GPU_PIPELINE_THRESHOLD`: The threshold to start using
-GPU pipelining. Default is 1MB.
+* `MPIR_CVAR_CH4_OFI_EAGER_THRESHOLD`: This CVAR enables enables the RNDV
+(rendezvous) path for large messages above the threshold. Recommended value
+is 1MB.
 
-* `MPIR_CVAR_CH4_OFI_GPU_PIPELINE_BUFFER_SZ`: Specifies the chunk size
-(in bytes) for GPU pipeline data transfer.
+* `MPIR_CVAR_CH4_OFI_RNDV_PROTOCOL=pipeline`: Forces the RNDV algorithm to
+use pipelining. The default is "auto", which will select best algorithms
+based on message attributes. Other include protocols include "read" - RDMA
+read, and "direct", which relies on underlying network library implementations.
 
-* `MPIR_CVAR_CH4_OFI_GPU_PIPELINE_NUM_BUFFERS_PER_CHUNK`: Specifies the
+* `MPIR_CVAR_CH4_OFI_PIPELINE_CHUNK_SZ`: Specifies the chunk size
+(in bytes) for pipeline data transfer.
+
+* `MPIR_CVAR_CH4_OFI_PIPELINE_NUM_CHUNKS`: Specifies the
 number of buffers for GPU pipeline data transfer in each block/chunk of
 the pool.
 
-* `MPIR_CVAR_CH4_OFI_GPU_PIPELINE_MAX_NUM_BUFFERS`: Specifies the maximum
-total number of buffers MPICH buffer pool can allocate.
-
-* `MPIR_CVAR_CH4_OFI_GPU_PIPELINE_D2H_ENGINE_TYPE`: Specify engine type
-for copying from device to host (sender side), default 0
-
-* `MPIR_CVAR_CH4_OFI_GPU_PIPELINE_H2D_ENGINE_TYPE`: Specify engine type
-for copying from host to device (receiver side), default 0
-
 To enable GPU Direct RDMA support for pt2pt communication, use the
 following CVARs:
 * `MPIR_CVAR_CH4_OFI_ENABLE_HMEM`: This CVAR with a value of `1` enables

src/include/mpiimpl.h

@@ -155,7 +155,6 @@ typedef struct MPIR_Stream MPIR_Stream;
 /******************* PART 3: DEVICE INDEPENDENT HEADERS **********************/
 /*****************************************************************************/
 
-#include "mpir_misc.h"
 #include "mpir_dbg.h"
 #include "mpir_objects.h"
 #include "mpir_strerror.h"
@@ -166,6 +165,7 @@ typedef struct MPIR_Stream MPIR_Stream;
 #include "mpir_mem.h"
 #include "mpir_info.h"
 #include "mpir_errcodes.h"
+#include "mpir_misc.h"
 #include "mpir_errhandler.h"
 #include "mpir_attr_generic.h"
 #include "mpir_contextid.h"

src/include/mpir_misc.h

@@ -74,6 +74,26 @@ typedef struct {
     MPIR_request_type_t type;
 } MPIR_gpu_req;
 
+MPL_STATIC_INLINE_PREFIX void MPIR_async_test(MPIR_gpu_req * areq, int *is_done)
+{
+    int err;
+    switch (areq->type) {
+        case MPIR_NULL_REQUEST:
+            /* a dummy, immediately complete */
+            *is_done = 1;
+            break;
+        case MPIR_TYPEREP_REQUEST:
+            MPIR_Typerep_test(areq->u.y_req, is_done);
+            break;
+        case MPIR_GPU_REQUEST:
+            err = MPL_gpu_test(&areq->u.gpu_req, is_done);
+            MPIR_Assertp(err == MPL_SUCCESS);
+            break;
+        default:
+            MPIR_Assert(0);
+    }
+}
+
 int MPIR_Localcopy(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype,
                    void *recvbuf, MPI_Aint recvcount, MPI_Datatype recvtype);
 int MPIR_Ilocalcopy(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype,

src/include/mpir_typerep.h

@@ -78,8 +78,6 @@ int MPIR_Typerep_ipack(const void *inbuf, MPI_Aint incount, MPI_Datatype datatyp
 int MPIR_Typerep_iunpack(const void *inbuf, MPI_Aint insize, void *outbuf, MPI_Aint outcount,
                          MPI_Datatype datatype, MPI_Aint outoffset, MPI_Aint * actual_unpack_bytes,
                          MPIR_Typerep_req * typerep_req, uint32_t flags);
-int MPIR_Typerep_wait(MPIR_Typerep_req typerep_req);
-int MPIR_Typerep_test(MPIR_Typerep_req typerep_req, int *completed);
 
 int MPIR_Typerep_size_external32(MPI_Datatype type);
 int MPIR_Typerep_pack_external(const void *inbuf, MPI_Aint incount, MPI_Datatype datatype,

src/mpi/datatype/typerep/src/typerep_pre.h

@@ -28,4 +28,7 @@ typedef struct {
 #define MPIR_TYPEREP_HANDLE_NULL NULL
 #endif
 
+int MPIR_Typerep_wait(MPIR_Typerep_req typerep_req);
+int MPIR_Typerep_test(MPIR_Typerep_req typerep_req, int *completed);
+
 #endif /* TYPEREP_PRE_H_INCLUDED */

src/mpi/misc/utils.c

@@ -500,7 +500,12 @@ int MPIR_Ilocalcopy_gpu(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype se
         do_localcopy(sendbuf, sendcount, sendtype, sendoffset, recvbuf, recvcount, recvtype,
                      recvoffset, LOCALCOPY_NONBLOCKING, &req->u.y_req);
     MPIR_ERR_CHECK(mpi_errno);
-    req->type = MPIR_TYPEREP_REQUEST;
+
+    if (req->u.y_req.req == MPIR_TYPEREP_REQ_NULL) {
+        req->type = MPIR_NULL_REQUEST;
+    } else {
+        req->type = MPIR_TYPEREP_REQUEST;
+    }
 #endif
 
   fn_exit:

src/mpid/ch4/ch4_api.txt

@@ -71,8 +71,8 @@ Non Native API:
       NM*: am_hdr_sz, data_sz, data, count, datatype, sreq
      SHM*: am_hdr_sz, data_sz, data, count, datatype, sreq
   am_can_do_tag: bool
-      NM*: void
-     SHM*: void
+      NM*: rreq
+     SHM*: rreq
   am_tag_send : int
       NM*: rank, comm, handler_id, tag, buf, count, datatype, src_vci, dst_vci, sreq
      SHM*: rank, comm, handler_id, tag, buf, count, datatype, src_vci, dst_vci, sreq

src/mpid/ch4/netmod/ofi/Makefile.mk

@@ -18,7 +18,9 @@ mpi_core_sources   += src/mpid/ch4/netmod/ofi/func_table.c \
                       src/mpid/ch4/netmod/ofi/ofi_part.c \
                       src/mpid/ch4/netmod/ofi/ofi_events.c \
                       src/mpid/ch4/netmod/ofi/ofi_rndv.c \
-                      src/mpid/ch4/netmod/ofi/ofi_huge.c \
+		      src/mpid/ch4/netmod/ofi/ofi_rndv_read.c \
+		      src/mpid/ch4/netmod/ofi/ofi_rndv_write.c \
+                      src/mpid/ch4/netmod/ofi/ofi_pipeline.c \
                       src/mpid/ch4/netmod/ofi/ofi_progress.c \
                       src/mpid/ch4/netmod/ofi/ofi_am_events.c \
                       src/mpid/ch4/netmod/ofi/ofi_nic.c \

src/mpid/ch4/netmod/ofi/ofi_am.h

@@ -215,9 +215,17 @@ MPL_STATIC_INLINE_PREFIX bool MPIDI_NM_am_check_eager(MPI_Aint am_hdr_sz, MPI_Ai
     }
 }
 
-MPL_STATIC_INLINE_PREFIX bool MPIDI_NM_am_can_do_tag(void)
+MPL_STATIC_INLINE_PREFIX bool MPIDI_NM_am_can_do_tag(MPIR_Request * rreq)
 {
-    return MPIDI_OFI_ENABLE_TAGGED;
+    if (MPIDI_OFI_ENABLE_TAGGED) {
+        MPI_Aint data_sz;
+        MPIR_Datatype_get_size_macro(MPIDIG_REQUEST(rreq, datatype), data_sz);
+        data_sz *= MPIDIG_REQUEST(rreq, count);
+        if (data_sz <= MPIDI_OFI_global.max_msg_size) {
+            return true;
+        }
+    }
+    return false;
 }
 
 MPL_STATIC_INLINE_PREFIX MPIDIG_recv_data_copy_cb MPIDI_NM_am_get_data_copy_cb(uint32_t attr)