MTL MXM: MPI_Mprobe, MPI_Mrecv implementation for MXM adding.

This commit was SVN r26866.

ompi/mca/mtl/mxm/mtl_mxm.c

@@ -399,3 +399,9 @@ int ompi_mtl_mxm_progress(void)
     }
     return 1;
 }
+
+OBJ_CLASS_INSTANCE(
+        ompi_mtl_mxm_message_t,
+        ompi_free_list_item_t,
+        NULL,
+        NULL);

ompi/mca/mtl/mxm/mtl_mxm.h

@@ -20,6 +20,8 @@
 #include "ompi/mca/pml/pml.h"
 #include "ompi/mca/mtl/mtl.h"
 #include "ompi/mca/mtl/base/base.h"
+#include "ompi/class/ompi_free_list.h"
+
 #include "opal/datatype/opal_convertor.h"
 
 #include "mtl_mxm_debug.h"
@@ -82,6 +84,19 @@ extern int ompi_mtl_mxm_finalize(struct mca_mtl_base_module_t* mtl);
 
 int ompi_mtl_mxm_module_init(void);
 
+struct ompi_mtl_mxm_message_t {
+    ompi_free_list_item_t super;
+
+    mxm_mq_h mq;
+    mxm_conn_h conn;
+    mxm_message_h mxm_msg;
+
+    mxm_tag_t tag;
+    mxm_tag_t tag_mask;
+};
+typedef struct ompi_mtl_mxm_message_t ompi_mtl_mxm_message_t;
+OBJ_CLASS_DECLARATION(ompi_mtl_mxm_message_t);
+
 END_C_DECLS
 
 #endif

ompi/mca/mtl/mxm/mtl_mxm_component.c

@@ -82,6 +82,7 @@ static int ompi_mtl_mxm_component_register(void)
 
 static int ompi_mtl_mxm_component_open(void)
 {
+    int rc;
     mxm_error_t err;
 
     mca_mtl_mxm_output = opal_output_open(NULL);
@@ -99,6 +100,22 @@ static int ompi_mtl_mxm_component_open(void)
         return OPAL_ERR_NOT_AVAILABLE;
     }
 
+    OBJ_CONSTRUCT(&mca_mtl_mxm_component.mxm_messages, ompi_free_list_t);
+    rc = ompi_free_list_init_new(&mca_mtl_mxm_component.mxm_messages,
+                                  sizeof(ompi_mtl_mxm_message_t),
+                                  opal_cache_line_size,
+                                  OBJ_CLASS(ompi_mtl_mxm_message_t),
+                                  0, opal_cache_line_size,
+                                  32 /* free list num */,
+                                  -1 /* free list max */,
+                                  32 /* free list inc */,
+                                  NULL);
+    if (OMPI_SUCCESS != rc) {
+        orte_show_help("help-mtl-mxm.txt", "mxm init", true,
+                    mxm_error_string(err));
+        return OPAL_ERR_NOT_AVAILABLE;
+    }
+
     return OMPI_SUCCESS;
 }
 
@@ -106,6 +123,7 @@ static int ompi_mtl_mxm_component_close(void)
 {
     mxm_cleanup(ompi_mtl_mxm.mxm_context);
     ompi_mtl_mxm.mxm_context = NULL;
+    OBJ_DESTRUCT(&mca_mtl_mxm_component.mxm_messages);
     return OMPI_SUCCESS;
 }
 

ompi/mca/mtl/mxm/mtl_mxm_probe.c

@@ -11,6 +11,7 @@
 #include "mtl_mxm.h"
 #include "mtl_mxm_types.h"
 
