Allow parallel writes to use zlib

configure.ac

@@ -1038,7 +1038,7 @@ if test "x$enable_hdf5" = xyes; then
 
    # H5Pset_fapl_mpiposix and H5Pget_fapl_mpiposix have been removed since HDF5 1.8.12.
    # Use H5Pset_fapl_mpio and H5Pget_fapl_mpio, instead.
-   AC_CHECK_FUNCS([H5Pget_fapl_mpio H5Pset_deflate H5Z_SZIP H5free_memory H5resize_memory H5allocate_memory H5Pset_libver_bounds H5Pset_all_coll_metadata_ops H5Z_SZIP])
+   AC_CHECK_FUNCS([H5Pget_fapl_mpio H5Pset_deflate H5Z_SZIP H5free_memory H5resize_memory H5allocate_memory H5Pset_libver_bounds H5Pset_all_coll_metadata_ops H5Z_SZIP H5DOread_chunk])
 
    # Check to see if HDF5 library has collective metadata APIs, (HDF5 >= 1.10.0)
    if test "x$ac_cv_func_H5Pset_all_coll_metadata_ops" = xyes; then
@@ -1053,7 +1053,14 @@ if test "x$enable_hdf5" = xyes; then
    AC_MSG_CHECKING([whether parallel io is enabled in hdf5])
    AC_MSG_RESULT([$hdf5_parallel])
 
-   # Check to see if we need to search for and link against szlib.
+   # Check to see if HDF5 library is 1.10.2 or greater. If so, allows parallel_zip.
+   if test "x$ac_cv_func_H5DOread_chunk" = xyes; then
+      AC_DEFINE([HDF5_1_10_2], [1], [if true, HDF5 is at least version 1.10.2 and allows parallel I/O with zip])
+   fi
+   AC_MSG_CHECKING([whether HDF5 is version 1.10.2 or greater])
+   AC_MSG_RESULT([$ac_cv_func_H5DOread_chunk])
+
+# Check to see if we need to search for and link against szlib.
    if test "x$ac_cv_func_H5Z_SZIP" = xyes; then
       AC_SEARCH_LIBS([SZ_BufftoBuffCompress], [szip sz], [],
       [AC_MSG_ERROR([libhdf5 installed with szip support, but cannot find or link to the szip library.])])

libdispatch/dvar.c

@@ -312,7 +312,7 @@ nc_def_var_fill(int ncid, int varid, int no_fill, const void *fill_value)
 }
 
 /**
-   Set the compression settings for a netCDF-4/HDF5 variable.
+   Set the zlib compression settings for a netCDF-4/HDF5 variable.
 
    This function must be called after nc_def_var and before nc_enddef
    or any functions which writes data to the file.
@@ -324,15 +324,18 @@ nc_def_var_fill(int ncid, int varid, int no_fill, const void *fill_value)
 
    If this function is called on a scalar variable, it is ignored.
 
+   @note Parallel I/O reads work with compressed data. Parallel I/O
+   writes work with compressed data starting in netcdf-c-4.7.4 and
+   later release, using hdf5-1.10.2 and later releases.
+
    @param ncid NetCDF or group ID, from a previous call to nc_open(),
    nc_create(), nc_def_grp(), or associated inquiry functions such as
    nc_inq_ncid().
    @param varid Variable ID
    @param shuffle True to turn on the shuffle filter. The shuffle
-   filter can assist with the compression of integer data by changing
-   the byte order in the data stream. It makes no sense to use the
-   shuffle filter without setting a deflate level, or to use shuffle
-   on non-integer data.
+   filter can assist with the compression of data by changing the byte
+   order in the data stream. It makes no sense to use the shuffle
+   filter without setting a deflate level.
    @param deflate True to turn on deflation for this variable.
    @param deflate_level A number between 0 (no compression) and 9
    (maximum compression).
@@ -347,11 +350,8 @@ nc_def_var_fill(int ncid, int varid, int no_fill, const void *fill_value)
    @return ::NC_ELATEDEF Too late to change settings for this variable.
    @return ::NC_ENOTINDEFINE Not in define mode.
    @return ::NC_EPERM File is read only.
-   @return ::NC_EMAXDIMS Classic model file exceeds ::NC_MAX_VAR_DIMS.
    @return ::NC_ESTRICTNC3 Attempting to create netCDF-4 type var in
    classic model file
-   @return ::NC_EBADTYPE Bad type.
-   @return ::NC_ENOMEM Out of memory.
    @return ::NC_EHDFERR Error returned by HDF5 layer.
    @return ::NC_EINVAL Invalid input. Deflate can't be set unless
    variable storage is NC_CHUNK.

libhdf5/hdf5var.c

@@ -658,10 +658,13 @@ nc_def_var_extra(int ncid, int varid, int *shuffle, int *deflate,
         return NC_ENOTVAR;
     assert(var && var->hdr.id == varid);
 
-    /* Can't turn on parallel and deflate/fletcher32/szip/shuffle (for now). */
+    /* Can't turn on parallel and deflate/fletcher32/szip/shuffle
+     * before HDF5 1.10.2. */
+#ifndef HDF5_1_10_2
     if (h5->parallel == NC_TRUE)
         if (deflate || fletcher32 || shuffle)
             return NC_EINVAL;
