path: root/src/H5FDsubfiling/H5FDioc_threads.c

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
 * 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 COPYING file, which can be found at the root of the source code       *
 * distribution tree, or in https://support.hdfgroup.org/ftp/HDF5/releases.  *
 * If you do not have access to either file, you may request a copy from     *
 * help@hdfgroup.org.                                                        *
 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#include "H5FDioc_priv.h"

#include "H5FDsubfiling.h"

#define MIN_READ_RETRIES 10

/*
 * The amount of time (in nanoseconds) for the IOC main
 * thread to sleep when there are no incoming I/O requests
 * to process
 */
#define IOC_MAIN_SLEEP_DELAY (20000)

/*
 * IOC data for a file that is stored in that
 * file's subfiling context object
 */
typedef struct ioc_data_t {
    ioc_io_queue_t    io_queue;
    hg_thread_t       ioc_main_thread;
    hg_thread_pool_t *io_thread_pool;
    int64_t           sf_context_id;

    /* sf_io_ops_pending is use to track the number of I/O operations pending so that we can wait
     * until all I/O operations have been serviced before shutting down the worker thread pool.
     * The value of this variable must always be non-negative.
     *
     * Note that this is a convenience variable -- we could use io_queue.q_len instead.
     * However, accessing this field requires locking io_queue.q_mutex.
     */
    atomic_int sf_ioc_ready;
    atomic_int sf_shutdown_flag;
    atomic_int sf_io_ops_pending;
    atomic_int sf_work_pending;
} ioc_data_t;

/*
 * NOTES:
 * Rather than re-create the code for creating and managing a thread pool,
 * I'm utilizing a reasonably well tested implementation from the mercury
 * project.  At some point, we should revisit this decision or possibly
 * directly link against the mercury library.  This would make sense if
 * we move away from using MPI as the messaging infrastructure and instead
 * use mercury for that purpose...
 */

static hg_thread_mutex_t ioc_thread_mutex = PTHREAD_MUTEX_INITIALIZER;

#ifdef H5FD_IOC_COLLECT_STATS
static int    sf_write_ops        = 0;
static int    sf_read_ops         = 0;
static double sf_pwrite_time      = 0.0;
static double sf_pread_time       = 0.0;
static double sf_write_wait_time  = 0.0;
static double sf_read_wait_time   = 0.0;
static double sf_queue_delay_time = 0.0;
#endif

/* Prototypes */
static HG_THREAD_RETURN_TYPE ioc_thread_main(void *arg);
static int                   ioc_main(ioc_data_t *ioc_data);

static int ioc_file_queue_write_indep(sf_work_request_t *msg, int ioc_idx, int source, MPI_Comm comm,
                                      uint32_t counter);
static int ioc_file_queue_read_indep(sf_work_request_t *msg, int ioc_idx, int source, MPI_Comm comm,
                                     uint32_t counter);

static int ioc_file_write_data(int fd, int64_t file_offset, void *data_buffer, int64_t data_size,
                               int ioc_idx);
static int ioc_file_read_data(int fd, int64_t file_offset, void *data_buffer, int64_t data_size, int ioc_idx);
static int ioc_file_truncate(sf_work_request_t *msg);
static int ioc_file_report_eof(sf_work_request_t *msg, MPI_Comm comm);

static ioc_io_queue_entry_t *ioc_io_queue_alloc_entry(void);
static void ioc_io_queue_complete_entry(ioc_data_t *ioc_data, ioc_io_queue_entry_t *entry_ptr);
static void ioc_io_queue_dispatch_eligible_entries(ioc_data_t *ioc_data, hbool_t try_lock);
static void ioc_io_queue_free_entry(ioc_io_queue_entry_t *q_entry_ptr);
static void ioc_io_queue_add_entry(ioc_data_t *ioc_data, sf_work_request_t *wk_req_ptr);

/*-------------------------------------------------------------------------
 * Function:    initialize_ioc_threads
 *
 * Purpose:     The principal entry point to initialize the execution
 *              context for an I/O Concentrator (IOC). The main thread
 *              is responsible for receiving I/O requests from each
 *              HDF5 "client" and distributing those to helper threads
 *              for actual processing. We initialize a fixed number
 *              of helper threads by creating a thread pool.
 *
 * Return:      SUCCESS (0) or FAIL (-1) if any errors are detected
 *              for the multi-threaded initialization.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
int
initialize_ioc_threads(void *_sf_context)
{
    subfiling_context_t *sf_context       = _sf_context;
    ioc_data_t          *ioc_data         = NULL;
    unsigned             thread_pool_size = H5FD_IOC_DEFAULT_THREAD_POOL_SIZE;
    char                *env_value;
    int                  ret_value = 0;
#ifdef H5FD_IOC_COLLECT_STATS
    double t_start = 0.0, t_end = 0.0;
#endif

    HDassert(sf_context);

    /*
     * Allocate and initialize IOC data that will be passed
     * to the IOC main thread
     */
    if (NULL == (ioc_data = HDmalloc(sizeof(*ioc_data))))
        H5_SUBFILING_GOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, (-1),
                                "can't allocate IOC data for IOC main thread");
    ioc_data->sf_context_id  = sf_context->sf_context_id;
    ioc_data->io_thread_pool = NULL;
    ioc_data->io_queue       = (ioc_io_queue_t){/* magic               = */ H5FD_IOC__IO_Q_MAGIC,
                                          /* q_head              = */ NULL,
                                          /* q_tail              = */ NULL,
                                          /* num_pending         = */ 0,
                                          /* num_in_progress     = */ 0,
                                          /* num_failed          = */ 0,
                                          /* q_len               = */ 0,
                                          /* req_counter         = */ 0,
                                          /* q_mutex             = */
                                          PTHREAD_MUTEX_INITIALIZER,
#ifdef H5FD_IOC_COLLECT_STATS
                                          /* max_q_len           = */ 0,
                                          /* max_num_pending     = */ 0,
                                          /* max_num_in_progress = */ 0,
                                          /* ind_read_requests   = */ 0,
                                          /* ind_write_requests  = */ 0,
                                          /* truncate_requests   = */ 0,
                                          /* get_eof_requests    = */ 0,
                                          /* requests_queued     = */ 0,
                                          /* requests_dispatched = */ 0,
                                          /* requests_completed  = */ 0
#endif
    };

    sf_context->ioc_data = ioc_data;

    /* Initialize atomic vars */
    atomic_init(&ioc_data->sf_ioc_ready, 0);
    atomic_init(&ioc_data->sf_shutdown_flag, 0);
    atomic_init(&ioc_data->sf_io_ops_pending, 0);
    atomic_init(&ioc_data->sf_work_pending, 0);

#ifdef H5FD_IOC_COLLECT_STATS
    t_start = MPI_Wtime();
#endif

    if (hg_thread_mutex_init(&ioc_data->io_queue.q_mutex) < 0)
        H5_SUBFILING_GOTO_ERROR(H5E_RESOURCE, H5E_CANTINIT, (-1), "can't initialize IOC thread queue mutex");

    /* Allow experimentation with the number of helper threads */
    if ((env_value = HDgetenv(H5FD_IOC_THREAD_POOL_SIZE)) != NULL) {
        int value_check = HDatoi(env_value);
        if (value_check > 0) {
            thread_pool_size = (unsigned int)value_check;
        }
    }

    /* Initialize a thread pool for the I/O concentrator's worker threads */
    if (hg_thread_pool_init(thread_pool_size, &ioc_data->io_thread_pool) < 0)
        H5_SUBFILING_GOTO_ERROR(H5E_RESOURCE, H5E_CANTINIT, (-1), "can't initialize IOC worker thread pool");

    /* Create the main IOC thread that will receive and dispatch I/O requests */
    if (hg_thread_create(&ioc_data->ioc_main_thread, ioc_thread_main, ioc_data) < 0)
        H5_SUBFILING_GOTO_ERROR(H5E_RESOURCE, H5E_CANTINIT, (-1), "can't create IOC main thread");

    /* Wait until ioc_main() reports that it is ready */
    while (atomic_load(&ioc_data->sf_ioc_ready) != 1) {
        usleep(20);
    }

