avoid bcast and send of extent cache

Author: MichaelBrim

server/src/unifyfs_group_rpc.c

@@ -1270,20 +1270,20 @@ int unifyfs_invoke_broadcast_extents_cache(int gfid)
     hg_id_t op_hgid = unifyfsd_rpc_context->rpcs.extent_cache_bcast_id;
     hg_size_t buf_size = n_extents * sizeof(*extents);
     hg_bulk_t extents_bulk;
-    //void* buf = (void*) extents; 
+    void* buf = (void*) extents; 
     // MJB TESTING: make a copy to avoid reuse of cache as bulk across
     //              concurrent bcasts
-    void* buf = malloc((size_t)buf_size);
-    if (NULL != buf) {
-        memcpy(buf, (void*)extents, (size_t)buf_size);
+    //void* buf = malloc((size_t)buf_size);
+    //if (NULL != buf) {
+        //memcpy(buf, (void*)extents, (size_t)buf_size);
         hg_return_t hret = margo_bulk_create(unifyfsd_rpc_context->svr_mid, 1,
                                              &buf, &buf_size,
                                              HG_BULK_READ_ONLY, &extents_bulk);
         if (hret != HG_SUCCESS) {
             LOGERR("margo_bulk_create() failed - %s",
                    HG_Error_to_string(hret));
             ret = UNIFYFS_ERROR_MARGO;
-            free(buf);
+            //free(buf);
         } else {
             coll_request* coll = NULL;
             extent_cache_bcast_in_t* in = calloc(1, sizeof(*in));
@@ -1301,18 +1301,27 @@ int unifyfs_invoke_broadcast_extents_cache(int gfid)
                 coll = collective_create(rpc, HG_HANDLE_NULL, op_hgid,
                                          glb_pmi_rank, (void*)in, NULL,
                                          sizeof(extent_cache_bcast_out_t),
-                                         HG_BULK_NULL, extents_bulk, buf);
+                                         HG_BULK_NULL, extents_bulk, NULL//buf
+                                         );
                 if (NULL == coll) {
                     ret = ENOMEM;
                 } else {
+                    /* start the broadcast */
                     ret = collective_forward(coll);
                     if (ret == UNIFYFS_SUCCESS) {
-                        ret = invoke_bcast_progress_rpc(coll);
+                        /* progress/finish the bcast operation */
+                        LOGDBG("BCAST_RPC: bcast progress collective(%p)",
+                               coll);
+                        ret = collective_finish(coll);
+                        if (ret != UNIFYFS_SUCCESS) {
+                            LOGERR("finish failed for coll(%p) (rc=%d)",
+                                   coll, ret);
+                        }
                     }
                 }
             }
         }
-    }
+    //}
 
     return ret;
 }

server/src/unifyfs_p2p_rpc.c

@@ -1056,6 +1056,8 @@ int unifyfs_invoke_get_extents_rpc(int gfid,
     return ret;
 }
 
+static struct timespec last_cache_bcast = (struct timespec) {0};
+
 static void process_get_extents_rpc(server_rpc_req_t* sreq)
 {
     int ret;
@@ -1090,9 +1092,16 @@ static void process_get_extents_rpc(server_rpc_req_t* sreq)
                 LOGDBG("owner has newer extents metadata");
                 send_extents = 1;
                 if (src_stamp.tv_sec == 0) {
-                    /* source timestamp is zero, time to broadcast */
-                    LOGDBG("broadcasting extents metadata to cache");
-                    ret = unifyfs_invoke_broadcast_extents_cache(gfid);
+                    /* source timestamp is zero, need to broadcast? */
+                    cmp = compare_timespec(&owner_stamp, &last_cache_bcast);
+                    if (0 != cmp) {
+                        last_cache_bcast = owner_stamp;
+                        LOGDBG("broadcasting extents metadata to cache");
+                        ret = unifyfs_invoke_broadcast_extents_cache(gfid);
+                        if (UNIFYFS_SUCCESS == ret) {
+                            send_extents = 0;
+                        }
+                    }
                 }
             } else if (-1 == cmp) {
                 /* source timestamp is newer, which should not happen.