In multithreaded application sending of initial portion of a request may overlap

with RDMAing the rest of it. Also more than one RDMA writes can be performed
simultaneously by different threads. To make this code thread safe this patch
clones original request convertor for each RDMA fragment.

This commit was SVN r14574.

ompi/mca/pml/ob1/pml_ob1_rdmafrag.h

@@ -42,7 +42,8 @@ struct mca_pml_ob1_rdma_frag_t {
     size_t rdma_length;
     mca_btl_base_segment_t rdma_segs[MCA_BTL_DES_MAX_SEGMENTS];
     void *rdma_req;
-    struct mca_bml_base_endpoint_t* rdma_ep; 
+    struct mca_bml_base_endpoint_t* rdma_ep;
+    ompi_convertor_t convertor;
 };
 typedef struct mca_pml_ob1_rdma_frag_t mca_pml_ob1_rdma_frag_t;
 

ompi/mca/pml/ob1/pml_ob1_sendreq.c

@@ -1071,7 +1071,6 @@ int mca_pml_ob1_send_request_put_frag( mca_pml_ob1_rdma_frag_t* frag )
     mca_mpool_base_registration_t* reg = NULL;
     mca_bml_base_btl_t* bml_btl;
     mca_btl_base_descriptor_t* des;
-    size_t offset = (size_t)frag->rdma_hdr.hdr_rdma.hdr_rdma_offset;
     size_t i, save_size = frag->rdma_length;
     int rc;
 
@@ -1086,19 +1085,18 @@ int mca_pml_ob1_send_request_put_frag( mca_pml_ob1_rdma_frag_t* frag )
        }
     } 
 
-    /* set convertor at current offset */
-    ompi_convertor_set_position(&sendreq->req_send.req_convertor, &offset);
-
     /* setup descriptor */
     mca_bml_base_prepare_src( bml_btl, 
                               reg,
-                              &sendreq->req_send.req_convertor, 
+                              &frag->convertor, 
                               0,
                               &frag->rdma_length, 
                               &des );
     
     if(NULL == des) {
+        size_t offset = (size_t)frag->rdma_hdr.hdr_rdma.hdr_rdma_offset;
         frag->rdma_length = save_size; 
+        ompi_convertor_set_position(&frag->convertor, &offset);
         OPAL_THREAD_LOCK(&mca_pml_ob1.lock);
         opal_list_append(&mca_pml_ob1.rdma_pending, (opal_list_item_t*)frag);
         OPAL_THREAD_UNLOCK(&mca_pml_ob1.lock);
@@ -1175,5 +1173,11 @@ void mca_pml_ob1_send_request_put(
     frag->rdma_length = size;
     frag->rdma_state = MCA_PML_OB1_RDMA_PUT;
 
+    /*  RDMA writes may proceed in parallel to send and to each other, so
+     *  create clone of the convertor for each RDMA fragment
+     */
+    ompi_convertor_clone_with_position(&sendreq->req_send.req_convertor,
+            &frag->convertor, 0, &hdr->hdr_rdma_offset);
+
     mca_pml_ob1_send_request_put_frag(frag);
 }