#ifdef H5FD_IOC_COLLECT_STATS
    t_end = MPI_Wtime();

#ifdef H5FD_IOC_DEBUG
    if (sf_context->topology->ioc_idx == 0) {
        HDprintf("%s: time = %lf seconds\n", __func__, (t_end - t_start));
        HDfflush(stdout);
    }
#endif

#endif

done:
    H5_SUBFILING_FUNC_LEAVE;
}

int
finalize_ioc_threads(void *_sf_context)
{
    subfiling_context_t *sf_context = _sf_context;
    ioc_data_t          *ioc_data   = NULL;
    int                  ret_value  = 0;

    HDassert(sf_context);
    HDassert(sf_context->topology->rank_is_ioc);

    ioc_data = sf_context->ioc_data;
    if (ioc_data) {
        HDassert(0 == atomic_load(&ioc_data->sf_shutdown_flag));

        /* Shutdown the main IOC thread */
        atomic_store(&ioc_data->sf_shutdown_flag, 1);

        /* Allow ioc_main to exit.*/
        do {
            usleep(20);
        } while (0 != atomic_load(&ioc_data->sf_shutdown_flag));

        /* Tear down IOC worker thread pool */
        HDassert(0 == atomic_load(&ioc_data->sf_io_ops_pending));
        hg_thread_pool_destroy(ioc_data->io_thread_pool);

        hg_thread_mutex_destroy(&ioc_data->io_queue.q_mutex);

        /* Wait for IOC main thread to exit */
        hg_thread_join(ioc_data->ioc_main_thread);
    }

    if (ioc_data->io_queue.num_failed > 0)
        H5_SUBFILING_DONE_ERROR(H5E_IO, H5E_CLOSEERROR, -1, "%" PRId32 " I/O requests failed",
                                ioc_data->io_queue.num_failed);

    HDfree(ioc_data);
    sf_context->ioc_data = NULL;

    H5_SUBFILING_FUNC_LEAVE;
}

/*-------------------------------------------------------------------------
 * Function:    ioc_thread_main
 *
 * Purpose:     An IO Concentrator instance is initialized with the
 *              specified subfiling context.
 *
 * Return:      The IO concentrator thread executes as long as the HDF5
 *              file associated with this context is open.  At file close,
 *              the thread will return from 'ioc_main' and the thread
 *              exit status will be checked by the main program.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
static HG_THREAD_RETURN_TYPE
ioc_thread_main(void *arg)
{
    hg_thread_ret_t thread_ret = (hg_thread_ret_t)0;
    ioc_data_t     *ioc_data   = (ioc_data_t *)arg;

    /* Pass along the ioc_data_t */
    ioc_main(ioc_data);

    return thread_ret;
}

/*-------------------------------------------------------------------------
 * Function:    ioc_main
 *
 * Purpose:     This is the principal function run by the I/O Concentrator
 *              main thread.  It remains within a loop until allowed to
 *              exit by means of setting the 'sf_shutdown_flag'. This is
 *              usually accomplished as part of the file close operation.
 *
 *              The function implements an asynchronous polling approach
 *              for incoming messages. These messages can be thought of
 *              as a primitive RPC which utilizes MPI tags to code and
 *              implement the desired subfiling functionality.
 *
 *              As each incoming message is received, it gets added to
 *              a queue for processing by a thread_pool thread. The
 *              message handlers are dispatched via the
 *              "handle_work_request" routine.

 *              Subfiling is effectively a software RAID-0 implementation
 *              where having multiple I/O Concentrators and independent
 *              subfiles is equated to the multiple disks and a true
 *              hardware base RAID implementation.
 *
 *              I/O Concentrators are ordered according to their MPI rank.
 *              In the simplest interpretation, IOC(0) will always contain
 *              the initial bytes of the logical disk image.  Byte 0 of
 *              IOC(1) will contain the byte written to the logical disk
 *              offset "stripe_size" X IOC(number).
 *
 *              Example: If the stripe size is defined to be 256K, then
 *              byte 0 of subfile(1) is at logical offset 262144 of the
 *              file.   Similarly, byte 0 of subfile(2) represents the
 *              logical file offset = 524288.   For logical files larger
 *              than 'N' X stripe_size, we simply "wrap around" back to
 *              subfile(0).  The following shows the mapping of 30
 *              logical blocks of data over 3 subfiles:
 *              +--------+--------+--------+--------+--------+--------+
 *              | blk(0 )| blk(1) | blk(2 )| blk(3 )| blk(4 )| blk(5 )|
 *              | IOC(0) | IOC(1) | IOC(2) | IOC(0) | IOC(1) | IOC(2) |
 *              +--------+--------+--------+--------+--------+--------+
 *              | blk(6 )| blk(7) | blk(8 )| blk(9 )| blk(10)| blk(11)|
 *              | IOC(0) | IOC(1) | IOC(2) | IOC(0) | IOC(1) | IOC(2) |
 *              +--------+--------+--------+--------+--------+--------+
 *              | blk(12)| blk(13)| blk(14)| blk(15)| blk(16)| blk(17)|
 *              | IOC(0) | IOC(1) | IOC(2) | IOC(0) | IOC(1) | IOC(2) |
 *              +--------+--------+--------+--------+--------+--------+
 *              | blk(18)| blk(19)| blk(20)| blk(21)| blk(22)| blk(23)|
 *              | IOC(0) | IOC(1) | IOC(2) | IOC(0) | IOC(1) | IOC(2) |
 *              +--------+--------+--------+--------+--------+--------+
 *              | blk(24)| blk(25)| blk(26)| blk(27)| blk(28)| blk(29)|
 *              | IOC(0) | IOC(1) | IOC(2) | IOC(0) | IOC(1) | IOC(2) |
 *              +--------+--------+--------+--------+--------+--------+
 *
 * Return:      None
 * Errors:      None
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *-------------------------------------------------------------------------
 */
