Skip to content

Misc async thread fixes #697

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jayeshkrishna wants to merge 16 commits into
base: master
Choose a base branch
from
Draft

Misc async thread fixes #697

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
aad0104
Get num threads from pool manager
jayeshkrishna May 18, 2026
0688592
Using tpool manager to get num threads
jayeshkrishna May 18, 2026
6de9320
Disable waits for hdf5 ops on close
jayeshkrishna May 22, 2026
a4f7e90
Disable wait for pending ops on open
jayeshkrishna May 22, 2026
66bcfd9
Avoid duping unused comms for threads
jayeshkrishna Jun 4, 2026
6ab9044
Code reformatting pio_err()
jayeshkrishna Jun 5, 2026
20209c2
Updating err print logic for NULL global comms
jayeshkrishna Jun 5, 2026
ca7317c
Fix barrier on global comm in thread
jayeshkrishna Jun 5, 2026
90f0cb7
When closing file check thread-specific comm
jayeshkrishna Jun 6, 2026
cf7c025
When closing file use return error handler
jayeshkrishna Jun 6, 2026
ef48f4d
Test creates mahy decomps and files
jayeshkrishna Jun 6, 2026
facf7f0
Adding a simple stdout logger
jayeshkrishna Jun 7, 2026
17de686
Adding config opt to trace I/O decomps
jayeshkrishna Jun 7, 2026
89dd7dc
Wait on writes before read
jayeshkrishna Jun 30, 2026
95739d9
Adding missing static keyword for mtx
jayeshkrishna Jul 11, 2026
cdb4d19
Test using many iodecomps with many iotypes
jayeshkrishna Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ option (PIO_ENABLE_INTERNAL_TIMING "Gather and print GPTL timing stats" OFF)
option (PIO_ENABLE_API_TIMING "Enable timer for all (costly) SCORPIO APIs (requires PIO_ENABLE_TIMING=ON)" ON)
option (PIO_ENABLE_API_TRACING "Enable tracing of SCORPIO APIs" OFF)
option (PIO_ENABLE_API_VAR_TRACING "Enable tracing of variables when tracing SCORPIO APIs" OFF)
option (PIO_ENABLE_API_DECOMP_TRACING "Enable tracing of I/O decompositions when tracing SCORPIO APIs" OFF)
option (PIO_ENABLE_IO_STATS "Gather and print I/O performance stats" ON)
option (PIO_ENABLE_LOGGING "Enable debug logging (large output possible)" OFF)
option (PIO_ENABLE_DOC "Enable building SCORPIO documentation" ON)
Expand Down
8 changes: 8 additions & 0 deletions src/clib/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -438,10 +438,18 @@ if (PIO_ENABLE_API_TRACING)
message(STATUS "Disabling variable tracing in APIs (default)")
set(ENABLE_API_VAR_TRACING 0)
endif ()
if (PIO_ENABLE_API_DECOMP_TRACING)
message(STATUS "Enabling I/O decomposition tracing in APIs")
set(ENABLE_API_DECOMP_TRACING 1)
else ()
message(STATUS "Disabling I/O decomposition tracing in APIs (default)")
set(ENABLE_API_DECOMP_TRACING 0)
endif ()
else ()
message(STATUS "API tracing disabled (default)")
set(ENABLE_API_TRACING 0)
set(ENABLE_API_VAR_TRACING 0)
set(ENABLE_API_DECOMP_TRACING 0)
endif ()

# The MPI library detection was done in the top level
Expand Down
4 changes: 4 additions & 0 deletions src/clib/api/spio_io_decomp_api.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,9 @@ int PIOc_InitDecomp(int iosysid, int pio_type, int ndims, const int *gdimlen, in
compmap, ioidp, rearr, iostart, iocount);

#if SPIO_ENABLE_API_TRACING
#if SPIO_ENABLE_API_DECOMP_TRACING
tr.set_decomp_info((ioidp) ? (*ioidp) : -1, compmap, maplen);
#endif
tr.add_rval("*ioidp", (ioidp) ? (*ioidp) : -1);
#endif
return ret;
Expand Down Expand Up @@ -80,7 +82,9 @@ int PIOc_init_decomp(int iosysid, int pio_type, int ndims, const int *gdimlen, i
compmap, ioidp, rearranger, iostart, iocount);

