Implement selection vector I/O with collective chunk filling (#3826)

* Changes for ECP-344: Implement selection vector I/O with collective chunk filling.
Also fix a bug in H5FD__mpio_write_vector() to account for fixed size optimization
when computing max address.

* Fixes based on PR review comments:
For H5Dchunk.c: fix H5MM_xfree()
For H5FDmpio.c:
1) Revert the fix to H5FD__mpio_write_vector()
2) Apply the patch from Neil on the proper length of s_sizes reported by H5FD__mpio_vector_build_types()

* Put back the logic of dividing up the work among all the mpi ranks similar to the
original H5D__chunk_collective_fill() routine.

* Add a test to verify the fix for the illegal reference problem in H5FD__mpio_write_vector().

2 files changed, 312 insertions, 19 deletions

diff --git a/testpar/t_filters_parallel.c b/testpar/t_filters_parallel.c
index 6c05408..f60431b 100644
--- a/testpar/t_filters_parallel.c
+++ b/testpar/t_filters_parallel.c
@@ -547,8 +547,9 @@ verify_chunk_opt_status(size_t num_dsets, test_mode_t test_mode, bool any_io, bo
 
         /* Verify selection I/O mode on rank 0 */
         if (mpi_rank == 0) {
-            /* No actual I/O performed, only reported I/O will be from allocation, even if "no" datasets were
-             * involved (num_dsets == 0 implies the call was expected to fail, but it fails after allocation).
+            /* No actual I/O performed, the only reported I/O will be from allocation which is vector I/O,
+             * even if "no" datasets were involved (num_dsets == 0 implies the call was expected to fail,
+             * but it fails after allocation).
              * Also if the test mode is mixed filtered and unfiltered and the call did not fail, then there
              * will always be an I/O callback made with raw data. This is because unfiltered datasets fall
              * back to scalar I/O when mixed with filtered, and scalar I/O reports an I/O call was made even
@@ -557,9 +558,18 @@ verify_chunk_opt_status(size_t num_dsets, test_mode_t test_mode, bool any_io, bo
              * filtered dataset with no selection. Vector I/O does report an I/O call was made if passed a raw
              * data element of size 0, so this is consistent. */
             if (!any_io) {
-                if (did_alloc || (num_dsets > 0 && test_mode == USE_MULTIPLE_DATASETS_MIXED_FILTERED))
+                if (did_alloc && (num_dsets > 0 && test_mode == USE_MULTIPLE_DATASETS_MIXED_FILTERED)) {
+                    VRFY((H5D_VECTOR_IO | H5D_SCALAR_IO) == actual_sel_io_mode_reduced,
+                         "verified actual selection I/O mode was vector and scalar I/O");
+                }
+                else if (did_alloc) {
+                    VRFY(H5D_VECTOR_IO == actual_sel_io_mode_reduced,
+                         "verified actual selection I/O mode was vector I/O");
+                }
+                else if (num_dsets > 0 && test_mode == USE_MULTIPLE_DATASETS_MIXED_FILTERED) {
                     VRFY(H5D_SCALAR_IO == actual_sel_io_mode_reduced,
                          "verified actual selection I/O mode was scalar I/O");
+                }
                 else
                     VRFY(0 == actual_sel_io_mode_reduced,
                          "verified actual selection I/O mode was 0 (no I/O)");
@@ -588,19 +598,18 @@ verify_chunk_opt_status(size_t num_dsets, test_mode_t test_mode, bool any_io, bo
                 switch (test_mode) {
                     case USE_SINGLE_DATASET:
                     case USE_MULTIPLE_DATASETS:
-                        /* Collective case with only filtered datasets. If we performed allocation then there
-                         * should be scalar I/O for allocation in addition to vector I/O for the actual data.
-                         * If we're reading from an unallocated dataset then there should be no actual I/O.
-                         * Otherwise there should only be vector I/O. */
-                        if (did_alloc)
-                            VRFY((H5D_SCALAR_IO | H5D_VECTOR_IO) == actual_sel_io_mode_reduced,
-                                 "verified actual selection I/O mode was scalar and vector I/O");
-                        else if (unalloc_read)
+                        /* Collective case with only filtered datasets.
+                         * If we're reading from an unallocated dataset then there
+                         * should be no actual I/O.
+                         * Otherwise, only vector I/O is reported whether or not
+                         * allocation happened. */
+                        if (unalloc_read)
                             VRFY(0 == actual_sel_io_mode_reduced,
                                  "verified actual selection I/O mode was 0 (no I/O)");
-                        else
+                        else { /* did_alloc || !unalloc_read */
                             VRFY(H5D_VECTOR_IO == actual_sel_io_mode_reduced,
                                  "verified actual selection I/O mode was vector I/O");
+                        }
                         break;
 
                     case USE_MULTIPLE_DATASETS_MIXED_FILTERED:
diff --git a/testpar/t_vfd.c b/testpar/t_vfd.c
index ac524ac..79b7e01 100644
--- a/testpar/t_vfd.c
+++ b/testpar/t_vfd.c
@@ -40,13 +40,14 @@ const char *FILENAMES[] = {"mpio_vfd_test_file_0",      /*0*/
                            "mpio_vfd_test_file_4",      /*4*/
                            "mpio_vfd_test_file_5",      /*5*/
                            "mpio_vfd_test_file_6",      /*6*/
-                           "subfiling_vfd_test_file_0", /*7*/
-                           "subfiling_vfd_test_file_1", /*8*/
-                           "subfiling_vfd_test_file_2", /*9*/
-                           "subfiling_vfd_test_file_3", /*10*/
-                           "subfiling_vfd_test_file_4", /*11*/
-                           "subfiling_vfd_test_file_5", /*12*/
-                           "subfiling_vfd_test_file_6", /*13*/
+                           "mpio_vfd_test_file_7",      /*7*/
+                           "subfiling_vfd_test_file_0", /*8*/
+                           "subfiling_vfd_test_file_1", /*9*/
+                           "subfiling_vfd_test_file_2", /*10*/
+                           "subfiling_vfd_test_file_3", /*11*/
+                           "subfiling_vfd_test_file_4", /*12*/
+                           "subfiling_vfd_test_file_5", /*13*/
+                           "subfiling_vfd_test_file_6", /*14*/
                            NULL};
 
 /* File Test Images
@@ -100,6 +101,8 @@ static unsigned vector_write_test_6(int file_name_id, int mpi_rank, int mpi_size
                                     H5FD_mpio_collective_opt_t coll_opt_mode, const char *vfd_name);
 static unsigned vector_write_test_7(int file_name_id, int mpi_rank, int mpi_size, H5FD_mpio_xfer_t xfer_mode,
                                     H5FD_mpio_collective_opt_t coll_opt_mode, const char *vfd_name);
+static unsigned vector_write_test_8(int file_name_id, int mpi_rank, int mpi_size, H5FD_mpio_xfer_t xfer_mode,
+                                    H5FD_mpio_collective_opt_t coll_opt_mode, const char *vfd_name);
 /*
  * Tests for selection I/O:
  * They are derived from test_selection_io() in test/vfd.c and modified for parallel testing.
@@ -4159,6 +4162,280 @@ vector_write_test_7(int file_name_id, int mpi_rank, int mpi_size, H5FD_mpio_xfer
 
 } /* vector_write_test_7() */
 
+/*-------------------------------------------------------------------------
+ * Function:    vector_write_test_8()
+ *
+ * Purpose:     This test is to verify the fix for the following problem
+ *              in H5FD__mpio_write_vector when calculating max_addr:
+ *              --illegal reference occurs when referencing the s_sizes array
+ *                with <count - 1> due to <count> exceeding the length of the
+ *                size array which uses the compressed feature.
+ *
+ *              1) Open the test file with the specified VFD, and set
+ *                 the eoa.
+ *
+ *              2) Set the test file in a known state by writing zeros
+ *                 to all bytes in the test file.  Since we have already
+ *                 tested this, do this via a vector write of zero_fi_buf.
+ *
+ *              3) Barrier
+ *
+ *              4) For each rank, define base_index equal to:
+ *
+ *                     mpi_rank * INTS_PER_RANK
+ *
+ *                 and define base_addr equal to
+ *
+ *                     base_index * sizeof(int32_t).
+ *
+ *                 Setup a vector of length INTS_PER_RANK - 1.
+ *                 Set up the size array with the compressed feature:
+ *                  --The first element has size (2 * sizeof(int32_t))
+ *                  --The second and third elements are of size sizeof(int32_t)
+ *                  --The fourth element is zero.
+ *                  Set up addrs and bufs accordingly.
+ *
+ *                 Write the vector.
+ *
+ *              5) Barrier
+ *
+ *              6) On each rank, read the entire file into the read_fi_buf,
+ *                 and compare against increasing_fi_buf.
+ *                 Report failure if any differences are detected.
+ *
+ *              7) Close the test file.  On rank 0, delete the test file.
+ *
+ * Return:      false on success, true if any errors are detected.
+ *
+ *-------------------------------------------------------------------------
+ */
+static unsigned
+vector_write_test_8(int file_name_id, int mpi_rank, int mpi_size, H5FD_mpio_xfer_t xfer_mode,
+                    H5FD_mpio_collective_opt_t coll_opt_mode, const char *vfd_name)
+{
+    const char *fcn_name = "vector_write_test_8()";
+    char        test_title[120];
+    char        filename[512];
+    haddr_t     eoa;
+    haddr_t     base_addr;
+    bool        show_progress = false;
+    hid_t       fapl_id       = H5I_INVALID_HID; /* file access property list ID */
+    hid_t       dxpl_id       = H5I_INVALID_HID; /* data access property list ID */
+    H5FD_t     *lf            = NULL;            /* VFD struct ptr               */
+    int         cp            = 0;
+    int         i;
+    int         base_index;
+    uint32_t    count = 0;
+    size_t      sizes[4];
+    H5FD_mem_t  types[2];
+
+    haddr_t     *tt_addrs = NULL; /* For storing addrs */
+    const void **tt_bufs  = NULL; /* For storing buf pointers */
+
+    pass = true;
+
+    if (mpi_rank == 0) {
+
+        if (xfer_mode == H5FD_MPIO_INDEPENDENT) {
+
+            snprintf(test_title, sizeof(test_title), "parallel vector write test 8 -- %s / independent",
+                     vfd_name);
+        }
+        else if (coll_opt_mode == H5FD_MPIO_INDIVIDUAL_IO) {
+
+            snprintf(test_title, sizeof(test_title), "parallel vector write test 8 -- %s / col op / ind I/O",
+                     vfd_name);
+        }
+        else {
+
+            assert(coll_opt_mode == H5FD_MPIO_COLLECTIVE_IO);
+
+            snprintf(test_title, sizeof(test_title), "parallel vector write test 8 -- %s / col op / col I/O",
+                     vfd_name);
+        }
+
+        TESTING(test_title);
+    }
+
+    show_progress = ((show_progress) && (mpi_rank == 0));
+
+    if (show_progress)
+        fprintf(stdout, "\n%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 1) Allocate local buffers for addrs and bufs,
+          open the test file with the specified VFD, set the eoa, and setup the dxpl */
+    if (pass) {
+
+        tt_addrs = (haddr_t *)malloc((INTS_PER_RANK) * sizeof(haddr_t *));
+        tt_bufs  = (const void **)malloc((INTS_PER_RANK) * sizeof(void *));
+
+        if (tt_addrs == NULL || tt_bufs == NULL) {
+            pass         = false;
+            failure_mssg = "Can't allocate local addrs and bufs buffers.";
+        }
+
+        if (pass) {
+            eoa = (haddr_t)mpi_size * (haddr_t)INTS_PER_RANK * (haddr_t)(sizeof(int32_t));
+
+            setup_vfd_test_file(file_name_id, filename, mpi_size, xfer_mode, coll_opt_mode, vfd_name, eoa,
+                                &lf, &fapl_id, &dxpl_id);
+        }
+    }
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 2) Using rank zero, write the entire negative_fi_buf to
+     *    the file.
+     */
+    if (pass) {
+
+        size_t image_size = (size_t)mpi_size * (size_t)INTS_PER_RANK * sizeof(int32_t);
+
+        if (mpi_rank == 0) {
+
+            if (H5FDwrite(lf, H5FD_MEM_DRAW, H5P_DEFAULT, (haddr_t)0, image_size, (void *)zero_fi_buf) < 0) {
+
+                pass         = false;
+                failure_mssg = "H5FDwrite() on rank 0 failed.\n";
+            }
+        }
+    }
+
+    /* 3) Barrier */
+    MPI_Barrier(comm);
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 4) For each rank, define base_index equal to:
+     *
+     *        mpi_rank * INTS_PER_RANK
+     *
+     *    and define base_addr equal to
+     *
+     *        base_index * sizeof(int32_t).
+     *
+     *    Set up the array of sizes and types with the compressed feature
+     *    as described in the routine header description.
+     */
+    if (pass) {
+
+        base_index = (mpi_rank * INTS_PER_RANK);
+        base_addr  = (haddr_t)base_index * (haddr_t)sizeof(int32_t);
+
+        count = INTS_PER_RANK - 1;
+
+        types[0] = H5FD_MEM_DRAW;
+        types[1] = H5FD_MEM_NOLIST;
+
+        sizes[0] = 2 * sizeof(int32_t);
+        sizes[1] = sizeof(int32_t);
+        sizes[2] = sizeof(int32_t);
+        sizes[3] = 0;
+
+        tt_addrs[0] = base_addr;
+        tt_bufs[0]  = (const void *)(&(increasing_fi_buf[base_index]));
+
+        tt_addrs[0] = base_addr;
+        base_index += 2;
+        base_addr = (haddr_t)base_index * (haddr_t)sizeof(int32_t);
+
+        for (i = 1; i < (INTS_PER_RANK - 1); i++) {
+
+            tt_addrs[i] = base_addr + ((haddr_t)(i - 1) * (haddr_t)sizeof(int32_t));
+            tt_bufs[i]  = (const void *)(&(increasing_fi_buf[base_index + (i - 1)]));
+        }
+
+        if (H5FDwrite_vector(lf, dxpl_id, count, types, tt_addrs, sizes, tt_bufs) < 0) {
+
+            pass         = false;
+            failure_mssg = "H5FDwrite_vector() failed (1).\n";
+        }
+    }
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 5) Barrier */
+    MPI_Barrier(comm);
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 6) On each rank, read the entire file into the read_fi_buf,
+     *    and compare against increasing_fi_buf
+     *    Report failure if any differences are detected.
+     */
+    if (pass) {
+
+        size_t image_size = (size_t)mpi_size * (size_t)INTS_PER_RANK * sizeof(int32_t);
+
+        if (H5FDread(lf, H5FD_MEM_DRAW, H5P_DEFAULT, (haddr_t)0, image_size, (void *)read_fi_buf) < 0) {
+
+            pass         = false;
+            failure_mssg = "H5FDread() failed.\n";
+        }
+
+        for (i = 0; ((pass) && (i < mpi_size * INTS_PER_RANK)); i++) {
+
+            if (read_fi_buf[i] != increasing_fi_buf[i]) {
+
+                pass         = false;
+                failure_mssg = "unexpected data read from file (1)";
+            }
+        }
+    } /* end if */
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 7) Barrier */
+    MPI_Barrier(comm);
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* 8) Close the test file and delete it (on rank 0 only).
+     *    Close FAPL and DXPL.
+     */
+    takedown_vfd_test_file(mpi_rank, filename, &lf, &fapl_id, &dxpl_id);
+
+    /* Free the local buffers */
+    if (tt_addrs) {
+        free(tt_addrs);
+        tt_addrs = NULL;
+    }
+
+    if (tt_bufs) {
+        free(tt_bufs);
+        tt_bufs = NULL;
+    }
+
+    if (show_progress)
+        fprintf(stdout, "%s: cp = %d, pass = %d.\n", fcn_name, cp++, pass);
+
+    /* report results */
+    if (mpi_rank == 0) {
+
+        if (pass) {
+
+            PASSED();
+        }
+        else {
+
+            H5_FAILED();
+
+            if ((disp_failure_mssgs) || (show_progress)) {
+                fprintf(stdout, "%s: failure_mssg = \"%s\"\n", fcn_name, failure_mssg);
+            }
+        }
+    }
+
+    return (!pass);
+
+} /* vector_write_test_8() */
+
 static void
 test_vector_io(int mpi_rank, int mpi_size)
 {
@@ -4249,6 +4526,13 @@ test_vector_io(int mpi_rank, int mpi_size)
     nerrs +=
         vector_write_test_7(6, mpi_rank, mpi_size, H5FD_MPIO_COLLECTIVE, H5FD_MPIO_COLLECTIVE_IO, "mpio");
 
+    nerrs +=
+        vector_write_test_8(7, mpi_rank, mpi_size, H5FD_MPIO_INDEPENDENT, H5FD_MPIO_INDIVIDUAL_IO, "mpio");
+    nerrs +=
+        vector_write_test_8(7, mpi_rank, mpi_size, H5FD_MPIO_COLLECTIVE, H5FD_MPIO_INDIVIDUAL_IO, "mpio");
+    nerrs +=
+        vector_write_test_8(7, mpi_rank, mpi_size, H5FD_MPIO_COLLECTIVE, H5FD_MPIO_COLLECTIVE_IO, "mpio");
+
     MPI_Barrier(comm);
 
 #ifdef H5_HAVE_SUBFILING_VFD