static int
ioc_main(ioc_data_t *ioc_data)
{
    subfiling_context_t *context = NULL;
    sf_work_request_t    wk_req;
    int                  shutdown_requested;
    int                  ret_value = 0;

    HDassert(ioc_data);

    context = H5_get_subfiling_object(ioc_data->sf_context_id);
    HDassert(context);

    /* We can't have opened any files at this point..
     * The file open approach has changed so that the normal
     * application rank (hosting this thread) does the file open.
     * We can simply utilize the file descriptor (which should now
     * represent an open file).
     */

    /* tell initialize_ioc_threads() that ioc_main() is ready to enter its main loop */
    atomic_store(&ioc_data->sf_ioc_ready, 1);

    shutdown_requested = 0;

    while ((!shutdown_requested) || (0 < atomic_load(&ioc_data->sf_io_ops_pending)) ||
           (0 < atomic_load(&ioc_data->sf_work_pending))) {
        MPI_Status status;
        int        flag = 0;
        int        mpi_code;

        /* Probe for incoming work requests */
        if (MPI_SUCCESS !=
            (mpi_code = (MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, context->sf_msg_comm, &flag, &status))))
            H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Iprobe failed", mpi_code);

        if (flag) {
            double queue_start_time;
            int    count;
            int    source = status.MPI_SOURCE;
            int    tag    = status.MPI_TAG;

            if ((tag != READ_INDEP) && (tag != WRITE_INDEP) && (tag != TRUNC_OP) && (tag != GET_EOF_OP))
                H5_SUBFILING_GOTO_ERROR(H5E_ARGS, H5E_BADVALUE, -1, "invalid work request operation (%d)",
                                        tag);

            if (MPI_SUCCESS != (mpi_code = MPI_Get_count(&status, MPI_BYTE, &count)))
                H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Get_count failed", mpi_code);

            if (count < 0)
                H5_SUBFILING_GOTO_ERROR(H5E_ARGS, H5E_BADVALUE, -1, "invalid work request message size (%d)",
                                        count);

            if ((size_t)count > sizeof(sf_work_request_t))
                H5_SUBFILING_GOTO_ERROR(H5E_ARGS, H5E_BADVALUE, -1, "work request message is too large (%d)",
                                        count);

            /*
             * Zero out work request, since the received message should
             * be smaller than sizeof(sf_work_request_t)
             */
            HDmemset(&wk_req, 0, sizeof(sf_work_request_t));

            if (MPI_SUCCESS != (mpi_code = MPI_Recv(&wk_req, count, MPI_BYTE, source, tag,
                                                    context->sf_msg_comm, MPI_STATUS_IGNORE)))
                H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Recv failed", mpi_code);

            /* Dispatch work request to worker threads in thread pool */

            queue_start_time = MPI_Wtime();

            wk_req.tag        = tag;
            wk_req.source     = source;
            wk_req.ioc_idx    = context->topology->ioc_idx;
            wk_req.context_id = ioc_data->sf_context_id;
            wk_req.start_time = queue_start_time;

            ioc_io_queue_add_entry(ioc_data, &wk_req);

            HDassert(atomic_load(&ioc_data->sf_io_ops_pending) >= 0);
        }
        else {
            struct timespec sleep_spec = {0, IOC_MAIN_SLEEP_DELAY};

            HDnanosleep(&sleep_spec, NULL);
        }

        ioc_io_queue_dispatch_eligible_entries(ioc_data, flag ? 0 : 1);

        shutdown_requested = atomic_load(&ioc_data->sf_shutdown_flag);
    }

    /* Reset the shutdown flag */
    atomic_store(&ioc_data->sf_shutdown_flag, 0);

done:
    H5_SUBFILING_FUNC_LEAVE;
} /* ioc_main() */

#ifdef H5_SUBFILING_DEBUG
static const char *
translate_opcode(io_op_t op)
{
    switch (op) {
        case READ_OP:
            return "READ_OP";
            break;
        case WRITE_OP:
            return "WRITE_OP";
            break;
        case OPEN_OP:
            return "OPEN_OP";
            break;
        case CLOSE_OP:
            return "CLOSE_OP";
            break;
        case TRUNC_OP:
            return "TRUNC_OP";
            break;
        case GET_EOF_OP:
            return "GET_EOF_OP";
            break;
        case FINI_OP:
            return "FINI_OP";
            break;
        case LOGGING_OP:
            return "LOGGING_OP";
            break;
    }
    return "unknown";
}
#endif

/*-------------------------------------------------------------------------
 * Function:    handle_work_request
 *
 * Purpose:     Handle a work request from the thread pool work queue.
 *              We dispatch the specific function as indicated by the
 *              TAG that has been added to the work request by the
 *              IOC main thread (which is just a copy of the MPI tag
 *              associated with the RPC message) and provide the subfiling
 *              context associated with the HDF5 file.
 *
 *              Any status associated with the function processing is
 *              returned directly to the client via ACK or NACK messages.
 *
 * Return:      (none) Doesn't fail.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
static HG_THREAD_RETURN_TYPE
handle_work_request(void *arg)
{
    ioc_io_queue_entry_t *q_entry_ptr     = (ioc_io_queue_entry_t *)arg;
    subfiling_context_t  *sf_context      = NULL;
    sf_work_request_t    *msg             = &(q_entry_ptr->wk_req);
    ioc_data_t           *ioc_data        = NULL;
    int64_t               file_context_id = msg->context_id;
    int                   op_ret;
    hg_thread_ret_t       ret_value = 0;

    HDassert(q_entry_ptr);
    HDassert(q_entry_ptr->magic == H5FD_IOC__IO_Q_ENTRY_MAGIC);
    HDassert(q_entry_ptr->in_progress);

    sf_context = H5_get_subfiling_object(file_context_id);
    HDassert(sf_context);

    ioc_data = sf_context->ioc_data;
    HDassert(ioc_data);

    atomic_fetch_add(&ioc_data->sf_work_pending, 1);

    switch (msg->tag) {
        case WRITE_INDEP:
            op_ret = ioc_file_queue_write_indep(msg, msg->ioc_idx, msg->source, sf_context->sf_data_comm,
                                                q_entry_ptr->counter);
            break;

        case READ_INDEP:
            op_ret = ioc_file_queue_read_indep(msg, msg->ioc_idx, msg->source, sf_context->sf_data_comm,
                                               q_entry_ptr->counter);
            break;

        case TRUNC_OP:
            op_ret = ioc_file_truncate(msg);
            break;

        case GET_EOF_OP:
            op_ret = ioc_file_report_eof(msg, sf_context->sf_eof_comm);
            break;

        default:
#ifdef H5_SUBFILING_DEBUG
            H5_subfiling_log(file_context_id, "%s: IOC %d received unknown message with tag %x from rank %d",
                             __func__, msg->ioc_idx, msg->tag, msg->source);
#endif

            op_ret = -1;
            break;
    }

    atomic_fetch_sub(&ioc_data->sf_work_pending, 1);

    if (op_ret < 0) {
#ifdef H5_SUBFILING_DEBUG
        H5_subfiling_log(file_context_id,
                         "%s: IOC %d request(%s) from rank(%d), (%" PRId64 ", %" PRId64 ", %" PRId64
                         ") FAILED with ret %d",
                         __func__, msg->ioc_idx, translate_opcode((io_op_t)msg->tag), msg->source,
                         msg->header[0], msg->header[1], msg->header[2], op_ret);
#endif

        q_entry_ptr->wk_ret = op_ret;
    }

#ifdef H5FD_IOC_DEBUG
    {
        int curr_io_ops_pending = atomic_load(&ioc_data->sf_io_ops_pending);
        HDassert(curr_io_ops_pending > 0);
    }
#endif

    /* complete the I/O request */
    ioc_io_queue_complete_entry(ioc_data, q_entry_ptr);

    HDassert(atomic_load(&ioc_data->sf_io_ops_pending) >= 0);

    /* Check the I/O Queue to see if there are any dispatchable entries */
    ioc_io_queue_dispatch_eligible_entries(ioc_data, 1);

    H5_SUBFILING_FUNC_LEAVE;
}

/*-------------------------------------------------------------------------
 * Function:    H5FD_ioc_begin_thread_exclusive
 *
 * Purpose:     Mutex lock to restrict access to code or variables.
 *
 * Return:      integer result of mutex_lock request.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
void
H5FD_ioc_begin_thread_exclusive(void)
{
    hg_thread_mutex_lock(&ioc_thread_mutex);
}

/*-------------------------------------------------------------------------
 * Function:    H5FD_ioc_end_thread_exclusive
 *
 * Purpose:     Mutex unlock.  Should only be called by the current holder
 *              of the locked mutex.
 *
 * Return:      result of mutex_unlock operation.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
void
H5FD_ioc_end_thread_exclusive(void)
{
    hg_thread_mutex_unlock(&ioc_thread_mutex);
}

static herr_t
send_ack_to_client(int ack_val, int dest_rank, int source_rank, int msg_tag, MPI_Comm comm)
{
    int    mpi_code;
    herr_t ret_value = SUCCEED;

    HDassert(ack_val > 0);

    (void)source_rank;

    if (MPI_SUCCESS != (mpi_code = MPI_Send(&ack_val, 1, MPI_INT, dest_rank, msg_tag, comm)))
        H5_SUBFILING_MPI_GOTO_ERROR(FAIL, "MPI_Send", mpi_code);

done:
    H5_SUBFILING_FUNC_LEAVE;
}

static herr_t
send_nack_to_client(int dest_rank, int source_rank, int msg_tag, MPI_Comm comm)
{
    int    nack = 0;
    int    mpi_code;
    herr_t ret_value = SUCCEED;

    (void)source_rank;

    if (MPI_SUCCESS != (mpi_code = MPI_Send(&nack, 1, MPI_INT, dest_rank, msg_tag, comm)))
        H5_SUBFILING_MPI_GOTO_ERROR(FAIL, "MPI_Send", mpi_code);

done:
    H5_SUBFILING_FUNC_LEAVE;
}

/*
=========================================
queue_xxx functions that should be run
from the thread pool threads...
=========================================
*/

