adjust address for dist_cdo and fragmentation

transport/rdma.c

@@ -395,7 +395,7 @@ mstro_transport_rdma__read(const struct mstro_pm_app_registry_entry *app_entry,
                            mstro_cdo cdo_dst,
                            void *mr_desc,
                            uint64_t mr_addr, uint64_t mr_key,
-                           size_t offset, size_t len,
+                           size_t offset, size_t len, size_t dst_offset, 
                            struct mstro_transport_rdma_cb_args *closure)
 {
   mstro_status status;
@@ -432,7 +432,7 @@ mstro_transport_rdma__read(const struct mstro_pm_app_registry_entry *app_entry,
   size_t num_retries = NUM_READ_RETRIES;
 RETRY_RDMA_TRANSPORT_READ: ;
   int ret = fi_read(app_entry->ep->ep,
-                    (uint8_t*)cdo_dst->raw_ptr+offset, len,
+                    (uint8_t*)cdo_dst->raw_ptr+offset+dst_offset, len,
                     mr_desc,
                     app_entry->addr, mr_addr+offset, mr_key, ctx);
   if(ret==-FI_EAGAIN) {
@@ -521,7 +521,7 @@ mstro_transport_rdma_dst_execute(mstro_cdo cdo_dst, Mstro__Pool__TransferTicket*
   }
 
   uint64_t requested_key = app_entry->ep->fi->domain_attr->mr_mode & FI_MR_PROV_KEY ? 0 : mstro_memory_new_key();
-  int err = fi_mr_reg(app_entry->ep->domain, cdo_dst->raw_ptr, len,
+  int err = fi_mr_reg(app_entry->ep->domain, cdo_dst->raw_ptr+ticket->dst_offset, len,
                   FI_READ, 0, requested_key, 0, &mr, NULL);
   if (err) {
 	ERR("Couldn't register memory region for RDMA transport (err: %d, %s)\n", 
@@ -601,14 +601,14 @@ mstro_transport_rdma_dst_execute(mstro_cdo cdo_dst, Mstro__Pool__TransferTicket*
   }
 
   /* read full-fragment blocks */
-  size_t offset=ticket->dst_offset; // write dst cdos at the correct position
+  size_t offset=0; 
   size_t num_retries;
   for(size_t i=closure->num_fragments;
       i-->1; ) {
     status = mstro_transport_rdma__read(app_entry, cdo_dst,
                                         local_buf_mr_desc, mr_addr, mr_key,
-                                        offset, fragment_size,
-                                        closure);
+                                        offset, fragment_size, ticket->dst_offset, 
+                                        closure); //add dst_offset for dist_cdos
     if(status!=MSTRO_OK) {
       ERR("Failed to read fragment %zu\n", closure->num_fragments - i);
       goto BAILOUT;
@@ -619,7 +619,7 @@ mstro_transport_rdma_dst_execute(mstro_cdo cdo_dst, Mstro__Pool__TransferTicket*
   /* final (partial) read */
   status = mstro_transport_rdma__read(app_entry, cdo_dst,
                                       local_buf_mr_desc, mr_addr, mr_key,
-                                      offset, len-offset,
+                                      offset, len-offset, ticket->dst_offset,
                                       closure); 
   if(status!=MSTRO_OK) {
     ERR("Failed to read last fragment\n");