#if SPIO_ENABLE_API_TRACING
#if SPIO_ENABLE_API_DECOMP_TRACING
tr.set_decomp_info((ioidp) ? (*ioidp) : -1, compmap, maplen);
#endif
tr.add_rval("*ioidp", (ioidp) ? (*ioidp) : -1);
#endif
return ret;
Expand Down
18 changes: 17 additions & 1 deletion src/clib/core/iolib/hdf5/spio_async_hdf5_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ extern "C"{
#include "spio_dbg_utils.hpp"
#include "spio_dt_converter.hpp"
#include "spio_hdf5_utils.hpp"
#include "spio_async_tcomm.hpp"

struct Hdf5_create_info{
file_desc_t *file;
Expand Down Expand Up @@ -158,6 +159,7 @@ static inline void update_reg_infos_start_frame(std::vector<Util::RInfo> &reg_in

void spio_iosys_async_op_hdf5_create_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
if(pdata){
delete(static_cast<Hdf5_create_info *>(pdata));
}
Expand All @@ -170,6 +172,7 @@ int spio_iosys_async_op_hdf5_create(void *pdata)
Hdf5_create_info *cinfo = static_cast<Hdf5_create_info *>(pdata);

assert(cinfo && cinfo->file && cinfo->file->iosystem);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + cinfo->fname);

ret = spio_hdf5_create(cinfo->file->iosystem, cinfo->file, cinfo->fname.c_str());

Expand Down Expand Up @@ -217,6 +220,7 @@ int spio_iosys_async_hdf5_create_op_add(file_desc_t *file, const char *filename)

void spio_iosys_async_op_hdf5_def_var_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
if(pdata){
delete(static_cast<Hdf5_def_var_info *>(pdata));
}
Expand All @@ -227,6 +231,7 @@ int spio_iosys_async_op_hdf5_def_var(void *pdata)
int ret = PIO_NOERR;

Hdf5_def_var_info *def_var_info = static_cast<Hdf5_def_var_info *>(pdata);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ": fname = " + def_var_info->file->fname + ", vname: " + def_var_info->vname);

assert(def_var_info && def_var_info->file && def_var_info->file->iosystem);

Expand Down Expand Up @@ -280,6 +285,7 @@ int spio_iosys_async_hdf5_def_var_op_add(file_desc_t *file, const char *name,

void spio_iosys_async_op_hdf5_put_att_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
if(pdata){
Hdf5_put_att_info *info = static_cast<Hdf5_put_att_info *>(pdata);
free(info->abuf);
Expand All @@ -296,6 +302,7 @@ int spio_iosys_async_op_hdf5_put_att(void *pdata)
assert(info && info->file && info->file->iosystem);
assert((info->varid >= 0) || (info->varid == PIO_GLOBAL));
assert(info->alen >= 0);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ": fname = " + info->file->fname + ", aname: " + info->aname);

ret = spio_hdf5_put_att(info->file->iosystem, info->file,
info->varid, info->aname.c_str(), info->atype,
Expand Down Expand Up @@ -355,6 +362,7 @@ int spio_iosys_async_hdf5_put_att_op_add(file_desc_t *file, int varid,

void spio_iosys_async_op_hdf5_enddef_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
if(pdata){
delete(static_cast<Hdf5_enddef_info *>(pdata));
}
Expand All @@ -367,6 +375,7 @@ int spio_iosys_async_op_hdf5_enddef(void *pdata)
Hdf5_enddef_info *info = static_cast<Hdf5_enddef_info *>(pdata);

assert(info && info->file && info->file->iosystem);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ": fname = " + info->file->fname);

ret = spio_hdf5_enddef(info->file->iosystem, info->file);

Expand Down Expand Up @@ -414,6 +423,7 @@ int spio_iosys_async_hdf5_enddef_op_add(file_desc_t *file)

void spio_iosys_async_op_hdf5_put_var_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
if(pdata){
Hdf5_put_var_info *info = static_cast<Hdf5_put_var_info *>(pdata);
free(info->vbuf);
Expand All @@ -429,6 +439,7 @@ int spio_iosys_async_op_hdf5_put_var(void *pdata)

assert(info && info->file && info->file->iosystem);
assert((info->varid >= 0) || (info->varid == PIO_GLOBAL));
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ": fname = " + info->file->fname + ", varid=" + std::to_string(info->varid));

ret = spio_hdf5_put_var(info->file->iosystem, info->file,
info->varid,
Expand Down Expand Up @@ -510,6 +521,7 @@ int spio_iosys_async_hdf5_put_var_op_add(file_desc_t *file, int varid,

void spio_iosys_async_op_hdf5_set_frame_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
if(pdata){
delete(static_cast<Hdf5_set_frame_info *>(pdata));
}
Expand All @@ -522,6 +534,7 @@ int spio_iosys_async_op_hdf5_set_frame(void *pdata)
Hdf5_set_frame_info *info = static_cast<Hdf5_set_frame_info *>(pdata);

assert(info && info->file && info->file->iosystem);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ": fname = " + info->file->fname + ", frame=" + std::to_string(info->frame));

ret = spio_hdf5_set_frame(info->file, info->varid, info->frame);

Expand Down Expand Up @@ -571,6 +584,7 @@ int spio_iosys_async_hdf5_set_frame_op_add(file_desc_t *file, int varid, int fra

int spio_wait_all_hdf5_async_ops(int iosysid)
{
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ", iosysid=" + std::to_string(iosysid));
unsigned long long int sleep_time = 0;
/* Sleep for 0.5 seconds */
const int SLEEP_TIME_IN_MILLISECONDS = 500;
Expand All @@ -596,6 +610,7 @@ int pio_iosys_async_op_hdf5_write(void *pdata)
int ret = PIO_NOERR;
Hdf5_wcache *wcache = static_cast<struct Hdf5_wcache *>(pdata);
assert(wcache);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + ", fname=" + wcache->file->fname + ", nvars=" + std::to_string(wcache->nvars));

file_desc_t *file = wcache->file;
int nvars = wcache->nvars;
Expand Down Expand Up @@ -802,7 +817,7 @@ int pio_iosys_async_op_hdf5_write(void *pdata)
}

/* FIXME: Is this barrier needed ? */
MPI_Barrier(ios->io_comm);
MPI_Barrier(ios->tcomm_info->get_io_comm());

iodesc->nasync_pend_ops--;
file->npend_ops--;
Expand All @@ -817,6 +832,7 @@ int pio_iosys_async_op_hdf5_write(void *pdata)

void pio_iosys_async_op_hdf5_write_free(void *pdata)
{
//SPIO_Util::Dbg_Util::Stdout_logger(__func__);
#ifdef _HDF5
Hdf5_wcache *wcache = static_cast<struct Hdf5_wcache *>(pdata);
assert(wcache);
Expand Down
33 changes: 15 additions & 18 deletions src/clib/core/pio_file.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
#include <functional>
#include "spio_hdf5_utils.hpp"
#include "spio_async_tcomm.hpp"
#include "spio_dbg_utils.hpp"

#ifdef _ADIOS2
#include "../../tools/adios2pio-nm/adios2pio-nm-lib-c.h"
Expand Down Expand Up @@ -549,6 +550,7 @@ int spio_hard_closefile(iosystem_desc_t *ios, file_desc_t *file,
#endif

assert(ios && file);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + file->fname);

/* Get the lock before proceeding */
std::lock_guard<std::mutex> lg(*(file->pmtx));
Expand All @@ -557,7 +559,11 @@ int spio_hard_closefile(iosystem_desc_t *ios, file_desc_t *file,

if(sync_with_ioprocs){
//if(ios->ioproc) { while(file->npend_ops){} }
MPI_Barrier(ios->union_comm);
/* For I/O async threads the union comm is NULL comm */
MPI_Comm sync_comm = ios->tcomm_info->get_union_comm();
if(sync_comm != MPI_COMM_NULL){
MPI_Barrier(sync_comm);
}
}

if(file->is_hard_closed) { return PIO_NOERR; }
Expand Down Expand Up @@ -859,7 +865,12 @@ int spio_hard_closefile(iosystem_desc_t *ios, file_desc_t *file,
}
}

#if PIO_USE_ASYNC_WR_THREAD
/* FIXME: Use the iosystem/file error handlers once error handlers are multi-threaded */
ierr = spio_handle_err(NULL, file, PIO_RETURN_ERROR, ierr, __FILE__, __LINE__);
#else
ierr = check_netcdf(NULL, file, ierr, __FILE__, __LINE__);
#endif
if(ierr != PIO_NOERR){
LOG((1, "nc*_close failed, ierr = %d", ierr));
return pio_err(NULL, file, ierr, __FILE__, __LINE__,
Expand Down Expand Up @@ -919,6 +930,7 @@ int spio_hard_closefile(iosystem_desc_t *ios, file_desc_t *file,
int spio_soft_closefile(iosystem_desc_t *ios, file_desc_t *file)
{
assert(ios && file && ios->ioproc);
//SPIO_Util::Dbg_Util::Stdout_logger(std::string(__func__) + file->fname);

#if PIO_USE_ASYNC_WR_THREAD
return pio_iosys_async_file_close_op_add(file);
Expand Down Expand Up @@ -1035,24 +1047,9 @@ int PIOc_closefile_impl(int ncid)
if(ios->ioproc){
soft_close = true;
}
else{
/* Wait on all hdf5 async ops before a "hard close" */
/*
/* To wait on all hdf5 async ops before a "hard close" (for debugging)
ierr = spio_wait_all_hdf5_async_ops(ios->iosysid);
if(ierr != PIO_NOERR){
return pio_err(ios, file, ierr, __FILE__, __LINE__,
"Closing file (%s, ncid=%d) failed. Error sending async msg PIO_MSG_CLOSE_FILE", pio_get_fname_from_file(file), ncid);
}
*/
}
}
else{
/* Wait on all hdf5 async ops before a "hard close" */
ierr = spio_wait_all_hdf5_async_ops(ios->iosysid);
if(ierr != PIO_NOERR){
return pio_err(ios, file, ierr, __FILE__, __LINE__,
"Closing file (%s, ncid=%d) failed. Error sending async msg PIO_MSG_CLOSE_FILE", pio_get_fname_from_file(file), ncid);
}
*/
}
#endif

Expand Down
Loading
Loading