/*-------------------------------------------------------------------------
 * Function:    ioc_file_queue_write_indep
 *
 * Purpose:     Implement the IOC independent write function.  The
 *              function is invoked as a result of the IOC receiving the
 *              "header"/RPC.  What remains is to allocate memory for the
 *              data sent by the client and then write the data to our
 *              subfile.  We utilize pwrite for the actual file writing.
 *              File flushing is done at file close.
 *
 * Return:      The integer status returned by the Internal read_independent
 *              function.  Successful operations will return 0.
 * Errors:      An MPI related error value.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
static int
ioc_file_queue_write_indep(sf_work_request_t *msg, int ioc_idx, int source, MPI_Comm comm, uint32_t counter)
{
    subfiling_context_t *sf_context = NULL;
    MPI_Status           msg_status;
    hbool_t              send_nack = FALSE;
    int64_t              file_context_id;
    int64_t              data_size;
    int64_t              file_offset;
    int64_t              subfile_idx;
    int64_t              stripe_id;
    haddr_t              sf_eof;
#ifdef H5FD_IOC_COLLECT_STATS
    double t_start;
    double t_end;
    double t_write;
    double t_wait;
    double t_queue_delay;
#endif
    char *recv_buf = NULL;
    int   rcv_tag;
    int   sf_fid;
    int   data_bytes_received;
    int   write_ret;
    int   mpi_code;
    int   ret_value = 0;

    HDassert(msg);

    file_context_id = msg->context_id;

    /* Retrieve the fields of the RPC message for the write operation */
    data_size   = msg->header[0];
    file_offset = msg->header[1];
    subfile_idx = msg->header[2];

    if (data_size < 0) {
        send_nack = TRUE;
        H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_BADVALUE, -1, "invalid data size for write");
    }

    sf_context = H5_get_subfiling_object(file_context_id);
    HDassert(sf_context);

    stripe_id = file_offset + data_size;
    sf_eof    = (haddr_t)(stripe_id % sf_context->sf_stripe_size);

    stripe_id /= sf_context->sf_stripe_size;
    sf_eof += (haddr_t)((stripe_id * sf_context->sf_blocksize_per_stripe) + sf_context->sf_base_addr);

    /* Flag that we've attempted to write data to the file */
    sf_context->sf_write_count++;

#ifdef H5FD_IOC_COLLECT_STATS
    /* For debugging performance */
    sf_write_ops++;

    t_start       = MPI_Wtime();
    t_queue_delay = t_start - msg->start_time;

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(file_context_id,
                     "[ioc(%d) %s]: msg from %d: datasize=%ld\toffset=%ld, queue_delay = %lf seconds\n",
                     ioc_idx, __func__, source, data_size, file_offset, t_queue_delay);
#endif

#endif

    /* Allocate space to receive data sent from the client */
    if (NULL == (recv_buf = HDmalloc((size_t)data_size))) {
        send_nack = TRUE;
        H5_SUBFILING_GOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, -1, "couldn't allocate receive buffer for data");
    }

    /*
     * Calculate message tag for the client to use for sending
     * data, then send an ACK message to the client with the
     * calculated message tag. This calculated message tag
     * allows us to distinguish between multiple concurrent
     * writes from a single rank.
     */
    HDassert(H5FD_IOC_tag_ub_val_ptr && (*H5FD_IOC_tag_ub_val_ptr >= IO_TAG_BASE));
    rcv_tag = (int)(counter % (INT_MAX - IO_TAG_BASE));
    rcv_tag %= (*H5FD_IOC_tag_ub_val_ptr - IO_TAG_BASE);
    rcv_tag += IO_TAG_BASE;

    if (send_ack_to_client(rcv_tag, source, ioc_idx, WRITE_INDEP_ACK, comm) < 0)
        H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_WRITEERROR, -1, "couldn't send ACK to client");

    /* Receive data from client */
    H5_CHECK_OVERFLOW(data_size, int64_t, int);
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Recv(recv_buf, (int)data_size, MPI_BYTE, source, rcv_tag, comm, &msg_status)))
        H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Recv failed", mpi_code);

    if (MPI_SUCCESS != (mpi_code = MPI_Get_count(&msg_status, MPI_BYTE, &data_bytes_received)))
        H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Get_count failed", mpi_code);

    if (data_bytes_received != data_size)
        H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_WRITEERROR, -1,
                                "message size mismatch -- expected = %" PRId64 ", actual = %d", data_size,
                                data_bytes_received);

#ifdef H5FD_IOC_COLLECT_STATS
    t_end  = MPI_Wtime();
    t_wait = t_end - t_start;
    sf_write_wait_time += t_wait;

    t_start = t_end;

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(file_context_id, "[ioc(%d) %s] MPI_Recv(%ld bytes, from = %d) status = %d\n", ioc_idx,
                     __func__, data_size, source, mpi_code);
#endif

#endif

    HDassert(subfile_idx < sf_context->sf_num_fids);
    sf_fid = sf_context->sf_fids[subfile_idx];

#ifdef H5_SUBFILING_DEBUG
    if (sf_fid < 0)
        H5_subfiling_log(file_context_id, "%s: WARNING: attempt to write data to closed subfile FID %d",
                         __func__, sf_fid);
#endif

    if (sf_fid >= 0) {
        /* Actually write data received from client into subfile */
        if ((write_ret = ioc_file_write_data(sf_fid, file_offset, recv_buf, data_size, ioc_idx)) < 0)
            H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_WRITEERROR, -1,
                                    "write function(FID=%d, Source=%d) returned an error (%d)", sf_fid,
                                    source, write_ret);

#ifdef H5FD_IOC_COLLECT_STATS
        t_end   = MPI_Wtime();
        t_write = t_end - t_start;
        sf_pwrite_time += t_write;
#endif
    }

#ifdef H5FD_IOC_COLLECT_STATS
    sf_queue_delay_time += t_queue_delay;
#endif

    H5FD_ioc_begin_thread_exclusive();

    /* Adjust EOF if necessary */
    if (sf_eof > sf_context->sf_eof)
        sf_context->sf_eof = sf_eof;

    H5FD_ioc_end_thread_exclusive();

    /*
     * Send a message back to the client that the I/O call has
     * completed and it is safe to return from the write call
     */
    if (MPI_SUCCESS != (mpi_code = MPI_Send(&rcv_tag, 1, MPI_INT, source, WRITE_DATA_DONE, comm)))
        H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Send failed", mpi_code);

done:
    if (send_nack) {
        /* Send NACK back to client so client can handle failure gracefully */
        if (send_nack_to_client(source, ioc_idx, WRITE_INDEP_ACK, comm) < 0)
            H5_SUBFILING_DONE_ERROR(H5E_IO, H5E_WRITEERROR, -1, "couldn't send NACK to client");
    }

    HDfree(recv_buf);

    H5_SUBFILING_FUNC_LEAVE;
} /* ioc_file_queue_write_indep() */

