Merge pull request #597 from E3SM-Project/dqwu/adios_decomp_enhancement

Storing large decomposition maps in ADIOS BP files can
significantly increase disk usage. This PR introduces
a build-time configuration option for the ADIOS type,
allowing users to conserve disk space by excluding
decomposition data from BP files.

Also see PR #589 (on maint branch)

Author: dqwu

CMakeLists.txt

@@ -53,6 +53,7 @@ option (WITH_NETCDF          "Require the use of NetCDF"                    ON)
 option (WITH_ADIOS2          "Require the use of ADIOS 2.x"                 OFF)
 option (WITH_HDF5            "Require the use of HDF5"                      OFF)
 option (ADIOS_BP2NC_TEST     "Enable testing of BP to NetCDF conversion"    OFF)
+option (ADIOS_NO_DECOMPS     "Save no decomposition data to ADIOS BP files" OFF)
 #===== Testing Options =====
 option (PIO_ENABLE_TESTS  "Enable the testing builds"                       OFF)
 option (PIO_ENABLE_LARGE_TESTS  "Enable large (file, processes) tests"      OFF)

ctest/CTestEnvironment-anlgce.cmake

@@ -86,6 +86,11 @@ if (DEFINED ENV{ENABLE_ADIOS_BP2NC_TEST})
     set (CTEST_CONFIGURE_OPTIONS "${CTEST_CONFIGURE_OPTIONS} -DADIOS_BP2NC_TEST=ON")
 endif ()
 
+# If ENABLE_ADIOS_NO_DECOMPS environment variable is set, then save no decomposition data to ADIOS BP files
+if (DEFINED ENV{ENABLE_ADIOS_NO_DECOMPS})
+    set (CTEST_CONFIGURE_OPTIONS "${CTEST_CONFIGURE_OPTIONS} -DADIOS_NO_DECOMPS=ON")
+endif ()
+
 # If ENABLE_TESTS environment variable is set, then enable the testing builds
 if (DEFINED ENV{ENABLE_TESTS})
     set (CTEST_CONFIGURE_OPTIONS "${CTEST_CONFIGURE_OPTIONS} -DPIO_ENABLE_TESTS=ON")

ctest/CTestScript-Test.cmake

@@ -32,7 +32,7 @@ ctest_test(INCLUDE "pio_unit_test|^init|pio_file\
 elseif (DEFINED ENV{ADIOS_READ_CTEST})
 ctest_test(INCLUDE "pio_unit_test|^init|pio_file\
 |ncdf_simple_tests\
-|pio_decomp_tests_|pio_decomp_frame_tests_|pio_decomp_extra_dims_\
+|pio_decomp_tests_np|pio_decomp_extra_dims_\
 |pio_iosystem_tests|examplePio|example1|darray_no_async|test_adios")
 elseif (DEFINED ENV{HDF5_CTEST})
 ctest_test(INCLUDE "test_hdf5")

src/clib/CMakeLists.txt

@@ -317,6 +317,12 @@ if (ADIOS_BP2NC_TEST)
     PUBLIC _ADIOS_BP2NC_TEST)
 endif ()
 
+if (ADIOS_NO_DECOMPS)
+  message(STATUS "No decomposition data will be saved to ADIOS BP files to reduce disk usage")
+  target_compile_definitions (pioc
+    PUBLIC _SPIO_ADIOS_NO_DECOMPS)
+endif ()
+
 if (PIO_ENABLE_API_TRACING)
   message(STATUS "Enabling API tracing")
   set(ENABLE_API_TRACING 1)

src/clib/pio.h

@@ -1084,6 +1084,10 @@ typedef struct file_desc_t
     struct spio_hmap *cache_darray_info;
 
     char io_name_reader[PIO_MAX_NAME + 1]; /* Name of io object, for ADIOS read */
+    size_t adios_reader_num_decomp_blocks; /* Number of decomposition blocks, for ADIOS read */
+
+    /* Indicates whether the decomposition maps (for ADIOS write) need to be stored in BP files. Default is true. */
+    bool store_adios_decomp;
 #endif /* _ADIOS2 */
 
 #ifdef _HDF5

src/clib/pio_darray.cpp

