Region transfer all optimization, update vpicio_mts and bdcats_mts, enable IDIOMS in cmake, reduce printfs

.github/workflows/ubuntu-cache-idioms.yml

@@ -0,0 +1,31 @@
+name: Ubuntu (cache,IDIOMS)
+
+on:
+  pull_request:
+    branches: [ stable, develop ]
+
+  push:
+    branches: [ stable, develop ]
+
+  workflow_dispatch:
+
+jobs:
+  PDC:
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Dependencies
+        run: .github/workflows/dependencies-linux.sh
+
+      - name: Build PDC with IDIOMS
+        run: |
+          mkdir build && cd build
+          cmake ../ -DBUILD_MPI_TESTING=ON -DBUILD_SHARED_LIBS=ON -DPDC_ENABLE_IDIOMS=ON -DPDC_SERVER_CACHE=ON -DBUILD_TESTING=ON -DPDC_ENABLE_MPI=ON -DCMAKE_C_COMPILER=mpicc -DCMAKE_POLICY_VERSION_MINIMUM=3.5
+          make -j2
+
+      - name: Test PDC
+        working-directory: build
+        run: ctest -L serial --output-on-failure

CMakeLists.txt

@@ -328,12 +328,18 @@ if(PDC_USE_CRAY_DRC)
 endif()
 
 #-----------------------------------------------------------------------------
-# DART Suffix Tree mode
+# IDIOMS Index
 #-----------------------------------------------------------------------------
-option(PDC_DART_SUFFIX_TREE_MODE "Enable DART Suffix Tree mode." ON)
-if(PDC_DART_SUFFIX_TREE_MODE)
-  set(PDC_DART_SFX_TREE 1)
-  # add_compile_definitions(PDC_DART_SFX_TREE=${PDC_DART_SFX_TREE})
+option(PDC_ENABLE_IDIOMS "Enable IDIOMS metadata." OFF)
+if(PDC_ENABLE_IDIOMS)
+    set(PDC_ENABLE_IDIOMS 1)
+    option(PDC_DART_SUFFIX_TREE_MODE "Enable DART Suffix Tree mode." ON)
+    if(PDC_DART_SUFFIX_TREE_MODE)
+      set(PDC_DART_SFX_TREE 1)
+      # add_compile_definitions(PDC_DART_SFX_TREE=${PDC_DART_SFX_TREE})
+endif()
+
+
 endif()
 
 #-----------------------------------------------------------------------------

src/api/pdc_client_connect.c

