Do more work at RDMA frag preparation time and less work at RDMA frag sending

time.

This commit was SVN r14627.

ompi/mca/pml/ob1/pml_ob1_rdmafrag.h

@@ -36,7 +36,7 @@ typedef enum {
 
 struct mca_pml_ob1_rdma_frag_t {
     ompi_free_list_item_t super;
-    mca_btl_base_module_t* rdma_btl;
+    mca_bml_base_btl_t* rdma_bml;
     mca_pml_ob1_hdr_t rdma_hdr;
     mca_pml_ob1_rdma_state_t rdma_state;
     size_t rdma_length;
@@ -44,6 +44,7 @@ struct mca_pml_ob1_rdma_frag_t {
     void *rdma_req;
     struct mca_bml_base_endpoint_t* rdma_ep;
     ompi_convertor_t convertor;
+    mca_mpool_base_registration_t* reg;
 };
 typedef struct mca_pml_ob1_rdma_frag_t mca_pml_ob1_rdma_frag_t;
 

ompi/mca/pml/ob1/pml_ob1_recvreq.c

@@ -332,19 +332,11 @@ int mca_pml_ob1_recv_request_get_frag(
         mca_pml_ob1_rdma_frag_t* frag)
 {
     mca_pml_ob1_recv_request_t* recvreq = (mca_pml_ob1_recv_request_t*)frag->rdma_req;
-    mca_bml_base_endpoint_t* bml_endpoint = frag->rdma_ep;
+    mca_bml_base_btl_t* bml_btl = frag->rdma_bml;
-    mca_bml_base_btl_t* bml_btl;
     mca_btl_base_descriptor_t* descriptor;
     size_t save_size = frag->rdma_length;
     int rc;
 
-    bml_btl = mca_bml_base_btl_array_find(&bml_endpoint->btl_rdma,
-            frag->rdma_btl);
-    if(NULL == bml_btl) {
-        opal_output(0, "[%s:%d] invalid bml for rdma get", __FILE__, __LINE__);
-        orte_errmgr.abort();
-    }
-
     /* prepare descriptor */
     mca_bml_base_prepare_dst(
         bml_btl, 
@@ -422,12 +414,17 @@ static void mca_pml_ob1_recv_request_rget(
         size += hdr->hdr_segs[i].seg_len;
         frag->rdma_segs[i] = hdr->hdr_segs[i];
     }
+    frag->rdma_bml = mca_bml_base_btl_array_find(&bml_endpoint->btl_rdma, btl);
+    if(NULL == frag->rdma_bml) {
+        opal_output(0, "[%s:%d] invalid bml for rdma get", __FILE__, __LINE__);
+        orte_errmgr.abort();
+    }
     frag->rdma_hdr.hdr_rget = *hdr;
     frag->rdma_req = recvreq;
     frag->rdma_ep = bml_endpoint;
-    frag->rdma_btl = btl;
     frag->rdma_length = size;
     frag->rdma_state = MCA_PML_OB1_RDMA_GET;
+    frag->reg = NULL;
 
     mca_pml_ob1_recv_request_get_frag(frag);
 }

ompi/mca/pml/ob1/pml_ob1_sendreq.c

@@ -1067,24 +1067,12 @@ static void mca_pml_ob1_put_completion( mca_btl_base_module_t* btl,
 
 int mca_pml_ob1_send_request_put_frag( mca_pml_ob1_rdma_frag_t* frag )
 {
-    mca_pml_ob1_send_request_t* sendreq = (mca_pml_ob1_send_request_t*)frag->rdma_req;
     mca_mpool_base_registration_t* reg = NULL;
-    mca_bml_base_btl_t* bml_btl;
+    mca_bml_base_btl_t* bml_btl = frag->rdma_bml;
     mca_btl_base_descriptor_t* des;
-    size_t i, save_size = frag->rdma_length;
+    size_t save_size = frag->rdma_length;
     int rc;
 
-    bml_btl = mca_bml_base_btl_array_find(&frag->rdma_ep->btl_rdma,
-                                          frag->rdma_btl);
-
-    /* lookup the corresponding registration */
-    for(i=0; i<sendreq->req_rdma_cnt; i++) {
-       if(sendreq->req_rdma[i].bml_btl == bml_btl) {
-           reg = sendreq->req_rdma[i].btl_reg;
-           break;
-       }
-    } 
-
     /* setup descriptor */
     mca_bml_base_prepare_src( bml_btl, 
                               reg,
@@ -1166,12 +1154,21 @@ void mca_pml_ob1_send_request_put(
         size += frag->rdma_segs[i].seg_len;
     }
 
+    frag->rdma_bml = mca_bml_base_btl_array_find(&bml_endpoint->btl_rdma, btl);
     frag->rdma_hdr.hdr_rdma = *hdr;
     frag->rdma_req = sendreq; 
     frag->rdma_ep = bml_endpoint;
-    frag->rdma_btl = btl;
     frag->rdma_length = size;
     frag->rdma_state = MCA_PML_OB1_RDMA_PUT;
+    frag->reg = NULL;
+
+    /* lookup the corresponding registration */
+    for(i=0; i<sendreq->req_rdma_cnt; i++) {
+       if(sendreq->req_rdma[i].bml_btl == frag->rdma_bml) {
+           frag->reg = sendreq->req_rdma[i].btl_reg;
+           break;
+       }
+    } 
 
     /*  RDMA writes may proceed in parallel to send and to each other, so
      *  create clone of the convertor for each RDMA fragment