+#endif
 
     /* If the HDF5 dataset has already been created, then it is too
      * late to set all the extra stuff. */

nc_test4/CMakeLists.txt

@@ -85,6 +85,7 @@ IF(TEST_PARALLEL4)
   build_bin_test(tst_parallel3)
   build_bin_test(tst_parallel4)
   build_bin_test(tst_parallel5)
+  build_bin_test(tst_parallel_zlib)
   build_bin_test(tst_nc4perf)
   build_bin_test(tst_mode)
   build_bin_test(tst_simplerw_coll_r)

nc_test4/Makefile.am

@@ -84,7 +84,7 @@ endif # BUILD_UTILITIES
 if TEST_PARALLEL4
 check_PROGRAMS += tst_mpi_parallel tst_parallel tst_parallel3		\
 tst_parallel4 tst_parallel5 tst_nc4perf tst_mode tst_simplerw_coll_r	\
-tst_mode
+tst_mode tst_parallel_zlib tst_parallel_zlib2
 TESTS += run_par_test.sh
 endif
 

nc_test4/run_par_test.sh.in

@@ -8,8 +8,6 @@
 set -x
 set -e
 echo
-echo "Testing MPI parallel I/O with various other mode flags..."
-@MPIEXEC@ -n 1 ./tst_mode
 echo
 echo "Testing MPI parallel I/O without netCDF..."
 @MPIEXEC@ -n 4 ./tst_mpi_parallel
@@ -42,3 +40,8 @@ echo "Parallel I/O test for Collective I/O, contributed by HDF Group."
 @MPIEXEC@ -n 1 ./tst_simplerw_coll_r
 @MPIEXEC@ -n 2 ./tst_simplerw_coll_r
 @MPIEXEC@ -n 4 ./tst_simplerw_coll_r
+
+echo
+echo "Parallel I/O test with zlib."
+@MPIEXEC@ -n 1 ./tst_parallel_zlib
+@MPIEXEC@ -n 4 ./tst_parallel_zlib

nc_test4/tst_mode.c

@@ -1,16 +1,16 @@
 /*! \file
 
-Copyright 1993, 1994, 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002,
-2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014,
-2015, 2016, 2017, 2018
-University Corporation for Atmospheric Research/Unidata.
+  Copyright 1993, 1994, 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002,
+  2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014,
+  2015, 2016, 2017, 2018
+  University Corporation for Atmospheric Research/Unidata.
 
-See \ref copyright file for more info.
+  See \ref copyright file for more info.
 
 
- * Test some illegal mode combinations
- *
- */
+  * Test some illegal mode combinations
+  *
+  */
 
 #include "nc_tests.h"
 #include "err_macros.h"