@@ -842,6 +842,11 @@ static int MPI_BigAdios_Gatherv(const void *sendbuf, int sendcount, MPI_Datatype
 static int needs_to_write_decomp(file_desc_t *file, int ioid)
 {
     assert(file != NULL);
+
+    /* No need to write the decomposition maps to ADIOS BP files. */
+    if (!file->store_adios_decomp)
+        return 0;
+
     int ret = 1; // Yes
     for (int i = 0; i < file->n_written_ioids; i++)
     {
@@ -2756,6 +2761,8 @@ static int PIOc_read_darray_adios(file_desc_t *file, int fndims, io_desc_t *iode
     decomp_info_buff = (const int *) file->cache_darray_info->get(file->cache_darray_info, decomp_name);
     if (decomp_info_buff == NULL)
     {
+      if (file->store_adios_decomp)
+      {
         /* We should search decomposition array from the step 0 if the decomp info is not in the cache, so we close and open the bp file */
         /* Should not reopen opened file at step 0 */
         if ((file->engineH != NULL && current_adios_step != 0) ||
@@ -2871,7 +2878,8 @@ static int PIOc_read_darray_adios(file_desc_t *file, int fndims, io_desc_t *iode
                            pio_get_vname_from_file(file, varid), varid, pio_get_fname_from_file(file), file->pio_ncid);
         }
 
-        int32_t decomp_blocks_size = decomp_blocks->nblocks;
+        size_t decomp_blocks_size = decomp_blocks->nblocks;
+        assert(decomp_blocks_size == file->adios_reader_num_decomp_blocks);
 
         /* Free memory */
         for (size_t i = 0; i < decomp_blocks->nblocks; i++)
@@ -3044,6 +3052,55 @@ static int PIOc_read_darray_adios(file_desc_t *file, int fndims, io_desc_t *iode
                            "Cannot determine block for the decomposition map (ADIOS read and write must use the same decomposition map)",
                            pio_get_vname_from_file(file, varid), varid, pio_get_fname_from_file(file), file->pio_ncid);
         }
+      } /* End if (file->store_adios_decomp) */
+      else
+      {
+          /* In case the write decomposition map is not available for verification,
+           * we simply use the anticipated block and start/end indices to read data. */
+          iosystem_desc_t *ios = file->iosystem; /* Pointer to io system information. */
+          assert(ios != NULL);
+
+          /* Determine the maximum number of compute tasks assigned to each block. */
+          size_t decomp_blocks_size = file->adios_reader_num_decomp_blocks;
+          assert(decomp_blocks_size > 0);
+          int max_comptasks_per_block = ios->num_comptasks / decomp_blocks_size;
+          if (ios->num_comptasks % decomp_blocks_size != 0)
+              max_comptasks_per_block++;
+          assert(max_comptasks_per_block >= 1);
+
+          /* Calculate the most probable target block we try to find. */
+          int guess_block = ios->comp_rank / max_comptasks_per_block;
+          assert(guess_block >= 0 && guess_block < decomp_blocks_size);
+
+          target_block = guess_block;
+
+          /* Store the length of the decomposition map on each computation task. */
+          int task_maplen[ios->num_comptasks];
+
+          /* Gather map lengths from all computation tasks and fill the task_maplen array on all tasks. */
+          int mpierr = MPI_SUCCESS;
+          if ((mpierr = MPI_Allgather(&iodesc->maplen, 1, MPI_INT, task_maplen, 1, MPI_INT, ios->comp_comm)))
+              return check_mpi(ios, NULL, mpierr, __FILE__, __LINE__);
+          assert(task_maplen[file->all_rank] == iodesc->maplen);
+
+          start_idx_in_target_block = 0;
+          for (int proc = 0; proc < ios->num_comptasks; proc++)
+          {
+              int proc_block = proc / max_comptasks_per_block;
+              if (proc_block == target_block && proc < file->all_rank)
+              {
+                  /* Handle the special case where the map length is 0 or 1.
+                   * The ADIOS writer always ensures that the map length is
+                   * at least 2 by padding if the original length is 0 or 1. */
+                  if (task_maplen[proc] > 1)
+                      start_idx_in_target_block += task_maplen[proc];
+                  else
+                      start_idx_in_target_block += 2;
+              }
+          }
+
+          end_idx_in_target_block = start_idx_in_target_block + iodesc->maplen - 1;
+      }
 
         /* Add to cache */
         int *buff = (int*)calloc(3, sizeof(int));

src/clib/pioc_support.cpp

@@ -2961,6 +2961,14 @@ int spio_createfile_int(int iosysid, int *ncidp, const int *iotype, const char *
         file->num_adiostasks = ios->num_adiostasks;
         file->adios_io_process = ios->adios_io_process;
 
+        file->adios_reader_num_decomp_blocks = 0;
+
+#ifdef _SPIO_ADIOS_NO_DECOMPS
+        file->store_adios_decomp = false;
+#else
+        file->store_adios_decomp = true;
+#endif
+
         /* Create a new ADIOS group */
         char declare_name[PIO_MAX_NAME];
         snprintf(declare_name, PIO_MAX_NAME, "%s%lu", file->filename, get_adios2_io_cnt());
@@ -3073,7 +3081,35 @@ int spio_createfile_int(int iosysid, int *ncidp, const int *iotype, const char *
                                    "Putting (ADIOS) variable (name=/__pio__/info/nproc) failed (adios2_error=%s) for file (%s)",
                                    convert_adios2_error_to_string(adiosErr), pio_get_fname_from_file(file));
                 }
-                (file->num_written_blocks)++;
+
+                variableH = adios2_inquire_variable(file->ioH, "/__pio__/info/decomp_stored");
+                if (variableH == NULL)
+                {
+                    variableH = adios2_define_variable(file->ioH,
+                                                       "/__pio__/info/decomp_stored", adios2_type_int32_t,
+                                                       0, NULL, NULL, NULL,
+                                                       adios2_constant_dims_true);
+                    if (variableH == NULL)
+                    {
+                        spio_ltimer_stop(file->io_fstats->wr_timer_name);
+                        spio_ltimer_stop(file->io_fstats->tot_timer_name);
+                        return pio_err(ios, NULL, PIO_EADIOS2ERR, __FILE__, __LINE__,
+                                       "Defining (ADIOS) variable (name=/__pio__/info/decomp_stored) failed for file (%s)",
+                                       pio_get_fname_from_file(file));
+                    }
+                }
+
+                int info_decomp_stored = (file->store_adios_decomp)? 1 : 0;
+                adiosErr = adios2_put(file->engineH, variableH, &info_decomp_stored, adios2_mode_sync);
+                if (adiosErr != adios2_error_none)
+                {
+                    spio_ltimer_stop(file->io_fstats->wr_timer_name);
+                    spio_ltimer_stop(file->io_fstats->tot_timer_name);
+                    return pio_err(ios, NULL, PIO_EADIOS2ERR, __FILE__, __LINE__,
+                                   "Putting (ADIOS) variable (name=/__pio__/info/decomp_stored) failed (adios2_error=%s) for file (%s)",
+                                   convert_adios2_error_to_string(adiosErr), pio_get_fname_from_file(file));
+                }
+                file->num_written_blocks += 2;
             }
 
             /* Write the number and list of processes in block merges */
@@ -3537,6 +3573,8 @@ static int adios_get_step_info(file_desc_t *file, int, size_t, size_t);
 static int adios_get_dim_ids(file_desc_t *file, int);
 static int adios_get_nc_op_tag(file_desc_t *file, int);
 static int adios_get_attr(file_desc_t *file, int current_var_cnt, char *const *attr_names, size_t);
+static int adios_get_num_decomp_blocks(file_desc_t *file);
+static int adios_get_decomp_storage_info(file_desc_t *file);
 static size_t adios_read_vars_vars(file_desc_t *file, size_t var_size, char *const *var_names);
 static size_t adios_read_vars_attrs(file_desc_t *file, size_t attr_size, char *const *attr_names);
 
@@ -4113,6 +4151,62 @@ static int adios_get_step_info(file_desc_t *file, int varid, size_t adios_step,
     return 0;
 }
 
+static int adios_get_num_decomp_blocks(file_desc_t *file)
+{
+    adios2_variable *variableH = adios2_inquire_variable(file->ioH, "/__pio__/info/block_list");
+    if (variableH == NULL)
+    {
+        return pio_err(NULL, file, PIO_EADIOS2ERR, __FILE__, __LINE__,
+                       "Getting number of decomposition blocks in file (%s, ncid=%d) using ADIOS iotype failed. "
+                       "/__pio__/info/block_list is missing",
+                       pio_get_fname_from_file(file), file->pio_ncid);
+    }
+
+    adios2_varinfo *info_block_list = adios2_inquire_blockinfo(file->engineH, variableH, 0);
+    if (info_block_list == NULL)
+    {
+        return pio_err(NULL, file, PIO_EADIOS2ERR, __FILE__, __LINE__,
+                       "Getting number of decomposition blocks (%s, ncid=%d) using ADIOS iotype failed. "
+                       "The low level (ADIOS) I/O library call failed to get the list of blocks for a variable in a given step (NULL pointer returned)",
+                       pio_get_fname_from_file(file), file->pio_ncid);
+    }
+
+    file->adios_reader_num_decomp_blocks = info_block_list->nblocks;
+
+    /* Free adios2_varinfo structure (this ADIOS2 API returns void) */
+    adios2_free_blockinfo(info_block_list);
+
+    return PIO_NOERR;
+}
+
+static int adios_get_decomp_storage_info(file_desc_t *file)
+{
+    adios2_variable *variableH = adios2_inquire_variable(file->ioH, "/__pio__/info/decomp_stored");
+
+    /* Maintains backward compatibility with earlier generated BP files where decomposition storage
+     * information is absent. */
+    if (variableH == NULL)
+    {
+        file->store_adios_decomp = true;
+        return PIO_NOERR;
+    }
+
+    int info_decomp_stored = -1;
+    adios2_error adiosErr = adios2_get(file->engineH, variableH, &info_decomp_stored, adios2_mode_sync);
+    if (adiosErr != adios2_error_none)
+    {
+        return pio_err(NULL, file, PIO_EADIOS2ERR, __FILE__, __LINE__,
+                       "Getting decomposition storage info in file (%s, ncid=%d) using ADIOS iotype failed. "
+                       "The low level (ADIOS) I/O library call failed to get data associated with a variable from an engine (adios2_error=%s)",
+                       pio_get_fname_from_file(file), file->pio_ncid, convert_adios2_error_to_string(adiosErr));
+    }
+    assert(info_decomp_stored == 0 || info_decomp_stored == 1);
+
+    file->store_adios_decomp = (info_decomp_stored == 0)? false : true;
+
+    return PIO_NOERR;
+}
+
 static size_t adios_read_vars_attrs(file_desc_t *file, size_t attr_size, char *const *attr_names)
 {
     size_t current_var_cnt = 0;
@@ -4604,6 +4698,10 @@ int PIOc_openfile_retry_impl(int iosysid, int *ncidp, int *iotype, const char *f
         file->cache_block_sizes = spio_hash(10000);
         file->cache_darray_info = spio_hash(10000);
 
+        file->adios_reader_num_decomp_blocks = 0;
+
+        file->store_adios_decomp = true;
+
         while (step < nsteps && adios2_begin_step(file->engineH, adios2_step_mode_read, -1.0, &status) == adios2_error_none)
         {
             file->begin_step_called = 1;
@@ -4669,6 +4767,12 @@ int PIOc_openfile_retry_impl(int iosysid, int *ncidp, int *iotype, const char *f
                 adios_get_step_info(file, var_id, step, nsteps);
             }
 
+            if (step == 0)
+            {
+                adios_get_num_decomp_blocks(file);
+                adios_get_decomp_storage_info(file);
+            }
+
             adiosErr = adios2_end_step(file->engineH);
             if (adiosErr != adios2_error_none)
             {

tools/adios2pio-nm/adios2pio-nm-lib.cxx

@@ -2184,6 +2184,7 @@ int ConvertBPFile(const string &infilepath, const string &outfilename,
     string err_msg = "No errors";
     int ncid = -1;
     int n_bp_writers;
+    int decomp_stored_flag = -1;
     int ret = PIO_NOERR;
     int iosysid;
 
@@ -2252,6 +2253,26 @@ int ConvertBPFile(const string &infilepath, const string &outfilename,
             return BP2PIO_ERROR;
         }
 
+        /* Get the value of /__pio__/info/decomp_stored flag */
+        adios2::Variable<int> bpDecompStored = bpIO[0].InquireVariable<int>("/__pio__/info/decomp_stored");
+        if (bpDecompStored)
+        {
+            bpReader[0].Get(bpDecompStored, &decomp_stored_flag, adios2::Mode::Sync);
+            assert(decomp_stored_flag == 0 || decomp_stored_flag == 1);
+        }
+        else
+        {
+            /* Maintains backward compatibility with earlier generated BP files where decomposition storage
+             * information is absent. */
+            decomp_stored_flag = 1;
+        }
+
+        if (decomp_stored_flag == 0)
+        {
+            if (w_mpirank == 0)
+                fprintf(stdout, "WARNING: Skipping conversion of distributed array variables due to missing decomposition data for BP file : \"%s\"\n", infilepath.c_str());
+        }
+
         uint64_t time_step = 0;
         std::vector<int> block_procs;
         adios2::Variable<int> blockProcs = bpIO[0].InquireVariable<int>("/__pio__/info/block_nprocs");
@@ -2408,6 +2429,9 @@ int ConvertBPFile(const string &infilepath, const string &outfilename,
                     }
                     else if (var.op == "darray")
                     {
+                      /* Convert distributed array variables only when decomposition maps are available. */
+                      if (decomp_stored_flag == 1)
+                      {
                         /* Variable was written with pio_write_darray() with a decomposition */
                         if (debug_out)
                         {
@@ -2418,6 +2442,7 @@ int ConvertBPFile(const string &infilepath, const string &outfilename,
                                                      iosysid, file0, adios_new, time_step,
                                                      comm, mpirank, nproc, block_procs, local_proc_blocks,
                                                      block_list, processed_attrs, decomp_cache);
+                      }
                     }
                 }