OSHMEM: spml ikrit: fixes zero copy

Allow mxm to use zero copy in put() and get() for the large messages.

Author: alex-mikheev

oshmem/mca/spml/ikrit/spml_ikrit.c

@@ -902,12 +902,8 @@ int mca_spml_ikrit_get(void *src_addr, size_t size, void *dst_addr, int src)
         return OSHMEM_ERROR;
     }
 
-#if MXM_API < MXM_VERSION(2,0)
-    sreq.base.flags = MXM_REQ_FLAG_BLOCKING;
-#else
-    sreq.flags = MXM_REQ_SEND_FLAG_BLOCKING;
-#endif
     sreq.base.completed_cb = NULL;
+    sreq.flags = 0;
 
     SPML_IKRIT_MXM_POST_SEND(sreq);
 
@@ -1161,7 +1157,11 @@ static inline int mca_spml_ikrit_put_internal(void* dst_addr,
         put_req->mxm_req.opcode = MXM_REQ_OP_PUT;
     }
     if (!zcopy) {
-        put_req->mxm_req.flags |= MXM_REQ_SEND_FLAG_BLOCKING;
+        if (size < mca_spml_ikrit.put_zcopy_threshold) {
+            put_req->mxm_req.flags |= MXM_REQ_SEND_FLAG_BLOCKING;
+        } else {
+            put_req->mxm_req.opcode = MXM_REQ_OP_PUT_SYNC;
+        }
     }
 #endif
 

oshmem/mca/spml/ikrit/spml_ikrit.h

@@ -104,6 +104,8 @@ struct mca_spml_ikrit_t {
 #if MXM_API >= MXM_VERSION(2,0)
     int unsync_conn_max;
 #endif
+    size_t put_zcopy_threshold; /* enable zcopy in put if message size is
+                                   greater than the threshold */
 };
 
 typedef struct mca_spml_ikrit_t mca_spml_ikrit_t;

oshmem/mca/spml/ikrit/spml_ikrit_component.c

@@ -167,6 +167,21 @@ static inline void mca_spml_ikrit_param_register_int(const char* param_name,
                                            storage);
 }
 
+static inline void mca_spml_ikrit_param_register_size_t(const char* param_name,
+                                                        size_t default_value,
+                                                        const char *help_msg,
+                                                        size_t *storage)
+{
+    *storage = default_value;
+    (void) mca_base_component_var_register(&mca_spml_ikrit_component.spmlm_version,
+                                           param_name,
+                                           help_msg,
+                                           MCA_BASE_VAR_TYPE_SIZE_T, NULL, 0, 0,
+                                           OPAL_INFO_LVL_9,
+                                           MCA_BASE_VAR_SCOPE_READONLY,
+                                           storage);
+}
+
 static inline void  mca_spml_ikrit_param_register_string(const char* param_name,
                                                     char* default_value,
                                                     const char *help_msg,
@@ -230,6 +245,9 @@ static int mca_spml_ikrit_component_register(void)
                                       &mca_spml_ikrit.unsync_conn_max);
 #endif
 
+    mca_spml_ikrit_param_register_size_t("put_zcopy_threshold", 16384ULL,
+                                         "[size_t] Use zero copy put if message size is greater than the threshold",
+                                      &mca_spml_ikrit.put_zcopy_threshold);
     if (oshmem_num_procs() < mca_spml_ikrit.np) {
         SPML_VERBOSE(1,
                      "Not enough ranks (%d<%d), disqualifying spml/ikrit",