/*-------------------------------------------------------------------------
 * Function:    ioc_file_queue_read_indep
 *
 * Purpose:     Implement the IOC independent read function.  The
 *              function is invoked as a result of the IOC receiving the
 *              "header"/RPC.  What remains is to allocate memory for
 *              reading the data and then to send this to the client.
 *              We utilize pread for the actual file reading.
 *
 * Return:      The integer status returned by the Internal read_independent
 *              function.  Successful operations will return 0.
 * Errors:      An MPI related error value.
 *
 * Programmer:  Richard Warren
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */
static int
ioc_file_queue_read_indep(sf_work_request_t *msg, int ioc_idx, int source, MPI_Comm comm, uint32_t counter)
{
    subfiling_context_t *sf_context     = NULL;
    hbool_t              send_empty_buf = TRUE;
    hbool_t              send_nack      = FALSE;
    hbool_t              need_data_tag  = FALSE;
    int64_t              file_context_id;
    int64_t              data_size;
    int64_t              file_offset;
    int64_t              subfile_idx;
#ifdef H5FD_IOC_COLLECT_STATS
    double t_start;
    double t_end;
    double t_read;
    double t_queue_delay;
#endif
    char *send_buf = NULL;
    int   send_tag;
    int   sf_fid;
    int   read_ret;
    int   mpi_code;
    int   ret_value = 0;

    HDassert(msg);

    file_context_id = msg->context_id;

    sf_context = H5_get_subfiling_object(file_context_id);
    HDassert(sf_context);

    /*
     * If we are using 1 subfile per IOC, we can optimize reads
     * a little since each read will go to a separate IOC and we
     * won't be in danger of data being received in an
     * unpredictable order. However, if some IOCs own more than
     * 1 subfile, we need to associate each read with a unique
     * message tag to make sure the data is received in the
     * correct order.
     */
    need_data_tag = sf_context->sf_num_subfiles != sf_context->topology->n_io_concentrators;
    if (!need_data_tag)
        send_tag = READ_INDEP_DATA;

    /* Retrieve the fields of the RPC message for the read operation */
    data_size   = msg->header[0];
    file_offset = msg->header[1];
    subfile_idx = msg->header[2];

    if (data_size < 0) {
        if (need_data_tag) {
            send_nack      = TRUE;
            send_empty_buf = FALSE;
        }
        H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_BADVALUE, -1, "invalid data size for read");
    }

    /* Flag that we've attempted to read data from the file */
    sf_context->sf_read_count++;

#ifdef H5FD_IOC_COLLECT_STATS
    /* For debugging performance */
    sf_read_ops++;

    t_start       = MPI_Wtime();
    t_queue_delay = t_start - msg->start_time;

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(file_context_id,
                     "[ioc(%d) %s] msg from %d: datasize=%ld\toffset=%ld queue_delay=%lf seconds\n", ioc_idx,
                     __func__, source, data_size, file_offset, t_queue_delay);
#endif

#endif

    /* Allocate space to send data read from file to client */
    if (NULL == (send_buf = HDmalloc((size_t)data_size))) {
        if (need_data_tag) {
            send_nack      = TRUE;
            send_empty_buf = FALSE;
        }
        H5_SUBFILING_GOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, -1, "couldn't allocate send buffer for data");
    }

    if (need_data_tag) {
        /*
         * Calculate message tag for the client to use for receiving
         * data, then send an ACK message to the client with the
         * calculated message tag. This calculated message tag
         * allows us to distinguish between multiple concurrent
         * reads from a single rank, which can happen when a rank
         * owns multiple subfiles.
         */
        HDassert(H5FD_IOC_tag_ub_val_ptr && (*H5FD_IOC_tag_ub_val_ptr >= IO_TAG_BASE));
        send_tag = (int)(counter % (INT_MAX - IO_TAG_BASE));
        send_tag %= (*H5FD_IOC_tag_ub_val_ptr - IO_TAG_BASE);
        send_tag += IO_TAG_BASE;

        if (send_ack_to_client(send_tag, source, ioc_idx, READ_INDEP_ACK, comm) < 0) {
            send_empty_buf = FALSE;
            H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_READERROR, -1, "couldn't send ACK to client");
        }
    }

    /* Read data from the subfile */
    HDassert(subfile_idx < sf_context->sf_num_fids);
    sf_fid = sf_context->sf_fids[subfile_idx];
    if (sf_fid < 0)
        H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_BADVALUE, -1, "subfile file descriptor %d is invalid", sf_fid);

    if ((read_ret = ioc_file_read_data(sf_fid, file_offset, send_buf, data_size, ioc_idx)) < 0) {
        H5_SUBFILING_GOTO_ERROR(H5E_IO, H5E_READERROR, read_ret,
                                "read function(FID=%d, Source=%d) returned an error (%d)", sf_fid, source,
                                read_ret);
    }

    send_empty_buf = FALSE;

    /* Send read data to the client */
    H5_CHECK_OVERFLOW(data_size, int64_t, int);
    if (MPI_SUCCESS != (mpi_code = MPI_Send(send_buf, (int)data_size, MPI_BYTE, source, send_tag, comm)))
        H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Send failed", mpi_code);

#ifdef H5FD_IOC_COLLECT_STATS
    t_end  = MPI_Wtime();
    t_read = t_end - t_start;
    sf_pread_time += t_read;
    sf_queue_delay_time += t_queue_delay;

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(sf_context->sf_context_id, "[ioc(%d)] MPI_Send to source(%d) completed\n", ioc_idx,
                     source);
#endif

#endif

done:
    if (need_data_tag && send_nack) {
        /* Send NACK back to client so client can handle failure gracefully */
        if (send_nack_to_client(source, ioc_idx, READ_INDEP_ACK, comm) < 0)
            H5_SUBFILING_DONE_ERROR(H5E_IO, H5E_READERROR, -1, "couldn't send NACK to client");
    }
    if (send_empty_buf) {
        /*
         * Send an empty message back to client on failure. The client will
         * likely get a message truncation error, but at least shouldn't hang.
         */
        if (MPI_SUCCESS != (mpi_code = MPI_Send(NULL, 0, MPI_BYTE, source, send_tag, comm)))
            H5_SUBFILING_MPI_DONE_ERROR(-1, "MPI_Send failed", mpi_code);
    }

    HDfree(send_buf);

    return ret_value;
} /* end ioc_file_queue_read_indep() */

/*
======================================================
File functions

The pread and pwrite posix functions are described as
being thread safe.
======================================================
*/

static int
ioc_file_write_data(int fd, int64_t file_offset, void *data_buffer, int64_t data_size, int ioc_idx)
{
    ssize_t bytes_remaining = (ssize_t)data_size;
    ssize_t bytes_written   = 0;
    char   *this_data       = (char *)data_buffer;
    int     ret_value       = 0;

#ifndef H5FD_IOC_DEBUG
    (void)ioc_idx;
#endif

    HDcompile_assert(H5_SIZEOF_OFF_T == sizeof(file_offset));

    while (bytes_remaining) {
        errno = 0;

        bytes_written = HDpwrite(fd, this_data, (size_t)bytes_remaining, file_offset);

        if (bytes_written >= 0) {
            bytes_remaining -= bytes_written;

#ifdef H5FD_IOC_DEBUG
            HDprintf("[ioc(%d) %s]: wrote %ld bytes, remaining=%ld, file_offset=%" PRId64 "\n", ioc_idx,
                     __func__, bytes_written, bytes_remaining, file_offset);
#endif

            this_data += bytes_written;
            file_offset += bytes_written;
        }
        else {
            H5_SUBFILING_SYS_GOTO_ERROR(H5E_IO, H5E_WRITEERROR, -1, "HDpwrite failed");
        }
    }

    /* We don't usually use this for each file write.  We usually do the file
     * flush as part of file close operation.
     */
#ifdef H5FD_IOC_REQUIRE_FLUSH
    fdatasync(fd);
#endif

done:
    H5_SUBFILING_FUNC_LEAVE;
} /* end ioc_file_write_data() */