+#include "ompi/message/message.h"
 #include "ompi/communicator/communicator.h"
 
 int ompi_mtl_mxm_iprobe(struct mca_mtl_base_module_t* mtl,
@@ -50,5 +51,64 @@ int ompi_mtl_mxm_improbe(struct mca_mtl_base_module_t *mtl,
                          struct ompi_message_t **message,
                          struct ompi_status_public_t *status)
 {
-    return OMPI_ERR_NOT_IMPLEMENTED;
+#if MXM_API >= 0x01010000
+    int rc;
+    mxm_error_t err;
+    mxm_recv_req_t req;
+
+    ompi_free_list_item_t *item;
+    ompi_mtl_mxm_message_t *msgp;
+
+    OMPI_FREE_LIST_WAIT(&mca_mtl_mxm_component.mxm_messages, item, rc);
+    if (OPAL_UNLIKELY(OMPI_SUCCESS != rc)) {
+        return rc;
+    }
+
+    msgp = (ompi_mtl_mxm_message_t *) item;
+
+    req.base.state = MXM_REQ_NEW;
+    ompi_mtl_mxm_set_recv_envelope(&req, comm, src, tag);
+
+    msgp->mq       = req.base.mq;
+    msgp->conn     = req.base.conn;
+    msgp->tag      = req.tag;
+    msgp->tag_mask = req.tag_mask;
+
+    err = mxm_req_mprobe(&req, &msgp->mxm_msg);
+    if (MXM_OK == err) {
+        if (MPI_STATUS_IGNORE != status) {
+            *matched = 1;
+            ompi_mtl_mxm_to_mpi_status(err, status);
+            status->MPI_SOURCE = req.completion.sender_imm;
+            status->MPI_TAG    = req.completion.sender_tag;
+            status->_ucount    = req.completion.sender_len;
+        } else{
+			*matched = 0;
+			*message = MPI_MESSAGE_NULL;
+			return OMPI_SUCCESS;
+		}
+    } else if (MXM_ERR_NO_MESSAGE == err) {
+        *matched = 0;
+        *message = MPI_MESSAGE_NULL;
+        return OMPI_SUCCESS;
+    } else {
+        return OMPI_ERROR;
+    }
+
+	(*message) = ompi_message_alloc();
+	if (OPAL_UNLIKELY(NULL == (*message))) {
+        *matched = 0;
+		*message = MPI_MESSAGE_NULL;
+		return OMPI_ERR_OUT_OF_RESOURCE;
+	}
+
+	(*message)->comm = comm;
+	(*message)->req_ptr = msgp;
+	(*message)->peer = status->MPI_SOURCE;
+	(*message)->count = status->_ucount;
+
+    return OMPI_SUCCESS;
+#else
+    return OMPI_ERR_NOT_IMPLEMENTED;
+#endif
 }

ompi/mca/mtl/mxm/mtl_mxm_recv.c

@@ -8,6 +8,7 @@
  */
 
 #include "ompi_config.h"
+#include "ompi/message/message.h"
 #include "opal/datatype/opal_convertor.h"
 #include "ompi/mca/mtl/base/mtl_base_datatype.h"
 
@@ -17,7 +18,6 @@
 #include "mtl_mxm_types.h"
 #include "mtl_mxm_request.h"
 
-
 static void ompi_mtl_mxm_recv_completion_cb(void *context)
 {
     mca_mtl_mxm_request_t *req = (mca_mtl_mxm_request_t *) context;
@@ -36,34 +36,23 @@ static void ompi_mtl_mxm_recv_completion_cb(void *context)
     req->super.completion_callback(&req->super);
 }
 
-
+static inline __opal_attribute_always_inline__ int
-int ompi_mtl_mxm_irecv(struct mca_mtl_base_module_t* mtl,
+        ompi_mtl_mxm_recv_init(mca_mtl_mxm_request_t *mtl_mxm_request,
-                       struct ompi_communicator_t *comm, int src, int tag,
+                               opal_convertor_t *convertor,
-                       struct opal_convertor_t *convertor,
+                               mxm_recv_req_t *mxm_recv_req)
-                       struct mca_mtl_request_t *mtl_request)
 {
-    mca_mtl_mxm_request_t * mtl_mxm_request;
-    mxm_error_t err;
-    mxm_recv_req_t *mxm_recv_req;
     int ret;
 
-    mtl_mxm_request = (mca_mtl_mxm_request_t*) mtl_request;
-
     mtl_mxm_request->convertor = convertor;
-    ret = ompi_mtl_datatype_recv_buf(mtl_mxm_request->convertor,
+    ret = ompi_mtl_datatype_recv_buf(convertor,
                                      &mtl_mxm_request->buf,
                                      &mtl_mxm_request->length,
                                      &mtl_mxm_request->free_after);
-    if (OMPI_SUCCESS != ret) {
+    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
         return ret;
     }
 
-
-    /* prepare a receive request embedded in the MTL request */
-    mxm_recv_req = &mtl_mxm_request->mxm.recv;
-
     mxm_recv_req->base.state               = MXM_REQ_NEW;
-    ompi_mtl_mxm_set_recv_envelope(mxm_recv_req, comm, src, tag);
 
     mxm_recv_req->base.flags               = 0;
     mxm_recv_req->base.data_type           = MXM_REQ_DATA_BUFFER;
@@ -73,9 +62,33 @@ int ompi_mtl_mxm_irecv(struct mca_mtl_base_module_t* mtl,
     mxm_recv_req->base.context             = mtl_mxm_request;
     mxm_recv_req->base.completed_cb        = ompi_mtl_mxm_recv_completion_cb;
 
+    return OMPI_SUCCESS;
+}
+
+int ompi_mtl_mxm_irecv(struct mca_mtl_base_module_t* mtl,
+                       struct ompi_communicator_t *comm, int src, int tag,
+                       struct opal_convertor_t *convertor,
+                       struct mca_mtl_request_t *mtl_request)
+{
+    int ret;
+    mxm_error_t err;
+    mxm_recv_req_t *mxm_recv_req;
+    mca_mtl_mxm_request_t * mtl_mxm_request;
+
+    mtl_mxm_request = (mca_mtl_mxm_request_t*) mtl_request;
+    mxm_recv_req = &mtl_mxm_request->mxm.recv;
+
+    ompi_mtl_mxm_set_recv_envelope(mxm_recv_req, comm, src, tag);
+
+    /* prepare a receive request embedded in the MTL request */
+    ret = ompi_mtl_mxm_recv_init(mtl_mxm_request, convertor, mxm_recv_req);
+    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
+        return ret;
+    }
+
     /* post-recv */
     err = mxm_req_recv(mxm_recv_req);
-    if (MXM_OK != err) {
+    if (OPAL_UNLIKELY(MXM_OK != err)) {
         orte_show_help("help-mtl-mxm.txt", "error posting receive", true,
                        mxm_error_string(err), mtl_mxm_request->buf, mtl_mxm_request->length);
         return OMPI_ERROR;
@@ -84,11 +97,48 @@ int ompi_mtl_mxm_irecv(struct mca_mtl_base_module_t* mtl,
     return OMPI_SUCCESS;
 }
 
-
 int ompi_mtl_mxm_imrecv(struct mca_mtl_base_module_t* mtl,
                         struct opal_convertor_t *convertor,
                         struct ompi_message_t **message,
                         struct mca_mtl_request_t *mtl_request)
 {
-    return OMPI_ERR_NOT_IMPLEMENTED;
+#if MXM_API >= 0x01010000
+    int ret;
+    mxm_error_t err;
+    mxm_recv_req_t *mxm_recv_req;
+    mca_mtl_mxm_request_t *mtl_mxm_request;
+
+    ompi_mtl_mxm_message_t *msgp =
+                        (ompi_mtl_mxm_message_t *) (*message)->req_ptr;
+
+    mtl_mxm_request = (mca_mtl_mxm_request_t*) mtl_request;
+    mxm_recv_req = &mtl_mxm_request->mxm.recv;
+
+    /* prepare a receive request embedded in the MTL request */
+    ret = ompi_mtl_mxm_recv_init(mtl_mxm_request, convertor, mxm_recv_req);
+    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
+        return ret;
+    }
+
+    mxm_recv_req->tag       = msgp->tag;
+    mxm_recv_req->tag_mask  = msgp->tag_mask;
+    mxm_recv_req->base.mq   = msgp->mq;
+    mxm_recv_req->base.conn = msgp->conn;
+
+    err = mxm_message_recv(mxm_recv_req, msgp->mxm_msg);
+    if (OPAL_UNLIKELY(MXM_OK != err)) {
+        orte_show_help("help-mtl-mxm.txt", "error posting message receive", true,
+                       mxm_error_string(err), mtl_mxm_request->buf, mtl_mxm_request->length);
+        return OMPI_ERROR;
+    }
+
+    OMPI_FREE_LIST_RETURN(&mca_mtl_mxm_component.mxm_messages,
+                         (ompi_free_list_item_t *) msgp);
+
+    (*message) = MPI_MESSAGE_NULL;
+
+    return OMPI_SUCCESS;
+#else
+    return OMPI_ERR_NOT_IMPLEMENTED;
+#endif
 }

ompi/mca/mtl/mxm/mtl_mxm_types.h

@@ -42,6 +42,7 @@ extern mca_mtl_mxm_module_t ompi_mtl_mxm;
 
 typedef struct mca_mtl_mxm_component_t {
     mca_mtl_base_component_2_0_0_t super; /**< base MTL component */
+    ompi_free_list_t mxm_messages; /* will be used for MPI_Mprobe and MPI_Mrecv calls */
 } mca_mtl_mxm_component_t;