@@ -21,37 +21,37 @@ See \ref copyright file for more info.
 int
 main(int argc, char** argv)
 {
-   int ncid,varid;
-   int retval;
-
-   printf("\n*** Testing illegal mode combinations\n");
-
-   MPI_Init(&argc,&argv);
-
-   printf("*** Testing create + MPIO + fletcher32\n");
-   if ((retval = nc_create_par(FILE_NAME, NC_CLOBBER|NC_NETCDF4, MPI_COMM_WORLD, MPI_INFO_NULL, &ncid))) ERR;
-   if ((retval = nc_def_var(ncid,"whatever",NC_INT,0,NULL,&varid))) ERR;
-   retval = nc_def_var_fletcher32(ncid,varid,NC_FLETCHER32);
-   if(retval != NC_EINVAL) ERR;
-   if ((retval = nc_abort(ncid)))
-{
-fprintf(stderr,"XXX: err=%d\n",retval);
-fflush(stderr);
-ERR;
-}
-
-   printf("*** Testing create + MPIO + deflation\n");
-   if ((retval = nc_create_par(FILE_NAME, NC_CLOBBER|NC_NETCDF4, MPI_COMM_WORLD, MPI_INFO_NULL, &ncid))) ERR;
-   if ((retval = nc_def_var(ncid,"whatever",NC_INT,0,NULL,&varid))) ERR;
-   retval = nc_def_var_deflate(ncid,varid, NC_NOSHUFFLE, 1, 1);
-   if(retval != NC_EINVAL) ERR;
-   if ((retval = nc_abort(ncid))) {
-	fprintf(stderr,"XXX: nc_abort: %d\n",retval); fflush(stderr);
-	ERR;
-   }
-
-   MPI_Finalize();
-
-   SUMMARIZE_ERR;
-   FINAL_RESULTS;
+    int ncid,varid;
+    int retval;
+
+    printf("\n*** Testing illegal mode combinations\n");
+
+    MPI_Init(&argc,&argv);
+
+    printf("*** Testing create + MPIO + fletcher32\n");
+    if ((retval = nc_create_par(FILE_NAME, NC_CLOBBER|NC_NETCDF4, MPI_COMM_WORLD, MPI_INFO_NULL, &ncid))) ERR;
+    if ((retval = nc_def_var(ncid,"whatever",NC_INT,0,NULL,&varid))) ERR;
+    retval = nc_def_var_fletcher32(ncid,varid,NC_FLETCHER32);
+    if(retval != NC_EINVAL) ERR;
+    if ((retval = nc_abort(ncid)))
+    {
+        fprintf(stderr,"XXX: err=%d\n",retval);
+        fflush(stderr);
+        ERR;
+    }
+
+    printf("*** Testing create + MPIO + deflation\n");
+    if ((retval = nc_create_par(FILE_NAME, NC_CLOBBER|NC_NETCDF4, MPI_COMM_WORLD, MPI_INFO_NULL, &ncid))) ERR;
+    if ((retval = nc_def_var(ncid,"whatever",NC_INT,0,NULL,&varid))) ERR;
+    retval = nc_def_var_deflate(ncid,varid, NC_NOSHUFFLE, 1, 1);
+    if(retval != NC_EINVAL) ERR;
+    if ((retval = nc_abort(ncid))) {
+        fprintf(stderr,"XXX: nc_abort: %d\n",retval); fflush(stderr);
+        ERR;
+    }
+
+    MPI_Finalize();
+
+    SUMMARIZE_ERR;
+    FINAL_RESULTS;
 }

nc_test4/tst_parallel_zlib.c