static int
ioc_file_read_data(int fd, int64_t file_offset, void *data_buffer, int64_t data_size, int ioc_idx)
{
    useconds_t delay           = 100;
    ssize_t    bytes_remaining = (ssize_t)data_size;
    ssize_t    bytes_read      = 0;
    char      *this_buffer     = (char *)data_buffer;
    int        retries         = MIN_READ_RETRIES;
    int        ret_value       = 0;

#ifndef H5FD_IOC_DEBUG
    (void)ioc_idx;
#endif

    HDcompile_assert(H5_SIZEOF_OFF_T == sizeof(file_offset));

    while (bytes_remaining) {
        errno = 0;

        bytes_read = HDpread(fd, this_buffer, (size_t)bytes_remaining, file_offset);

        if (bytes_read > 0) {
            /* Reset retry params */
            retries = MIN_READ_RETRIES;
            delay   = 100;

            bytes_remaining -= bytes_read;

#ifdef H5FD_IOC_DEBUG
            HDprintf("[ioc(%d) %s]: read %ld bytes, remaining=%ld, file_offset=%" PRId64 "\n", ioc_idx,
                     __func__, bytes_read, bytes_remaining, file_offset);
#endif

            this_buffer += bytes_read;
            file_offset += bytes_read;
        }
        else if (bytes_read == 0) {
            HDassert(bytes_remaining > 0);

            /* end of file but not end of format address space */
            HDmemset(this_buffer, 0, (size_t)bytes_remaining);
            break;
        }
        else {
            if (retries == 0) {
#ifdef H5FD_IOC_DEBUG
                HDprintf("[ioc(%d) %s]: TIMEOUT: file_offset=%" PRId64 ", data_size=%ld\n", ioc_idx, __func__,
                         file_offset, data_size);
#endif

                H5_SUBFILING_SYS_GOTO_ERROR(H5E_IO, H5E_READERROR, -1, "HDpread failed");
            }

            retries--;
            usleep(delay);
            delay *= 2;
        }
    }

done:
    H5_SUBFILING_FUNC_LEAVE;
} /* end ioc_file_read_data() */

static int
ioc_file_truncate(sf_work_request_t *msg)
{
    subfiling_context_t *sf_context = NULL;
    int64_t              file_context_id;
    int64_t              length;
    int64_t              subfile_idx;
    int                  fd;
    int                  ioc_idx;
    int                  ret_value = 0;

    HDassert(msg);

    file_context_id = msg->context_id;
    ioc_idx         = msg->ioc_idx;

    length      = msg->header[0];
    subfile_idx = msg->header[1];

#ifndef H5FD_IOC_DEBUG
    (void)ioc_idx;
#endif

    if (NULL == (sf_context = H5_get_subfiling_object(file_context_id)))
        H5_SUBFILING_GOTO_ERROR(H5E_FILE, H5E_CANTGET, -1, "couldn't retrieve subfiling context");

    HDassert(subfile_idx < sf_context->sf_num_fids);

    fd = sf_context->sf_fids[subfile_idx];

    if (HDftruncate(fd, (off_t)length) != 0)
        H5_SUBFILING_SYS_GOTO_ERROR(H5E_FILE, H5E_SEEKERROR, -1, "HDftruncate failed");

#ifdef H5FD_IOC_DEBUG
    HDprintf("[ioc(%d) %s]: truncated subfile to %lld bytes. ret = %d\n", ioc_idx, __func__,
             (long long)length, errno);
    HDfflush(stdout);
#endif

done:
    H5_SUBFILING_FUNC_LEAVE;
} /* end ioc_file_truncate() */

/*-------------------------------------------------------------------------
 * Function:    ioc_file_report_eof
 *
 * Purpose:     Determine the target subfile's eof and report this value
 *              to the requesting rank.
 *
 *              Notes: This function will have to be reworked once we solve
 *                     the IOC error reporting problem.
 *
 *                     This function mixes functionality that should be
 *                     in two different VFDs.
 *
 * Return:      0 if successful, 1 or an MPI error code on failure.
 *
 * Programmer:  John Mainzer
 *              7/17/2020
 *
 * Changes:     Initial Version/None.
 *
 *-------------------------------------------------------------------------
 */

static int
ioc_file_report_eof(sf_work_request_t *msg, MPI_Comm comm)
{
    subfiling_context_t *sf_context = NULL;
    h5_stat_t            sb;
    int64_t              eof_req_reply[3];
    int64_t              file_context_id;
    int64_t              subfile_idx;
    int                  fd;
    int                  source;
    int                  ioc_idx;
    int                  mpi_code;
    int                  ret_value = 0;

    HDassert(msg);

    file_context_id = msg->context_id;
    source          = msg->source;
    ioc_idx         = msg->ioc_idx;

    subfile_idx = msg->header[0];

    if (NULL == (sf_context = H5_get_subfiling_object(file_context_id)))
        H5_SUBFILING_GOTO_ERROR(H5E_FILE, H5E_CANTGET, -1, "couldn't retrieve subfiling context");

    HDassert(subfile_idx < sf_context->sf_num_fids);

    fd = sf_context->sf_fids[subfile_idx];

    if (HDfstat(fd, &sb) < 0)
        H5_SUBFILING_SYS_GOTO_ERROR(H5E_FILE, H5E_SYSERRSTR, -1, "HDfstat failed");

    eof_req_reply[0] = (int64_t)ioc_idx;
    eof_req_reply[1] = (int64_t)(sb.st_size);
    eof_req_reply[2] = subfile_idx;

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(file_context_id, "%s: reporting file EOF as %" PRId64 ".", __func__, eof_req_reply[1]);
#endif

    /* return the subfile EOF to the querying rank */
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Send(eof_req_reply, 1, H5_subfiling_rpc_msg_type, source, GET_EOF_COMPLETED, comm)))
        H5_SUBFILING_MPI_GOTO_ERROR(-1, "MPI_Send", mpi_code);

done:
    H5_SUBFILING_FUNC_LEAVE;
} /* ioc_file_report_eof() */

/*-------------------------------------------------------------------------
 * Function:    ioc_io_queue_alloc_entry
 *
 * Purpose:     Allocate and initialize an instance of
 *              ioc_io_queue_entry_t.  Return pointer to the new
 *              instance on success, and NULL on failure.
 *
 * Return:      Pointer to new instance of ioc_io_queue_entry_t
 *              on success, and NULL on failure.
 *
 * Programmer:  JRM -- 11/6/21
 *
 * Changes:     None.
 *
 *-------------------------------------------------------------------------
 */
static ioc_io_queue_entry_t *
ioc_io_queue_alloc_entry(void)
{
    ioc_io_queue_entry_t *q_entry_ptr = NULL;

    q_entry_ptr = (ioc_io_queue_entry_t *)HDmalloc(sizeof(ioc_io_queue_entry_t));

    if (q_entry_ptr) {

        q_entry_ptr->magic       = H5FD_IOC__IO_Q_ENTRY_MAGIC;
        q_entry_ptr->next        = NULL;
        q_entry_ptr->prev        = NULL;
        q_entry_ptr->in_progress = FALSE;
        q_entry_ptr->counter     = 0;

        /* will memcpy the wk_req field, so don't bother to initialize */
        /* will initialize thread_wk field before use */

        q_entry_ptr->wk_ret = 0;

#ifdef H5FD_IOC_COLLECT_STATS
        q_entry_ptr->q_time        = 0;
        q_entry_ptr->dispatch_time = 0;
#endif
    }

    return q_entry_ptr;
} /* ioc_io_queue_alloc_entry() */

/*-------------------------------------------------------------------------
 * Function:    ioc_io_queue_add_entry
 *
 * Purpose:     Add an I/O request to the tail of the IOC I/O Queue.
 *
 *              To do this, we must:
 *
 *              1) allocate a new instance of ioc_io_queue_entry_t
 *
 *              2) Initialize the new instance and copy the supplied
 *                 instance of sf_work_request_t into it.
 *
 *              3) Append it to the IOC I/O queue.
 *
 *              Note that this does not dispatch the request even if it
 *              is eligible for immediate dispatch.  This is done with
 *              a call to ioc_io_queue_dispatch_eligible_entries().
 *
 * Return:      void.
 *
 * Programmer:  JRM -- 11/7/21
 *
 * Changes:     None.
 *
 *-------------------------------------------------------------------------
 */
