diff --git a/docs/INSTALL_CMake_options.md b/docs/INSTALL_CMake_options.md index 63a9372c30a..d116b60b544 100644 --- a/docs/INSTALL_CMake_options.md +++ b/docs/INSTALL_CMake_options.md @@ -58,7 +58,7 @@ These options concern the general build process of the main HDF5 libraries, util | `BUILD_STATIC_EXECS` | `BOOL` | `OFF` | If `ON`, builds statically-linked executables. **NOTE:** The `BUILD_STATIC_EXECS` option is only valid on some UNIX operating systems. It adds the `-static` flag to `CMAKE_EXE_LINKER_FLAGS`. This flag is not available on Windows and some modern Linux systems will ignore the flag. | | `HDF5_DEFAULT_API_VERSION` | `STRING` | `v200` | Specifies the default HDF5 API version to use when compiling HDF5 libraries. Valid values are `v200` (2.x API), `v114` (1.14.x API), `v112` (1.12.x API), `v110` (1.10.x API), `v18` (1.8.x API) and `v16` (1.6.x API). See [API Compatibility Macros](https://support.hdfgroup.org/documentation/hdf5/latest/api-compat-macros.html#title5) for more information on this option. | | `HDF5_ALLOW_UNSUPPORTED` | `BOOL` | `OFF` | If `ON`, allows configuring and building HDF5 with unsupported combinations of features. Otherwise, causes a configuration error if an unsupported combination is enabled. See [Unsupported option combinations](#unsupported_combos) for a list of unsupported combinations. | -| `HDF5_ENABLE_CONCURRENCY` | `BOOL` | `OFF` | If `ON`, enables building of a multi-thread concurrent HDF5 library. Requires C11 threads, Win32 threads or Pthreads. Requires shared HDF5 libraries on Windows. **NOTE:** Currently non-functional and experimental. | +| `HDF5_ENABLE_CONCURRENCY` | `BOOL` | `OFF` | If `ON`, enables building of a multi-thread concurrent HDF5 library. Requires C11 threads, Win32 threads or Pthreads. Requires shared HDF5 libraries on Windows. **NOTE:** Currently only used to enable internal multithreading where the library spawns its own threads and internally parallelizes a single operation. Does not yet allow multiple concurrent application threads inside the library. | | `HDF5_ENABLE_THREADSAFE` | `BOOL` | `OFF` | If `ON`, enables building of a thread-safe HDF5 library. Requires C11 threads, Win32 threads or Pthreads. Requires shared HDF5 libraries on Windows. | | `HDF5_ENABLE_NONSTANDARD_FEATURES` | `BOOL` | `ON` | If `ON`, enables non-standard programming language features. If `OFF`, disables all non-standard programming language features. Each feature has its own separate option. | | `HDF5_ENABLE_NONSTANDARD_FEATURE_FLOAT16` | `BOOL` | `ON` (if `_Float16` type is supported) | If `ON`, enables building of support for the `_Float16` 16-bit floating-point datatype. | diff --git a/docs/doxygen/CMakeLists.txt b/docs/doxygen/CMakeLists.txt index e06fac28e07..c10dba932ee 100644 --- a/docs/doxygen/CMakeLists.txt +++ b/docs/doxygen/CMakeLists.txt @@ -43,6 +43,7 @@ if (DOXYGEN_FOUND) # This ensures the Doxygen configuration stays synchronized with the # actual default versions defined in src/H5version.h. set (_doxygen_predefined_entries + H5_HAVE_CONCURRENCY H5_HAVE_DIRECT H5_HAVE_LIBHDFS H5_HAVE_MAP_API diff --git a/docs/doxygen/dox/ReferenceManual.dox b/docs/doxygen/dox/ReferenceManual.dox index a30ded40b08..3c6d163ac09 100644 --- a/docs/doxygen/dox/ReferenceManual.dox +++ b/docs/doxygen/dox/ReferenceManual.dox @@ -84,6 +84,10 @@ The functions provided by the HDF5 API are grouped into the following VOL Connector (H5VL)@ref H5VL "C"C++@ref FH5VL "Fortran"@ref JH5VL "Java"Manage HDF5 VOL connector plugins. + +Thread Safety (H5TS)@ref H5TS "C"C++FortranJavaThreadsafety and threading-related operations. + + diff --git a/docs/doxygen/examples/tables/propertyLists.dox b/docs/doxygen/examples/tables/propertyLists.dox index 6323bd740a9..2aff32a3a3a 100644 --- a/docs/doxygen/examples/tables/propertyLists.dox +++ b/docs/doxygen/examples/tables/propertyLists.dox @@ -723,6 +723,10 @@ of the library for reading or writing the actual data. + + + + diff --git a/release_docs/CHANGELOG.md b/release_docs/CHANGELOG.md index 74d4de1c54c..c0e3139ac07 100644 --- a/release_docs/CHANGELOG.md +++ b/release_docs/CHANGELOG.md @@ -54,6 +54,21 @@ We would like to thank the many HDF5 community members who contributed to this r ## Library +### Added support for internally concurrent multithreaded reads of chunked datasets + + Added 3 new functions to support this: H5TSset_internal_threads(), + H5Pset_io_threads(), and H5Pget_io_threads(). + + This feature internally parallelizes read operations on chunked datasets. + H5TSset_internal_threads() is used to enable the feature globally, while + H5Pset_io_threads() can be used to disable the feature on a per-operation + basis. These functions are only available when the library is configured with + HDF5_ENABLE_CONCURRENCY=ON. When performing an internally threaded read, the + library will concurrently read from disk, unfilter, and scatter to memory all + chunks in a read operation on a chunked dataset. Currently each of these + sub-operations is serialized (protected by a mutex), so there is not yet + likely to be any performance improvement. + ## Parallel Library ## Fortran Library diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 967eec1b08b..96ec9181ff1 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -900,6 +900,7 @@ set (H5TS_SOURCES ) set (H5TS_PUBLIC_HDRS ${HDF5_SRC_DIR}/H5TSdevelop.h + ${HDF5_SRC_DIR}/H5TSpublic.h ) IDE_GENERATED_PROPERTIES ("H5TS" "${H5TS_HDRS}" "${H5TS_SOURCES}" ) diff --git a/src/H5.c b/src/H5.c index 9f007d27abf..4f9727d01a1 100644 --- a/src/H5.c +++ b/src/H5.c @@ -400,6 +400,9 @@ H5_term_library(void) pending += DOWN(M_top); pending += DOWN(S_top); pending += DOWN(T_top); +#ifdef H5_HAVE_THREADSAFE_API + pending += DOWN(TS_top); +#endif /* H5_HAVE_THREADSAFE_API */ } /* end if */ /* Don't shut down the file code until objects in files are shut down */ diff --git a/src/H5CX.c b/src/H5CX.c index c6a43801d0b..802968d1e80 100644 --- a/src/H5CX.c +++ b/src/H5CX.c @@ -175,6 +175,9 @@ typedef struct H5CX_dxpl_cache_t { uint32_t actual_selection_io_mode; /* Actual selection I/O mode (H5D_XFER_ACTUAL_SELECTION_IO_MODE_NAME) */ bool modify_write_buf; /* Whether the library can modify write buffers */ +#ifdef H5_HAVE_CONCURRENCY + bool io_threads_enabled; /* Whether the library can use concurrent threads to accelerate I/O */ +#endif /* H5_HAVE_CONCURRENCY */ } H5CX_dxpl_cache_t; /* Typedef for cached default link creation property list information */ @@ -377,6 +380,12 @@ H5CX__init_package(void) if (H5P_get(dx_plist, H5D_XFER_MODIFY_WRITE_BUF_NAME, &H5CX_def_dxpl_cache.modify_write_buf) < 0) HGOTO_ERROR(H5E_CONTEXT, H5E_CANTGET, FAIL, "Can't retrieve modify write buffer property"); +#ifdef H5_HAVE_CONCURRENCY + /* Get the modify write buffer property */ + if (H5P_get(dx_plist, H5D_XFER_IO_THREADS_ENABLED_NAME, &H5CX_def_dxpl_cache.io_threads_enabled) < 0) + HGOTO_ERROR(H5E_CONTEXT, H5E_CANTGET, FAIL, "Can't retrieve I/O threads enabled property"); +#endif /* H5_HAVE_CONCURRENCY */ + /* Reset the "default LCPL cache" information */ memset(&H5CX_def_lcpl_cache, 0, sizeof(H5CX_lcpl_cache_t)); @@ -2307,6 +2316,41 @@ done: FUNC_LEAVE_NOAPI(ret_value) } /* end H5CX_get_selection_io_mode() */ +#ifdef H5_HAVE_CONCURRENCY +/*------------------------------------------------------------------------- + * Function: H5CX_get_io_threads + * + * Purpose: Retrieves the I/O threads enabled property for the current API call context. + * + * Return: Non-negative on success / Negative on failure + * + *------------------------------------------------------------------------- + */ +herr_t +H5CX_get_io_threads(bool *io_threads_enabled) +{ + H5CX_node_t **head = NULL; /* Pointer to head of API context list */ + herr_t ret_value = SUCCEED; /* Return value */ + + FUNC_ENTER_NOAPI(FAIL) + + /* Sanity check */ + assert(io_threads_enabled); + head = H5CX_get_my_context(); /* Get the pointer to the head of the API context, for this thread */ + assert(head && *head); + assert(H5P_DEFAULT != (*head)->ctx.dxpl_id); + + H5CX_RETRIEVE_PROP_VALID(dxpl, H5P_DATASET_XFER_DEFAULT, H5D_XFER_IO_THREADS_ENABLED_NAME, + io_threads_enabled) + + /* Get the value */ + *io_threads_enabled = (*head)->ctx.io_threads_enabled; + +done: + FUNC_LEAVE_NOAPI(ret_value) +} /* end H5CX_get_io_threads() */ +#endif /* H5_HAVE_CONCURRENCY */ + /*------------------------------------------------------------------------- * Function: H5CX_get_encoding * diff --git a/src/H5CXprivate.h b/src/H5CXprivate.h index ad61a1a430e..d974c9570d3 100644 --- a/src/H5CXprivate.h +++ b/src/H5CXprivate.h @@ -175,6 +175,10 @@ typedef struct H5CX_t { bool selection_io_mode_valid; /* Whether selection I/O mode is valid */ bool modify_write_buf; /* Whether the library can modify write buffers (H5D_XFER_MODIFY_WRITE_BUF_NAME)*/ bool modify_write_buf_valid; /* Whether the modify_write_buf field is valid */ +#ifdef H5_HAVE_CONCURRENCY + bool io_threads_enabled; /* Whether the library can use concurrent threads to accelerate I/O */ + bool io_threads_enabled_valid; /* Whether the io_threads_enabled field is valid */ +#endif /* H5_HAVE_CONCURRENCY */ /* Return-only DXPL properties to return to application */ #ifdef H5_HAVE_PARALLEL @@ -348,6 +352,9 @@ H5_DLL herr_t H5CX_get_selection_io_mode(H5D_selection_io_mode_t *selection_io_m H5_DLL herr_t H5CX_get_no_selection_io_cause(uint32_t *no_selection_io_cause); H5_DLL herr_t H5CX_get_actual_selection_io_mode(uint32_t *actual_selection_io_mode); H5_DLL herr_t H5CX_get_modify_write_buf(bool *modify_write_buf); +#ifdef H5_HAVE_CONCURRENCY +H5_DLL herr_t H5CX_get_io_threads(bool *io_threads_enabled); +#endif /* H5_HAVE_CONCURRENCY */ /* "Getter" routines for LCPL properties cached in API context */ H5_DLL herr_t H5CX_get_encoding(H5T_cset_t *encoding); diff --git a/src/H5Dchunk.c b/src/H5Dchunk.c index 9e1d20cc52e..3292686cfb7 100644 --- a/src/H5Dchunk.c +++ b/src/H5Dchunk.c @@ -57,6 +57,7 @@ #include "H5MFprivate.h" /* File memory management */ #include "H5PBprivate.h" /* Page Buffer */ #include "H5SLprivate.h" /* Skip Lists */ +#include "H5TSprivate.h" /* Threadsafety */ #include "H5VMprivate.h" /* Vector and array functions */ /****************/ @@ -160,6 +161,14 @@ 0x02U /* Filters have been disabled since \ * the last flush */ +#define H5D_CHUNK_THREADED_INITIAL_ALLOC_COUNT 64 + +#ifdef H5_HAVE_CONCURRENCY +#define H5D_CHUNK_LOCK_NO_THREADING_PARAMS , NULL, NULL +#else /* H5_HAVE_CONCURRENCY */ +#define H5D_CHUNK_LOCK_NO_THREADING_PARAMS +#endif /* H5_HAVE_CONCURRENCY */ + /******************/ /* Local Typedefs */ /******************/ @@ -292,6 +301,37 @@ typedef struct H5D_chunk_iter_ud_t { haddr_t base_addr; /* Base address of the file, taking user block into account */ } H5D_chunk_iter_ud_t; +#ifdef H5_HAVE_CONCURRENCY +/* Information about a single chunk in an internally concurrent operation */ +typedef struct H5D_threaded_chunk_info_t { + struct H5D_threaded_io_info_t + *threaded_io_info; /* Pointer to the dataset-global struct for this operation */ + H5O_pline_t *old_pline; /* Filter pipeline used to compress this chunk */ + H5D_chunk_ud_t udata; /* Chunk udata struct (from lookup */ + size_t chunk_nbytes; /* Size of chunk on disk */ + size_t buf_alloc; /* Allocated size of chunk buffer */ + void *chunk; /* Chunk buffer */ + hsize_t src_accessed_bytes; /* Number of bytes accessed in the chunk */ + H5D_dset_io_info_t chk_dset_io_info; /* Temporary dataset I/O info for individual chunk I/O */ +} H5D_threaded_chunk_info_t; + +/* Dataset-global information about an internally concurrent operation */ +typedef struct H5D_threaded_io_info_t { + H5D_threaded_chunk_info_t *chunk_info; /* Array of info structs for each chunk */ + H5D_dset_io_info_t *dset_info; /* Dataset I/O info */ + H5D_io_info_t cpt_io_info; /* I/O info struct for memory scatter */ + size_t num_chunks; /* Number of chunks in concurrent operation */ + size_t chunk_nalloc; /* Allocated size of chunk_info array */ + size_t chunk_size; /* Size of an unfiltered chunk */ + size_t chunks_left; /* Number of chunks left to process */ + bool chunks_locked; /* Whether any chunks are locked */ + H5TS_cond_t cond; /* Condition variable for waiting on threads */ + H5TS_mutex_t cond_mutex; /* Mutex associated with cond */ + bool failed; /* Whether any threads failed */ + hid_t dxpl_id; /* Dataset transfer property list ID */ +} H5D_threaded_io_info_t; +#endif /* H5_HAVE_CONCURRENCY */ + /********************/ /* Local Prototypes */ /********************/ @@ -318,31 +358,39 @@ static int H5D__chunk_iter_cb(const H5D_chunk_rec_t *chunk_rec, void *udata); static int H5D__chunk_format_convert_cb(const H5D_chunk_rec_t *chunk_rec, void *_udata); /* Helper routines */ -static herr_t H5D__chunk_set_info_real(H5O_layout_chunk_t *layout, unsigned ndims, const hsize_t *curr_dims, - const hsize_t *max_dims); -static herr_t H5D__chunk_set_sizes(H5D_t *dset); -static herr_t H5D__chunk_cinfo_cache_reset(H5D_chunk_cached_t *last); -static herr_t H5D__chunk_cinfo_cache_update(H5D_chunk_cached_t *last, const H5D_chunk_ud_t *udata); -static bool H5D__chunk_cinfo_cache_found(const H5D_chunk_cached_t *last, H5D_chunk_ud_t *udata); -static herr_t H5D__create_piece_map_single(H5D_dset_io_info_t *di, H5D_io_info_t *io_info); -static herr_t H5D__create_piece_file_map_all(H5D_dset_io_info_t *di, H5D_io_info_t *io_info); -static herr_t H5D__create_piece_file_map_hyper(H5D_dset_io_info_t *di, H5D_io_info_t *io_info); -static herr_t H5D__create_piece_mem_map_1d(const H5D_dset_io_info_t *di); -static herr_t H5D__create_piece_mem_map_hyper(const H5D_dset_io_info_t *di); -static herr_t H5D__piece_file_cb(void *elem, const H5T_t *type, unsigned ndims, const hsize_t *coords, - void *_opdata); -static herr_t H5D__piece_mem_cb(void *elem, const H5T_t *type, unsigned ndims, const hsize_t *coords, - void *_opdata); -static herr_t H5D__chunk_may_use_select_io(H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info); +static herr_t H5D__chunk_set_info_real(H5O_layout_chunk_t *layout, unsigned ndims, const hsize_t *curr_dims, + const hsize_t *max_dims); +static herr_t H5D__chunk_set_sizes(H5D_t *dset); +static herr_t H5D__chunk_cinfo_cache_reset(H5D_chunk_cached_t *last); +static herr_t H5D__chunk_cinfo_cache_update(H5D_chunk_cached_t *last, const H5D_chunk_ud_t *udata); +static bool H5D__chunk_cinfo_cache_found(const H5D_chunk_cached_t *last, H5D_chunk_ud_t *udata); +static herr_t H5D__create_piece_map_single(H5D_dset_io_info_t *di, H5D_io_info_t *io_info); +static herr_t H5D__create_piece_file_map_all(H5D_dset_io_info_t *di, H5D_io_info_t *io_info); +static herr_t H5D__create_piece_file_map_hyper(H5D_dset_io_info_t *di, H5D_io_info_t *io_info); +static herr_t H5D__create_piece_mem_map_1d(const H5D_dset_io_info_t *di); +static herr_t H5D__create_piece_mem_map_hyper(const H5D_dset_io_info_t *di); +static herr_t H5D__piece_file_cb(void *elem, const H5T_t *type, unsigned ndims, const hsize_t *coords, + void *_opdata); +static herr_t H5D__piece_mem_cb(void *elem, const H5T_t *type, unsigned ndims, const hsize_t *coords, + void *_opdata); +static herr_t H5D__chunk_may_use_select_io(H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info); +#ifdef H5_HAVE_CONCURRENCY +static H5TS_THREAD_RETURN_TYPE H5D__chunk_thread_read(void *_threaded_chunk_info); +#endif /* H5_HAVE_CONCURRENCY */ static unsigned H5D__chunk_hash_val(const H5D_shared_t *shared, const hsize_t *scaled); static herr_t H5D__chunk_flush_entry(const H5D_t *dset, H5D_rdcc_ent_t *ent, bool reset); static herr_t H5D__chunk_cache_evict(const H5D_t *dset, H5D_rdcc_ent_t *ent, bool flush); static void *H5D__chunk_lock(const H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info, - H5D_chunk_ud_t *udata, bool relax, bool prev_unfilt_chunk); -static herr_t H5D__chunk_unlock(const H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info, - const H5D_chunk_ud_t *udata, bool dirty, void *chunk, hsize_t naccessed); -static herr_t H5D__chunk_cache_prune(const H5D_t *dset, size_t size); -static herr_t H5D__chunk_prune_fill(H5D_chunk_it_ud1_t *udata, bool new_unfilt_chunk); + H5D_chunk_ud_t *udata, bool relax, bool prev_unfilt_chunk +#ifdef H5_HAVE_CONCURRENCY + , + H5D_threaded_io_info_t *threaded_io_info, bool *threaded_chunk +#endif /* H5_HAVE_CONCURRENCY */ +); +static herr_t H5D__chunk_unlock(const H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info, + const H5D_chunk_ud_t *udata, bool dirty, void *chunk, hsize_t naccessed); +static herr_t H5D__chunk_cache_prune(const H5D_t *dset, size_t size); +static herr_t H5D__chunk_prune_fill(H5D_chunk_it_ud1_t *udata, bool new_unfilt_chunk); #ifdef H5_HAVE_PARALLEL static herr_t H5D__chunk_collective_fill(const H5D_t *dset, H5D_chunk_coll_fill_info_t *chunk_fill_info, const void *fill_buf, const void *partial_chunk_fill_buf); @@ -2719,7 +2767,7 @@ done: */ htri_t H5D__chunk_cacheable(const H5D_io_info_t H5_ATTR_PARALLEL_USED *io_info, H5D_dset_io_info_t *dset_info, - haddr_t caddr, bool write_op) + const hsize_t *scaled, haddr_t caddr, bool write_op) { const H5D_t *dataset = NULL; /* Local pointer to dataset info */ bool has_filters = false; /* Whether there are filters on the chunk or not */ @@ -2739,7 +2787,7 @@ H5D__chunk_cacheable(const H5D_io_info_t H5_ATTR_PARALLEL_USED *io_info, H5D_dse if (dataset->shared->layout.u.chunk.flags & H5O_LAYOUT_CHUNK_DONT_FILTER_PARTIAL_BOUND_CHUNKS) { has_filters = !H5D__chunk_is_partial_edge_chunk(dataset->shared->ndims, dataset->shared->layout.u.chunk.dim, - dset_info->store->chunk.scaled, dataset->shared->curr_dims); + scaled, dataset->shared->curr_dims); } /* end if */ else has_filters = true; @@ -2904,7 +2952,10 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) void *chunk = NULL; /* Pointer to locked chunk buffer */ bool chunk_locked = false; /* Indicates whether the chunk is locked */ H5D_chunk_ud_t udata; /* Chunk index pass-through */ - herr_t ret_value = SUCCEED; /*return value */ +#ifdef H5_HAVE_CONCURRENCY + H5D_threaded_io_info_t *threaded_io_info = NULL; /* Info for concurrent threaded execution */ +#endif /* H5_HAVE_CONCURRENCY */ + herr_t ret_value = SUCCEED; /*return value */ FUNC_ENTER_PACKAGE @@ -2932,9 +2983,9 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) /* Different blocks depending on whether we're using selection I/O */ if (io_info->use_select_io == H5D_SELECTION_IO_MODE_ON) { - size_t num_chunks = 0; size_t element_sizes[2] = {dset_info->type_info.src_type_size, 0}; void *bufs[2] = {dset_info->buf.vp, NULL}; + size_t num_chunks = 0; /* Number of chunks selected */ /* Only create selection I/O arrays if not performing multi dataset I/O, * otherwise the higher level will handle it */ @@ -3063,6 +3114,11 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) H5D_io_info_t cpt_io_info; /* Compact I/O info object */ H5D_storage_t cpt_store; /* Chunk storage information as compact dataset */ bool cpt_dirty; /* Temporary placeholder for compact storage "dirty" flag */ +#ifdef H5_HAVE_CONCURRENCY + bool do_threading; /* Whether to do internal thread spawning */ + bool threaded_chunk; /* Whether to do internal thread spawning for this chunk */ + size_t init_chunk_nalloc = 0; /* Initial allocation size of array of chunks in threaded I/O */ +#endif /* H5_HAVE_CONCURRENCY */ /* Set up contiguous I/O info object */ H5MM_memcpy(&ctg_io_info, io_info, sizeof(ctg_io_info)); @@ -3096,37 +3152,111 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) /* Initialize temporary compact storage info */ cpt_store.compact.dirty = &cpt_dirty; +#ifdef H5_HAVE_CONCURRENCY + /* Get number of chunks */ + init_chunk_nalloc = H5D_CHUNK_GET_NODE_COUNT(dset_info); + + /* Check if we're using threads - first check if the global thread pool exists , we're not already in + * a concurrency event, and there's at least one chunk selected */ + do_threading = (H5TS_pool_g != NULL) && !H5TS_currently_concurrent_g && (init_chunk_nalloc > 0); + + if (do_threading) { + /* Now check if threading is disabled by the context (DXPL) */ + if (H5CX_get_io_threads(&do_threading) < 0) + HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't check if threading is enabled"); + + if (do_threading) { + /* Allocate threaded I/O info struct */ + if (NULL == (threaded_io_info = H5MM_calloc(sizeof(H5D_threaded_io_info_t)))) + HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL, "can't allocate threaded I/O info struct"); + + /* Calculate initial allocation size */ + init_chunk_nalloc = MIN(H5D_CHUNK_THREADED_INITIAL_ALLOC_COUNT, init_chunk_nalloc); + + /* Allocate array of threaded chunk info structs */ + if (NULL == (threaded_io_info->chunk_info = (H5D_threaded_chunk_info_t *)H5MM_malloc( + init_chunk_nalloc * sizeof(H5D_threaded_chunk_info_t)))) + HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL, + "can't allocate array of threaded chunk info structs"); + threaded_io_info->chunk_nalloc = init_chunk_nalloc; + + /* Store dataset I/O info */ + threaded_io_info->dset_info = dset_info; + + /* Copy compact I/O info to threaded I/O info struct */ + H5MM_memcpy(&threaded_io_info->cpt_io_info, &cpt_io_info, sizeof(cpt_io_info)); + + /* Initialize num chunks in threaded_io_info to 0 (will be incremented as info is filled in + * for each chunk */ + threaded_io_info->num_chunks = 0; + + /* Cache chunk size */ + threaded_io_info->chunk_size = dset_info->dset->shared->layout.u.chunk.size; + + /* If a non-default DXPL is set, save the DXPL for the threads to use. Note that this means + * that H5CX_get* operations must be protected by a mutex unless we're sure they've been + * cached previously */ + if (!H5CX_is_def_dxpl()) + threaded_io_info->dxpl_id = H5CX_get_dxpl(); + else + threaded_io_info->dxpl_id = H5I_INVALID_HID; + } + } +#endif /* H5_HAVE_CONCURRENCY */ + /* Iterate through nodes in chunk skip list */ chunk_node = H5D_CHUNK_GET_FIRST_NODE(dset_info); while (chunk_node) { H5D_piece_info_t *chunk_info; /* Chunk information */ + H5D_chunk_ud_t *udata_p; /* Pointer to chunk udata */ htri_t cacheable; /* Whether the chunk is cacheable */ +#ifdef H5_HAVE_CONCURRENCY + if (do_threading) { + void *tmp_list; + + /* Expand chunk array if necessary */ + if (threaded_io_info->num_chunks == threaded_io_info->chunk_nalloc) { + if (NULL == (tmp_list = H5MM_realloc(threaded_io_info->chunk_info, + 2 * threaded_io_info->chunk_nalloc * + sizeof(*threaded_io_info->chunk_info)))) + HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL, + "memory reallocation failed for chunk list"); + threaded_io_info->chunk_info = (H5D_threaded_chunk_info_t *)tmp_list; + + threaded_io_info->chunk_nalloc *= 2; + } + + /* Set udata_p */ + udata_p = &(threaded_io_info->chunk_info[threaded_io_info->num_chunks].udata); + } + else +#endif /* H5_HAVE_CONCURRENCY */ + /* Set udata_p */ + udata_p = &udata; + /* Get the actual chunk information from the skip list node */ chunk_info = H5D_CHUNK_GET_NODE_INFO(dset_info, chunk_node); /* Get the info for the chunk in the file */ - if (H5D__chunk_lookup(dset_info->dset, chunk_info->scaled, &udata) < 0) + if (H5D__chunk_lookup(dset_info->dset, chunk_info->scaled, udata_p) < 0) HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "error looking up chunk address"); /* Sanity check */ - assert((H5_addr_defined(udata.chunk_block.offset) && udata.chunk_block.length > 0) || - (!H5_addr_defined(udata.chunk_block.offset) && udata.chunk_block.length == 0)); + assert((H5_addr_defined(udata_p->chunk_block.offset) && udata_p->chunk_block.length > 0) || + (!H5_addr_defined(udata_p->chunk_block.offset) && udata_p->chunk_block.length == 0)); /* Check for non-existent chunk & skip it if appropriate */ - if (H5_addr_defined(udata.chunk_block.offset) || UINT_MAX != udata.idx_hint || + if (H5_addr_defined(udata_p->chunk_block.offset) || UINT_MAX != udata_p->idx_hint || !skip_missing_chunks) { H5D_io_info_t *chk_io_info = NULL; /* Pointer to I/O info object for this chunk */ - /* Set chunk's [scaled] coordinates */ - dset_info->store->chunk.scaled = chunk_info->scaled; - /* Don't lock the chunk if it doesn't exist on disk or in cache, to avoid unnecessary * allocation and conversion */ - if (H5_addr_defined(udata.chunk_block.offset) || UINT_MAX != udata.idx_hint) { + if (H5_addr_defined(udata_p->chunk_block.offset) || UINT_MAX != udata_p->idx_hint) { /* Determine if we should use the chunk cache */ - if ((cacheable = - H5D__chunk_cacheable(io_info, dset_info, udata.chunk_block.offset, false)) < 0) + if ((cacheable = H5D__chunk_cacheable(io_info, dset_info, chunk_info->scaled, + udata_p->chunk_block.offset, false)) < 0) HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't tell if chunk is cacheable"); if (cacheable) { /* Load the chunk into cache and lock it. */ @@ -3138,7 +3268,14 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) chunk_info->piece_points * (hsize_t)dset_info->type_info.src_type_size; /* Lock the chunk into the cache */ - if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, &udata, false, false))) +#ifdef H5_HAVE_CONCURRENCY + threaded_chunk = false; + if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, udata_p, false, false, + do_threading ? threaded_io_info : NULL, + &threaded_chunk))) +#else /* H5_HAVE_CONCURRENCY */ + if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, udata_p, false, false))) +#endif /* H5_HAVE_CONCURRENCY */ HGOTO_ERROR(H5E_IO, H5E_CANTLOCK, FAIL, "unable to lock raw data chunk"); chunk_locked = true; @@ -3151,32 +3288,76 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) else { /* Since the chunk isn't cacheable it must not be in cache, therefore it must exist on * disk if it made it into the outer if statement */ - assert(H5_addr_defined(udata.chunk_block.offset)); + assert(H5_addr_defined(udata_p->chunk_block.offset)); + +#ifdef H5_HAVE_CONCURRENCY + /* Disable threading for now since there's likely no performance gain and if it isn't + * cacheable it may be because it's very large and we don't want to load the whole + * thing into memory */ + threaded_chunk = false; +#endif /* H5_HAVE_CONCURRENCY */ /* Set up the storage address information for this chunk */ - ctg_store.contig.dset_addr = udata.chunk_block.offset; + ctg_store.contig.dset_addr = udata_p->chunk_block.offset; /* Point I/O info at temporary I/O info for this chunk */ chk_io_info = &ctg_io_info; } - /* Perform the actual read operation */ - assert(chk_io_info); - assert(chk_io_info->count == 1); - chk_io_info->dsets_info[0].layout_io_info.contig_piece_info = chunk_info; - chk_io_info->dsets_info[0].file_space = chunk_info->fspace; - chk_io_info->dsets_info[0].mem_space = chunk_info->mspace; - chk_io_info->dsets_info[0].nelmts = chunk_info->piece_points; - if ((dset_info->io_ops.single_read)(chk_io_info, &chk_io_info->dsets_info[0]) < 0) - HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "chunked read failed"); +#ifdef H5_HAVE_CONCURRENCY + /* Delay actual I/O for threaded chunks */ + if (threaded_chunk) { + /* Set up chunk I/O info. old_pline, chunk_nbytes, and buf_alloc were set up in + * H5D__chunk_lock. udata was set up in H5D__chunk_lookup. */ + assert(chk_io_info); + assert(chk_io_info == &cpt_io_info); + assert(chk_io_info->count == 1); + threaded_io_info->chunk_info[threaded_io_info->num_chunks].chunk = chunk; + chunk = NULL; + chunk_locked = false; + threaded_io_info->chunks_locked = true; + threaded_io_info->chunk_info[threaded_io_info->num_chunks].threaded_io_info = + threaded_io_info; + H5MM_memcpy( + &threaded_io_info->chunk_info[threaded_io_info->num_chunks].chk_dset_io_info, + dset_info, + sizeof( + threaded_io_info->chunk_info[threaded_io_info->num_chunks].chk_dset_io_info)); + threaded_io_info->chunk_info[threaded_io_info->num_chunks] + .chk_dset_io_info.layout_ops = *H5D_LOPS_COMPACT; + threaded_io_info->chunk_info[threaded_io_info->num_chunks] + .chk_dset_io_info.layout_io_info.contig_piece_info = chunk_info; + threaded_io_info->chunk_info[threaded_io_info->num_chunks] + .chk_dset_io_info.file_space = chunk_info->fspace; + threaded_io_info->chunk_info[threaded_io_info->num_chunks] + .chk_dset_io_info.mem_space = chunk_info->mspace; + threaded_io_info->chunk_info[threaded_io_info->num_chunks].chk_dset_io_info.nelmts = + chunk_info->piece_points; + threaded_io_info->chunk_info[threaded_io_info->num_chunks].src_accessed_bytes = + src_accessed_bytes; + threaded_io_info->num_chunks++; + } + else +#endif /* H5_HAVE_CONCURRENCY */ + { + /* Perform the actual read operation */ + assert(chk_io_info); + assert(chk_io_info->count == 1); + chk_io_info->dsets_info[0].layout_io_info.contig_piece_info = chunk_info; + chk_io_info->dsets_info[0].file_space = chunk_info->fspace; + chk_io_info->dsets_info[0].mem_space = chunk_info->mspace; + chk_io_info->dsets_info[0].nelmts = chunk_info->piece_points; + if ((dset_info->io_ops.single_read)(chk_io_info, &chk_io_info->dsets_info[0]) < 0) + HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "chunked read failed"); - /* Release the cache lock on the chunk */ - chunk_locked = false; - if (chunk && - H5D__chunk_unlock(io_info, dset_info, &udata, false, chunk, src_accessed_bytes) < 0) - HGOTO_ERROR(H5E_IO, H5E_CANTUNLOCK, FAIL, "unable to unlock raw data chunk"); - chunk = NULL; - } /* end if */ + /* Release the cache lock on the chunk */ + chunk_locked = false; + if (chunk && H5D__chunk_unlock(io_info, dset_info, udata_p, false, chunk, + src_accessed_bytes) < 0) + HGOTO_ERROR(H5E_IO, H5E_CANTUNLOCK, FAIL, "unable to unlock raw data chunk"); + chunk = NULL; + } + } else /* Write fill values to memory buffer */ if (H5D__fill(dset_info->dset->shared->dcpl_cache.fill.buf, dset_info->dset->shared->type, @@ -3187,14 +3368,132 @@ H5D__chunk_read(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) /* Advance to next chunk in list */ chunk_node = H5D_CHUNK_GET_NEXT_NODE(dset_info, chunk_node); } /* end while */ - } /* end else */ + +#ifdef H5_HAVE_CONCURRENCY + /* Handle chunks that were deferred for concurrent processing */ + if (threaded_io_info) { + assert(threaded_io_info->chunk_info); + + if (threaded_io_info->num_chunks > 0) { + size_t threads_launched = 0; + + assert(do_threading); + assert(H5TS_pool_g); + + /* Create condition variable for signaling task completion */ + if (H5_UNLIKELY(H5TS_cond_init(&threaded_io_info->cond) < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, + "can't create condition variable for completed tasks"); + + /* Create mutex for condition variable */ + if (H5_UNLIKELY(H5TS_mutex_init(&threaded_io_info->cond_mutex, H5TS_MUTEX_TYPE_PLAIN) < 0)) { + if (H5TS_cond_destroy(&threaded_io_info->cond) < 0) + HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "can't destroy condition variable"); + HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "can't create mutex for completed tasks"); + } + + /* Defer errors from now until the end of this block, to minimize cleanup code needed in the + * done section */ + + /* Store number of threads launched */ + threaded_io_info->chunks_left = threaded_io_info->num_chunks; + + /* Mark that we are concurrent */ + H5TS_currently_concurrent_g = true; + + /* Loop over threaded chunks, launching worker task function for each */ + for (size_t i = 0; i < threaded_io_info->num_chunks; i++) { + if (H5_UNLIKELY(H5TS_pool_add_task(H5TS_pool_g, H5D__chunk_thread_read, + &(threaded_io_info->chunk_info[i])) < 0)) { + HDONE_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "can't launch worker thread"); + break; + } + threads_launched++; + } + + /* Acquire cond_mutex so the signal for task completion doesn't get sent before we start + * waiting on it */ + if (H5_UNLIKELY(H5TS_mutex_lock(&threaded_io_info->cond_mutex) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock mutex for condition variable"); + + /* If we failed, subtract any unlaunched threads from chunks_left */ + if (H5_UNLIKELY(ret_value < 0)) { + assert(threaded_io_info->num_chunks >= threads_launched); + threaded_io_info->chunks_left -= threaded_io_info->num_chunks - threads_launched; + } + else + assert(threaded_io_info->num_chunks == threads_launched); + + /* Wait on condition variable for all worker tasks to complete */ + while (threaded_io_info->chunks_left) { + if (H5_UNLIKELY(H5TS_cond_wait(&threaded_io_info->cond, &threaded_io_info->cond_mutex) < + 0)) { + HDONE_ERROR(H5E_DATASET, H5E_CANTWAIT, FAIL, "can't wait for worker threads"); + break; + } + } + + /* Mark that we are no longer concurrent */ + H5TS_currently_concurrent_g = false; + + /* Unlock cond_mutex */ + if (H5_UNLIKELY(H5TS_mutex_unlock(&threaded_io_info->cond_mutex) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, + "can't unlock mutex for condition variable"); + + /* Unlock all chunks */ + for (size_t i = 0; i < threaded_io_info->num_chunks; i++) { + if (H5_UNLIKELY( + H5D__chunk_unlock(io_info, dset_info, &threaded_io_info->chunk_info[i].udata, + false, threaded_io_info->chunk_info[i].chunk, + threaded_io_info->chunk_info[i].src_accessed_bytes) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "unable to unlock raw data chunk"); + threaded_io_info->chunk_info[i].chunk = NULL; + } + threaded_io_info->chunks_locked = false; + + /* Check for thread failure */ + if (H5_UNLIKELY(threaded_io_info->failed)) { + HDONE_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "threaded read worker(s) failed"); + + /* We must evict all chunks if a worker failed, because the chunk may be in an + * inconsistent state in memory */ + for (size_t i = 0; i < threaded_io_info->num_chunks; i++) + if (H5D__chunk_cache_evict(dset_info->dset, + dset_info->dset->shared->cache.chunk.slot[udata.idx_hint], + false) < 0) + HDONE_ERROR(H5E_DATASET, H5E_CANTREMOVE, FAIL, "unable to evict chunk"); + } + + /* Prune chunk cache to maximum size */ + if (H5_UNLIKELY(H5D__chunk_cache_prune(dset_info->dset, 0) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "unable to preempt chunk(s) from cache"); + + /* Destroy condition variable and mutex */ + if (H5_UNLIKELY(H5TS_mutex_destroy(&threaded_io_info->cond_mutex) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, + "can't destroy mutex for condition variable"); + if (H5_UNLIKELY(H5TS_cond_destroy(&threaded_io_info->cond) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "can't destroy condition variable"); + + /* Check for failure (not technically necessary right now, but include in case anything gets + * added after this) */ + if (H5_UNLIKELY(ret_value < 0)) + HGOTO_DONE(ret_value); + } + + /* Free chunk_info array */ + H5MM_free(threaded_io_info->chunk_info); + threaded_io_info->chunk_info = NULL; + + /* Free threaded I/O info struct */ + H5MM_free(threaded_io_info); + threaded_io_info = NULL; + } +#endif /* H5_HAVE_CONCURRENCY */ + } /* end else */ done: - /* Release chunk lock if we failed while holding it */ - if (chunk_locked && chunk) - if (H5D__chunk_unlock(io_info, dset_info, &udata, false, chunk, src_accessed_bytes) < 0) - HDONE_ERROR(H5E_IO, H5E_CANTUNLOCK, FAIL, "unable to unlock raw data chunk"); - /* Free dataset sieve buffer and reset cached fields */ if (dset_info->dset->shared->cache.sieve.sieve_buf) { dset_info->dset->shared->cache.sieve.sieve_loc = HADDR_UNDEF; @@ -3211,16 +3510,255 @@ done: chunk_file_spaces = H5MM_xfree(chunk_file_spaces); if (chunk_addrs != chunk_addrs_local) chunk_addrs = H5MM_xfree(chunk_addrs); - } /* end if */ + + /* Release chunk lock if we failed while holding it */ + if (chunk_locked && chunk) { + if (H5D__chunk_unlock(io_info, dset_info, &udata, false, chunk, src_accessed_bytes) < 0) + HDONE_ERROR(H5E_IO, H5E_CANTUNLOCK, FAIL, "unable to unlock raw data chunk"); + chunk_locked = false; + chunk = NULL; + } + +#ifdef H5_HAVE_CONCURRENCY + if (threaded_io_info) { + if (threaded_io_info->chunk_info && threaded_io_info->chunks_locked) { + /* Unlock all chunks */ + for (size_t i = 0; i < threaded_io_info->num_chunks; i++) + if (threaded_io_info->chunk_info[i].chunk && + H5D__chunk_unlock(io_info, dset_info, &threaded_io_info->chunk_info[i].udata, false, + threaded_io_info->chunk_info[i].chunk, + threaded_io_info->chunk_info[i].src_accessed_bytes) < 0) + HDONE_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "unable to unlock raw data chunk"); + + /* Prune chunk cache to maximum size */ + if (H5D__chunk_cache_prune(dset_info->dset, 0) < 0) + HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "unable to preempt chunk(s) from cache"); + } + + /* Free chunk_info array */ + H5MM_xfree(threaded_io_info->chunk_info); + + /* Free threaded I/O info struct */ + threaded_io_info = H5MM_xfree(threaded_io_info); + } +#endif /* H5_HAVE_CONCURRENCY */ + } /* end if */ /* Make sure we cleaned up */ assert(!chunk_mem_spaces || chunk_mem_spaces == chunk_mem_spaces_local); assert(!chunk_file_spaces || chunk_file_spaces == chunk_file_spaces_local); assert(!chunk_addrs || chunk_addrs == chunk_addrs_local); + assert(!chunk_locked); + assert(!chunk); +#ifdef H5_HAVE_CONCURRENCY + assert(!threaded_io_info); + assert(!H5TS_currently_concurrent_g); +#endif /* H5_HAVE_CONCURRENCY */ FUNC_LEAVE_NOAPI(ret_value) } /* H5D__chunk_read() */ +#ifdef H5_HAVE_CONCURRENCY +/*------------------------------------------------------------------------- + * Function: H5D__chunk_thread_read + * + * Purpose: Thread worker function for internally concurrent chunk + * reads. Reads the chunk from disk, unfilters it, and + * scatters it to the user's buffer. + * + * Return: 0 (return value not checked - error status reported in + * udata struct) + * + *------------------------------------------------------------------------- + */ +static H5TS_THREAD_RETURN_TYPE +H5D__chunk_thread_read(void *_threaded_chunk_info) +{ + H5D_threaded_chunk_info_t *threaded_chunk_info = + (H5D_threaded_chunk_info_t *)_threaded_chunk_info; /* Threaded info for this chunk */ + H5CX_node_t api_ctx = {{0}, NULL}; /* API context node to push */ + H5D_storage_t cpt_store; /* Chunk storage information as compact dataset */ + bool cpt_dirty; /* Temporary placeholder for compact storage "dirty" flag */ + bool mutex_held = false; /* Whether we hold the internal mutex */ + bool api_ctx_pushed = false; /* Whether API context pushed */ + herr_t ret_value = SUCCEED; /* "Return value" for error macros (actual return value is always + (H5TS_thread_ret_t)0) */ + + assert(threaded_chunk_info); + assert(threaded_chunk_info->threaded_io_info); + + FUNC_ENTER_PACKAGE + + /* Set API context */ + if (H5CX_push(&api_ctx) < 0) + HGOTO_ERROR(H5E_SYM, H5E_CANTSET, FAIL, "can't set API context"); + api_ctx_pushed = true; + + /* Set DXPL on context if it exists */ + if (threaded_chunk_info->threaded_io_info->dxpl_id != H5I_INVALID_HID) + H5CX_set_dxpl(threaded_chunk_info->threaded_io_info->dxpl_id); + + /* Lock internal mutex */ + if (H5_UNLIKELY(H5TS_internal_lock() < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock internal mutex"); + mutex_held = true; + + /* Read chunk from disk */ + if (H5_UNLIKELY(H5F_shared_block_read(H5F_SHARED(threaded_chunk_info->chk_dset_io_info.dset->oloc.file), + H5FD_MEM_DRAW, threaded_chunk_info->udata.chunk_block.offset, + threaded_chunk_info->udata.chunk_block.length, + threaded_chunk_info->chunk) < 0)) + HGOTO_ERROR(H5E_IO, H5E_READERROR, FAIL, "unable to read raw data chunk"); + + /* Unlock internal mutex */ + if (H5_UNLIKELY(H5TS_internal_unlock() < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "can't unlock internal mutex"); + mutex_held = false; + + if (threaded_chunk_info->old_pline) { + H5Z_EDC_t err_detect; + H5Z_cb_t filter_cb; +#ifndef NDEBUG + void *old_chunk = threaded_chunk_info->chunk; +#endif /* NDEBUG */ + +#ifndef H5_UNSAFE_CONCURRENCY + /* Lock internal mutex */ + if (H5_UNLIKELY(H5TS_internal_lock() < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock internal mutex"); + mutex_held = true; +#endif /* H5_UNSAFE_CONCURRENCY */ + + /* Retrieve filter settings from API context. These must be protected by a mutex if using a + * non-default DXPL. */ + if (H5_UNLIKELY(H5CX_get_err_detect(&err_detect) < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get error detection info"); + if (H5_UNLIKELY(H5CX_get_filter_cb(&filter_cb) < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get I/O filter callback function"); + + /* Perform filter pipeline. Defer going to done on error so the chunk cache is always patched. */ + if (H5_UNLIKELY(H5Z_pipeline(threaded_chunk_info->old_pline, H5Z_FLAG_REVERSE, + &(threaded_chunk_info->udata.filter_mask), err_detect, filter_cb, + &threaded_chunk_info->chunk_nbytes, &threaded_chunk_info->buf_alloc, + &threaded_chunk_info->chunk) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTFILTER, FAIL, "data pipeline read failed"); + +#ifndef H5_UNSAFE_CONCURRENCY + /* Unlock internal mutex. Also defer going to done here. */ + if (H5_UNLIKELY(H5TS_internal_unlock() < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "can't unlock internal mutex"); + mutex_held = false; +#endif /* H5_UNSAFE_CONCURRENCY */ + + /* Patch chunk pointer in chunk cache. Ok to do in concurrent section since no other thread will touch + * this chunk. */ + if (UINT_MAX != threaded_chunk_info->udata.idx_hint) { + H5D_rdcc_t *rdcc = &threaded_chunk_info->threaded_io_info->dset_info->dset->shared->cache.chunk; + + assert(threaded_chunk_info->udata.idx_hint < rdcc->nslots); + assert(rdcc->slot[threaded_chunk_info->udata.idx_hint]); + assert(rdcc->slot[threaded_chunk_info->udata.idx_hint]->chunk == old_chunk); + rdcc->slot[threaded_chunk_info->udata.idx_hint]->chunk = threaded_chunk_info->chunk; + } + + /* Now goto done if the call to H5Z_pipeline() failed */ + if (H5_UNLIKELY(ret_value < 0)) + HGOTO_DONE(ret_value); + + /* Make sure the chunk is the correct size after being unfiltered */ + if (H5_UNLIKELY(threaded_chunk_info->chunk_nbytes != + threaded_chunk_info->threaded_io_info->chunk_size)) + HGOTO_ERROR(H5E_DATASET, H5E_BADVALUE, FAIL, "chunk size is incorrect after being unfiltered"); + + /* The new_unfilt_chunk flag should only be set during an H5Dset_extent operation. This is not + * currently supported with internal threading. If this support is added, we will need to realloc + * the buffer in the associated threaded worker function. */ + assert(!threaded_chunk_info->udata.new_unfilt_chunk); + } + + /* Set up the storage buffer information for this chunk */ + cpt_store.compact.buf = threaded_chunk_info->chunk; + cpt_store.compact.dirty = &cpt_dirty; + threaded_chunk_info->chk_dset_io_info.store = &cpt_store; + + /* Lock internal mutex */ + if (H5_UNLIKELY(H5TS_internal_lock() < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock internal mutex"); + mutex_held = true; + + /* Scatter data from chunk buffer to application buffer. Even though this is protected by a mutex it must + * still not interfere with the threads in H5Z_pipeline() */ + if (H5_UNLIKELY((threaded_chunk_info->threaded_io_info->dset_info->io_ops.single_read)( + &threaded_chunk_info->threaded_io_info->cpt_io_info, + &threaded_chunk_info->chk_dset_io_info) < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "chunked read failed"); + + /* Unlock internal mutex */ + if (H5_UNLIKELY(H5TS_internal_unlock() < 0)) + HGOTO_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "can't unlock internal mutex"); + mutex_held = false; + + if (H5_UNLIKELY(H5CX_pop(false) < 0)) + HDONE_ERROR(H5E_SYM, H5E_CANTRESET, FAIL, "can't reset API context"); + api_ctx_pushed = false; + +done: + /* Report failure to task invoker (actual return value is ignored by the thread pool) */ + if (H5_UNLIKELY(ret_value < 0)) + threaded_chunk_info->threaded_io_info->failed = true; + + /* Acquire condition variable mutex */ + if (H5_UNLIKELY(H5TS_mutex_lock(&threaded_chunk_info->threaded_io_info->cond_mutex) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock mutex for condition variable"); + + /* Decrement the number of chunks left, and signal the main thread if this was the last */ + if (1 == threaded_chunk_info->threaded_io_info->chunks_left--) + /* Signal condition variable */ + if (H5_UNLIKELY(H5TS_cond_signal(&threaded_chunk_info->threaded_io_info->cond) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't signal condition variable"); + + /* Unlock condition variable mutex */ + if (H5_UNLIKELY(H5TS_mutex_unlock(&threaded_chunk_info->threaded_io_info->cond_mutex) < 0)) + HDONE_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "can't unlock mutex for condition variable"); + + /* Handle failures */ + if (H5_UNLIKELY(ret_value < 0)) { + /* Set failed again in case something failed in the cond_signal block. This is best effort, it's + * possible at this point for the main thread to race ahead and not see the failure. */ + threaded_chunk_info->threaded_io_info->failed = true; + + /* Clean up */ + if (api_ctx_pushed && H5CX_pop(false) < 0) + HDONE_ERROR(H5E_SYM, H5E_CANTRESET, FAIL, "can't reset API context"); + api_ctx_pushed = false; + + /* Acquire mutex */ + if (!mutex_held) { + if (H5TS_internal_lock() < 0) + HDONE_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock internal mutex"); + else + mutex_held = true; + } + + /* Print and clear error stack */ + (void)H5E_dump_api_stack(); + (void)H5E_clear_stack(); + + /* Release mutex */ + if (mutex_held) { + if (H5TS_internal_unlock() < 0) + HDONE_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "can't unlock internal mutex"); + mutex_held = false; + } + } + + assert(!mutex_held); + assert(!api_ctx_pushed); + + FUNC_LEAVE_NOAPI((H5TS_thread_ret_t)0); +} /* end H5D__chunk_thread_read() */ +#endif /* H5_HAVE_CONCURRENCY */ + /*------------------------------------------------------------------------- * Function: H5D__chunk_write * @@ -3353,11 +3891,9 @@ H5D__chunk_write(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) assert((H5_addr_defined(udata.chunk_block.offset) && udata.chunk_block.length > 0) || (!H5_addr_defined(udata.chunk_block.offset) && udata.chunk_block.length == 0)); - /* Set chunk's [scaled] coordinates */ - dset_info->store->chunk.scaled = chunk_info->scaled; - /* Determine if we should use the chunk cache */ - if ((cacheable = H5D__chunk_cacheable(io_info, dset_info, udata.chunk_block.offset, true)) < 0) + if ((cacheable = H5D__chunk_cacheable(io_info, dset_info, chunk_info->scaled, + udata.chunk_block.offset, true)) < 0) HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't tell if chunk is cacheable"); if (cacheable) { /* Load the chunk into cache. But if the whole chunk is written, @@ -3376,7 +3912,8 @@ H5D__chunk_write(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) entire_chunk = false; /* Lock the chunk into the cache */ - if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, &udata, entire_chunk, false))) + if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, &udata, entire_chunk, + false H5D_CHUNK_LOCK_NO_THREADING_PARAMS))) HGOTO_ERROR(H5E_IO, H5E_CANTLOCK, FAIL, "unable to lock raw data chunk"); chunk_locked = true; @@ -3522,11 +4059,9 @@ H5D__chunk_write(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) assert((H5_addr_defined(udata.chunk_block.offset) && udata.chunk_block.length > 0) || (!H5_addr_defined(udata.chunk_block.offset) && udata.chunk_block.length == 0)); - /* Set chunk's [scaled] coordinates */ - dset_info->store->chunk.scaled = chunk_info->scaled; - /* Determine if we should use the chunk cache */ - if ((cacheable = H5D__chunk_cacheable(io_info, dset_info, udata.chunk_block.offset, true)) < 0) + if ((cacheable = H5D__chunk_cacheable(io_info, dset_info, chunk_info->scaled, + udata.chunk_block.offset, true)) < 0) HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't tell if chunk is cacheable"); if (cacheable) { /* Load the chunk into cache. But if the whole chunk is written, @@ -3545,7 +4080,8 @@ H5D__chunk_write(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info) entire_chunk = false; /* Lock the chunk into the cache */ - if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, &udata, entire_chunk, false))) + if (NULL == (chunk = H5D__chunk_lock(io_info, dset_info, &udata, entire_chunk, + false H5D_CHUNK_LOCK_NO_THREADING_PARAMS))) HGOTO_ERROR(H5E_IO, H5E_CANTLOCK, FAIL, "unable to lock raw data chunk"); chunk_locked = true; @@ -4598,7 +5134,12 @@ done: */ static void * H5D__chunk_lock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_dset_io_info_t *dset_info, - H5D_chunk_ud_t *udata, bool relax, bool prev_unfilt_chunk) + H5D_chunk_ud_t *udata, bool relax, bool prev_unfilt_chunk +#ifdef H5_HAVE_CONCURRENCY + , + H5D_threaded_io_info_t *threaded_io_info, bool *threaded_chunk +#endif /* H5_HAVE_CONCURRENCY */ +) { const H5D_t *dset; /* Convenience pointer to the dataset */ H5O_pline_t *pline; /* I/O pipeline info - always equal to the pline passed to H5D__chunk_mem_alloc */ @@ -4655,7 +5196,7 @@ H5D__chunk_lock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_ds /* Make sure this is the right chunk */ for (u = 0; u < layout->u.chunk.ndims - 1; u++) - assert(dset_info->store->chunk.scaled[u] == ent->scaled[u]); + assert(udata->common.scaled[u] == ent->scaled[u]); } #endif /* NDEBUG */ @@ -4771,8 +5312,7 @@ H5D__chunk_lock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_ds else if (layout->u.chunk.flags & H5O_LAYOUT_CHUNK_DONT_FILTER_PARTIAL_BOUND_CHUNKS) { /* Check if this is an edge chunk */ if (H5D__chunk_is_partial_edge_chunk(dset->shared->ndims, layout->u.chunk.dim, - dset_info->store->chunk.scaled, - dset->shared->curr_dims)) { + udata->common.scaled, dset->shared->curr_dims)) { /* Disable the filters for both writing and reading */ disable_filters = true; old_pline = NULL; @@ -4797,7 +5337,7 @@ H5D__chunk_lock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_ds rdcc->stats.nhits++; if (NULL == (chunk = H5D__chunk_mem_alloc(chunk_size, pline))) - HGOTO_ERROR(H5E_RESOURCE, H5E_NOSPACE, NULL, "memory allocation failed for raw data chunk"); + HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, NULL, "memory allocation failed for raw data chunk"); /* In the case that some dataset functions look through this data, * clear it to all 0s. */ @@ -4830,49 +5370,68 @@ H5D__chunk_lock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_ds HGOTO_ERROR(H5E_RESOURCE, H5E_NOSPACE, NULL, "memory allocation failed for raw data chunk"); - /* Read chunk from disk */ - if (H5F_shared_block_read(H5F_SHARED(dset->oloc.file), H5FD_MEM_DRAW, chunk_addr, - chunk_disk_size, chunk) < 0) - HGOTO_ERROR(H5E_IO, H5E_READERROR, NULL, "unable to read raw data chunk"); +#ifdef H5_HAVE_CONCURRENCY + /* Check if we're doing threaded I/O, if so add chunk to read list */ + if (threaded_io_info) { + assert(threaded_chunk); + assert(!*threaded_chunk); - /* Unfilter chunk */ - if (old_pline && old_pline->nused) { - H5Z_EDC_t err_detect; /* Error detection info */ - H5Z_cb_t filter_cb; /* I/O filter callback function */ + /* Store chunk I/O information */ + threaded_io_info->chunk_info[threaded_io_info->num_chunks].old_pline = old_pline; + threaded_io_info->chunk_info[threaded_io_info->num_chunks].chunk_nbytes = chunk_nbytes; + threaded_io_info->chunk_info[threaded_io_info->num_chunks].buf_alloc = buf_alloc; - /* Retrieve filter settings from API context */ - if (H5CX_get_err_detect(&err_detect) < 0) - HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, NULL, "can't get error detection info"); - if (H5CX_get_filter_cb(&filter_cb) < 0) - HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, NULL, "can't get I/O filter callback function"); + /* Report that this chunk is threaded */ + *threaded_chunk = true; + } + else +#endif /* H5_HAVE_CONCURRENCY */ + { + /* Read chunk from disk */ + if (H5F_shared_block_read(H5F_SHARED(dset->oloc.file), H5FD_MEM_DRAW, chunk_addr, + chunk_disk_size, chunk) < 0) + HGOTO_ERROR(H5E_IO, H5E_READERROR, NULL, "unable to read raw data chunk"); - /* Perform filter pipeline */ - if (H5Z_pipeline(old_pline, H5Z_FLAG_REVERSE, &(udata->filter_mask), err_detect, - filter_cb, &chunk_nbytes, &buf_alloc, &chunk) < 0) - HGOTO_ERROR(H5E_DATASET, H5E_CANTFILTER, NULL, "data pipeline read failed"); + /* Unfilter chunk */ + if (old_pline && old_pline->nused) { + H5Z_EDC_t err_detect; /* Error detection info */ + H5Z_cb_t filter_cb; /* I/O filter callback function */ - /* Make sure the chunk is the correct size after being unfiltered */ - if (chunk_nbytes != chunk_size) - HGOTO_ERROR(H5E_DATASET, H5E_BADVALUE, NULL, - "chunk size is incorrect after being unfiltered"); + /* Retrieve filter settings from API context */ + if (H5CX_get_err_detect(&err_detect) < 0) + HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, NULL, "can't get error detection info"); + if (H5CX_get_filter_cb(&filter_cb) < 0) + HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, NULL, + "can't get I/O filter callback function"); - /* Reallocate chunk if necessary */ - if (udata->new_unfilt_chunk) { - void *tmp_chunk = chunk; + /* Perform filter pipeline */ + if (H5Z_pipeline(old_pline, H5Z_FLAG_REVERSE, &(udata->filter_mask), err_detect, + filter_cb, &chunk_nbytes, &buf_alloc, &chunk) < 0) + HGOTO_ERROR(H5E_DATASET, H5E_CANTFILTER, NULL, "data pipeline read failed"); - if (NULL == (chunk = H5D__chunk_mem_alloc(chunk_nbytes, pline))) { + /* Make sure the chunk is the correct size after being unfiltered */ + if (chunk_nbytes != chunk_size) + HGOTO_ERROR(H5E_DATASET, H5E_BADVALUE, NULL, + "chunk size is incorrect after being unfiltered"); + + /* Reallocate chunk if necessary */ + if (udata->new_unfilt_chunk) { + void *tmp_chunk = chunk; + + if (NULL == (chunk = H5D__chunk_mem_alloc(chunk_nbytes, pline))) { + (void)H5D__chunk_mem_xfree(tmp_chunk, old_pline); + HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, NULL, + "memory allocation failed for raw data chunk"); + } /* end if */ + H5MM_memcpy(chunk, tmp_chunk, chunk_size); (void)H5D__chunk_mem_xfree(tmp_chunk, old_pline); - HGOTO_ERROR(H5E_RESOURCE, H5E_NOSPACE, NULL, - "memory allocation failed for raw data chunk"); } /* end if */ - H5MM_memcpy(chunk, tmp_chunk, chunk_size); - (void)H5D__chunk_mem_xfree(tmp_chunk, old_pline); - } /* end if */ - } /* end if */ + } /* end if */ - /* Assert that the chunk is the correct size and the buffer is big enough */ - assert(chunk_nbytes == chunk_size); - assert(buf_alloc >= chunk_size); + /* Assert that the chunk is the correct size and the buffer is big enough */ + assert(chunk_nbytes == chunk_size); + assert(buf_alloc >= chunk_size); + } /* Increment # of cache misses */ rdcc->stats.nmisses++; @@ -4885,7 +5444,7 @@ H5D__chunk_lock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_ds /* Allocate chunk buffer large enough to hold an unfiltered (in cache) chunk */ if (NULL == (chunk = H5D__chunk_mem_alloc(chunk_size, pline))) - HGOTO_ERROR(H5E_RESOURCE, H5E_NOSPACE, NULL, + HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, NULL, "memory allocation failed for raw data chunk"); if (H5P_is_fill_value_defined(fill, &fill_status) < 0) @@ -5066,9 +5625,8 @@ H5D__chunk_unlock(const H5D_io_info_t H5_ATTR_NDEBUG_UNUSED *io_info, const H5D_ } /* end if */ else if (layout->u.chunk.flags & H5O_LAYOUT_CHUNK_DONT_FILTER_PARTIAL_BOUND_CHUNKS) { /* Check if the chunk is an edge chunk, and disable filters if so */ - is_unfiltered_edge_chunk = - H5D__chunk_is_partial_edge_chunk(dset->shared->ndims, layout->u.chunk.dim, - dset_info->store->chunk.scaled, dset->shared->curr_dims); + is_unfiltered_edge_chunk = H5D__chunk_is_partial_edge_chunk( + dset->shared->ndims, layout->u.chunk.dim, udata->common.scaled, dset->shared->curr_dims); } /* end if */ if (dirty) { @@ -5729,11 +6287,7 @@ H5D__chunk_update_old_edge_chunks(H5D_t *dset, hsize_t old_dim[]) HGOTO_DONE(SUCCEED); } /* end if */ - /* Set up chunked I/O info object, for operations on chunks (in callback). - * Note that we only need to set chunk_offset once, as the array's address - * will never change. */ - chk_store.chunk.scaled = chunk_sc; - + /* Set up chunked I/O info object, for operations on chunks (in callback) */ chk_io_info.op_type = H5D_IO_OP_READ; chk_dset_info.dset = dset; @@ -5804,8 +6358,8 @@ H5D__chunk_update_old_edge_chunks(H5D_t *dset, hsize_t old_dim[]) if (H5_addr_defined(chk_udata.chunk_block.offset) || (UINT_MAX != chk_udata.idx_hint)) { /* Lock the chunk into cache. H5D__chunk_lock will take care of * updating the chunk to no longer be an edge chunk. */ - if (NULL == - (chunk = (void *)H5D__chunk_lock(&chk_io_info, &chk_dset_info, &chk_udata, false, true))) + if (NULL == (chunk = (void *)H5D__chunk_lock(&chk_io_info, &chk_dset_info, &chk_udata, false, + true H5D_CHUNK_LOCK_NO_THREADING_PARAMS))) HGOTO_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "unable to lock raw data chunk"); /* Unlock the chunk */ @@ -6132,7 +6686,8 @@ H5D__chunk_prune_fill(H5D_chunk_it_ud1_t *udata, bool new_unfilt_chunk) HGOTO_ERROR(H5E_DATASET, H5E_CANTSELECT, FAIL, "unable to select hyperslab"); /* Lock the chunk into the cache, to get a pointer to the chunk buffer */ - if (NULL == (chunk = (void *)H5D__chunk_lock(io_info, udata->dset_info, &chk_udata, false, false))) + if (NULL == (chunk = (void *)H5D__chunk_lock(io_info, udata->dset_info, &chk_udata, false, + false H5D_CHUNK_LOCK_NO_THREADING_PARAMS))) HGOTO_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "unable to lock raw data chunk"); chunk_locked = true; @@ -6365,11 +6920,7 @@ H5D__chunk_prune_by_extent(H5D_t *dset, const hsize_t *old_dim) /* (hyperslabs will always start from origin) */ memset(hyper_start, 0, sizeof(hyper_start)); - /* Set up chunked I/O info object, for operations on chunks (in callback) - * Note that we only need to set scaled once, as the array's address - * will never change. */ - chk_store.chunk.scaled = scaled; - + /* Set up chunked I/O info object, for operations on chunks (in callback) */ chk_io_info.op_type = H5D_IO_OP_READ; chk_dset_info.dset = dset; diff --git a/src/H5Dmpio.c b/src/H5Dmpio.c index 7fcaeadc613..d44827fa070 100644 --- a/src/H5Dmpio.c +++ b/src/H5Dmpio.c @@ -2131,9 +2131,6 @@ H5D__multi_chunk_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_ if (NULL == (next_chunk_info = (H5D_piece_info_t *)H5SL_item(piece_node))) HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "couldn't get piece info from skip list"); } - - /* Pass in chunk's coordinates in a union. */ - store.chunk.scaled = chunk_info->scaled; } else chunk_info = NULL; diff --git a/src/H5Dpkg.h b/src/H5Dpkg.h index 1e153ed5232..2f5b64f90f8 100644 --- a/src/H5Dpkg.h +++ b/src/H5Dpkg.h @@ -267,18 +267,14 @@ typedef struct { hsize_t dset_size; /* Total size of dataset in file */ } H5D_contig_storage_t; -typedef struct { - hsize_t *scaled; /* Scaled coordinates for a chunk */ -} H5D_chunk_storage_t; - typedef struct { void *buf; /* Buffer for compact dataset */ bool *dirty; /* Pointer to dirty flag to mark */ } H5D_compact_storage_t; typedef union H5D_storage_t { - H5D_contig_storage_t contig; /* Contiguous information for dataset */ - H5D_chunk_storage_t chunk; /* Chunk information for dataset */ + H5D_contig_storage_t contig; /* Contiguous information for dataset */ + /* No global storage information for a chunked dataset for this union */ H5D_compact_storage_t compact; /* Compact information for dataset */ H5O_efl_t efl; /* External file list information for dataset */ } H5D_storage_t; @@ -757,8 +753,8 @@ H5_DLL herr_t H5D__contig_check(const H5F_t *f, const H5O_layout_t *layout, cons const H5T_t *dt); /* Functions that operate on chunked dataset storage */ -H5_DLL htri_t H5D__chunk_cacheable(const H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info, haddr_t caddr, - bool write_op); +H5_DLL htri_t H5D__chunk_cacheable(const H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info, + const hsize_t *scaled, haddr_t caddr, bool write_op); H5_DLL herr_t H5D__chunk_create(const H5D_t *dset /*in,out*/); H5_DLL herr_t H5D__chunk_set_info(const H5D_t *dset); H5_DLL bool H5D__chunk_is_space_alloc(const H5O_storage_t *storage); diff --git a/src/H5Dprivate.h b/src/H5Dprivate.h index a6175492016..9afd3deea9b 100644 --- a/src/H5Dprivate.h +++ b/src/H5Dprivate.h @@ -89,6 +89,9 @@ #define H5D_XFER_NO_SELECTION_IO_CAUSE_NAME "no_selection_io_cause" /* Cause for no selection I/O */ #define H5D_XFER_ACTUAL_SELECTION_IO_MODE_NAME "actual_selection_io_mode" /* Actual selection I/O mode */ #define H5D_XFER_MODIFY_WRITE_BUF_NAME "modify_write_buf" /* Modify write buffers */ +#ifdef H5_HAVE_CONCURRENCY +#define H5D_XFER_IO_THREADS_ENABLED_NAME "io_threads_enabled" /* I/O threads enabled */ +#endif /* H5_HAVE_CONCURRENCY */ #ifdef H5_HAVE_INSTRUMENTED_LIBRARY /* Collective chunk instrumentation properties */ #define H5D_XFER_COLL_CHUNK_LINK_HARD_NAME "coll_chunk_link_hard" diff --git a/src/H5E.c b/src/H5E.c index 936d80c771a..b4430578f76 100644 --- a/src/H5E.c +++ b/src/H5E.c @@ -47,6 +47,11 @@ /***********/ /* Headers */ /***********/ +/* Disable free lists in this package when concurrency is enabled until free lists are threadsafe */ +#ifdef H5_HAVE_CONCURRENCY +#define H5_NO_FREE_LISTS +#endif /* H5_HAVE_CONCURRENCY */ + #include "H5private.h" /* Generic Functions */ #include "H5Epkg.h" /* Error handling */ #include "H5FLprivate.h" /* Free lists */ diff --git a/src/H5Eint.c b/src/H5Eint.c index 4948fd7d987..5d31887eeba 100644 --- a/src/H5Eint.c +++ b/src/H5Eint.c @@ -28,6 +28,11 @@ /***********/ /* Headers */ /***********/ +/* Disable free lists in this package when concurrency is enabled until free lists are threadsafe */ +#ifdef H5_HAVE_CONCURRENCY +#define H5_NO_FREE_LISTS +#endif /* H5_HAVE_CONCURRENCY */ + #include "H5private.h" /* Generic Functions */ #include "H5Epkg.h" /* Error handling */ #include "H5FLprivate.h" /* Free lists */ diff --git a/src/H5Pdxpl.c b/src/H5Pdxpl.c index fd20131ccf3..b36000a3ab4 100644 --- a/src/H5Pdxpl.c +++ b/src/H5Pdxpl.c @@ -181,6 +181,13 @@ #define H5D_XFER_MODIFY_WRITE_BUF_DEF false #define H5D_XFER_MODIFY_WRITE_BUF_ENC H5P__dxfr_modify_write_buf_enc #define H5D_XFER_MODIFY_WRITE_BUF_DEC H5P__dxfr_modify_write_buf_dec +#ifdef H5_HAVE_CONCURRENCY +/* Definitions for modify write buffer property */ +#define H5D_XFER_IO_THREADS_ENABLED_SIZE sizeof(bool) +#define H5D_XFER_IO_THREADS_ENABLED_DEF true +#define H5D_XFER_IO_THREADS_ENABLED_ENC H5P__encode_bool +#define H5D_XFER_IO_THREADS_ENABLED_DEC H5P__decode_bool +#endif /* H5_HAVE_CONCURRENCY */ /******************/ /* Local Typedefs */ @@ -298,6 +305,9 @@ static const H5D_selection_io_mode_t H5D_def_selection_io_mode_g = H5D_XFER_ static const uint32_t H5D_def_no_selection_io_cause_g = H5D_XFER_NO_SELECTION_IO_CAUSE_DEF; static const uint32_t H5D_def_actual_selection_io_mode_g = H5D_XFER_ACTUAL_SELECTION_IO_MODE_DEF; static const bool H5D_def_modify_write_buf_g = H5D_XFER_MODIFY_WRITE_BUF_DEF; +#ifdef H5_HAVE_CONCURRENCY +static const bool H5D_def_io_threads_enabled_g = H5D_XFER_IO_THREADS_ENABLED_DEF; +#endif /* H5_HAVE_CONCURRENCY */ /*------------------------------------------------------------------------- * Function: H5P__dxfr_reg_prop @@ -485,6 +495,14 @@ H5P__dxfr_reg_prop(H5P_genclass_t *pclass) H5D_XFER_MODIFY_WRITE_BUF_DEC, NULL, NULL, NULL, NULL) < 0) HGOTO_ERROR(H5E_PLIST, H5E_CANTINSERT, FAIL, "can't insert property into class"); +#ifdef H5_HAVE_CONCURRENCY + /* Register the I/O threads enabled property */ + if (H5P__register_real(pclass, H5D_XFER_IO_THREADS_ENABLED_NAME, H5D_XFER_IO_THREADS_ENABLED_SIZE, + &H5D_def_io_threads_enabled_g, NULL, NULL, NULL, H5D_XFER_IO_THREADS_ENABLED_ENC, + H5D_XFER_IO_THREADS_ENABLED_DEC, NULL, NULL, NULL, NULL) < 0) + HGOTO_ERROR(H5E_PLIST, H5E_CANTINSERT, FAIL, "can't insert property into class"); +#endif /* H5_HAVE_CONCURRENCY */ + done: FUNC_LEAVE_NOAPI(ret_value) } /* end H5P__dxfr_reg_prop() */ @@ -2602,3 +2620,72 @@ H5Pget_modify_write_buf(hid_t plist_id, bool *modify_write_buf /*out*/) done: FUNC_LEAVE_API(ret_value) } /* end H5Pget_modify_write_buf() */ + +#ifdef H5_HAVE_CONCURRENCY +/*------------------------------------------------------------------------- + * Function: H5Pset_io_threads + * + * Purpose: Sets whether the library can use the global thread pool to + * accelerate I/O using concurrent execution. + * + * Return: Success: Non-negative + * Failure: Negative + * + *------------------------------------------------------------------------- + */ +herr_t +H5Pset_io_threads(hid_t plist_id, bool io_threads_enabled) +{ + H5P_genplist_t *plist; /* Property list pointer */ + herr_t ret_value = SUCCEED; /* Return value */ + + FUNC_ENTER_API(FAIL) + + /* Check arguments */ + if (plist_id == H5P_DEFAULT) + HGOTO_ERROR(H5E_PLIST, H5E_BADVALUE, FAIL, "can't set values in default property list"); + + if (NULL == (plist = H5P_object_verify(plist_id, H5P_DATASET_XFER, false))) + HGOTO_ERROR(H5E_PLIST, H5E_BADTYPE, FAIL, "not a dxpl"); + + /* Set the threads enabled property */ + if (H5P_set(plist, H5D_XFER_IO_THREADS_ENABLED_NAME, &io_threads_enabled) < 0) + HGOTO_ERROR(H5E_PLIST, H5E_CANTSET, FAIL, "unable to set value"); + +done: + FUNC_LEAVE_API(ret_value) +} /* end H5Pset_io_threads() */ + +/*------------------------------------------------------------------------- + * Function: H5Pget_io_threads + * + * Purpose: Retrieves the setting that determines whether the library + * can use the global thread pool to accelerate I/O using + * concurrent execution. + * + * Return: Success: Non-negative + * Failure: Negative + * + *------------------------------------------------------------------------- + */ +herr_t +H5Pget_io_threads(hid_t plist_id, bool *io_threads_enabled /*out*/) +{ + H5P_genplist_t *plist; /* Property list pointer */ + herr_t ret_value = SUCCEED; /* Return value */ + + FUNC_ENTER_API(FAIL) + + /* Check arguments */ + if (NULL == (plist = H5P_object_verify(plist_id, H5P_DATASET_XFER, true))) + HGOTO_ERROR(H5E_PLIST, H5E_BADTYPE, FAIL, "not a dxpl"); + + /* Get the threads enabled property */ + if (io_threads_enabled) + if (H5P_get(plist, H5D_XFER_IO_THREADS_ENABLED_NAME, io_threads_enabled) < 0) + HGOTO_ERROR(H5E_PLIST, H5E_CANTGET, FAIL, "unable to get value"); + +done: + FUNC_LEAVE_API(ret_value) +} /* end H5Pget_io_threads() */ +#endif /* H5_HAVE_CONCURRENCY */ diff --git a/src/H5Ppublic.h b/src/H5Ppublic.h index bc5d5b9ea99..85065ad026a 100644 --- a/src/H5Ppublic.h +++ b/src/H5Ppublic.h @@ -8800,6 +8800,54 @@ H5_DLL herr_t H5Pset_modify_write_buf(hid_t plist_id, bool modify_write_buf); */ H5_DLL herr_t H5Pget_modify_write_buf(hid_t plist_id, bool *modify_write_buf); +/** + * + * \ingroup DXPL + * + * \brief Allows the library to use internal multithreading to accelerate I/O + * + * \dxpl_id{plist_id} + * \param[in] io_threads_enabled Whether the library can use internal multithreading to accelerate I/O + * + * \return \herr_t + * + * \details H5Pset_io_threads() sets whether the library is allowed to use internal multithreading to + * accelerate I/O. By default this is set to true. However, internal threading must be enabled using + * H5TSset_internal_threads() before the library can use threads to accelerate I/O. + * H5Pset_io_threads() can be used to disable this acceleration for a specific operation even if + * internal threading is enabled globally. + * + * \note This function is only present when the library is compiled with HDF5_ENABLE_CONCURRENCY=ON. + * + * \since 2.3.0 + * + */ +H5_DLL herr_t H5Pset_io_threads(hid_t plist_id, bool io_threads_enabled); + +/** + * + * \ingroup DXPL + * + * \brief Retrieves the "I/O threads enabled" property + * + * \dxpl_id{plist_id} + * \param[out] io_threads_enabled Whether the library can use internal multithreading to accelerate I/O + * + * \return \herr_t + * + * \details H5Pget_io_threads() gets the "I/O threads enabled" property from the dataset transfer + * property list \p plist_id. This property determines whether the library is allowed to use internal + * multithreading to accelerate I/O. The default value for io_threads_enabled is true. However, + * internal threading must be enabled using H5TSset_internal_threads() before the library can use + * threads to accelerate I/O. + * + * \note This function is only present when the library is compiled with HDF5_ENABLE_CONCURRENCY=ON. + * + * \since 2.3.0 + * + */ +H5_DLL herr_t H5Pget_io_threads(hid_t plist_id, bool *io_threads_enabled); + /** * \ingroup LCPL * diff --git a/src/H5TS.c b/src/H5TS.c index 469d492acf3..d6b26bc7821 100644 --- a/src/H5TS.c +++ b/src/H5TS.c @@ -60,6 +60,14 @@ H5TS_api_info_t H5TS_api_info_p; /* Library Private Variables */ /*****************************/ +#ifdef H5_HAVE_CONCURRENCY +/* Global thread pool */ +H5TS_pool_t *H5TS_pool_g = NULL; + +/* Whether there are concurrent threads in the library (from internal spawning) */ +bool H5TS_currently_concurrent_g = false; +#endif /* H5_HAVE_CONCURRENCY */ + /*******************/ /* Local Variables */ /*******************/ @@ -147,4 +155,59 @@ H5TSmutex_release(unsigned *lock_count) FUNC_LEAVE_API_NAMECHECK_ONLY(ret_value) } /* end H5TSmutex_release() */ + +#ifdef H5_HAVE_CONCURRENCY +/*-------------------------------------------------------------------------- + * Function: H5TSset_internal_threads + * + * Purpose: Sets the number of threads that the library can use + * internally to accelerate operations that be parallelized. + * This number is in addition to the main thread, which the + * library currently does not use for this type of parallel + * execution. + * + * Creates a global thread pool for the HDF5 library to use to + * accelerate parallelizable operations, or destroys it if + * num_threads == 0. + * + * This function does use the error stack because it is not + * meant to be called within a concurrent section. + * + * Parameters: + * num_threads; IN: The number of threads to use for + * internally concurrent execution. + * + * Return: Non-negative on success / Negative on failure + *-------------------------------------------------------------------------- + */ +herr_t +H5TSset_internal_threads(unsigned num_threads) +{ + herr_t ret_value = SUCCEED; + + FUNC_ENTER_API(FAIL) + + /* Check if the pool already exists, destroy it if so */ + if (H5TS_pool_g) { + /* Check if the pool already has requested number of threads, if so we're done */ + if (num_threads == H5TS_pool_g->num_threads) + HGOTO_DONE(SUCCEED); + + /* Otherwise, destroy the pool (will recreate with the requested number of threads in the next step) + */ + if (H5TS_pool_destroy(H5TS_pool_g) < 0) + HGOTO_ERROR(H5E_LIB, H5E_CANTFREE, FAIL, "can't destroy thread pool"); + H5TS_pool_g = NULL; + } + + /* Create global thread pool if num_threads > 0 */ + if (num_threads > 0) + if (H5TS_pool_create(&H5TS_pool_g, num_threads) < 0) + HGOTO_ERROR(H5E_LIB, H5E_CANTINIT, FAIL, "can't create thread pool"); + +done: + FUNC_LEAVE_API(ret_value); +} /* end H5TSset_internal_threads() */ +#endif /* H5_HAVE_CONCURRENCY */ + #endif /* H5_HAVE_THREADSAFE_API */ diff --git a/src/H5TSint.c b/src/H5TSint.c index 7edc46f5ada..b99133151a4 100644 --- a/src/H5TSint.c +++ b/src/H5TSint.c @@ -138,6 +138,8 @@ H5TS__init_package(void) #else /* H5_HAVE_CONCURRENCY */ if (H5_UNLIKELY(H5TS_rwlock_init(&H5TS_api_info_p.api_lock) < 0)) HGOTO_DONE(FAIL); + if (H5_UNLIKELY(H5TS_mutex_init(&H5TS_api_info_p.internal_mutex, H5TS_MUTEX_TYPE_PLAIN) < 0)) + HGOTO_DONE(FAIL); #endif H5TS_atomic_init_uint(&H5TS_api_info_p.attempt_lock_count, 0); @@ -156,8 +158,7 @@ done: * all threads. * * Note: This function is currently registered via atexit() and is called - * AFTER H5_term_library(). H5TS_top_term_package() is called at library - * termination to clean up per-thread resources. + * AFTER H5_term_library(). * * Return: void * @@ -166,6 +167,8 @@ done: void H5TS_term_package(void) { + H5TS_tinfo_node_t *tinfo_node; + FUNC_ENTER_NOAPI_NOINIT_NOERR /* Reset global API lock info */ @@ -176,6 +179,14 @@ H5TS_term_package(void) #endif H5TS_atomic_destroy_uint(&H5TS_api_info_p.attempt_lock_count); + /* Check if info for thread has been created, free it if so */ + (void)H5TS_key_get_value(H5TS_thrd_info_key_g, (void **)&tinfo_node); + if (tinfo_node) + H5TS__tinfo_destroy(tinfo_node); + + /* Clean up per-thread library info */ + H5TS__tinfo_term(); + FUNC_LEAVE_NOAPI_VOID } /* end H5TS_term_package() */ @@ -446,6 +457,62 @@ done: FUNC_LEAVE_NOAPI_NAMECHECK_ONLY(ret_value) } /* H5TS_api_unlock */ +#ifdef H5_HAVE_CONCURRENCY +/*-------------------------------------------------------------------------- + * Function: H5TS_internal_lock + * + * Purpose: Acquire the internal mutex, which is meant for when + * internally concurrent code (threads spawned inside the + * library) enters a non-threadsafe section. + * + * Note: This is not currently a recursive lock, so the library must + * not spawn internal threads when recursively entering such a + * section while this mutex is locked. + * + * Return: Non-negative on success / Negative on failure + * + *-------------------------------------------------------------------------- + */ +herr_t +H5TS_internal_lock(void) +{ + herr_t ret_value = SUCCEED; + + FUNC_ENTER_NOAPI_NAMECHECK_ONLY + + /* Acquire the library's internal lock */ + if (H5_UNLIKELY(H5TS_mutex_lock(&H5TS_api_info_p.internal_mutex) < 0)) + HGOTO_DONE(FAIL); + +done: + FUNC_LEAVE_NOAPI_NAMECHECK_ONLY(ret_value) +} /* end H5TS_internal_lock() */ + +/*-------------------------------------------------------------------------- + * Function: H5TS_internal_unlock + * + * Purpose: Unlock the mutex locked by H5TS_internal_lock(). + * + * Return: Non-negative on success / Negative on failure + * + *-------------------------------------------------------------------------- + */ +herr_t +H5TS_internal_unlock(void) +{ + herr_t ret_value = SUCCEED; + + FUNC_ENTER_NOAPI_NAMECHECK_ONLY + + /* Release the library's internal lock */ + if (H5_UNLIKELY(H5TS_mutex_unlock(&H5TS_api_info_p.internal_mutex) < 0)) + HGOTO_DONE(FAIL); + +done: + FUNC_LEAVE_NOAPI_NAMECHECK_ONLY(ret_value) +} /* end H5TS_internal_unlock() */ +#endif + /*-------------------------------------------------------------------------- * Function: H5TS__tinfo_init * @@ -824,9 +891,8 @@ H5TS__tinfo_destroy(void *_tinfo_node) /*-------------------------------------------------------------------------- * Function: H5TS_top_term_package * - * Purpose: Terminate the threadlocal parts of the H5TS interface during library terminaton. - * - * Note: See H5TS_term_package for termination of the thread-global resources + * Purpose: Terminate the parts of the H5TS interface that can or must be done early during library + *terminaton. * * Return: Non-negative on success / Negative on failure * @@ -839,8 +905,13 @@ H5TS_top_term_package(void) FUNC_ENTER_NOAPI_NOINIT_NOERR - /* Clean up per-thread library info */ - H5TS__tinfo_term(); +#ifdef H5_HAVE_CONCURRENCY + /* Destroy global thread pool if it exists */ + if (H5TS_pool_g) { + (void)H5TS_pool_destroy(H5TS_pool_g); + H5TS_pool_g = NULL; + } +#endif /* H5_HAVE_CONCURRENCY */ FUNC_LEAVE_NOAPI(n) } diff --git a/src/H5TSmodule.h b/src/H5TSmodule.h index 6b15ddb35f0..619635dbe51 100644 --- a/src/H5TSmodule.h +++ b/src/H5TSmodule.h @@ -25,4 +25,11 @@ #define H5_MY_PKG H5TS #define H5_MY_PKG_INIT NO +/** + * \defgroup H5TS Thread Safety (H5TS) + * + * Use the functions in this module to manage threads in HDF5. + * + */ + #endif /* H5TSmodule_H */ diff --git a/src/H5TSpkg.h b/src/H5TSpkg.h index 0b5ab28558c..b6689289c04 100644 --- a/src/H5TSpkg.h +++ b/src/H5TSpkg.h @@ -55,6 +55,7 @@ typedef struct H5TS_api_info_t { #else /* H5_HAVE_CONCURRENCY */ /* API lock */ H5TS_rwlock_t api_lock; + H5TS_mutex_t internal_mutex; #endif /* Count of # of attempts to acquire API lock */ diff --git a/src/H5TSprivate.h b/src/H5TSprivate.h index 36a35de7823..9dd883b913b 100644 --- a/src/H5TSprivate.h +++ b/src/H5TSprivate.h @@ -25,6 +25,7 @@ #ifdef H5_HAVE_THREADSAFE_API /* Include package's public headers */ +#include "H5TSpublic.h" #include "H5TSdevelop.h" #endif /* H5_HAVE_THREADSAFE_API */ @@ -293,6 +294,14 @@ typedef atomic_flag H5TS_spinlock_t; /* Library-private Variables */ /*****************************/ +#ifdef H5_HAVE_CONCURRENCY +/* Global thread pool */ +extern H5TS_pool_t *H5TS_pool_g; + +/* Whether there are concurrent threads in the library (from internal spawning) */ +extern bool H5TS_currently_concurrent_g; +#endif /* H5_HAVE_CONCURRENCY */ + /***************************************/ /* Library-private Function Prototypes */ /***************************************/ @@ -311,11 +320,17 @@ H5_DLL herr_t H5TS_user_cb_restore(void); /* API locking */ #ifdef H5_HAVE_THREADSAFE H5_DLL herr_t H5TS_api_lock(void); -#else /* H5_HAVE_CONCURRENCY */ +#else /* H5_HAVE_THREADSAFE */ H5_DLL herr_t H5TS_api_lock(unsigned *dlftt); #endif H5_DLL herr_t H5TS_api_unlock(void); +/* Internal locking */ +#ifdef H5_HAVE_CONCURRENCY +H5_DLL herr_t H5TS_internal_lock(void); +H5_DLL herr_t H5TS_internal_unlock(void); +#endif /* H5_HAVE_CONCURRENCY */ + /* Retrieve per-thread info */ H5_DLL herr_t H5TS_thread_id(uint64_t *id); H5_DLL struct H5CX_node_t **H5TS_get_api_ctx_ptr(void); diff --git a/src/H5TSpublic.h b/src/H5TSpublic.h new file mode 100644 index 00000000000..c2a8fc06274 --- /dev/null +++ b/src/H5TSpublic.h @@ -0,0 +1,102 @@ +/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * + * Copyright by The HDF Group. * + * All rights reserved. * + * * + * This file is part of HDF5. The full HDF5 copyright notice, including * + * terms governing use, modification, and redistribution, is contained in * + * the LICENSE file, which can be found at the root of the source code * + * distribution tree, or in https://www.hdfgroup.org/licenses. * + * If you do not have access to either file, you may request a copy from * + * help@hdfgroup.org. * + * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ + +/* + * This file contains public declarations for the H5TS (threadsafety) module. + */ + +#ifndef H5TSpublic_H +#define H5TSpublic_H + +#include "H5public.h" /* Generic Functions */ + +/*****************/ +/* Public Macros */ +/*****************/ + +/*******************/ +/* Public Typedefs */ +/*******************/ + +/********************/ +/* Public Variables */ +/********************/ + +/*********************/ +/* Public Prototypes */ +/*********************/ + +#ifdef __cplusplus +extern "C" { +#endif + +/* HDF5 global thread pool routines */ + +#ifdef H5_HAVE_CONCURRENCY +/** + * \ingroup H5TS + * + * \brief Sets the number of internal threads to use for internal multithreading + * + * \param[in] num_threads Number of threads to use for internal multithreading + * + * \return \herr_t + * + * \details H5TSset_internal_threads() directs the HDF5 library to us + * \p num_threads threads to accelerate parallelizable operations. + * + * This is currently only used to accelerate read operations for + * chunked datasets that either have data filters applied or for which + * the chunks are small enough to fit in cache. In this case, the + * library parallelizes the reads from disk, the data filter + * operations, and the memory scatter operation. However, all of these + * operations are currently protected by a mutex so no performance gain + * is expected and this feature is purely experimental. These mutexes + * will be relaxed in the future to enable performance acceleration. + * + * Currently, the library will immediately create \p num_threads + * threads and retain them until this function is called again. Calling + * this function with \p num_threads set to \c 0 will terminate these + * threads and disable internal multithreading. + * + * This is currently only used to accelerate raw data reads of chunked + * datasets. This will occur when the following conditions are met: + * \li This function is called with \p num_threads > \c 0 . + * \li H5Pset_io_threads() was not called with \c threads_enabled set + * to \c false . + * \li Selection I/O is not used. See H5Pset_selection_io(). + * \li At least one chunk exists on disk and is not cached by the + * dataset chunk cache. + * \li For unfiltered datasets, the chunk cache is large enough to fit + * at least one chunk. See H5Pset_chunk_cache(). + * + * \note This function is only present when the library is compiled with HDF5_ENABLE_CONCURRENCY=ON. + * + * \note Use of this function may increase memory usage during concurrent + * operations due to the need to maintain state for each thread + * simultaneously. + * + * \warning Errors that are printed inside the threaded area, for example by the + * data filters, do not currently respect non-default error settings, + * and print their errors to stderr upon thread completion. + * + * \since 2.3.0 + * + */ +H5_DLL herr_t H5TSset_internal_threads(unsigned num_threads); +#endif /* H5_HAVE_CONCURRENCY */ + +#ifdef __cplusplus +} +#endif + +#endif /* H5TSpublic_H */ diff --git a/src/H5Z.c b/src/H5Z.c index 99106ed6552..bc344d3a475 100644 --- a/src/H5Z.c +++ b/src/H5Z.c @@ -1384,8 +1384,11 @@ H5Z_pipeline(const H5O_pline_t *pline, unsigned flags, unsigned *filter_mask /*i #endif unsigned failed = 0; unsigned tmp_flags; - size_t i; - herr_t ret_value = SUCCEED; /* Return value */ +#ifdef H5_HAVE_CONCURRENCY + bool mutex_held = false; +#endif /* H5_HAVE_CONCURRENCY */ + size_t i; + herr_t ret_value = SUCCEED; /* Return value */ FUNC_ENTER_NOAPI(FAIL) @@ -1409,6 +1412,17 @@ H5Z_pipeline(const H5O_pline_t *pline, unsigned flags, unsigned *filter_mask /*i continue; /* filter excluded */ } +#ifdef H5_UNSAFE_CONCURRENCY /* We currently take this lock before entering H5Z_pipeline(), and this is not a recursive lock, so don't take it again here */ +#ifdef H5_HAVE_CONCURRENCY + /* If we're using concurrent threads, lock the internal mutex to search for the plugin */ + if (H5TS_currently_concurrent_g) { + if (H5_UNLIKELY(H5TS_internal_lock() < 0)) + HGOTO_ERROR(H5E_PLINE, H5E_CANTLOCK, FAIL, "can't lock internal mutex"); + mutex_held = true; + } +#endif /* H5_HAVE_CONCURRENCY */ +#endif /* H5_UNSAFE_CONCURRENCY */ + /* If the filter isn't registered and the application doesn't * indicate no plugin through HDF5_PRELOAD_PLUG (using the symbol "::"), * try to load it dynamically and register it. Otherwise, return failure @@ -1446,6 +1460,19 @@ H5Z_pipeline(const H5O_pline_t *pline, unsigned flags, unsigned *filter_mask /*i } } /* end if */ +#ifdef H5_UNSAFE_CONCURRENCY +#ifdef H5_HAVE_CONCURRENCY + /* Unlock the internal mutex */ + if (H5TS_currently_concurrent_g) { + assert(mutex_held); + if (H5_UNLIKELY(H5TS_internal_unlock() < 0)) + HGOTO_ERROR(H5E_PLINE, H5E_CANTUNLOCK, FAIL, "can't unlock internal mutex"); + mutex_held = false; + } + assert(!mutex_held); +#endif /* H5_HAVE_CONCURRENCY */ +#endif /* H5_UNSAFE_CONCURRENCY */ + fclass = &H5Z_table_g[fclass_idx]; #ifdef H5Z_DEBUG @@ -1615,6 +1642,17 @@ H5Z_pipeline(const H5O_pline_t *pline, unsigned flags, unsigned *filter_mask /*i *filter_mask = failed; done: +#ifdef H5_HAVE_CONCURRENCY + /* Unlock the internal mutex */ + if (mutex_held) { + assert(H5TS_currently_concurrent_g); + assert(ret_value < 0); + if (H5TS_internal_unlock() < 0) + HDONE_ERROR(H5E_PLINE, H5E_CANTUNLOCK, FAIL, "can't unlock internal mutex"); + mutex_held = false; + } +#endif /* H5_HAVE_CONCURRENCY */ + FUNC_LEAVE_NOAPI(ret_value) /* clang-format on */ diff --git a/src/hdf5.h b/src/hdf5.h index 66dfb152a69..711ca25e5a8 100644 --- a/src/hdf5.h +++ b/src/hdf5.h @@ -37,6 +37,7 @@ #include "H5Rpublic.h" /* References */ #include "H5Spublic.h" /* Dataspaces */ #include "H5Tpublic.h" /* Datatypes */ +#include "H5TSpublic.h" /* Threadsafety */ #include "H5VLpublic.h" /* Virtual Object Layer */ #include "H5Zpublic.h" /* Data filters */ diff --git a/test/dsets.c b/test/dsets.c index 9ecb67c9f38..2cadea28dc2 100644 --- a/test/dsets.c +++ b/test/dsets.c @@ -82,6 +82,7 @@ static const char *FILENAME[] = {"dataset", /* 0 */ "chunk_expand2", /* 30 */ "scalar_datasets", /* 31 */ "read_only_vlen_fill", /* 32 */ + "threaded_chunks", /* 33 */ NULL}; #define OHMIN_FILENAME_A "ohdr_min_a" @@ -335,6 +336,9 @@ const char *OLD_FILENAME[] = { /* Declarations for test test_vds_shared_strings */ #define NUM_MAPPINGS_MANY 1000 +/* Whether to test with threads */ +unsigned threads = 0; + /* Local prototypes for filter functions */ static size_t filter_bogus(unsigned int flags, size_t cd_nelmts, const unsigned int *cd_values, size_t nbytes, size_t *buf_size, void **buf); @@ -2344,13 +2348,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (corrupted) { /* Default behavior is failure when data is corrupted. */ /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; /* Callback decides to continue in spite data is corrupted. */ if (H5Pset_filter_callback(dxpl, filter_cb_cont, NULL) < 0) @@ -2362,13 +2368,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (H5Pset_filter_callback(write_dxpl, filter_cb_fail, NULL) < 0) TEST_ERROR; /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; } else { if (H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, dxpl, check_data) < 0) @@ -2411,13 +2419,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (corrupted) { /* Default behavior is failure when data is corrupted. */ /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; /* Callback decides to continue in spite data is corrupted. */ if (H5Pset_filter_callback(dxpl, filter_cb_cont, NULL) < 0) @@ -2429,13 +2439,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (H5Pset_filter_callback(write_dxpl, filter_cb_fail, NULL) < 0) TEST_ERROR; /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; } else { /* Read the dataset back and check it */ @@ -2475,13 +2487,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (corrupted) { /* Default behavior is failure when data is corrupted. */ /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; /* Callback decides to continue in spite data is corrupted. */ if (H5Pset_filter_callback(dxpl, filter_cb_cont, NULL) < 0) @@ -2494,13 +2508,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, TEST_ERROR; /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; } /* end if */ else { if (H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, dxpl, check_data) < 0) @@ -2542,13 +2558,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (corrupted) { /* Default behavior is failure when data is corrupted. */ /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; /* Callback decides to continue in spite data is corrupted. */ if (H5Pset_filter_callback(dxpl, filter_cb_cont, NULL) < 0) @@ -2560,13 +2578,15 @@ test_filter_internal(hid_t fid, const char *name, hid_t dcpl, int if_fletcher32, if (H5Pset_filter_callback(write_dxpl, filter_cb_fail, NULL) < 0) TEST_ERROR; /* (Use the "write" DXPL in order to make certain corruption is seen) */ - H5E_BEGIN_TRY - { - status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + if (!threads) { + H5E_BEGIN_TRY + { + status = H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, write_dxpl, check_data); + } + H5E_END_TRY + if (status >= 0) + TEST_ERROR; } - H5E_END_TRY - if (status >= 0) - TEST_ERROR; } else { if (H5Dread(dataset, H5T_NATIVE_INT, sid, sid, dxpl, check_data) < 0) @@ -19408,6 +19428,221 @@ error: return FAIL; } /* end test_filter_error_msg() */ +#ifdef H5_HAVE_CONCURRENCY +/*------------------------------------------------------------------------- + * Function: test_threaded_chunks + * + * Purpose: Tests threading in simple chunk I/O. + * + * Return: Success: 0 + * Failure: -1 + *------------------------------------------------------------------------- + */ +#define THREADED_CHUNKS_CDIM1 10 +#define THREADED_CHUNKS_CDIM2 10 +static herr_t +test_threaded_chunks(void) +{ + char filename[FILENAME_BUF_SIZE] = ""; + hid_t file = H5I_INVALID_HID, dataset = H5I_INVALID_HID, dcpl = H5I_INVALID_HID, dapl = H5I_INVALID_HID, + space = H5I_INVALID_HID, xfer = H5I_INVALID_HID; + int i, j, n = 0; + hsize_t dims[2], cdims[2]; + unsigned pool_threads; + unsigned dxpl_setting; + unsigned cache_enabled; + + TESTING("threaded chunk I/O"); + + if (NULL == h5_fixname(FILENAME[33], H5P_DEFAULT, filename, sizeof(filename))) + TEST_ERROR; + + if ((file = H5Fcreate(filename, H5F_ACC_TRUNC, H5P_DEFAULT, H5P_DEFAULT)) < 0) + TEST_ERROR; + + /* Create the data space */ + dims[0] = DSET_DIM1; + dims[1] = DSET_DIM2; + if ((space = H5Screate_simple(2, dims, NULL)) < 0) + TEST_ERROR; + + /* Create DXPL, DCPL, and DAPL */ + if ((xfer = H5Pcreate(H5P_DATASET_XFER)) < 0) + TEST_ERROR; + if ((dcpl = H5Pcreate(H5P_DATASET_CREATE)) < 0) + TEST_ERROR; + if ((dapl = H5Pcreate(H5P_DATASET_ACCESS)) < 0) + TEST_ERROR; + + /* Set chunk dimensions */ + cdims[0] = THREADED_CHUNKS_CDIM1; + cdims[1] = THREADED_CHUNKS_CDIM2; + if (H5Pset_chunk(dcpl, 2, cdims) < 0) + TEST_ERROR; + + /* Disable chunk cache on dapl */ + if (H5Pset_chunk_cache(dapl, 0, 0, 0.) < 0) + TEST_ERROR; + + /* Create the dataset */ + if ((dataset = H5Dcreate2(file, "dset", H5T_NATIVE_INT, space, H5P_DEFAULT, dcpl, H5P_DEFAULT)) < 0) + TEST_ERROR; + if (H5Dclose(dataset) < 0) + TEST_ERROR; + dataset = H5I_INVALID_HID; + +#ifdef H5_HAVE_FILTER_DEFLATE + /* Create dataset with deflate filter */ + if (H5Pset_deflate(dcpl, 6) < 0) + TEST_ERROR; + if ((dataset = H5Dcreate2(file, "dset_deflate", H5T_NATIVE_INT, space, H5P_DEFAULT, dcpl, H5P_DEFAULT)) < + 0) + TEST_ERROR; + if (H5Dclose(dataset) < 0) + TEST_ERROR; + dataset = H5I_INVALID_HID; +#endif /* H5_HAVE_FILTER_DEFLATE */ + + if (H5Sclose(space) < 0) + TEST_ERROR; + space = H5I_INVALID_HID; + if (H5Pclose(dcpl) < 0) + TEST_ERROR; + dcpl = H5I_INVALID_HID; + + /* Loop over number of threads (0 = no thread pool) */ + for (pool_threads = 0; pool_threads <= 8; pool_threads++) { + /* Set number of threads */ + if (H5TSset_internal_threads(pool_threads) < 0) + TEST_ERROR; + + /* Loop over DXPL setting, 0 = off, 1 = on, 2 = use H5P_DEFAULT */ + for (dxpl_setting = 0; dxpl_setting <= 2; dxpl_setting++) { + /* Set DXPL IO threads and verify */ + if (dxpl_setting != 2) { + bool dxpl_setting_out; + + if (H5Pset_io_threads(xfer, (bool)dxpl_setting) < 0) + TEST_ERROR; + if (H5Pget_io_threads(xfer, &dxpl_setting_out) < 0) + TEST_ERROR; + if ((bool)dxpl_setting != dxpl_setting_out) + TEST_ERROR; + } + + /* Loop over chunk cache */ + for (cache_enabled = false; cache_enabled <= true; cache_enabled++) { + /* Initialize the write buffer */ + for (i = 0; i < DSET_DIM1; i++) + for (j = 0; j < DSET_DIM2; j++) + points[i][j] = n++; + + /* Open dataset */ + if ((dataset = H5Dopen2(file, "dset", cache_enabled ? H5P_DEFAULT : dapl)) < 0) + TEST_ERROR; + + /* Write the data to the dataset */ + if (H5Dwrite(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, + (dxpl_setting == 2) ? H5P_DEFAULT : xfer, points_data) < 0) + TEST_ERROR; + + /* Read the dataset back */ + if (H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, + (dxpl_setting == 2) ? H5P_DEFAULT : xfer, check_data) < 0) + TEST_ERROR; + + /* Check that the values read are the same as the values written */ + for (i = 0; i < DSET_DIM1; i++) { + for (j = 0; j < DSET_DIM2; j++) { + if (points[i][j] != check[i][j]) { + H5_FAILED(); + printf(" Read different values than written.\n"); + printf(" At index %d,%d\n", i, j); + goto error; + } + } + } + + /* Close dataset */ + if (H5Dclose(dataset) < 0) + TEST_ERROR; + dataset = H5I_INVALID_HID; + +#ifdef H5_HAVE_FILTER_DEFLATE + /* Initialize the write buffer */ + for (i = 0; i < DSET_DIM1; i++) + for (j = 0; j < DSET_DIM2; j++) + points[i][j] = n++; + + /* Open dataset with deflate filter */ + if ((dataset = H5Dopen2(file, "dset_deflate", cache_enabled ? H5P_DEFAULT : dapl)) < 0) + TEST_ERROR; + + /* Write the data to the dataset */ + if (H5Dwrite(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, + (dxpl_setting == 2) ? H5P_DEFAULT : xfer, points_data) < 0) + TEST_ERROR; + + /* Read the dataset back */ + if (H5Dread(dataset, H5T_NATIVE_INT, H5S_ALL, H5S_ALL, + (dxpl_setting == 2) ? H5P_DEFAULT : xfer, check_data) < 0) + TEST_ERROR; + + /* Check that the values read are the same as the values written */ + for (i = 0; i < DSET_DIM1; i++) { + for (j = 0; j < DSET_DIM2; j++) { + if (points[i][j] != check[i][j]) { + H5_FAILED(); + printf(" Read different values than written.\n"); + printf(" At index %d,%d\n", i, j); + goto error; + } + } + } + + /* Close dataset */ + if (H5Dclose(dataset) < 0) + TEST_ERROR; + dataset = H5I_INVALID_HID; +#endif /* H5_HAVE_FILTER_DEFLATE */ + } + } + } + + /* Disable threads */ + if (H5TSset_internal_threads(0) < 0) + TEST_ERROR; + + if (H5Pclose(xfer) < 0) + TEST_ERROR; + xfer = -1; + if (H5Pclose(dapl) < 0) + TEST_ERROR; + dapl = -1; + if (H5Fclose(file) < 0) + TEST_ERROR; + file = -1; + + PASSED(); + + return SUCCEED; + +error: + H5E_BEGIN_TRY + { + H5TSset_internal_threads(0); + H5Sclose(space); + H5Pclose(xfer); + H5Pclose(dcpl); + H5Pclose(dapl); + H5Dclose(dataset); + H5Fclose(file); + } + H5E_END_TRY + return FAIL; +} /* end test_threaded_chunks() */ +#endif /* H5_HAVE_CONCURRENCY */ + /*------------------------------------------------------------------------- * Function: main * @@ -19518,144 +19753,178 @@ main(void) for (low = H5F_LIBVER_EARLIEST; low < H5F_LIBVER_NBOUNDS; low++) { hid_t my_fcpl; - /* Set version bounds */ - if (H5Pset_libver_bounds(fapl, low, H5F_LIBVER_LATEST) < 0) - TEST_ERROR; +#ifdef H5_HAVE_CONCURRENCY + /* Test with and without threads */ + for (threads = false; threads <= true; threads++) { +#endif /* H5_HAVE_CONCURRENCY */ - /* Print partial message about file format */ - printf("\nTesting with %s file format ", h5_get_version_string(low)); + /* Print partial message about file format */ + printf("\nTesting with %s file format", h5_get_version_string(low)); - /* Set the FCPL and print the rest of the message depending on paged aggregation setting */ - if (paged) { - my_fcpl = fcpl2; - puts("and paged aggregation"); - } - else { - my_fcpl = fcpl; - puts("and non-paged aggregation"); - } +#ifdef H5_HAVE_CONCURRENCY + /* Set internal threading with 4 threads */ + if (threads) { + if (H5TSset_internal_threads(4) < 0) + goto error; - /* Create the file for this test */ - if ((file = H5Fcreate(filename, H5F_ACC_TRUNC, my_fcpl, fapl)) < 0) - goto error; + /* Print message about threads */ + printf(", with threads,"); + } + else + /* Print message about threads */ + printf(", without threads,"); +#endif /* H5_HAVE_CONCURRENCY */ - if (true == minimized_ohdr) { - if (0 > H5Fset_dset_no_attrs_hint(file, true)) + /* Set the FCPL and print the rest of the message depending on paged aggregation setting + */ + if (paged) { + my_fcpl = fcpl2; + puts(" and paged aggregation"); + } + else { + my_fcpl = fcpl; + puts(" and non-paged aggregation"); + } + + /* Set version bounds */ + if (H5Pset_libver_bounds(fapl, low, H5F_LIBVER_LATEST) < 0) + TEST_ERROR; + + /* Create the file for this test */ + if ((file = H5Fcreate(filename, H5F_ACC_TRUNC, my_fcpl, fapl)) < 0) goto error; - puts("(minimized dataset object headers with file setting)"); - } - /* Cause the library to emit initial messages */ - if ((grp = H5Gcreate2(file, "emit diagnostics", H5P_DEFAULT, H5P_DEFAULT, H5P_DEFAULT)) < 0) - goto error; - if (H5Oset_comment(grp, "Causes diagnostic messages to be emitted") < 0) - goto error; - if (H5Gclose(grp) < 0) - goto error; + if (true == minimized_ohdr) { + if (0 > H5Fset_dset_no_attrs_hint(file, true)) + goto error; + puts("(minimized dataset object headers with file setting)"); + } - nerrors += (test_create(file) < 0 ? 1 : 0); - nerrors += (test_simple_io(driver_name, fapl) < 0 ? 1 : 0); - nerrors += (test_scalar_io(fapl) < 0 ? 1 : 0); - nerrors += (test_compact_io(fapl) < 0 ? 1 : 0); - nerrors += (test_max_compact(fapl) < 0 ? 1 : 0); - nerrors += (test_compact_open_close_dirty(fapl) < 0 ? 1 : 0); - nerrors += (test_conv_buffer(file) < 0 ? 1 : 0); - nerrors += (test_tconv(file) < 0 ? 1 : 0); - nerrors += (test_filters(file) < 0 ? 1 : 0); - nerrors += (test_onebyte_shuffle(file) < 0 ? 1 : 0); - nerrors += (test_nbit_int(file) < 0 ? 1 : 0); - nerrors += (test_nbit_float(file) < 0 ? 1 : 0); - nerrors += (test_nbit_double(file) < 0 ? 1 : 0); - nerrors += (test_nbit_array(file) < 0 ? 1 : 0); - nerrors += (test_nbit_compound(file) < 0 ? 1 : 0); - nerrors += (test_nbit_compound_2(file) < 0 ? 1 : 0); - nerrors += (test_nbit_compound_3(file) < 0 ? 1 : 0); - nerrors += (test_nbit_int_size(file) < 0 ? 1 : 0); - nerrors += (test_nbit_flt_size(file) < 0 ? 1 : 0); - nerrors += (test_scaleoffset_int(file) < 0 ? 1 : 0); - nerrors += (test_scaleoffset_int_2(file) < 0 ? 1 : 0); - nerrors += (test_scaleoffset_float(file) < 0 ? 1 : 0); - nerrors += (test_scaleoffset_float_2(file) < 0 ? 1 : 0); - nerrors += (test_scaleoffset_double(file) < 0 ? 1 : 0); - nerrors += (test_scaleoffset_double_2(file) < 0 ? 1 : 0); - nerrors += (test_multiopen(file) < 0 ? 1 : 0); - nerrors += (test_types(file) < 0 ? 1 : 0); - nerrors += (test_floattypes(file) < 0 ? 1 : 0); - nerrors += (test_userblock_offset(driver_name, fapl, low >= H5F_LIBVER_V110) < 0 ? 1 : 0); + /* Cause the library to emit initial messages */ + if ((grp = H5Gcreate2(file, "emit diagnostics", H5P_DEFAULT, H5P_DEFAULT, H5P_DEFAULT)) < + 0) + goto error; + if (H5Oset_comment(grp, "Causes diagnostic messages to be emitted") < 0) + goto error; + if (H5Gclose(grp) < 0) + goto error; - if (driver_is_default_compatible) { - nerrors += (test_missing_filter(file) < 0 ? 1 : 0); - } + nerrors += (test_create(file) < 0 ? 1 : 0); + nerrors += (test_simple_io(driver_name, fapl) < 0 ? 1 : 0); + nerrors += (test_scalar_io(fapl) < 0 ? 1 : 0); + nerrors += (test_compact_io(fapl) < 0 ? 1 : 0); + nerrors += (test_max_compact(fapl) < 0 ? 1 : 0); + nerrors += (test_compact_open_close_dirty(fapl) < 0 ? 1 : 0); + nerrors += (test_conv_buffer(file) < 0 ? 1 : 0); + nerrors += (test_tconv(file) < 0 ? 1 : 0); + nerrors += (test_filters(file) < 0 ? 1 : 0); + nerrors += (test_onebyte_shuffle(file) < 0 ? 1 : 0); + nerrors += (test_nbit_int(file) < 0 ? 1 : 0); + nerrors += (test_nbit_float(file) < 0 ? 1 : 0); + nerrors += (test_nbit_double(file) < 0 ? 1 : 0); + nerrors += (test_nbit_array(file) < 0 ? 1 : 0); + nerrors += (test_nbit_compound(file) < 0 ? 1 : 0); + nerrors += (test_nbit_compound_2(file) < 0 ? 1 : 0); + nerrors += (test_nbit_compound_3(file) < 0 ? 1 : 0); + nerrors += (test_nbit_int_size(file) < 0 ? 1 : 0); + nerrors += (test_nbit_flt_size(file) < 0 ? 1 : 0); + nerrors += (test_scaleoffset_int(file) < 0 ? 1 : 0); + nerrors += (test_scaleoffset_int_2(file) < 0 ? 1 : 0); + nerrors += (test_scaleoffset_float(file) < 0 ? 1 : 0); + nerrors += (test_scaleoffset_float_2(file) < 0 ? 1 : 0); + nerrors += (test_scaleoffset_double(file) < 0 ? 1 : 0); + nerrors += (test_scaleoffset_double_2(file) < 0 ? 1 : 0); + nerrors += (test_multiopen(file) < 0 ? 1 : 0); + nerrors += (test_types(file) < 0 ? 1 : 0); + nerrors += (test_floattypes(file) < 0 ? 1 : 0); + nerrors += (test_userblock_offset(driver_name, fapl, low >= H5F_LIBVER_V110) < 0 ? 1 : 0); - nerrors += (test_can_apply(file) < 0 ? 1 : 0); - nerrors += (test_can_apply2(file) < 0 ? 1 : 0); - nerrors += (test_optional_filters(file) < 0 ? 1 : 0); - nerrors += (test_optional_filters_scalar(file) < 0 ? 1 : 0); - nerrors += (test_optional_filters_null(file) < 0 ? 1 : 0); - nerrors += (test_set_local(fapl) < 0 ? 1 : 0); - nerrors += (test_set_local_updates_cd(file) < 0 ? 1 : 0); - nerrors += (test_set_local_updates_cd_vlen(file) < 0 ? 1 : 0); - nerrors += (test_deflate_vlen(file) < 0 ? 1 : 0); - nerrors += (test_can_apply_szip(file) < 0 ? 1 : 0); - nerrors += (test_compare_dcpl(file) < 0 ? 1 : 0); - nerrors += (test_copy_dcpl(file, fapl) < 0 ? 1 : 0); - nerrors += (test_filter_delete(file) < 0 ? 1 : 0); + /* Don't test these with threads yet since H5E_BEGIN_TRY doesn't yet apply to threads and + * these tests will otherwise produce lots of irrelevant error messages */ +#ifdef H5_HAVE_CONCURRENCY + if (!threads) +#endif /* H5_HAVE_CONCURRENCY */ + { + nerrors += (test_missing_filter(file) < 0 ? 1 : 0); + nerrors += (test_bad_decode_size(file) < 0 ? 1 : 0); + nerrors += (test_bad_decode_size_vlen(file) < 0 ? 1 : 0); + nerrors += (test_bad_buf_size(file) < 0 ? 1 : 0); - if (driver_is_default_compatible) { - nerrors += (test_filters_endianess() < 0 ? 1 : 0); - nerrors += (test_chunk_dims_mismatch() < 0 ? 1 : 0); - } + /* This one will also fail because it doesn't detect the expected message in the main + * thread's stack (it is printed by the child thread) */ + nerrors += (test_filter_error_msg(file) < 0 ? 1 : 0); + } - nerrors += (test_zero_dims(file) < 0 ? 1 : 0); - nerrors += (test_missing_chunk(file) < 0 ? 1 : 0); - nerrors += (test_random_chunks(fapl) < 0 ? 1 : 0); + nerrors += (test_can_apply(file) < 0 ? 1 : 0); + nerrors += (test_can_apply2(file) < 0 ? 1 : 0); + nerrors += (test_optional_filters(file) < 0 ? 1 : 0); + nerrors += (test_optional_filters_scalar(file) < 0 ? 1 : 0); + nerrors += (test_optional_filters_null(file) < 0 ? 1 : 0); + nerrors += (test_set_local(fapl) < 0 ? 1 : 0); + nerrors += (test_set_local_updates_cd(file) < 0 ? 1 : 0); + nerrors += (test_set_local_updates_cd_vlen(file) < 0 ? 1 : 0); + nerrors += (test_deflate_vlen(file) < 0 ? 1 : 0); + nerrors += (test_can_apply_szip(file) < 0 ? 1 : 0); + nerrors += (test_compare_dcpl(file) < 0 ? 1 : 0); + nerrors += (test_copy_dcpl(file, fapl) < 0 ? 1 : 0); + nerrors += (test_filter_delete(file) < 0 ? 1 : 0); + + if (driver_is_default_compatible) { + nerrors += (test_filters_endianess() < 0 ? 1 : 0); + nerrors += (test_chunk_dims_mismatch() < 0 ? 1 : 0); + } + + nerrors += (test_zero_dims(file) < 0 ? 1 : 0); + nerrors += (test_missing_chunk(file) < 0 ? 1 : 0); + nerrors += (test_random_chunks(fapl) < 0 ? 1 : 0); #ifndef H5_NO_DEPRECATED_SYMBOLS - nerrors += (test_deprec(file) < 0 ? 1 : 0); + nerrors += (test_deprec(file) < 0 ? 1 : 0); #endif /* H5_NO_DEPRECATED_SYMBOLS */ - nerrors += (test_huge_chunks(fapl, low) < 0 ? 1 : 0); - nerrors += (test_chunk_cache(fapl) < 0 ? 1 : 0); - nerrors += (test_big_chunks_bypass_cache(fapl) < 0 ? 1 : 0); - nerrors += (test_chunk_fast(driver_name, fapl) < 0 ? 1 : 0); - nerrors += (test_reopen_chunk_fast(fapl) < 0 ? 1 : 0); - nerrors += (test_chunk_fast_bug1(fapl) < 0 ? 1 : 0); - if (low >= H5F_LIBVER_V200) - nerrors += (test_chunk_expand2(fapl) < 0 ? 1 : 0); - else - nerrors += (test_chunk_expand(fapl) < 0 ? 1 : 0); - nerrors += (test_layout_extend(fapl) < 0 ? 1 : 0); - nerrors += (test_fixed_array(fapl) < 0 ? 1 : 0); + nerrors += (test_huge_chunks(fapl, low) < 0 ? 1 : 0); + nerrors += (test_chunk_cache(fapl) < 0 ? 1 : 0); + nerrors += (test_big_chunks_bypass_cache(fapl) < 0 ? 1 : 0); + nerrors += (test_chunk_fast(driver_name, fapl) < 0 ? 1 : 0); + nerrors += (test_reopen_chunk_fast(fapl) < 0 ? 1 : 0); + nerrors += (test_chunk_fast_bug1(fapl) < 0 ? 1 : 0); + if (low >= H5F_LIBVER_V200) + nerrors += (test_chunk_expand2(fapl) < 0 ? 1 : 0); + else + nerrors += (test_chunk_expand(fapl) < 0 ? 1 : 0); + nerrors += (test_layout_extend(fapl) < 0 ? 1 : 0); + nerrors += (test_fixed_array(fapl) < 0 ? 1 : 0); - if (driver_is_default_compatible) { - nerrors += (test_idx_compatible() < 0 ? 1 : 0); + if (driver_is_default_compatible) { + nerrors += (test_idx_compatible() < 0 ? 1 : 0); + } + + nerrors += (test_unfiltered_edge_chunks(fapl) < 0 ? 1 : 0); + nerrors += (test_single_chunk(fapl) < 0 ? 1 : 0); + nerrors += (test_large_chunk_shrink(fapl) < 0 ? 1 : 0); + nerrors += (test_zero_dim_dset(fapl) < 0 ? 1 : 0); + nerrors += (test_storage_size(fapl) < 0 ? 1 : 0); + nerrors += (test_power2up(fapl) < 0 ? 1 : 0); + + nerrors += (test_swmr_non_latest(driver_name, fapl) < 0 ? 1 : 0); + nerrors += (test_earray_hdr_fd(driver_name, fapl) < 0 ? 1 : 0); + nerrors += (test_farray_hdr_fd(driver_name, fapl) < 0 ? 1 : 0); + nerrors += (test_bt2_hdr_fd(driver_name, fapl) < 0 ? 1 : 0); + + nerrors += (test_downsize_vlen_scalar_dataset(file) < 0 ? 1 : 0); + + nerrors += (test_readonly_chunk_vlen_fill(fapl, false) < 0 ? 1 : 0); + nerrors += (test_readonly_chunk_vlen_fill(fapl, true) < 0 ? 1 : 0); + + if (H5Fclose(file) < 0) + goto error; +#ifdef H5_HAVE_CONCURRENCY + /* Disable internal threading */ + if (threads && H5TSset_internal_threads(0) < 0) + goto error; } +#endif /* H5_HAVE_CONCURRENCY */ - nerrors += (test_unfiltered_edge_chunks(fapl) < 0 ? 1 : 0); - nerrors += (test_single_chunk(fapl) < 0 ? 1 : 0); - nerrors += (test_large_chunk_shrink(fapl) < 0 ? 1 : 0); - nerrors += (test_zero_dim_dset(fapl) < 0 ? 1 : 0); - nerrors += (test_storage_size(fapl) < 0 ? 1 : 0); - nerrors += (test_power2up(fapl) < 0 ? 1 : 0); - - nerrors += (test_swmr_non_latest(driver_name, fapl) < 0 ? 1 : 0); - nerrors += (test_earray_hdr_fd(driver_name, fapl) < 0 ? 1 : 0); - nerrors += (test_farray_hdr_fd(driver_name, fapl) < 0 ? 1 : 0); - nerrors += (test_bt2_hdr_fd(driver_name, fapl) < 0 ? 1 : 0); - - nerrors += (test_downsize_vlen_scalar_dataset(file) < 0 ? 1 : 0); - - nerrors += (test_bad_decode_size(file) < 0 ? 1 : 0); - nerrors += (test_bad_decode_size_vlen(file) < 0 ? 1 : 0); - nerrors += (test_bad_buf_size(file) < 0 ? 1 : 0); - - nerrors += (test_readonly_chunk_vlen_fill(fapl, false) < 0 ? 1 : 0); - nerrors += (test_readonly_chunk_vlen_fill(fapl, true) < 0 ? 1 : 0); - - nerrors += (test_filter_error_msg(file) < 0 ? 1 : 0); - - if (H5Fclose(file) < 0) - goto error; } /* end for low */ } /* end for minimized_ohdr */ } /* end for paged */ @@ -19681,6 +19950,9 @@ main(void) nerrors += (test_object_header_minimization_dcpl() < 0 ? 1 : 0); nerrors += (test_h5s_block() < 0 ? 1 : 0); nerrors += (test_h5s_plist() < 0 ? 1 : 0); +#ifdef H5_HAVE_CONCURRENCY + nerrors += (test_threaded_chunks() < 0 ? 1 : 0); +#endif /* H5_HAVE_CONCURRENCY */ /* Run misc tests */ nerrors += (dls_01_main() < 0 ? 1 : 0); @@ -19698,6 +19970,12 @@ main(void) /* Verify that source file/dataset names are shared properly */ nerrors += (test_vds_shared_strings(fapl) < 0 ? 1 : 0); +#ifdef H5_HAVE_CONCURRENCY + /* Set threading now to ensure the library can shut down cleanly with threading enabled */ + if (H5TSset_internal_threads(4) < 0) + TEST_ERROR; +#endif /* H5_HAVE_CONCURRENCY */ + if (nerrors) goto error; printf("All dataset tests passed.\n"); diff --git a/test/dtransform.c b/test/dtransform.c index 6c860881297..702005cbf32 100644 --- a/test/dtransform.c +++ b/test/dtransform.c @@ -329,100 +329,116 @@ main(void) /* inverses the utrans transform in init_test to get back original array */ const char *utrans_inv = "(x/3 - 25)*4"; - if ((file_id = H5Fcreate("dtransform.h5", H5F_ACC_TRUNC, H5P_DEFAULT, H5P_DEFAULT)) < 0) - TEST_ERROR; +#ifdef H5_HAVE_CONCURRENCY + /* Test with and without threads */ + for (unsigned threads = false; threads <= true; threads++) { + if (threads && H5TSset_internal_threads(4) < 0) + TEST_ERROR; +#endif /* H5_HAVE_CONCURRENCY */ - if ((dxpl_id_c_to_f = H5Pcreate(H5P_DATASET_XFER)) < 0) - TEST_ERROR; - if ((dxpl_id_simple = H5Pcreate(H5P_DATASET_XFER)) < 0) - TEST_ERROR; - if ((dxpl_id_utrans_inv = H5Pcreate(H5P_DATASET_XFER)) < 0) - TEST_ERROR; - if ((dxpl_id_polynomial = H5Pcreate(H5P_DATASET_XFER)) < 0) - TEST_ERROR; - if (H5Pset_data_transform(dxpl_id_c_to_f, c_to_f) < 0) - TEST_ERROR; - if (H5Pset_data_transform(dxpl_id_polynomial, polynomial) < 0) - TEST_ERROR; - if (H5Pset_data_transform(dxpl_id_simple, simple) < 0) - TEST_ERROR; - if (H5Pset_data_transform(dxpl_id_utrans_inv, utrans_inv) < 0) - TEST_ERROR; - if ((dxpl_id_polynomial_copy = H5Pcopy(dxpl_id_polynomial)) < 0) - TEST_ERROR; - if ((dxpl_id_c_to_f_copy = H5Pcopy(dxpl_id_c_to_f)) < 0) - TEST_ERROR; + if ((file_id = H5Fcreate("dtransform.h5", H5F_ACC_TRUNC, H5P_DEFAULT, H5P_DEFAULT)) < 0) + TEST_ERROR; - /* Run all the tests */ + if ((dxpl_id_c_to_f = H5Pcreate(H5P_DATASET_XFER)) < 0) + TEST_ERROR; + if ((dxpl_id_simple = H5Pcreate(H5P_DATASET_XFER)) < 0) + TEST_ERROR; + if ((dxpl_id_utrans_inv = H5Pcreate(H5P_DATASET_XFER)) < 0) + TEST_ERROR; + if ((dxpl_id_polynomial = H5Pcreate(H5P_DATASET_XFER)) < 0) + TEST_ERROR; + if (H5Pset_data_transform(dxpl_id_c_to_f, c_to_f) < 0) + TEST_ERROR; + if (H5Pset_data_transform(dxpl_id_polynomial, polynomial) < 0) + TEST_ERROR; + if (H5Pset_data_transform(dxpl_id_simple, simple) < 0) + TEST_ERROR; + if (H5Pset_data_transform(dxpl_id_utrans_inv, utrans_inv) < 0) + TEST_ERROR; + if ((dxpl_id_polynomial_copy = H5Pcopy(dxpl_id_polynomial)) < 0) + TEST_ERROR; + if ((dxpl_id_c_to_f_copy = H5Pcopy(dxpl_id_c_to_f)) < 0) + TEST_ERROR; - if (init_test(file_id) < 0) - TEST_ERROR; - if (test_set() < 0) - TEST_ERROR; - TEST_TYPE_CONTIG(dxpl_id_utrans_inv, char, H5T_NATIVE_CHAR, "char", transformData, 0); - TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned char, H5T_NATIVE_UCHAR, "uchar", transformData, 0); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, signed char, H5T_NATIVE_SCHAR, "schar", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, short, H5T_NATIVE_SHORT, "short", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned short, H5T_NATIVE_USHORT, "ushort", transformData, 0); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, int, H5T_NATIVE_INT, "int", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned int, H5T_NATIVE_UINT, "uint", transformData, 0); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, long, H5T_NATIVE_LONG, "long", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned long, H5T_NATIVE_ULONG, "ulong", transformData, 0); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, long long, H5T_NATIVE_LLONG, "llong", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned long long, H5T_NATIVE_ULLONG, "ullong", transformData, 0); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, float, H5T_NATIVE_FLOAT, "float", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, double, H5T_NATIVE_DOUBLE, "double", windchillFfloat, 1); - TEST_TYPE_CONTIG(dxpl_id_c_to_f, long double, H5T_NATIVE_LDOUBLE, "ldouble", windchillFfloat, 1); + /* Run all the tests */ - TEST_TYPE_CHUNK(dxpl_id_utrans_inv, char, H5T_NATIVE_CHAR, "char", transformData, 0); - TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned char, H5T_NATIVE_UCHAR, "uchar", transformData, 0); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, signed char, H5T_NATIVE_SCHAR, "schar", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, short, H5T_NATIVE_SHORT, "short", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned short, H5T_NATIVE_USHORT, "ushort", transformData, 0); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, int, H5T_NATIVE_INT, "int", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned int, H5T_NATIVE_UINT, "uint", transformData, 0); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, long, H5T_NATIVE_LONG, "long", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned long, H5T_NATIVE_ULONG, "ulong", transformData, 0); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, long long, H5T_NATIVE_LLONG, "llong", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned long long, H5T_NATIVE_ULLONG, "ullong", transformData, 0); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, float, H5T_NATIVE_FLOAT, "float", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, double, H5T_NATIVE_DOUBLE, "double", windchillFfloat, 1); - TEST_TYPE_CHUNK(dxpl_id_c_to_f, long double, H5T_NATIVE_LDOUBLE, "ldouble", windchillFfloat, 1); + if (init_test(file_id) < 0) + TEST_ERROR; + if (test_set() < 0) + TEST_ERROR; + TEST_TYPE_CONTIG(dxpl_id_utrans_inv, char, H5T_NATIVE_CHAR, "char", transformData, 0); + TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned char, H5T_NATIVE_UCHAR, "uchar", transformData, 0); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, signed char, H5T_NATIVE_SCHAR, "schar", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, short, H5T_NATIVE_SHORT, "short", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned short, H5T_NATIVE_USHORT, "ushort", transformData, 0); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, int, H5T_NATIVE_INT, "int", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned int, H5T_NATIVE_UINT, "uint", transformData, 0); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, long, H5T_NATIVE_LONG, "long", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned long, H5T_NATIVE_ULONG, "ulong", transformData, 0); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, long long, H5T_NATIVE_LLONG, "llong", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_utrans_inv, unsigned long long, H5T_NATIVE_ULLONG, "ullong", transformData, + 0); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, float, H5T_NATIVE_FLOAT, "float", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, double, H5T_NATIVE_DOUBLE, "double", windchillFfloat, 1); + TEST_TYPE_CONTIG(dxpl_id_c_to_f, long double, H5T_NATIVE_LDOUBLE, "ldouble", windchillFfloat, 1); - if (test_copy(dxpl_id_c_to_f_copy, dxpl_id_polynomial_copy) < 0) - TEST_ERROR; - if (test_trivial(dxpl_id_simple) < 0) - TEST_ERROR; - if (test_poly(dxpl_id_polynomial) < 0) - TEST_ERROR; - if (test_getset(dxpl_id_c_to_f) < 0) - TEST_ERROR; - if (test_specials(file_id) < 0) - TEST_ERROR; + TEST_TYPE_CHUNK(dxpl_id_utrans_inv, char, H5T_NATIVE_CHAR, "char", transformData, 0); + TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned char, H5T_NATIVE_UCHAR, "uchar", transformData, 0); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, signed char, H5T_NATIVE_SCHAR, "schar", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, short, H5T_NATIVE_SHORT, "short", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned short, H5T_NATIVE_USHORT, "ushort", transformData, 0); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, int, H5T_NATIVE_INT, "int", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned int, H5T_NATIVE_UINT, "uint", transformData, 0); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, long, H5T_NATIVE_LONG, "long", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned long, H5T_NATIVE_ULONG, "ulong", transformData, 0); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, long long, H5T_NATIVE_LLONG, "llong", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_utrans_inv, unsigned long long, H5T_NATIVE_ULLONG, "ullong", transformData, + 0); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, float, H5T_NATIVE_FLOAT, "float", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, double, H5T_NATIVE_DOUBLE, "double", windchillFfloat, 1); + TEST_TYPE_CHUNK(dxpl_id_c_to_f, long double, H5T_NATIVE_LDOUBLE, "ldouble", windchillFfloat, 1); - /* Close the objects we opened/created */ - if (H5Dclose(dset_id_int) < 0) - TEST_ERROR; - if (H5Dclose(dset_id_int_chunk) < 0) - TEST_ERROR; - if (H5Dclose(dset_id_float) < 0) - TEST_ERROR; - if (H5Dclose(dset_id_float_chunk) < 0) - TEST_ERROR; - if (H5Fclose(file_id) < 0) - TEST_ERROR; - if (H5Pclose(dxpl_id_c_to_f) < 0) - TEST_ERROR; - if (H5Pclose(dxpl_id_c_to_f_copy) < 0) - TEST_ERROR; - if (H5Pclose(dxpl_id_polynomial) < 0) - TEST_ERROR; - if (H5Pclose(dxpl_id_polynomial_copy) < 0) - TEST_ERROR; - if (H5Pclose(dxpl_id_simple) < 0) - TEST_ERROR; - if (H5Pclose(dxpl_id_utrans_inv) < 0) - TEST_ERROR; + if (test_copy(dxpl_id_c_to_f_copy, dxpl_id_polynomial_copy) < 0) + TEST_ERROR; + if (test_trivial(dxpl_id_simple) < 0) + TEST_ERROR; + if (test_poly(dxpl_id_polynomial) < 0) + TEST_ERROR; + if (test_getset(dxpl_id_c_to_f) < 0) + TEST_ERROR; + if (test_specials(file_id) < 0) + TEST_ERROR; + + /* Close the objects we opened/created */ + if (H5Dclose(dset_id_int) < 0) + TEST_ERROR; + if (H5Dclose(dset_id_int_chunk) < 0) + TEST_ERROR; + if (H5Dclose(dset_id_float) < 0) + TEST_ERROR; + if (H5Dclose(dset_id_float_chunk) < 0) + TEST_ERROR; + if (H5Fclose(file_id) < 0) + TEST_ERROR; + if (H5Pclose(dxpl_id_c_to_f) < 0) + TEST_ERROR; + if (H5Pclose(dxpl_id_c_to_f_copy) < 0) + TEST_ERROR; + if (H5Pclose(dxpl_id_polynomial) < 0) + TEST_ERROR; + if (H5Pclose(dxpl_id_polynomial_copy) < 0) + TEST_ERROR; + if (H5Pclose(dxpl_id_simple) < 0) + TEST_ERROR; + if (H5Pclose(dxpl_id_utrans_inv) < 0) + TEST_ERROR; + +#ifdef H5_HAVE_CONCURRENCY + /* Disable internal threading */ + if (threads && H5TSset_internal_threads(0) < 0) + goto error; + } +#endif /* H5_HAVE_CONCURRENCY */ return 0; diff --git a/test/vds.c b/test/vds.c index b17cbd08ae1..1887240363e 100644 --- a/test/vds.c +++ b/test/vds.c @@ -95,7 +95,12 @@ char vds_test_str_g[128] = ""; #define TEST_IO_REOPEN_VIRT 0x04U #define TEST_IO_FCLOSE_SEMI 0x08U #define TEST_IO_FCLOSE_STRONG 0x10U -#define TEST_IO_NTESTS 0x20U +#ifdef H5_HAVE_CONCURRENCY +#define TEST_IO_THREADS 0x20U +#define TEST_IO_NTESTS 0x40U +#else /* H5_HAVE_CONCURRENCY */ +#define TEST_IO_NTESTS 0x20U +#endif /* H5_HAVE_CONCURRENCY */ #define LIST_DOUBLE_SIZE (H5D_VIRTUAL_DEF_LIST_SIZE + 1) @@ -12397,7 +12402,18 @@ main(void) if ((bit_config & TEST_IO_FCLOSE_SEMI) && (bit_config & TEST_IO_FCLOSE_STRONG)) continue; - /* Print message */ + /* Print message */ +#ifdef H5_HAVE_CONCURRENCY + PRINT_CONFIG( + "%s%s%s, %s file close degree%s", + bit_config & TEST_IO_CLOSE_SRC ? "closed source dataset, " : "", + bit_config & TEST_IO_DIFFERENT_FILE ? "different source file" : "same source file", + bit_config & TEST_IO_REOPEN_VIRT ? ", reopen virtual file" : "", + bit_config & TEST_IO_FCLOSE_SEMI + ? "H5F_CLOSE_SEMI" + : (bit_config & TEST_IO_FCLOSE_STRONG ? "H5F_CLOSE_STRONG" : "H5F_CLOSE_WEAK"), + bit_config & TEST_IO_THREADS ? ", threads enabled" : ", threads disabled"); +#else /* H5_HAVE_CONCURRENCY */ PRINT_CONFIG( "%s%s%s, %s file close degree", bit_config & TEST_IO_CLOSE_SRC ? "closed source dataset, " : "", @@ -12406,6 +12422,7 @@ main(void) bit_config & TEST_IO_FCLOSE_SEMI ? "H5F_CLOSE_SEMI" : (bit_config & TEST_IO_FCLOSE_STRONG ? "H5F_CLOSE_STRONG" : "H5F_CLOSE_WEAK")); +#endif /* H5_HAVE_CONCURRENCY */ /* Set file close degree */ if (bit_config & TEST_IO_FCLOSE_SEMI) { @@ -12421,12 +12438,24 @@ main(void) TEST_ERROR; } +#ifdef H5_HAVE_CONCURRENCY + /* Enable threads if part of configuration */ + if ((bit_config & TEST_IO_THREADS) && H5TSset_internal_threads(4) < 0) + TEST_ERROR; +#endif /* H5_HAVE_CONCURRENCY */ + /* Run tests */ nerrors += test_basic_io(bit_config, vds_fapl, src_fapl); nerrors += test_vds_prefix_first(bit_config, vds_fapl, src_fapl); nerrors += test_unlim(bit_config, vds_fapl, src_fapl); nerrors += test_printf(bit_config, vds_fapl, src_fapl); nerrors += test_all(bit_config, vds_fapl, src_fapl); + +#ifdef H5_HAVE_CONCURRENCY + /* Disable threads if appropriate */ + if ((bit_config & TEST_IO_THREADS) && H5TSset_internal_threads(0) < 0) + TEST_ERROR; +#endif /* H5_HAVE_CONCURRENCY */ } #ifndef VDS_TEST_VERBOSE
High-level Reference Manual Modules
Sets/gets a flag allowing the library to modify the contents of the write buffer.
#H5Pset_io_threads/#H5Pget_io_threadsSets/gets a flag allowing the library to use the global thread pool to accelerate I/O.
H5Pset_preserve/H5Pget_preserve No longer available, deprecated as it no longer has any effect.