@@ -0,0 +1,167 @@
+/*
+Copyright 2018, UCAR/Unidata
+See COPYRIGHT file for copying and redistribution conditions.
+
+This program tests netcdf-4 parallel I/O using zlib compression while
+writing.
+
+Ed Hartnett, 2019/12/19
+*/
+
+/* Defining USE_MPE causes the MPE trace library to be used (and you
+ * must also relink with -llmpe -lmpe). This causes clog2 output to be
+ * written, which can be converted to slog2 (by the program
+ * clog2TOslog2) and then used in the analysis program jumpshot. */
+/*#define USE_MPE 1*/
+
+#include <nc_tests.h>
+#include "err_macros.h"
+#include <mpi.h>
+#ifdef USE_MPE
+#include <mpe.h>
+#endif /* USE_MPE */
+
+#define FILE "tst_parallel_zlib.nc"
+#define NDIMS 3
+#define DIMSIZE 24
+#define QTR_DATA (DIMSIZE * DIMSIZE / 4)
+#define NUM_PROC 4
+#define NUM_SLABS 10
+
+int
+main(int argc, char **argv)
+{
+    /* MPI stuff. */
+    int mpi_namelen;
+    char mpi_name[MPI_MAX_PROCESSOR_NAME];
+    int mpi_size, mpi_rank;
+    MPI_Comm comm = MPI_COMM_WORLD;
+    MPI_Info info = MPI_INFO_NULL;
+
+    /* Netcdf-4 stuff. */
+    int ncid, v1id, dimids[NDIMS];
+    size_t start[NDIMS], count[NDIMS];
+
+    int i, res;
+    int slab_data[DIMSIZE * DIMSIZE / 4]; /* one slab */
+    char file_name[NC_MAX_NAME + 1];
+
+#ifdef USE_MPE
+    int s_init, e_init, s_define, e_define, s_write, e_write, s_close, e_close;
+#endif /* USE_MPE */
+
+    /* Initialize MPI. */
+    MPI_Init(&argc, &argv);
+    MPI_Comm_size(MPI_COMM_WORLD, &mpi_size);
+    MPI_Comm_rank(MPI_COMM_WORLD, &mpi_rank);
+    MPI_Get_processor_name(mpi_name, &mpi_namelen);
+    /*printf("mpi_name: %s size: %d rank: %d\n", mpi_name,
+      mpi_size, mpi_rank);*/
+
+#ifdef USE_MPE
+    MPE_Init_log();
+    s_init = MPE_Log_get_event_number();
+    e_init = MPE_Log_get_event_number();
+    s_define = MPE_Log_get_event_number();
+    e_define = MPE_Log_get_event_number();
+    s_write = MPE_Log_get_event_number();
+    e_write = MPE_Log_get_event_number();
+    s_close = MPE_Log_get_event_number();
+    e_close = MPE_Log_get_event_number();
+    MPE_Describe_state(s_init, e_init, "Init", "red");
+    MPE_Describe_state(s_define, e_define, "Define", "yellow");
+    MPE_Describe_state(s_write, e_write, "Write", "green");
+    MPE_Describe_state(s_close, e_close, "Close", "purple");
+    MPE_Start_log();
+    MPE_Log_event(s_init, 0, "start init");
+#endif /* USE_MPE */
+
+    if (mpi_rank == 0)
+    {
+       printf("\n*** Testing parallel writes with zlib.\n");
+       printf("*** testing simple write with zlib...");
+    }
+
+    /* Create phony data. We're going to write a 24x24 array of ints,
+       in 4 sets of 144. */
+    for (i = 0; i < DIMSIZE * DIMSIZE / 4; i++)
+       slab_data[i] = mpi_rank;
+
+#ifdef USE_MPE
+    MPE_Log_event(e_init, 0, "end init");
+    MPE_Log_event(s_define, 0, "start define file");
+#endif /* USE_MPE */
+
+    /* Create a parallel netcdf-4 file. */
+    /*nc_set_log_level(3);*/
+    /* sprintf(file_name, "%s/%s", TEMP_LARGE, FILE); */
+    sprintf(file_name, "%s", FILE);
+    if ((res = nc_create_par(file_name, NC_NETCDF4, comm, info, &ncid))) ERR;
+
+    /* Create three dimensions. */
+    if (nc_def_dim(ncid, "d1", DIMSIZE, dimids)) ERR;
+    if (nc_def_dim(ncid, "d2", DIMSIZE, &dimids[1])) ERR;
+    if (nc_def_dim(ncid, "d3", NUM_SLABS, &dimids[2])) ERR;
+
+    /* Create one var. */
+    if ((res = nc_def_var(ncid, "v1", NC_INT, NDIMS, dimids, &v1id))) ERR;
+    if ((res = nc_def_var_deflate(ncid, 0, 0, 1, 1))) ERR;
+
+    /* Write metadata to file. */
+    if ((res = nc_enddef(ncid))) ERR;
+
+#ifdef USE_MPE
+    MPE_Log_event(e_define, 0, "end define file");
+    if (mpi_rank)
+       sleep(mpi_rank);
+#endif /* USE_MPE */
+
+    /* Set up slab for this process. */
+    start[0] = mpi_rank * DIMSIZE/mpi_size;
+    start[1] = 0;
+    count[0] = DIMSIZE/mpi_size;
+    count[1] = DIMSIZE;
+    count[2] = 1;
+    /*printf("mpi_rank=%d start[0]=%d start[1]=%d count[0]=%d count[1]=%d\n",
+      mpi_rank, start[0], start[1], count[0], count[1]);*/
+
+    if (nc_var_par_access(ncid, v1id, NC_COLLECTIVE)) ERR;
+/*    if (nc_var_par_access(ncid, v1id, NC_INDEPENDENT)) ERR;*/
+
+    for (start[2] = 0; start[2] < NUM_SLABS; start[2]++)
+    {
+#ifdef USE_MPE
+       MPE_Log_event(s_write, 0, "start write slab");
+#endif /* USE_MPE */
+
+       /* Write slabs of phoney data. */
+       if (nc_put_vara_int(ncid, v1id, start, count, slab_data)) ERR;
+#ifdef USE_MPE
+       MPE_Log_event(e_write, 0, "end write file");
+#endif /* USE_MPE */
+    }
+
+#ifdef USE_MPE
+    MPE_Log_event(s_close, 0, "start close file");
+#endif /* USE_MPE */
+
+    /* Close the netcdf file. */
+    if ((res = nc_close(ncid)))	ERR;
+
+#ifdef USE_MPE
+    MPE_Log_event(e_close, 0, "end close file");
+#endif /* USE_MPE */
+
+    /* Delete this large file. */
+    /* remove(file_name); */
+
+    /* Shut down MPI. */
+    MPI_Finalize();
+
+    if (mpi_rank == 0)
+    {
+       SUMMARIZE_ERR;
+       FINAL_RESULTS;
+    }
+    return 0;
+}