use the btls min rdma size as the cutoff (rdma offset) for the send protocol

This commit was SVN r6372.

ompi/mca/pml/ob1/pml_ob1.c

@@ -363,9 +363,12 @@ int mca_pml_ob1_add_procs(ompi_proc_t** procs, size_t nprocs)
 
             /* check flags - is rdma prefered */
             if(endpoint->btl->btl_flags & MCA_BTL_FLAGS_RDMA &&
-               proc->proc_arch == ompi_proc_local_proc->proc_arch) {
+                proc->proc_arch == ompi_proc_local_proc->proc_arch) {
                 mca_pml_ob1_endpoint_t* rdma_ep = mca_pml_ob1_ep_array_insert(&proc_pml->btl_rdma);
                 *rdma_ep = *endpoint;
+                if(proc_pml->proc_rdma_offset < rdma_ep->btl_min_rdma_size) {
+                    proc_pml->proc_rdma_offset = rdma_ep->btl_min_rdma_size;
+                }
             }
         }
     }

ompi/mca/pml/ob1/pml_ob1.h

@@ -57,7 +57,6 @@ struct mca_pml_ob1_t {
     int free_list_max;      /* maximum size of free list */
     int free_list_inc;      /* number of elements to grow free list */
     size_t eager_limit;     /* maximum eager limit size - overrides btl setting */
-    size_t rdma_offset;     /* offset at which we attempt to initiate rdma */
     size_t send_pipeline_depth;
     size_t recv_pipeline_depth;
     bool leave_pinned; 

ompi/mca/pml/ob1/pml_ob1_component.c

@@ -115,8 +115,6 @@ int mca_pml_ob1_component_open(void)
         mca_pml_ob1_param_register_int("send_pipeline_depth", 3);
     mca_pml_ob1.recv_pipeline_depth =
         mca_pml_ob1_param_register_int("recv_pipeline_depth", 5);
-    mca_pml_ob1.rdma_offset = 
-        mca_pml_ob1_param_register_int("rdma_offset", 512*1024); 
     
     mca_base_param_register_int("mpi", NULL, "leave_pinned", "leave_pinned", 0); 
     param = mca_base_param_find("mpi", NULL, "leave_pinned"); 

ompi/mca/pml/ob1/pml_ob1_proc.c

@@ -25,6 +25,7 @@ static void mca_pml_ob1_proc_construct(mca_pml_ob1_proc_t* proc)
 {
     proc->proc_ompi = NULL;
     proc->proc_sequence = 0;
+    proc->proc_rdma_offset = 0;
     OBJ_CONSTRUCT(&proc->proc_lock, opal_mutex_t);
     OBJ_CONSTRUCT(&proc->btl_eager, mca_pml_ob1_ep_array_t);
     OBJ_CONSTRUCT(&proc->btl_send,  mca_pml_ob1_ep_array_t);

ompi/mca/pml/ob1/pml_ob1_proc.h

@@ -41,6 +41,7 @@ struct mca_pml_proc_t {
    mca_pml_ob1_ep_array_t btl_eager;   /**< array of endpoints to use for first fragments */
    mca_pml_ob1_ep_array_t btl_send;    /**< array of endpoints to use for remaining fragments */
    mca_pml_ob1_ep_array_t btl_rdma;    /**< array of endpoints that support (prefer) rdma */
+   size_t proc_rdma_offset;            /**< max of min rdma size for available rmda btls */
 };
 typedef struct mca_pml_proc_t mca_pml_ob1_proc_t;
 

ompi/mca/pml/ob1/pml_ob1_recvreq.c

@@ -165,12 +165,12 @@ static void mca_pml_ob1_recv_request_ack(
      * - rdma devices are available
     */
     if(NULL == recvreq->req_mpool && !mca_pml_ob1.leave_pinned) {
-        if(recvreq->req_recv.req_bytes_packed > mca_pml_ob1.rdma_offset &&
+        if(recvreq->req_recv.req_bytes_packed > proc->proc_rdma_offset &&
            mca_pml_ob1_ep_array_get_size(&proc->btl_rdma) &&
            ompi_convertor_need_buffers(&recvreq->req_recv.req_convertor) == 0) {
     
             /* use convertor to figure out the rdma offset for this request */
-            recvreq->req_rdma_offset = mca_pml_ob1.rdma_offset;
+            recvreq->req_rdma_offset = proc->proc_rdma_offset;
             ompi_convertor_set_position(
                 &recvreq->req_recv.req_convertor,
                 &recvreq->req_rdma_offset);