static void
ioc_io_queue_add_entry(ioc_data_t *ioc_data, sf_work_request_t *wk_req_ptr)
{
    ioc_io_queue_entry_t *entry_ptr = NULL;

    HDassert(ioc_data);
    HDassert(ioc_data->io_queue.magic == H5FD_IOC__IO_Q_MAGIC);
    HDassert(wk_req_ptr);

    entry_ptr = ioc_io_queue_alloc_entry();

    HDassert(entry_ptr);
    HDassert(entry_ptr->magic == H5FD_IOC__IO_Q_ENTRY_MAGIC);

    HDmemcpy((void *)(&(entry_ptr->wk_req)), (const void *)wk_req_ptr, sizeof(sf_work_request_t));

    /* must obtain io_queue mutex before appending */
    hg_thread_mutex_lock(&ioc_data->io_queue.q_mutex);

    HDassert(ioc_data->io_queue.q_len == atomic_load(&ioc_data->sf_io_ops_pending));

    entry_ptr->counter = ioc_data->io_queue.req_counter++;

    ioc_data->io_queue.num_pending++;

    H5FD_IOC__Q_APPEND(&ioc_data->io_queue, entry_ptr);

    atomic_fetch_add(&ioc_data->sf_io_ops_pending, 1);

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(
        wk_req_ptr->context_id,
        "%s: request %d queued. op = %d, req = (%lld, %lld, %lld), q-ed/disp/ops_pend = %d/%d/%d.", __func__,
        entry_ptr->counter, (entry_ptr->wk_req.tag), (long long)(entry_ptr->wk_req.header[0]),
        (long long)(entry_ptr->wk_req.header[1]), (long long)(entry_ptr->wk_req.header[2]),
        ioc_data->io_queue.num_pending, ioc_data->io_queue.num_in_progress,
        atomic_load(&ioc_data->sf_io_ops_pending));
#endif

    HDassert(ioc_data->io_queue.num_pending + ioc_data->io_queue.num_in_progress == ioc_data->io_queue.q_len);

#ifdef H5FD_IOC_COLLECT_STATS
    entry_ptr->q_time = H5_now_usec();

    if (ioc_data->io_queue.q_len > ioc_data->io_queue.max_q_len) {
        ioc_data->io_queue.max_q_len = ioc_data->io_queue.q_len;
    }

    if (ioc_data->io_queue.num_pending > ioc_data->io_queue.max_num_pending) {
        ioc_data->io_queue.max_num_pending = ioc_data->io_queue.num_pending;
    }

    if (entry_ptr->wk_req.tag == READ_INDEP) {
        ioc_data->io_queue.ind_read_requests++;
    }
    else if (entry_ptr->wk_req.tag == WRITE_INDEP) {
        ioc_data->io_queue.ind_write_requests++;
    }
    else if (entry_ptr->wk_req.tag == TRUNC_OP) {
        ioc_data->io_queue.truncate_requests++;
    }
    else if (entry_ptr->wk_req.tag == GET_EOF_OP) {
        ioc_data->io_queue.get_eof_requests++;
    }

    ioc_data->io_queue.requests_queued++;
#endif

#ifdef H5_SUBFILING_DEBUG
    if (ioc_data->io_queue.q_len != atomic_load(&ioc_data->sf_io_ops_pending)) {
        H5_subfiling_log(
            wk_req_ptr->context_id,
            "%s: ioc_data->io_queue->q_len = %d != %d = atomic_load(&ioc_data->sf_io_ops_pending).", __func__,
            ioc_data->io_queue.q_len, atomic_load(&ioc_data->sf_io_ops_pending));
    }
#endif

    HDassert(ioc_data->io_queue.q_len == atomic_load(&ioc_data->sf_io_ops_pending));

    hg_thread_mutex_unlock(&ioc_data->io_queue.q_mutex);

    return;
} /* ioc_io_queue_add_entry() */

/*-------------------------------------------------------------------------
 * Function:    ioc_io_queue_dispatch_eligible_entries
 *
 * Purpose:     Scan the IOC I/O Queue for dispatchable entries, and
 *              dispatch any such entries found.
 *
 *              Do this by scanning the I/O queue from head to tail for
 *              entries that:
 *
 *              1) Have not already been dispatched
 *
 *              2) Either:
 *
 *                 a) do not intersect with any prior entries on the
 *                    I/O queue, or
 *
 *                 b) Are read requests, and all intersections are with
 *                    prior read requests.
 *
 *              Dispatch any such entries found.
 *
 *              Do this to maintain the POSIX semantics required by
 *              HDF5.
 *
 *              Note that TRUNC_OPs and GET_EOF_OPs are a special case.
 *              Specifically, no I/O queue entry can be dispatched if
 *              there is a truncate or get EOF operation between it and
 *              the head of the queue.  Further, a truncate or get EOF
 *              request cannot be executed unless it is at the head of
 *              the queue.
 *
 * Return:      void.
 *
 * Programmer:  JRM -- 11/7/21
 *
 * Changes:     None.
 *
 *-------------------------------------------------------------------------
 */
/* TODO: Keep an eye on statistics and optimize this algorithm if necessary.  While it is O(N)
 *       where N is the number of elements in the I/O Queue if there are are no-overlaps, it
 *       can become O(N**2) in the worst case.
 */