@@ -1255,17 +1255,19 @@ PDC_Client_mercury_init(hg_class_t **hg_class, hg_context_t **hg_context, int po
 #endif
 
     if ((hg_transport = getenv("HG_TRANSPORT")) == NULL) {
-        LOG_INFO("Environment variable HG_TRANSPORT was NOT set\n");
         hg_transport = default_hg_transport;
+        if (pdc_client_mpi_rank_g == 0)
+            LOG_INFO("Environment variable HG_TRANSPORT was NOT set, default to %s\n", default_hg_transport);
     }
     else
         LOG_INFO("Environment variable HG_TRANSPORT was set\n");
     if ((hostname = getenv("HG_HOST")) == NULL) {
-        LOG_INFO("Environment variable HG_HOST was NOT set\n");
         hostname = PDC_malloc(HOSTNAME_LEN);
         memset(hostname, 0, HOSTNAME_LEN);
         gethostname(hostname, HOSTNAME_LEN - 1);
         free_hostname = true;
+        if (pdc_client_mpi_rank_g == 0)
+            LOG_INFO("Environment variable HG_HOST was NOT set, default to %s\n", hostname);
     }
     else
         LOG_INFO("Environment variable HG_HOST was set\n");

src/api/pdc_region/pdc_region_transfer.c

@@ -855,51 +855,49 @@ register_metadata(pdc_transfer_request_start_all_pkg **transfer_request_input, i
     remain_size = input_size - size;
     output_size = 0;
 
-    index = 0;
     qsort(transfer_requests, size, sizeof(pdc_transfer_request_start_all_pkg *),
           sort_by_metadata_server_start_all);
-    for (i = 1; i < size; ++i) {
-        if (transfer_requests[i]->transfer_request->metadata_server_id !=
-            transfer_requests[i - 1]->transfer_request->metadata_server_id) {
-            n_objs = i - index;
-            pack_region_metadata_query(transfer_requests + index, n_objs, &buf, &total_buf_size);
-            PDC_Client_transfer_request_metadata_query(
-                &bulk_handle, buf, total_buf_size, n_objs,
-                transfer_requests[index]->transfer_request->metadata_server_id, is_write, &output_buf_size,
-                &query_id);
-            PDCregion_transfer_add_bulk_handle(transfer_requests[index]->transfer_request, bulk_handle);
-            buf = (char *)PDC_free(buf);
-            if (query_id) {
-                output_buf = (char *)PDC_malloc(output_buf_size);
-                PDC_Client_transfer_request_metadata_query2(
-                    &bulk_handle, output_buf, output_buf_size, query_id,
-                    transfer_requests[index]->transfer_request->metadata_server_id);
-                PDCregion_transfer_add_bulk_handle(transfer_requests[index]->transfer_request, bulk_handle);
-                unpack_region_metadata_query(output_buf, transfer_requests + index, &transfer_request_head,
-                                             &transfer_request_end, &output_size);
-                output_buf = (char *)PDC_free(output_buf);
-                if (transfer_request_front_head) {
-                    previous->next = transfer_request_head;
-                }
-                else {
-                    transfer_request_front_head = transfer_request_head;
-                }
-                previous = transfer_request_end;
-            }
-            index = i;
+
+    // Each iteration finds the first transfer that has a target meta server different from the previous one
+    // index is the first transfer index
+    int  current_unique_idx     = 0;
+    int *unique_server_xfer_idx = NULL;
+    int *unique_server_nboj     = NULL;
+    if (size > 0) {
+        unique_server_xfer_idx = (int *)PDC_calloc(size, sizeof(int));
+        unique_server_nboj     = (int *)PDC_calloc(size, sizeof(int));
+    }
+
+    // Iterate through the input array
+    for (i = 0; i < size; ++i) {
+        if (i == 0 || transfer_requests[i]->transfer_request->metadata_server_id !=
+                          transfer_requests[i - 1]->transfer_request->metadata_server_id) {
+            // Check if the current element is different from the previous one
+            // or if it's the first element
+            unique_server_xfer_idx[current_unique_idx] = i;
+            unique_server_nboj[current_unique_idx]     = 1;
+
+            current_unique_idx++;
+        }
+        else {
+            unique_server_nboj[current_unique_idx - 1]++;
         }
     }
+    int num_unique_server_ids = current_unique_idx;
+
+    // Now we will try to distribute the metadata requests to different servers across clients
+    for (i = 0; i < num_unique_server_ids; i++) {
+        int current_index = (pdc_client_mpi_rank_g + i) % num_unique_server_ids;
+        index             = unique_server_xfer_idx[current_index];
+        n_objs            = unique_server_nboj[current_index];
 
-    if (size) {
-        n_objs = size - index;
         pack_region_metadata_query(transfer_requests + index, n_objs, &buf, &total_buf_size);
         PDC_Client_transfer_request_metadata_query(
             &bulk_handle, buf, total_buf_size, n_objs,
             transfer_requests[index]->transfer_request->metadata_server_id, is_write, &output_buf_size,
             &query_id);
         PDCregion_transfer_add_bulk_handle(transfer_requests[index]->transfer_request, bulk_handle);
         buf = (char *)PDC_free(buf);
-        // If it is a valid query ID, then it means regions are overlapping.
         if (query_id) {
             output_buf = (char *)PDC_malloc(output_buf_size);
             PDC_Client_transfer_request_metadata_query2(
@@ -909,16 +907,21 @@ register_metadata(pdc_transfer_request_start_all_pkg **transfer_request_input, i
             unpack_region_metadata_query(output_buf, transfer_requests + index, &transfer_request_head,
                                          &transfer_request_end, &output_size);
             output_buf = (char *)PDC_free(output_buf);
-            if (transfer_request_front_head) {
+
+            if (transfer_request_front_head)
                 previous->next = transfer_request_head;
-            }
-            else {
+            else
                 transfer_request_front_head = transfer_request_head;
-            }
+
             previous = transfer_request_end;
         }
     }
 
+    if (unique_server_xfer_idx)
+        free(unique_server_xfer_idx);
+    if (unique_server_nboj)
+        free(unique_server_nboj);
+
     if (output_size) {
         transfer_request_output = (pdc_transfer_request_start_all_pkg **)PDC_malloc(
             sizeof(pdc_transfer_request_start_all_pkg *) * (output_size + remain_size));

src/server/CMakeLists.txt

@@ -23,6 +23,9 @@ if(PDC_ENABLE_ZFP)
     # find_path(ZFP_INCLUDE_DIR include/zfp.h)
 endif()
 
+if(PDC_ENABLE_IDIOMS)
+    add_definitions(-DPDC_ENABLE_IDIOMS=1)
+endif()
 
 include_directories(
   ${PDC_COMMON_INCLUDE_DIRS}

src/server/pdc_server.c

@@ -774,17 +774,19 @@ PDC_Server_init(int port, hg_class_t **hg_class, hg_context_t **hg_context)
     total_mem_usage_g += (sizeof(char) + sizeof(char *));
 
     if ((hg_transport = getenv("HG_TRANSPORT")) == NULL) {
-        LOG_INFO("Environment variable HG_TRANSPORT was NOT set\n");
         hg_transport = default_hg_transport;
+        if (pdc_server_rank_g == 0)
+            LOG_INFO("Environment variable HG_TRANSPORT was NOT set, default to %s\n", hg_transport);
     }
     else
         LOG_INFO("Environment variable HG_TRANSPORT was set\n");
     if ((hostname = getenv("HG_HOST")) == NULL) {
-        LOG_INFO("Environment variable HG_HOST was NOT set\n");
         hostname = PDC_malloc(HOSTNAME_LEN);
         memset(hostname, 0, HOSTNAME_LEN);
         gethostname(hostname, HOSTNAME_LEN - 1);
         free_hostname = true;
+        if (pdc_server_rank_g == 0)
+            LOG_INFO("Environment variable HG_HOST was NOT set, default to %s\n", hostname);
     }
     else
         LOG_INFO("Environment variable HG_HOST was set\n");
@@ -922,8 +924,12 @@ PDC_Server_init(int port, hg_class_t **hg_class, hg_context_t **hg_context)
         LOG_INFO("Read cache enabled\n");
 #endif
 
+#ifdef PDC_ENABLE_IDIOMS
     // Initialize IDIOMS
+    if (pdc_server_rank_g == 0)
+        LOG_INFO("IDIOMS index enabled\n");
     PDC_Server_metadata_index_init(pdc_server_size_g, pdc_server_rank_g);
+#endif
 
     // TODO: support restart with different number of servers than previous run
     char checkpoint_file[ADDR_MAX + sizeof(int) + 1];
@@ -934,7 +940,9 @@ PDC_Server_init(int port, hg_class_t **hg_class, hg_context_t **hg_context)
         ret_value = PDC_Server_restart(checkpoint_file);
         if (ret_value != SUCCEED)
             PGOTO_ERROR(FAIL, "Error with PDC_Server_restart");
+#ifdef PDC_ENABLE_IDIOMS
         metadata_index_recover(pdc_server_tmp_dir_g, pdc_server_size_g, pdc_server_rank_g);
+#endif
     }
     else {
         // We are starting a brand new server
@@ -1364,7 +1372,9 @@ PDC_Server_checkpoint()
         LOG_INFO("Checkpointed %10d objects, with %10d regions \n", all_metadata_size, all_region_count);
     }
 
+#ifdef PDC_ENABLE_IDIOMS
     metadata_index_dump(pdc_server_tmp_dir_g, pdc_server_rank_g);
+#endif
 
 done:
     FUNC_LEAVE(ret_value);
@@ -1417,11 +1427,6 @@ PDC_Server_restart(char *filename)
     if (file == NULL)
         PGOTO_ERROR(FAIL, "Error with fopen, filename: [%s]", filename);
 
-    char *slurm_jobid = getenv("SLURM_JOB_ID");
-    if (slurm_jobid == NULL) {
-        LOG_ERROR("Error getting slurm job id from SLURM_JOB_ID\n");
-    }
-
     if (fread(&n_cont, sizeof(int), 1, file) != 1) {
         LOG_ERROR("Read failed for n_count\n");
     }
@@ -1971,8 +1976,10 @@ PDC_Server_mercury_register()
     PDC_region_analysis_release_register(hg_class_g);
 
     // DART Index
+#ifdef PDC_ENABLE_IDIOMS
     PDC_dart_get_server_info_register(hg_class_g);
     PDC_dart_perform_one_server_register(hg_class_g);
+#endif
 
     // Server to client RPC
     server_lookup_client_register_id_g = PDC_server_lookup_client_register(hg_class_g);

src/tests/CMakeLists.txt

@@ -80,6 +80,7 @@ set(PROGRAMS
   misc/vpicio
   misc/vpicio_mts
   misc/bdcats
+  misc/bdcats_mts
   deprecated/vpicio_old
   deprecated/bdcats_old
   tags/kvtag_add_get
@@ -139,9 +140,14 @@ if(BUILD_MPI_TESTING)
     tags/kvtag_affix_query_scale
     tags/kvtag_add_get_benchmark
     tags/kvtag_add_get_scale
+  )
+
+if(PDC_ENABLE_IDIOMS)
+  list(APPEND MPI_PROGRAMS
     misc/llsm_idioms_bench
     misc/idioms_ci_test
   )
+endif()
 
   foreach(program ${MPI_PROGRAMS})
     get_filename_component(program_name ${program} NAME)
@@ -261,7 +267,9 @@ add_test(NAME obj_buf           WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTO
 add_test(NAME obj_tags          WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./obj_tags )
 add_test(NAME kvtag_add_get     WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./kvtag_add_get)
 add_test(NAME kvtag_query     WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./kvtag_query 100 1 10 0)
+if(PDC_ENABLE_IDIOMS)
 add_test(NAME idioms_ci_test     WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./idioms_ci_test)
+endif()
 add_test(NAME obj_info          WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./obj_info )
 add_test(NAME obj_put_data      WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./obj_put_data )
 add_test(NAME obj_get_data      WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY} COMMAND run_test.sh ./obj_get_data )
@@ -332,7 +340,9 @@ set_tests_properties(obj_buf            PROPERTIES LABELS serial)
 set_tests_properties(obj_tags           PROPERTIES LABELS serial)
 set_tests_properties(kvtag_add_get      PROPERTIES LABELS serial)
 set_tests_properties(kvtag_query        PROPERTIES LABELS serial)
+if(PDC_ENABLE_IDIOMS)
 set_tests_properties(idioms_ci_test     PROPERTIES LABELS serial)
+endif()
 set_tests_properties(obj_info           PROPERTIES LABELS serial)
 set_tests_properties(obj_put_data       PROPERTIES LABELS serial)
 set_tests_properties(obj_get_data       PROPERTIES LABELS serial)