Initial implementation of generic data (bulk) transfer from client to server

Author: houjun

src/api/include/pdc_client_connect.h

@@ -115,6 +115,10 @@ struct _pdc_transfer_request_metadata_query2_args {
     int32_t ret;
 };
 
+struct _generic_bulk_c2s_transfer_args {
+    int32_t ret;
+};
+
 struct _pdc_transfer_request_status_args {
     uint32_t status;
     int32_t  ret;
@@ -1135,4 +1139,6 @@ void report_avg_server_profiling_rst();
 perr_t PDC_Client_transfer_pthread_create();
 perr_t PDC_Client_transfer_pthread_terminate();
 perr_t PDC_Client_transfer_pthread_cnt_add(int n);
+
+perr_t PDC_data_transfer_c2s(uint32_t server_id, void *buf, uint64_t buf_size);
 #endif /* PDC_CLIENT_CONNECT_H */

src/api/pdc_client_connect.c

@@ -171,6 +171,7 @@ static hg_id_t transfer_request_status_register_id_g;
 static hg_id_t transfer_request_wait_register_id_g;
 static hg_id_t buf_map_register_id_g;
 static hg_id_t buf_unmap_register_id_g;
+static hg_id_t generic_bulk_c2s_transfer_register_id_g;
 
 static hg_id_t cont_add_del_objs_rpc_register_id_g;
 static hg_id_t cont_add_tags_rpc_register_id_g;
@@ -627,7 +628,6 @@ client_send_transfer_request_metadata_query_rpc_cb(const struct hg_cb_info *call
 }
 
 static hg_return_t
-
 client_send_transfer_request_metadata_query2_rpc_cb(const struct hg_cb_info *callback_info)
 {
     hg_return_t                                        ret_value = HG_SUCCESS;
@@ -1510,6 +1510,7 @@ PDC_Client_mercury_init(hg_class_t **hg_class, hg_context_t **hg_context, int po
     transfer_request_wait_register_id_g            = PDC_transfer_request_wait_register(*hg_class);
     buf_map_register_id_g                          = PDC_buf_map_register(*hg_class);
     buf_unmap_register_id_g                        = PDC_buf_unmap_register(*hg_class);
+    generic_bulk_c2s_transfer_register_id_g        = PDC_generic_bulk_c2s_transfer_register(*hg_class);
 
     // Analysis and Transforms
     analysis_ftn_register_id_g         = PDC_analysis_ftn_register(*hg_class);
@@ -9611,3 +9612,87 @@ PDC_Client_search_obj_ref_through_dart_mpi(dart_hash_algo_t hash_algo, char *que
 #endif
 
 /******************** Collective Object Selection Query Ends *******************************/
+
+// Generic bulk data transfer from client to server
+static hg_return_t
+generic_bulk_c2s_rpc_cb(const struct hg_cb_info *callback_info)
+{
+    hg_return_t ret_value = HG_SUCCESS;
+    hg_handle_t handle;
+    struct _generic_bulk_c2s_transfer_args *transfer_args;
+    generic_bulk_c2s_transfer_out_t output;
+
+    FUNC_ENTER(NULL);
+
+    fprintf(stderr, "entered %s\n", __func__);
+
+    transfer_args = (struct _generic_bulk_c2s_transfer_args*)callback_info->arg;
+    handle        = callback_info->info.forward.handle;
+
+    ret_value = HG_Get_output(handle, &output);
+    if (ret_value != HG_SUCCESS) {
+        PGOTO_ERROR(FAIL, "==CLIENT[%d]: ERROR with HG_Get_output @ line %d",
+                    pdc_client_mpi_rank_g, __LINE__);
+        transfer_args->ret = -1;
+        goto done;
+    }
+
+    transfer_args->ret = output.ret;
+    fprintf(stderr, "received return value %d\n", output.ret);
+
+done:
+    fflush(stdout);
+    hg_atomic_decr32(&atomic_work_todo_g);
+    HG_Free_output(handle, &output);
+
+    FUNC_LEAVE(ret_value);
+}
+
+perr_t
+PDC_data_transfer_c2s(uint32_t server_id, void *buf, uint64_t buf_size)
+{
+    perr_t ret_value = SUCCEED;
+    hg_return_t hg_ret = HG_SUCCESS;
+    hg_class_t *hg_class;
+    hg_handle_t rpc_handle;
+    generic_bulk_c2s_transfer_in_t in;
+    struct _generic_bulk_c2s_transfer_args transfer_args;
+
+    FUNC_ENTER(NULL);
+
+    debug_server_id_count[server_id]++;
+
+    hg_class = HG_Context_get_class(send_context_g);
+
+    if (PDC_Client_try_lookup_server(server_id, 0) != SUCCEED)
+        PGOTO_ERROR(FAIL, "==CLIENT[%d]: ERROR with PDC_Client_try_lookup_server @ line %d",
+                    pdc_client_mpi_rank_g, __LINE__);
+
+    hg_ret = HG_Create(send_context_g, pdc_server_info_g[server_id].addr,
+                       generic_bulk_c2s_transfer_register_id_g, &rpc_handle);
+    if (hg_ret != HG_SUCCESS)
+        PGOTO_ERROR(FAIL, "%s: Could not create rpc handle @ line %d\n", __func__, __LINE__);
+
+    in.buf_size = (hg_size_t) buf_size;
+
+    hg_ret = HG_Bulk_create(hg_class, 1, &buf, &in.buf_size, HG_BULK_READ_ONLY, &in.local_bulk_handle);
+    if (hg_ret != HG_SUCCESS)
+        PGOTO_ERROR(FAIL, "%s: Could not create local bulk data handle @ line %d\n", __func__, __LINE__);
+
+    hg_atomic_set32(&atomic_work_todo_g, 1);
+
+    hg_ret = HG_Forward(rpc_handle, generic_bulk_c2s_rpc_cb, &transfer_args, &in);
+    if (hg_ret != HG_SUCCESS)
+        PGOTO_ERROR(FAIL, "%s: Could not start HG_Forward() @ line %d\n", __func__, __LINE__);
+
+    PDC_Client_check_response(&send_context_g);
+
+    if (transfer_args.ret != 1)
+        PGOTO_ERROR(FAIL, "%s return failed @ line %d\n", __func__, __LINE__);
+
+    HG_Destroy(rpc_handle);
+
+done:
+    fflush(stdout);
+    FUNC_LEAVE(ret_value);
+}

src/server/include/pdc_client_server_common.h

@@ -851,6 +851,17 @@ typedef struct transfer_request_metadata_query2_out_t {
     int32_t ret;
 } transfer_request_metadata_query2_out_t;
 
+/* Define generic_bulk_c2s_transfer_in_t*/
+typedef struct generic_bulk_c2s_transfer_in_t {
+    hg_bulk_t local_bulk_handle;
+    hg_size_t buf_size;
+} generic_bulk_c2s_transfer_in_t;
+
+/* Define generic_bulk_c2s_transfer_out_t*/
+typedef struct generic_bulk_c2s_transfer_out_t {
+    int32_t ret;
+} generic_bulk_c2s_transfer_out_t;
+
 /* Define buf_map_in_t */
 typedef struct {
     uint32_t               meta_server_id;
@@ -2975,6 +2986,42 @@ hg_proc_transfer_request_metadata_query2_out_t(hg_proc_t proc, void *data)
     return ret;
 }
 
+/* Define hg_proc_generic_bulk_c2s_transfer_in_t */
+static HG_INLINE hg_return_t
+hg_proc_generic_bulk_c2s_transfer_in_t(hg_proc_t proc, void *data)
+{
+    hg_return_t ret;
+    generic_bulk_c2s_transfer_in_t *struct_data = (generic_bulk_c2s_transfer_in_t *)data;
+
+    ret = hg_proc_hg_bulk_t(proc, &struct_data->local_bulk_handle);
+    if (ret != HG_SUCCESS) {
+        fprintf(stderr, "Proc error @ %s:%d", __func__, __LINE__);
+        return ret;
+    }
+
+    ret = hg_proc_hg_size_t(proc, &struct_data->buf_size);
+    if (ret != HG_SUCCESS) {
+        fprintf(stderr, "Proc error @ %s:%d", __func__, __LINE__);
+        return ret;
+    }
+
+    return ret;
+}
+
+/* Define hg_proc_generic_bulk_c2s_transfer_out_t */
+static HG_INLINE hg_return_t
+hg_proc_generic_bulk_c2s_transfer_out_t(hg_proc_t proc, void *data)
+{
+    hg_return_t ret;
+    generic_bulk_c2s_transfer_out_t *struct_data = (generic_bulk_c2s_transfer_out_t *)data;
+
+    ret = hg_proc_int32_t(proc, &struct_data->ret);
+    if (ret != HG_SUCCESS) {
+        fprintf(stderr, "Proc error @ %s:%d", __func__, __LINE__);
+    }
+    return ret;
+}
+
 /* Define hg_proc_transfer_request_wait_in_t */
 static HG_INLINE hg_return_t
 hg_proc_transfer_request_wait_in_t(hg_proc_t proc, void *data)
@@ -4041,7 +4088,7 @@ struct bulk_args_t {
     uint64_t          cont_id;
     hg_handle_t       handle;
     hg_bulk_t         bulk_handle;
-    size_t            nbytes;
+    hg_size_t         nbytes;
     int               origin;
     size_t            ret;
     pdc_metadata_t ** meta_arr;
@@ -4326,6 +4373,7 @@ hg_id_t PDC_region_transform_release_register(hg_class_t *hg_class);
 hg_id_t PDC_transform_region_release_register(hg_class_t *hg_class);
 hg_id_t PDC_buf_map_server_register(hg_class_t *hg_class);
 hg_id_t PDC_buf_unmap_server_register(hg_class_t *hg_class);
+hg_id_t PDC_generic_bulk_c2s_transfer_register(hg_class_t *hg_class);
 
 hg_id_t PDC_test_bulk_xfer_register(hg_class_t *hg_class);
 hg_id_t PDC_server_lookup_remote_server_register(hg_class_t *hg_class);

src/server/pdc_client_server_common.c

@@ -4703,12 +4703,12 @@ HG_TEST_RPC_CB(query_kvtag, handle)
     FUNC_LEAVE(ret_value);
 }
 
+
 /*
  * Data server related
  */
 
 static hg_return_t
-
 update_storage_meta_bulk_cb(const struct hg_cb_info *hg_cb_info)
 {
     hg_return_t         ret_value         = HG_SUCCESS;
@@ -6534,6 +6534,86 @@ HG_TEST_RPC_CB(dart_perform_one_server, handle)
     FUNC_LEAVE(ret);
 }
 
+static hg_return_t
+process_generic_c2s_bulk_transfer_cb(const struct hg_cb_info *hg_cb_info)
+{
+    hg_return_t ret_value = HG_SUCCESS;
+    struct bulk_args_t *bulk_args = (struct bulk_args_t *)hg_cb_info->arg;
+    hg_bulk_t local_bulk_handle = hg_cb_info->info.bulk.local_handle;
+    void *buf;
+    generic_bulk_c2s_transfer_out_t out;
+
+    FUNC_ENTER(NULL);
+
+    ret_value = HG_Bulk_access(local_bulk_handle, 0, bulk_args->nbytes, HG_BULK_READWRITE, 1, &buf, NULL, NULL);
+    if (ret_value != HG_SUCCESS)
+        PGOTO_ERROR(ret_value, "Could not access bulk data @ %s:%d", __func__, __LINE__);
+
+    printf("==PDC_SERVER: process generic bulk data, size %lu\n", bulk_args->nbytes);
+    fprintf(stderr, "first 3 values: %d %d %d\n", ((int*)buf)[0], ((int*)buf)[1], ((int*)buf)[2]);
+    fprintf(stderr, "last values: %d %d %d\n", ((int*)buf)[bulk_args->nbytes/4-3], 
+            ((int*)buf)[bulk_args->nbytes/4-2], ((int*)buf)[bulk_args->nbytes/4-1]);
+
+    out.ret = 1;
+    HG_Respond(bulk_args->handle, NULL, NULL, &out);
+
+done:
+    fflush(stdout);
+    HG_Bulk_free(local_bulk_handle);
+    HG_Destroy(bulk_args->handle);
+    free(bulk_args);
+
+    FUNC_LEAVE(ret_value);
+}
+
+/* static hg_return_t */
+/* generic_bulk_c2s_transfer_cb(hg_handle_t handle) */
+HG_TEST_RPC_CB(generic_bulk_c2s_transfer, handle)
+{
+    hg_return_t ret_value = HG_SUCCESS;
+    hg_bulk_t origin_bulk_handle, local_bulk_handle;
+    const struct hg_info *hg_info = NULL;
+    struct bulk_args_t *bulk_args = NULL;
+    generic_bulk_c2s_transfer_in_t in;
+    generic_bulk_c2s_transfer_out_t out;
+    void *buf;
+
+    FUNC_ENTER(NULL);
+
+    /* Get input parameters and data */
+    ret_value = HG_Get_input(handle, &in);
+    if (ret_value != HG_SUCCESS)
+        PGOTO_ERROR(ret_value, "Could not get input @ %s:%d", __func__, __LINE__);
+
+    bulk_args = (struct bulk_args_t *)malloc(sizeof(struct bulk_args_t));
+    bulk_args->handle = handle;
+    origin_bulk_handle = in.local_bulk_handle;
+    bulk_args->nbytes = HG_Bulk_get_size(origin_bulk_handle);
+
+    buf = malloc(bulk_args->nbytes);
+
+    fprintf(stderr, "==PDC_SERVER: %s received nbytes %lu\n", __func__, bulk_args->nbytes);
+
+    hg_info = HG_Get_info(handle);
+    HG_Bulk_create(hg_info->hg_class, 1, &buf, &bulk_args->nbytes, HG_BULK_WRITE_ONLY, &local_bulk_handle);
+
+    // Data should be processed in process_generic_c2s_bulk_transfer_cb, after sending a reply to the client
+    // HG_Bulk_transfer is non-blocking
+    ret_value = HG_Bulk_transfer(hg_info->context, process_generic_c2s_bulk_transfer_cb, bulk_args, 
+                                 HG_BULK_PULL, hg_info->addr, origin_bulk_handle, 0, 
+                                 local_bulk_handle, 0, bulk_args->nbytes, HG_OP_ID_IGNORE);
+    if (ret_value != HG_SUCCESS)
+        PGOTO_ERROR(ret_value, "Could not read bulk data");
+
+    HG_Free_input(handle, &in);
+
+done:
+    fflush(stdout);
+    /* out.ret = ret_value; */
+    /* HG_Respond(handle, NULL, NULL, &out); */
+    FUNC_LEAVE(ret_value);
+}
+
 HG_TEST_THREAD_CB(server_lookup_client)
 HG_TEST_THREAD_CB(gen_obj_id)
 HG_TEST_THREAD_CB(gen_cont_id)
@@ -6594,6 +6674,7 @@ HG_TEST_THREAD_CB(send_client_storage_meta_rpc)
 HG_TEST_THREAD_CB(send_shm_bulk_rpc)
 HG_TEST_THREAD_CB(send_data_query_rpc)
 HG_TEST_THREAD_CB(send_rpc)
+HG_TEST_THREAD_CB(generic_bulk_c2s_transfer)
 
 HG_TEST_THREAD_CB(send_nhits)
 HG_TEST_THREAD_CB(send_bulk_rpc)
@@ -6639,6 +6720,7 @@ PDC_FUNC_DECLARE_REGISTER_IN_OUT(buf_map_server, buf_map_in_t, buf_map_out_t)
 PDC_FUNC_DECLARE_REGISTER_IN_OUT(buf_unmap_server, buf_unmap_in_t, buf_unmap_out_t)
 PDC_FUNC_DECLARE_REGISTER(buf_unmap)
 PDC_FUNC_DECLARE_REGISTER(region_lock)
+PDC_FUNC_DECLARE_REGISTER(generic_bulk_c2s_transfer)
 
 PDC_FUNC_DECLARE_REGISTER_IN_OUT(region_release, region_lock_in_t, region_lock_out_t)
 PDC_FUNC_DECLARE_REGISTER_IN_OUT(transform_region_release, region_transform_and_lock_in_t, region_lock_out_t)

src/server/pdc_server.c

@@ -2047,6 +2047,7 @@ PDC_Server_mercury_register()
     PDC_transform_region_release_register(hg_class_g);
     PDC_region_transform_release_register(hg_class_g);
     PDC_region_analysis_release_register(hg_class_g);
+    PDC_generic_bulk_c2s_transfer_register(hg_class_g);
 
     // DART Index
     PDC_dart_get_server_info_register(hg_class_g);

src/server/pdc_server_region/pdc_server_region_request_handler.h

@@ -22,7 +22,6 @@ transfer_request_all_bulk_transfer_read_cb2(const struct hg_cb_info *info)
     free(local_bulk_args2->transfer_request_id);
     HG_Bulk_free(local_bulk_args2->bulk_handle);
     HG_Destroy(local_bulk_args2->handle);
-    free(local_bulk_args2);
     // printf("finishing transfer_request_all_bulk_transfer_read_cb2\n");
 
 #ifdef PDC_TIMING
@@ -32,6 +31,7 @@ transfer_request_all_bulk_transfer_read_cb2(const struct hg_cb_info *info)
     pdc_server_timings->PDCreg_transfer_request_inner_read_all_bulk_rpc += end - start;
     pdc_timestamp_register(pdc_transfer_request_inner_read_all_bulk_timestamps, start, end);
 #endif
+    free(local_bulk_args2);
 
     FUNC_LEAVE(ret);
 }
@@ -331,15 +331,15 @@ transfer_request_wait_all_bulk_transfer_cb(const struct hg_cb_info *info)
 
     HG_Bulk_free(local_bulk_args->bulk_handle);
 
-    free(local_bulk_args);
-
 #ifdef PDC_TIMING
     double end = MPI_Wtime();
 
     pdc_server_timings->PDCreg_transfer_request_wait_all_rpc += end - local_bulk_args->start_time;
     pdc_timestamp_register(pdc_transfer_request_wait_all_timestamps, local_bulk_args->start_time, end);
 #endif
 
+    free(local_bulk_args);
+
     FUNC_LEAVE(ret);
 }
 

src/tests/CMakeLists.txt

@@ -43,6 +43,7 @@ include_directories(${TEST_EXT_INCLUDE_DIRS})
 
 set(PROGRAMS
   pdc_init
+  pdc_generic_bulk
 #  create_prop
 #  create_region
 #  create_obj_coll