ch4/ofi: add new pipeline implementation

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/netmod/ofi/Makefile.mk

@@ -18,6 +18,8 @@ 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_rndv_read.c \
+                      src/mpid/ch4/netmod/ofi/ofi_pipeline.c \
                       src/mpid/ch4/netmod/ofi/ofi_huge.c \
                       src/mpid/ch4/netmod/ofi/ofi_progress.c \
                       src/mpid/ch4/netmod/ofi/ofi_am_events.c \

src/mpid/ch4/netmod/ofi/ofi_events.c

@@ -669,6 +669,26 @@ int MPIDI_OFI_dispatch_function(int vci, struct fi_cq_tagged_entry *wc, MPIR_Req
                 mpi_errno = MPIDI_OFI_rndv_cts_event(vci, wc, req);
                 break;
 
+            case MPIDI_OFI_EVENT_PIPELINE_SEND_CHUNK:
+                mpi_errno = MPIDI_OFI_pipeline_send_chunk_event(wc, req);
+                break;
+
+            case MPIDI_OFI_EVENT_PIPELINE_RECV_CHUNK:
+                mpi_errno = MPIDI_OFI_pipeline_recv_chunk_event(wc, req);
+                break;
+
+            case MPIDI_OFI_EVENT_RNDVREAD_RECV_MRS:
+                mpi_errno = MPIDI_OFI_rndvread_recv_mrs_event(wc, req);
+                break;
+
+            case MPIDI_OFI_EVENT_RNDVREAD_READ_CHUNK:
+                mpi_errno = MPIDI_OFI_rndvread_read_chunk_event(wc, req);
+                break;
+
+            case MPIDI_OFI_EVENT_RNDVREAD_ACK:
+                mpi_errno = MPIDI_OFI_rndvread_ack_event(wc, req);
+                break;
+
             case MPIDI_OFI_EVENT_CHUNK_DONE:
                 mpi_errno = chunk_done_event(vci, wc, req);
                 break;

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

@@ -36,6 +36,11 @@ int MPIDI_OFI_dispatch_function(int vci, struct fi_cq_tagged_entry *wc, MPIR_Req
 int MPIDI_OFI_recv_rndv_event(int vci, struct fi_cq_tagged_entry *wc, MPIR_Request * rreq);
 int MPIDI_OFI_peek_rndv_event(int vci, struct fi_cq_tagged_entry *wc, MPIR_Request * rreq);
 int MPIDI_OFI_rndv_cts_event(int vci, struct fi_cq_tagged_entry *wc, MPIR_Request * req);
+int MPIDI_OFI_pipeline_send_chunk_event(struct fi_cq_tagged_entry *wc, MPIR_Request * r);
+int MPIDI_OFI_pipeline_recv_chunk_event(struct fi_cq_tagged_entry *wc, MPIR_Request * r);
+int MPIDI_OFI_rndvread_recv_mrs_event(struct fi_cq_tagged_entry *wc, MPIR_Request * r);
+int MPIDI_OFI_rndvread_read_chunk_event(struct fi_cq_tagged_entry *wc, MPIR_Request * r);
+int MPIDI_OFI_rndvread_ack_event(struct fi_cq_tagged_entry *wc, MPIR_Request * r);
 
 MPL_STATIC_INLINE_PREFIX MPL_gpu_engine_type_t MPIDI_OFI_gpu_get_recv_engine_type(void)
 {

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

@@ -56,6 +56,8 @@ ATTRIBUTE((unused));
                                     MPIDI_OFI_global.prov_use[nic]->domain_attr->name : "(n/a)")
 #define MPIDI_OFI_DEFAULT_NIC_NAME (MPIDI_OFI_NIC_NAME(0))
 
+#define MPIDI_OFI_EAGER_THRESH (MPIR_CVAR_CH4_OFI_EAGER_THRESHOLD == -1 ? MPIDI_OFI_global.max_msg_size : MPIR_CVAR_CH4_OFI_EAGER_THRESHOLD)
+
 int MPIDI_OFI_progress_uninlined(int vci);
 int MPIDI_OFI_handle_cq_error(int vci, int nic, ssize_t ret);
 int MPIDI_OFI_flush_send_queue(void);

src/mpid/ch4/netmod/ofi/ofi_init.c

@@ -470,6 +470,26 @@ categories :
       description : >-
         If true, enable OFI triggered ops for MPI collectives.
 
+    - name        : MPIR_CVAR_CH4_OFI_PIPELINE_CHUNK_SZ
+      category    : CH4_OFI
+      type        : int
+      default     : 1048576
+      class       : none
+      verbosity   : MPI_T_VERBOSITY_USER_BASIC
+      scope       : MPI_T_SCOPE_LOCAL
+      description : >-
+        Specifies the chunk size (in bytes) for pipeline data transfer.
+
+    - name        : MPIR_CVAR_CH4_OFI_PIPELINE_NUM_CHUNKS
+      category    : CH4_OFI
+      type        : int
+      default     : 32
+      class       : none
+      verbosity   : MPI_T_VERBOSITY_USER_BASIC
+      scope       : MPI_T_SCOPE_LOCAL
+      description : >-
+        Specifies the number of chunk buffers for pipeline data transfer.
+
     - name        : MPIR_CVAR_CH4_OFI_ENABLE_GPU_PIPELINE
       category    : CH4_OFI
       type        : boolean
@@ -655,7 +675,7 @@ int MPIDI_OFI_init_local(int *tag_bits)
     MPL_COMPILE_TIME_ASSERT(offsetof(struct MPIR_Request, dev.ch4.netmod) ==
                             offsetof(MPIDI_OFI_dynamic_process_request_t, context));
     MPL_COMPILE_TIME_ASSERT(offsetof(struct MPIR_Request, dev.ch4.am.netmod_am.ofi.context) ==
-                            offsetof(struct MPIR_Request, dev.ch4.netmod.ofi.context));
+                            offsetof(struct MPIR_Request, dev.ch4.netmod.ofi.direct.context));
     MPL_COMPILE_TIME_ASSERT(sizeof(MPIDI_Devreq_t) >= sizeof(MPIDI_OFI_request_t));
 
     int err;
@@ -794,8 +814,7 @@ int MPIDI_OFI_init_local(int *tag_bits)
     MPIR_Assert(MPIDI_OFI_DEFAULT_SHORT_SEND_SIZE <= MPIR_CVAR_CH4_PACK_BUFFER_SIZE);
 
     MPIDI_OFI_global.num_vcis = 1;
-    MPIDI_OFI_am_init(0);
-    MPIDI_OFI_am_post_recv(0, 0);
+    MPIDI_OFI_init_per_vci(0);
 
   fn_exit:
     *tag_bits = MPIDI_OFI_TAG_BITS;
@@ -984,6 +1003,10 @@ int MPIDI_OFI_mpi_finalize_hook(void)
 
     MPIDIU_map_destroy(MPIDI_OFI_global.win_map);
 
+    for (int vci = 0; vci < MPIDI_OFI_global.num_vcis; vci++) {
+        MPIDU_genq_private_pool_destroy(MPIDI_OFI_global.per_vci[vci].pipeline_pool);
+    }
+
     if (MPIDI_OFI_ENABLE_AM) {
         for (int vci = 0; vci < MPIDI_OFI_global.num_vcis; vci++) {
             while (MPIDI_OFI_global.per_vci[vci].am_unordered_msgs) {
@@ -1584,7 +1607,35 @@ static void dump_global_settings(void)
 
 /* static functions for AM */
 
-int MPIDI_OFI_am_init(int vci)
+static int am_init(int vci);
+static int am_post_recv(int vci, int nic);
+
+int MPIDI_OFI_init_per_vci(int vci)
+{
+    int mpi_errno = MPI_SUCCESS;
+
+    /* Create chunk buffer pool (for pipeline etc.) */
+    mpi_errno = MPIDU_genq_private_pool_create(MPIR_CVAR_CH4_OFI_PIPELINE_CHUNK_SZ,
+                                               MPIR_CVAR_CH4_OFI_PIPELINE_NUM_CHUNKS,
+                                               MPIR_CVAR_CH4_OFI_PIPELINE_NUM_CHUNKS,
+                                               host_alloc_registered,
+                                               host_free_registered,
+                                               &MPIDI_OFI_global.per_vci[vci].pipeline_pool);
+    MPIR_ERR_CHECK(mpi_errno);
+
+    mpi_errno = am_init(vci);
+    MPIR_ERR_CHECK(mpi_errno);
+
+    mpi_errno = am_post_recv(vci, 0);
+    MPIR_ERR_CHECK(mpi_errno);
+
+  fn_exit:
+    return mpi_errno;
+  fn_fail:
+    goto fn_exit;
+}
+
+static int am_init(int vci)
 {
     int mpi_errno = MPI_SUCCESS;
 
@@ -1633,7 +1684,7 @@ int MPIDI_OFI_am_init(int vci)
     goto fn_exit;
 }
 
-int MPIDI_OFI_am_post_recv(int vci, int nic)
+static int am_post_recv(int vci, int nic)
 {
     int mpi_errno = MPI_SUCCESS;
 

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

@@ -34,8 +34,7 @@ void MPIDI_OFI_update_global_settings(struct fi_info *prov);
 bool MPIDI_OFI_nic_already_used(const struct fi_info *prov, struct fi_info **others, int nic_count);
 
 int MPIDI_OFI_create_vci_context(int vci, int nic);
-int MPIDI_OFI_am_init(int vci);
-int MPIDI_OFI_am_post_recv(int vci, int nic);
+int MPIDI_OFI_init_per_vci(int vci);
 
 bool MPIDI_OFI_nic_is_up(struct fi_info *prov);