mirrors/hdf5
2
0
Fork 0
mirror of https://github.com/HDFGroup/hdf5.git synced 2026-09-25 04:09:44 +03:00
Code Issues Packages Projects Releases Wiki Activity

Implement internally concurrent multithreading for chunk dataset I/O reads (#6645)

Browse Source 
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.
This commit is contained in:
Neil Fortner
2026-09-15 10:51:57 -05:00
committed by GitHub
parent 7489f285be
commit 1d946c7028
28 changed files with 1799 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/INSTALL_CMake_options.md
+1 -1
View File
@@ -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. |
docs/doxygen/CMakeLists.txt
+1
View File
@@ -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
docs/doxygen/dox/ReferenceManual.dox
+4
View File
@@ -84,6 +84,10 @@ The functions provided by the HDF5 API are grouped into the following
<th>VOL Connector (H5VL)</th><td>@ref H5VL "C"</td><td style="white-space: nowrap;">C++</td><td>@ref FH5VL "Fortran"</td><td>@ref JH5VL "Java"</td><td>Manage HDF5 VOL connector plugins.
</td>
</tr>
<tr>
<th>Thread Safety (H5TS)</th><td>@ref H5TS "C"</td><td style="white-space: nowrap;">C++</td><td>Fortran</td><td>Java</td><td>Threadsafety and threading-related operations.
</td>
</tr>
</table>
<table>
<caption>High-level Reference Manual Modules</caption>
docs/doxygen/examples/tables/propertyLists.dox
+4
View File
@@ -723,6 +723,10 @@ of the library for reading or writing the actual data.</td>
<td>Sets/gets a flag allowing the library to modify the contents of the write buffer.</td>
</tr>
<tr>
<td>#H5Pset_io_threads/#H5Pget_io_threads</td>
<td>Sets/gets a flag allowing the library to use the global thread pool to accelerate I/O.</td>
</tr>
<tr>
<td>H5Pset_preserve/H5Pget_preserve</td>
<td>No longer available, deprecated as it no longer has any effect.</td>
</tr>
release_docs/CHANGELOG.md
+15
View File
@@ -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
src/CMakeLists.txt
+1
View File
@@ -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}" )
src/H5.c
+3
View File
@@ -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 */
src/H5CX.c
+44
View File
@@ -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
*
src/H5CXprivate.h
+7
View File
@@ -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);
src/H5Dchunk.c
+678 -127
View File
@@ -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;
src/H5Dmpio.c
-3
View File
@@ -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;
src/H5Dpkg.h
+4 -8
View File
@@ -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);
src/H5Dprivate.h
+3
View File
@@ -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"
src/H5E.c
+5
View File
@@ -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 */
src/H5Eint.c
+5
View File
@@ -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 */
src/H5Pdxpl.c
+87
View File
@@ -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 */
src/H5Ppublic.h
+48
View File
@@ -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
*
src/H5TS.c
+63
View File
@@ -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 */
src/H5TSint.c
+78 -7
View File
@@ -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)
}
src/H5TSmodule.h
+7
View File
@@ -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 */
src/H5TSpkg.h
+1
View File
@@ -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 */
src/H5TSprivate.h
+16 -1
View File
@@ -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);
src/H5TSpublic.h
+102
View File
@@ -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 */
src/H5Z.c
+40 -2
View File
@@ -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 */
src/hdf5.h
+1
View File
@@ -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 */
test/dsets.c
+446 -168
View File
@@ -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");
test/dtransform.c
+104 -88
View File
@@ -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;
test/vds.c
+31 -2
View File
@@ -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