reimplement client node-local support

MichaelBrim · MichaelBrim · commit 41415759b079 · 2025-09-19T15:04:52.000-04:00
diff --git a/client/src/client_read.c b/client/src/client_read.c
@@ -701,7 +701,7 @@ int process_gfid_reads(unifyfs_client* client,
 
                 meta->needs_reads_sync = 0;
 
-                /* MJB TODO - rewrite to request local extents for single gfid */
+                /* request local extents for single gfid */
                 size_t chunk_count = 0;
                 unifyfs_data_chunk_t* chunks = NULL;
                 rc = invoke_client_node_local_extents_get_rpc(client,
diff --git a/client/src/margo_client.c b/client/src/margo_client.c
@@ -958,8 +958,8 @@ int invoke_client_node_local_extents_get_rpc(unifyfs_client* client,
         LOGDBG("%s got response ret=%" PRIi32, rpc_name, out.ret);
         ret = (int) out.ret;
         if (ret == (int) UNIFYFS_SUCCESS) {
-            *chunk_count = out.chunk_count;
-            void* out_buffer = pull_margo_bulk(rpc->handle, out.bulk_data,
+            *chunk_count = out.ext_count;
+            void* out_buffer = pull_margo_bulk(rpc->handle, out.bulk_extents,
                                                out.bulk_size, NULL);
             *chunks = (unifyfs_data_chunk_t*) out_buffer;
         }
diff --git a/common/src/unifyfs_client_rpcs.h b/common/src/unifyfs_client_rpcs.h
@@ -332,9 +332,9 @@ MERCURY_GEN_PROC(unifyfs_node_local_extents_get_in_t,
                  ((int32_t)(gfid)))
 MERCURY_GEN_PROC(unifyfs_node_local_extents_get_out_t,
                  ((int32_t)(ret))
-                 ((hg_size_t)(chunk_count))
-                 ((hg_bulk_t)(bulk_data))
-                 ((hg_size_t)(bulk_size)))
+                 ((hg_size_t)(ext_count))
+                 ((hg_size_t)(bulk_size))
+                 ((hg_bulk_t)(bulk_extents)))
 DECLARE_MARGO_RPC_HANDLER(unifyfs_node_local_extents_get_rpc)
 
 /* unifyfs_get_gfids_rpc (client => server)
diff --git a/server/src/unifyfs_client_rpc.c b/server/src/unifyfs_client_rpc.c
@@ -748,7 +748,6 @@ void process_client_gfids_rpc(client_rpc_req_t* creq)
     //       files, but only use the gfids. The client must then issue
     //       a separate metaget request for each gfid.
     unifyfs_file_attr_t* file_attrs = NULL;
-    hg_bulk_t bulk_gfids = HG_BULK_NULL;
     int* new_gfid_list = NULL;
     int num_file_attrs = 0;
     ret = unifyfs_invoke_broadcast_metaget_all(&file_attrs,
@@ -759,24 +758,26 @@ void process_client_gfids_rpc(client_rpc_req_t* creq)
         // Package all the gfids up into one list
         new_gfid_list = (int*) calloc(num_file_attrs, sizeof(int));
         if (NULL != new_gfid_list) {
-            /* initialize bulk handle for the gfid_list */
+            for (int i=0; i < num_file_attrs; i++) {
+                new_gfid_list[i] = file_attrs[i].gfid;
+            }
             
-            hg_size_t sizes[1] = { num_file_attrs * sizeof(int) };
-            void* ptrs[1] = { (void*)new_gfid_list };
+            /* initialize bulk for the gfid_list */
+            hg_bulk_t bulk_gfids = HG_BULK_NULL;
+            hg_size_t bulk_sz = (hg_size_t) num_file_attrs * sizeof(int);
             hret = margo_bulk_create(unifyfsd_rpc_context->shm_mid,
-                                     1, ptrs, sizes,
+                                     1, (void**)&new_gfid_list, &bulk_sz,
                                      HG_BULK_READ_ONLY, &bulk_gfids);
             if (hret != HG_SUCCESS) {
                 LOGDBG("margo_bulk_create() failed - %s",
                        HG_Error_to_string(hret));
                 free(new_gfid_list);
                 ret = UNIFYFS_ERROR_MARGO;
             } else {
-                for (int i=0; i < num_file_attrs; i++) {
-                    new_gfid_list[i] = file_attrs[i].gfid;
-                }
                 out->bulk_gfids = bulk_gfids;
-                creq->req_state->bulk = bulk_gfids; // free on rpc cleanup
+
+                /* request bulk free on rpc cleanup */
+                creq->req_state->bulk = bulk_gfids;
             }
         } else {
             ret = ENOMEM;
@@ -806,9 +807,38 @@ void process_client_node_local_extents_rpc(client_rpc_req_t* creq)
     unifyfs_node_local_extents_get_out_t* out = creq->req_state->outputs;
     assert((in != NULL) && (out != NULL));
 
-    /* MJB TODO - rewrite to take a single gfid and return the local extents
-     *            if the file is laminated and has not been reverse-synced */
-    ret = UNIFYFS_ERROR_NYI;
+    unifyfs_fops_ctx_t ctx = {
+        .client_req = creq,
+        .app_id     = in->app_id,
+        .client_id  = in->client_id,
+    };
+
+    int gfid = (int) in->gfid;
+    size_t n_extents = 0;
+    unifyfs_data_chunk_t* extents = NULL;
+    ret = unifyfs_fops_local_extents(&ctx, gfid, &n_extents, &extents);
+    if (ret != UNIFYFS_SUCCESS) {
+        LOGERR("unifyfs_fops_local_extents() failed");
+    } else {
+        out->ext_count = (hg_size_t) n_extents;
+
+        /* register user buffer for bulk access */
+        hg_bulk_t bulk_local;
+        hg_size_t bulk_sz = (hg_size_t) n_extents * sizeof(unifyfs_data_chunk_t);
+        hg_return_t hret = margo_bulk_create(unifyfsd_rpc_context->shm_mid,
+                                             1, (void**)&extents, &bulk_sz,
+                                             HG_BULK_READ_ONLY, &bulk_local);
+        if (hret != HG_SUCCESS) {
+            LOGERR("margo_bulk_create() failed");
+            ret = UNIFYFS_ERROR_MARGO;
+        } else {
+            out->bulk_extents = bulk_local;
+            out->bulk_size = bulk_sz;
+
+            /* request bulk free on rpc cleanup */
+            creq->req_state->bulk = bulk_local;
+        }
+    }
 
     out->ret = (int32_t) ret;
 
@@ -1163,7 +1193,7 @@ static void unifyfs_node_local_extents_get_rpc(hg_handle_t handle)
     if (NULL == creq) {
         unifyfs_node_local_extents_get_out_t out;
         out.ret = (int32_t) ENOMEM;
-        out.chunk_count = 0;
+        out.ext_count = 0;
         hg_return_t hret = margo_respond(handle, &out);
         if (hret != HG_SUCCESS) {
             LOGERR("margo_respond() failed - %s", HG_Error_to_string(hret));
diff --git a/server/src/unifyfs_fops.h b/server/src/unifyfs_fops.h
@@ -58,7 +58,6 @@ typedef int (*unifyfs_fops_read_t)(unifyfs_fops_ctx_t* ctx,
                                    size_t* coverage_begin_offset,
                                    size_t* coverage_end_offset);
 
-
 typedef int (*unifyfs_fops_transfer_t)(unifyfs_fops_ctx_t* ctx,
                                        int transfer_id,
                                        int gfid,
@@ -72,13 +71,19 @@ typedef int (*unifyfs_fops_unlink_t)(unifyfs_fops_ctx_t* ctx, int gfid);
 
 typedef int (*unifyfs_fops_get_gfids_t)(int** gfid_list, int* num_gfids);
 
+typedef int (*unifyfs_fops_local_extents_t)(unifyfs_fops_ctx_t* ctx,
+                                            int gfid,
+                                            size_t* ext_count,
+                                            unifyfs_data_chunk_t** extents);
+
 struct unifyfs_fops {
     const char* name;
     unifyfs_fops_init_t init;
     unifyfs_fops_filesize_t filesize;
     unifyfs_fops_fsync_t fsync;
     unifyfs_fops_get_gfids_t get_gfids;
     unifyfs_fops_laminate_t laminate;
+    unifyfs_fops_local_extents_t local_extents;
     unifyfs_fops_metaget_t metaget;
     unifyfs_fops_metaset_t metaset;
     unifyfs_fops_mread_t mread;
@@ -145,6 +150,18 @@ static inline int unifyfs_fops_laminate(unifyfs_fops_ctx_t* ctx, int gfid)
     return global_fops_tab->laminate(ctx, gfid);
 }
 
+static inline int unifyfs_fops_local_extents(unifyfs_fops_ctx_t* ctx,
+                                             int gfid,
+                                             size_t* ext_count,
+                                             unifyfs_data_chunk_t** extents)
+{
+    if (!global_fops_tab->local_extents) {
+        return ENOSYS;
+    }
+
+    return global_fops_tab->local_extents(ctx, gfid, ext_count, extents);
+}
+
 static inline int unifyfs_fops_metaget(unifyfs_fops_ctx_t* ctx,
                                        int gfid, unifyfs_file_attr_t* attr)
 {
diff --git a/server/src/unifyfs_fops_rpc.c b/server/src/unifyfs_fops_rpc.c
@@ -490,20 +490,86 @@ int rpc_get_gfids(
     return unifyfs_get_gfids(num_gfids, gfid_list);
 }
 
+static
+int rpc_local_extents(unifyfs_fops_ctx_t* ctx,
+                      int gfid,
+                      size_t* ext_count,
+                      unifyfs_data_chunk_t** extents)
+{
+    int ret;
+
+    assert((NULL != ext_count) && (NULL != extents));
+
+    *ext_count = 0;
+    *extents = NULL;
+
+    /* define extent covering whole file */
+    unifyfs_file_attr_t attrs;
+    memset(&attrs, 0, sizeof(attrs));
+    int rc = sm_get_fileattr(gfid, &attrs);
+    if (rc == UNIFYFS_SUCCESS) {
+        unifyfs_extent_t whole_file = {
+            .gfid = gfid,
+            .length = attrs.size,
+            .offset = 0
+        };
+
+        /* get all file extents */
+        unsigned int n_chunks = 0;
+        unifyfs_data_chunk_t* chunks = NULL;
+        int coverage;
+        rc = unifyfs_inode_get_extent_chunks(&whole_file,
+                                             &n_chunks, &chunks,
+                                             &coverage);
+        if ((UNIFYFS_SUCCESS == rc) && (n_chunks > 0)) {
+            /* find local chunks */
+            unsigned int n_local = 0;
+            unifyfs_data_chunk_t* local_chunks = (unifyfs_data_chunk_t*)
+                calloc(n_chunks, sizeof(unifyfs_data_chunk_t));
+            if (NULL == local_chunks) {
+                ret = ENOMEM;
+            } else {
+                for (unsigned int i=0; i < n_chunks; i++) {
+                    unifyfs_data_chunk_t* chk = chunks + i;
+                    unifyfs_data_chunk_t* lchk = local_chunks + n_local;
+                    if (chk->log_server == glb_pmi_rank) {
+                        *lchk = *chk;
+                        n_local++;
+                    }
+                }
+                ret = UNIFYFS_SUCCESS;
+            }
+            *ext_count = (size_t) n_local;
+            *extents = local_chunks;
+
+            /* release chunks array */
+            free(chunks);
+        } else {
+            ret = rc;
+        }
+    } else {
+        /* gfid not found, so we have no local extents */
+        ret = UNIFYFS_SUCCESS;
+    }
+
+    return ret;
+}
+
 static struct unifyfs_fops _fops_rpc = {
-    .name      = "rpc",
-    .init      = rpc_init,
-    .filesize  = rpc_filesize,
-    .fsync     = rpc_fsync,
-    .get_gfids = rpc_get_gfids,
-    .laminate  = rpc_laminate,
-    .metaget   = rpc_metaget,
-    .metaset   = rpc_metaset,
-    .mread     = rpc_mread,
-    .read      = rpc_read,
-    .transfer  = rpc_transfer,
-    .truncate  = rpc_truncate,
-    .unlink    = rpc_unlink
+    .name          = "rpc",
+    .init          = rpc_init,
+    .filesize      = rpc_filesize,
+    .fsync         = rpc_fsync,
+    .get_gfids     = rpc_get_gfids,
+    .laminate      = rpc_laminate,
+    .local_extents = rpc_local_extents,
+    .metaget       = rpc_metaget,
+    .metaset       = rpc_metaset,
+    .mread         = rpc_mread,
+    .read          = rpc_read,
+    .transfer      = rpc_transfer,
+    .truncate      = rpc_truncate,
+    .unlink        = rpc_unlink
 };
 
 struct unifyfs_fops* unifyfs_fops_impl = &_fops_rpc;
diff --git a/server/src/unifyfs_inode.c b/server/src/unifyfs_inode.c
@@ -777,7 +777,8 @@ int get_extent_cache_chunks(unifyfs_extent_t* extent,
         /* found begin and end extents, convert to chunks */
         int gap_found = 0;
         int n_chk = 1 + (em_end - em_begin);
-        unifyfs_data_chunk_t* chks = calloc(n_chk, sizeof(*chks));
+        unifyfs_data_chunk_t* chks = (unifyfs_data_chunk_t*) 
+            calloc(n_chk, sizeof(unifyfs_data_chunk_t));
         if (NULL != chks) {
             extent_metadata* em_prev;
             extent_metadata* em_iter = em_begin;
