MTL: make MXM work with read (in blocking send case) call-backs.

This commit was SVN r26807.

ompi/mca/mtl/mxm/mtl_mxm_send.c

@@ -16,6 +16,58 @@
 #include "mtl_mxm_request.h"
 #include "ompi/mca/mtl/base/mtl_base_datatype.h"
 
+static size_t ompi_mtl_mxm_stream_send(void *buffer, size_t length, size_t offset, void *context)
+{
+    struct iovec iov;
+    uint32_t iov_count = 1;
+
+    opal_convertor_t *convertor = (opal_convertor_t *) context;
+
+    iov.iov_len = length;
+    iov.iov_base = buffer;
+
+    opal_convertor_set_position(convertor, &offset);
+    opal_convertor_pack(convertor, &iov, &iov_count, &length);
+
+    return length;
+}
+
+static inline __opal_attribute_always_inline__ int
+               ompi_mtl_mxm_choose_send_datatype(mxm_send_req_t *mxm_send_req,
+                                           opal_convertor_t *convertor)
+{
+    struct iovec iov;
+    uint32_t iov_count = 1;
+
+    size_t *buffer_len = &mxm_send_req->base.data.buffer.length;
+
+    opal_convertor_get_packed_size(convertor, buffer_len);
+    if (0 == *buffer_len) {
+        mxm_send_req->base.data.buffer.ptr = NULL;
+        mxm_send_req->base.data_type = MXM_REQ_DATA_BUFFER;
+
+        return OMPI_SUCCESS;
+    }
+
+    if (opal_convertor_need_buffers(convertor)) {
+        mxm_send_req->base.context = convertor;
+        mxm_send_req->base.data_type = MXM_REQ_DATA_STREAM;
+        mxm_send_req->base.data.stream.length = *buffer_len;
+        mxm_send_req->base.data.stream.cb = ompi_mtl_mxm_stream_send;
+
+        return OMPI_SUCCESS;
+    }
+
+    mxm_send_req->base.data_type = MXM_REQ_DATA_BUFFER;
+
+    iov.iov_base = NULL;
+    iov.iov_len = *buffer_len;
+
+    opal_convertor_pack(convertor, &iov, &iov_count, buffer_len);
+    mxm_send_req->base.data.buffer.ptr = iov.iov_base;
+
+    return OMPI_SUCCESS;
+}
 
 static void ompi_mtl_mxm_send_completion_cb(void *context)
 {
@@ -42,7 +94,6 @@ int ompi_mtl_mxm_send(struct mca_mtl_base_module_t* mtl,
 {
     mxm_send_req_t mxm_send_req;
     mxm_wait_t wait;
-    bool free_after;
     mxm_error_t err;
     int ret;
 
@@ -51,16 +102,14 @@ int ompi_mtl_mxm_send(struct mca_mtl_base_module_t* mtl,
     mxm_send_req.base.mq            = ompi_mtl_mxm_mq_lookup(comm);
     mxm_send_req.base.conn          = ompi_mtl_mxm_conn_lookup(comm, dest);
     mxm_send_req.base.flags         = MXM_REQ_FLAG_BLOCKING;
-    mxm_send_req.base.data_type     = MXM_REQ_DATA_BUFFER;
-    ret = ompi_mtl_datatype_pack(convertor, &mxm_send_req.base.data.buffer.ptr,
-                                 &mxm_send_req.base.data.buffer.length,
-                                 &free_after);
-    if (OMPI_SUCCESS != ret) {
+
+    mxm_send_req.base.context       = NULL;
+    ret = ompi_mtl_mxm_choose_send_datatype(&mxm_send_req, convertor);
+    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
         return ret;
     }
 
     mxm_send_req.base.data.buffer.mkey   = MXM_MKEY_NONE;
-    mxm_send_req.base.context            = NULL;
     mxm_send_req.base.completed_cb       = NULL;
     if (mode == MCA_PML_BASE_SEND_SYNCHRONOUS) {
         mxm_send_req.base.flags          |= MXM_REQ_FLAG_SEND_SYNC;