static void
ioc_io_queue_dispatch_eligible_entries(ioc_data_t *ioc_data, hbool_t try_lock)
{
    hbool_t               conflict_detected;
    int64_t               entry_offset;
    int64_t               entry_len;
    int64_t               scan_offset;
    int64_t               scan_len;
    ioc_io_queue_entry_t *entry_ptr = NULL;
    ioc_io_queue_entry_t *scan_ptr  = NULL;

    HDassert(ioc_data);
    HDassert(ioc_data->io_queue.magic == H5FD_IOC__IO_Q_MAGIC);

    if (try_lock) {
        if (hg_thread_mutex_try_lock(&ioc_data->io_queue.q_mutex) < 0)
            return;
    }
    else
        hg_thread_mutex_lock(&ioc_data->io_queue.q_mutex);

    entry_ptr = ioc_data->io_queue.q_head;

    /* sanity check on first element in the I/O queue */
    HDassert((entry_ptr == NULL) || (entry_ptr->prev == NULL));

    while ((entry_ptr) && (ioc_data->io_queue.num_pending > 0)) {

        HDassert(entry_ptr->magic == H5FD_IOC__IO_Q_ENTRY_MAGIC);

        /* Check for a get EOF or truncate operation at head of queue */
        if (ioc_data->io_queue.q_head->in_progress) {
            if ((ioc_data->io_queue.q_head->wk_req.tag == TRUNC_OP) ||
                (ioc_data->io_queue.q_head->wk_req.tag == GET_EOF_OP)) {

                /* we have a truncate or get eof operation in progress -- thus no other operations
                 * can be dispatched until the truncate or get eof operation completes.  Just break
                 * out of the loop.
                 */

                break;
            }
        }

        if (!entry_ptr->in_progress) {

            entry_offset = entry_ptr->wk_req.header[1];
            entry_len    = entry_ptr->wk_req.header[0];

            conflict_detected = FALSE;

            scan_ptr = entry_ptr->prev;

            HDassert((scan_ptr == NULL) || (scan_ptr->magic == H5FD_IOC__IO_Q_ENTRY_MAGIC));

            if ((entry_ptr->wk_req.tag == TRUNC_OP) || (entry_ptr->wk_req.tag == GET_EOF_OP)) {

                if (scan_ptr != NULL) {

                    /* the TRUNC_OP or GET_EOF_OP is not at the head of the queue, and thus cannot
                     * be dispatched.  Further, no operation can be dispatched if a truncate request
                     * appears before it in the queue.  Thus we have done all we can and will break
                     * out of the loop.
                     */
                    break;
                }
            }

            while ((scan_ptr) && (!conflict_detected)) {

                /* check for overlaps */
                scan_offset = scan_ptr->wk_req.header[1];
                scan_len    = scan_ptr->wk_req.header[0];

                /* at present, I/O requests are scalar -- i.e. single blocks specified by offset and length.
                 * when this changes, this if statement will have to be updated accordingly.
                 */
                if (((scan_offset + scan_len) > entry_offset) && ((entry_offset + entry_len) > scan_offset)) {

                    /* the two request overlap -- unless they are both reads, we have detected a conflict */

                    /* TODO: update this if statement when we add collective I/O */
                    if ((entry_ptr->wk_req.tag != READ_INDEP) || (scan_ptr->wk_req.tag != READ_INDEP)) {

                        conflict_detected = TRUE;
                    }
                }

                scan_ptr = scan_ptr->prev;
            }

            if (!conflict_detected) { /* dispatch I/O request */

                HDassert(scan_ptr == NULL);
                HDassert(!entry_ptr->in_progress);

                entry_ptr->in_progress = TRUE;

                HDassert(ioc_data->io_queue.num_pending > 0);

                ioc_data->io_queue.num_pending--;
                ioc_data->io_queue.num_in_progress++;

                HDassert(ioc_data->io_queue.num_pending + ioc_data->io_queue.num_in_progress ==
                         ioc_data->io_queue.q_len);

                entry_ptr->thread_wk.func = handle_work_request;
                entry_ptr->thread_wk.args = entry_ptr;

#ifdef H5_SUBFILING_DEBUG
                H5_subfiling_log(
                    entry_ptr->wk_req.context_id,
                    "%s: request %d dispatched. op = %d, req = (%lld, %lld, %lld), "
                    "q-ed/disp/ops_pend = %d/%d/%d.",
                    __func__, entry_ptr->counter, (entry_ptr->wk_req.tag),
                    (long long)(entry_ptr->wk_req.header[0]), (long long)(entry_ptr->wk_req.header[1]),
                    (long long)(entry_ptr->wk_req.header[2]), ioc_data->io_queue.num_pending,
                    ioc_data->io_queue.num_in_progress, atomic_load(&ioc_data->sf_io_ops_pending));
#endif

#ifdef H5FD_IOC_COLLECT_STATS
                if (ioc_data->io_queue.num_in_progress > ioc_data->io_queue.max_num_in_progress) {
                    ioc_data->io_queue.max_num_in_progress = ioc_data->io_queue.num_in_progress;
                }

                ioc_data->io_queue.requests_dispatched++;

                entry_ptr->dispatch_time = H5_now_usec();
#endif

                hg_thread_pool_post(ioc_data->io_thread_pool, &(entry_ptr->thread_wk));
            }
        }

        entry_ptr = entry_ptr->next;
    }

    HDassert(ioc_data->io_queue.q_len == atomic_load(&ioc_data->sf_io_ops_pending));

    hg_thread_mutex_unlock(&ioc_data->io_queue.q_mutex);
} /* ioc_io_queue_dispatch_eligible_entries() */

/*-------------------------------------------------------------------------
 * Function:    ioc_io_queue_complete_entry
 *
 * Purpose:     Update the IOC I/O Queue for the completion of an I/O
 *              request.
 *
 *              To do this:
 *
 *              1) Remove the entry from the I/O Queue
 *
 *              2) If so configured, update statistics
 *
 *              3) Discard the instance of ioc_io_queue_entry_t.
 *
 * Return:      void.
 *
 * Programmer:  JRM -- 11/7/21
 *
 * Changes:     None.
 *
 *-------------------------------------------------------------------------
 */
static void
ioc_io_queue_complete_entry(ioc_data_t *ioc_data, ioc_io_queue_entry_t *entry_ptr)
{
#ifdef H5FD_IOC_COLLECT_STATS
    uint64_t queued_time;
    uint64_t execution_time;
#endif

    HDassert(ioc_data);
    HDassert(ioc_data->io_queue.magic == H5FD_IOC__IO_Q_MAGIC);
    HDassert(entry_ptr);
    HDassert(entry_ptr->magic == H5FD_IOC__IO_Q_ENTRY_MAGIC);

    /* must obtain io_queue mutex before deleting and updating stats */
    hg_thread_mutex_lock(&ioc_data->io_queue.q_mutex);

    HDassert(ioc_data->io_queue.num_pending + ioc_data->io_queue.num_in_progress == ioc_data->io_queue.q_len);
    HDassert(ioc_data->io_queue.num_in_progress > 0);

    if (entry_ptr->wk_ret < 0)
        ioc_data->io_queue.num_failed++;

    H5FD_IOC__Q_REMOVE(&ioc_data->io_queue, entry_ptr);

    ioc_data->io_queue.num_in_progress--;

    HDassert(ioc_data->io_queue.num_pending + ioc_data->io_queue.num_in_progress == ioc_data->io_queue.q_len);

    atomic_fetch_sub(&ioc_data->sf_io_ops_pending, 1);

#ifdef H5_SUBFILING_DEBUG
    H5_subfiling_log(entry_ptr->wk_req.context_id,
                     "%s: request %d completed with ret %d. op = %d, req = (%lld, %lld, %lld), "
                     "q-ed/disp/ops_pend = %d/%d/%d.",
                     __func__, entry_ptr->counter, entry_ptr->wk_ret, (entry_ptr->wk_req.tag),
                     (long long)(entry_ptr->wk_req.header[0]), (long long)(entry_ptr->wk_req.header[1]),
                     (long long)(entry_ptr->wk_req.header[2]), ioc_data->io_queue.num_pending,
                     ioc_data->io_queue.num_in_progress, atomic_load(&ioc_data->sf_io_ops_pending));

    /*
     * If this I/O request is a truncate or "get eof" op, make sure
     * there aren't other operations in progress
     */
    if ((entry_ptr->wk_req.tag == GET_EOF_OP) || (entry_ptr->wk_req.tag == TRUNC_OP))
        HDassert(ioc_data->io_queue.num_in_progress == 0);
#endif

    HDassert(ioc_data->io_queue.q_len == atomic_load(&ioc_data->sf_io_ops_pending));

#ifdef H5FD_IOC_COLLECT_STATS
    /* Compute the queued and execution time */
    queued_time    = entry_ptr->dispatch_time - entry_ptr->q_time;
    execution_time = H5_now_usec() - entry_ptr->dispatch_time;

    ioc_data->io_queue.requests_completed++;

    entry_ptr->q_time = H5_now_usec();

#endif

    hg_thread_mutex_unlock(&ioc_data->io_queue.q_mutex);

    ioc_io_queue_free_entry(entry_ptr);

    entry_ptr = NULL;

    return;
} /* ioc_io_queue_complete_entry() */

/*-------------------------------------------------------------------------
 * Function:    ioc_io_queue_free_entry
 *
 * Purpose:     Free the supplied instance of ioc_io_queue_entry_t.
 *
 *              Verify that magic field is set to
 *              H5FD_IOC__IO_Q_ENTRY_MAGIC, and that the next and prev
 *              fields are NULL.
 *
 * Return:      void.
 *
 * Programmer:  JRM -- 11/6/21
 *
 * Changes:     None.
 *
 *-------------------------------------------------------------------------
 */
static void
ioc_io_queue_free_entry(ioc_io_queue_entry_t *q_entry_ptr)
{
    /* use assertions for error checking, since the following should never fail. */
    HDassert(q_entry_ptr);
    HDassert(q_entry_ptr->magic == H5FD_IOC__IO_Q_ENTRY_MAGIC);
    HDassert(q_entry_ptr->next == NULL);
    HDassert(q_entry_ptr->prev == NULL);

    q_entry_ptr->magic = 0;

    HDfree(q_entry_ptr);

    q_entry_ptr = NULL;

    return;
} /* H5FD_ioc__free_c_io_q_entry() */