path: root/src/H5Dmpio.c

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
 * Copyright by The HDF Group.                                               *
 * Copyright by the Board of Trustees of the University of Illinois.         *
 * 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://www.hdfgroup.org/licenses.               *
 * If you do not have access to either file, you may request a copy from     *
 * help@hdfgroup.org.                                                        *
 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

/*
 * Programmer:  rky 980813
 * KY 2005 revised the code and made the change to support and optimize
 * collective IO support.
 * Purpose:    Functions to read/write directly between app buffer and file.
 *
 *         Beware of the ifdef'ed print statements.
 *         I didn't make them portable.
 */

/****************/
/* Module Setup */
/****************/

#include "H5Dmodule.h" /* This source code file is part of the H5D module */

/***********/
/* Headers */
/***********/
#include "H5private.h"   /* Generic Functions */
#include "H5CXprivate.h" /* API Contexts      */
#include "H5Dpkg.h"      /* Datasets          */
#include "H5Eprivate.h"  /* Error handling    */
#include "H5Fprivate.h"  /* File access       */
#include "H5FDprivate.h" /* File drivers      */
#include "H5FLprivate.h" /* Free Lists        */
#include "H5Iprivate.h"  /* IDs               */
#include "H5MMprivate.h" /* Memory management */
#include "H5Oprivate.h"  /* Object headers    */
#include "H5Pprivate.h"  /* Property lists    */
#include "H5Sprivate.h"  /* Dataspaces        */
#include "H5VMprivate.h" /* Vector            */

#ifdef H5_HAVE_PARALLEL

/****************/
/* Local Macros */
/****************/

/* Macros to represent different IO options */
#define H5D_ONE_LINK_CHUNK_IO          0
#define H5D_MULTI_CHUNK_IO             1
#define H5D_ONE_LINK_CHUNK_IO_MORE_OPT 2
#define H5D_MULTI_CHUNK_IO_MORE_OPT    3
#define H5D_NO_IO                      4

/***** Macros for One linked collective IO case. *****/
/* The default value to do one linked collective IO for all chunks.
 * If the average number of chunks per process is greater than this
 * value, the library will create an MPI derived datatype to link all
 * chunks to do collective IO.  The user can set this value through an
 * API.
 */

/* Macros to represent options on how to obtain chunk address for one linked-chunk IO case */
#define H5D_OBTAIN_ONE_CHUNK_ADDR_IND 0
#define H5D_OBTAIN_ALL_CHUNK_ADDR_COL 2

/* Macros to define the default ratio of obtaining all chunk addresses for one linked-chunk IO case */
#define H5D_ALL_CHUNK_ADDR_THRES_COL     30
#define H5D_ALL_CHUNK_ADDR_THRES_COL_NUM 10000

/***** Macros for multi-chunk collective IO case. *****/
/* The default value of the threshold to do collective IO for this
 *  chunk.  If the average number of processes per chunk is greater
 *  than the default value, collective IO is done for this chunk.
 */

/* Macros to represent different IO modes(NONE, Independent or collective)for multiple chunk IO case */
#define H5D_CHUNK_IO_MODE_COL 1

/* Macros to represent the regularity of the selection for multiple chunk IO case. */
#define H5D_CHUNK_SELECT_REG 1

/*
 * Threshold value for redistributing shared filtered chunks
 * on all MPI ranks, or just MPI rank 0
 */
#define H5D_CHUNK_REDISTRIBUTE_THRES ((size_t)((25 * H5_MB) / sizeof(H5D_chunk_redistribute_info_t)))

/*
 * Initial allocation size for the arrays that hold
 * buffers for chunk modification data that is sent
 * to other ranks and the MPI_Request objects for
 * those send operations
 */
#define H5D_CHUNK_NUM_SEND_MSGS_INIT 64

/*
 * Define a tag value for the MPI messages sent/received for
 * chunk modification data
 */
#define H5D_CHUNK_MOD_DATA_TAG 64

/*
 * Macro to initialize a H5D_chk_idx_info_t
 * structure, given a pointer to a H5D_io_info_t
 * structure
 */
#define H5D_MPIO_INIT_CHUNK_IDX_INFO(index_info, dset)                                                       \
    do {                                                                                                     \
        index_info.f       = (dset)->oloc.file;                                                              \
        index_info.pline   = &((dset)->shared->dcpl_cache.pline);                                            \
        index_info.layout  = &((dset)->shared->layout.u.chunk);                                              \
        index_info.storage = &((dset)->shared->layout.storage.u.chunk);                                      \
    } while (0)

/*
 * Macro to initialize a H5D_chunk_ud_t structure
 * given a pointer to a H5D_chk_idx_info_t structure
 */
#define H5D_MPIO_INIT_CHUNK_UD_INFO(chunk_ud, index_info_ptr)                                                \
    do {                                                                                                     \
        HDmemset(&chunk_ud, 0, sizeof(H5D_chunk_ud_t));                                                      \
        chunk_ud.common.layout  = (index_info_ptr)->layout;                                                  \
        chunk_ud.common.storage = (index_info_ptr)->storage;                                                 \
    } while (0)

/******************/
/* Local Typedefs */
/******************/

/* Combine chunk/piece address and chunk/piece info into a struct for
 * better performance. */
typedef struct H5D_chunk_addr_info_t {
    /* piece for multi-dset */
    haddr_t          piece_addr;
    H5D_piece_info_t piece_info;
} H5D_chunk_addr_info_t;

/* Rank 0 Bcast values */
typedef enum H5D_mpio_no_rank0_bcast_cause_t {
    H5D_MPIO_RANK0_BCAST            = 0x00,
    H5D_MPIO_RANK0_NOT_H5S_ALL      = 0x01,
    H5D_MPIO_RANK0_NOT_CONTIGUOUS   = 0x02,
    H5D_MPIO_RANK0_NOT_FIXED_SIZE   = 0x04,
    H5D_MPIO_RANK0_GREATER_THAN_2GB = 0x08
} H5D_mpio_no_rank0_bcast_cause_t;

/*
 * Information necessary for re-allocating file space for a chunk
 * during a parallel write of a chunked dataset with filters
 * applied.
 */
typedef struct H5D_chunk_alloc_info_t {
    H5F_block_t chunk_current;
    H5F_block_t chunk_new;
    hsize_t     chunk_idx;
} H5D_chunk_alloc_info_t;

/*
 * Information for a chunk pertaining to the dataset's chunk
 * index entry for the chunk
 */
typedef struct H5D_chunk_index_info_t {
    hsize_t  chunk_idx;
    unsigned filter_mask;
    hbool_t  need_insert;
} H5D_chunk_index_info_t;

/*
 * Information about a single chunk when performing collective filtered I/O. All
 * of the fields of one of these structs are initialized at the start of collective
 * filtered I/O in the function H5D__mpio_collective_filtered_chunk_io_setup(). This
 * struct's fields are as follows:
 *
 * index_info - A structure containing the information needed when collectively
 *              re-inserting the chunk into the dataset's chunk index. The structure
 *              is distributed to all ranks during the re-insertion operation. Its fields
 *              are as follows:
 *
 *     chunk_idx - The index of the chunk in the dataset's chunk index.
 *
 *     filter_mask - A bit-mask that indicates which filters are to be applied to the
 *                   chunk. Each filter in a chunk's filter pipeline has a bit position
 *                   that can be masked to disable that particular filter for the chunk.
 *                   This filter mask is saved alongside the chunk in the file.
 *
 *     need_insert - A flag which determines whether or not a chunk needs to be re-inserted into
 *                   the chunk index after the write operation.
 *
 * chunk_info - A pointer to the chunk's H5D_piece_info_t structure, which contains useful
 *              information like the dataspaces containing the selection in the chunk.
 *
 * chunk_current - The address in the file and size of this chunk before the filtering
 *                 operation. When reading a chunk from the file, this field is used to
 *                 read the correct amount of bytes. It is also used when redistributing
 *                 shared chunks among MPI ranks and as a parameter to the chunk file
 *                 space reallocation function.
 *
 * chunk_new - The address in the file and size of this chunk after the filtering
 *             operation. This field is relevant when collectively re-allocating space
 *             in the file for all of the chunks written to in the I/O operation, as
 *             their sizes may have changed after their data has been filtered.
 *
 * need_read - A flag which determines whether or not a chunk needs to be read from the
 *             file. During writes, if a chunk is being fully overwritten (the entire extent
 *             is selected in its file dataspace), then it is not necessary to read the chunk
 *             from the file. However, if the chunk is not being fully overwritten, it has to
 *             be read from the file in order to update the chunk without trashing the parts
 *             of the chunk that are not selected. During reads, this field should generally
 *             be true, but may be false if the chunk isn't allocated, for example.
 *
 * skip_filter_pline - A flag which determines whether to skip calls to the filter pipeline
 *                     for this chunk. This flag is mostly useful for correct handling of
 *                     partial edge chunks when the "don't filter partial edge chunks" flag
 *                     is set on the dataset's DCPL.
 *
 * io_size - The total size of I/O to this chunk. This field is an accumulation of the size of
 *           I/O to the chunk from each MPI rank which has the chunk selected and is used to
 *           determine the value for the previous `full_overwrite` flag.
 *
 * chunk_buf_size - The size in bytes of the data buffer allocated for the chunk
 *
 * orig_owner - The MPI rank which originally had this chunk selected at the beginning of
 *              the collective filtered I/O operation. This field is currently used when
 *              redistributing shared chunks among MPI ranks.
 *
 * new_owner - The MPI rank which has been selected to perform the modifications to this chunk.
 *
 * num_writers - The total number of MPI ranks writing to this chunk. This field is used when
 *               the new owner of a chunk is receiving messages from other MPI ranks that
 *               contain their selections in the chunk and the data to update the chunk with.
 *               The new owner must know how many MPI ranks it should expect messages from so
 *               that it can post an equal number of receive calls.
 *
 * buf - A pointer which serves the dual purpose of holding either the chunk data which is to be
 *       written to the file or the chunk data which has been read from the file.
 *
 * hh - A handle for hash tables provided by the uthash.h header
 *
 */
typedef struct H5D_filtered_collective_io_info_t {
    H5D_chunk_index_info_t index_info;

    H5D_piece_info_t *chunk_info;
    H5F_block_t       chunk_current;
    H5F_block_t       chunk_new;
    hbool_t           need_read;
    hbool_t           skip_filter_pline;
    size_t            io_size;
    size_t            chunk_buf_size;
    int               orig_owner;
    int               new_owner;
    int               num_writers;
    void             *buf;

    UT_hash_handle hh;
} H5D_filtered_collective_io_info_t;

/*
 * Information necessary for redistributing shared chunks during
 * a parallel write of a chunked dataset with filters applied.
 */
typedef struct H5D_chunk_redistribute_info_t {
    H5F_block_t chunk_block;
    hsize_t     chunk_idx;
    int         orig_owner;
    int         new_owner;
    int         num_writers;
} H5D_chunk_redistribute_info_t;

/*
 * Information used when re-inserting a chunk into a dataset's
 * chunk index during a parallel write of a chunked dataset with
 * filters applied.
 */
typedef struct H5D_chunk_insert_info_t {
    H5F_block_t            chunk_block;
    H5D_chunk_index_info_t index_info;
} H5D_chunk_insert_info_t;

/********************/
/* Local Prototypes */
/********************/
static herr_t H5D__piece_io(H5D_io_info_t *io_info);
static herr_t H5D__multi_chunk_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info,
                                             int mpi_rank, int mpi_size);
static herr_t H5D__multi_chunk_filtered_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info,
                                                      int mpi_rank, int mpi_size);
static herr_t H5D__link_piece_collective_io(H5D_io_info_t *io_info, int mpi_rank);
static herr_t H5D__link_chunk_filtered_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info,
                                                     int mpi_rank, int mpi_size);
static herr_t H5D__inter_collective_io(H5D_io_info_t *io_info, const H5D_dset_io_info_t *di,
                                       H5S_t *file_space, H5S_t *mem_space);
static herr_t H5D__final_collective_io(H5D_io_info_t *io_info, hsize_t mpi_buf_count,
                                       MPI_Datatype mpi_file_type, MPI_Datatype mpi_buf_type);
static herr_t H5D__obtain_mpio_mode(H5D_io_info_t *io_info, H5D_dset_io_info_t *di, uint8_t assign_io_mode[],
                                    haddr_t chunk_addr[], int mpi_rank, int mpi_size);
static herr_t H5D__mpio_get_sum_chunk(const H5D_io_info_t *io_info, int *sum_chunkf);
static herr_t H5D__mpio_get_sum_chunk_dset(const H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info,
                                           int *sum_chunkf);
static herr_t H5D__mpio_collective_filtered_chunk_io_setup(const H5D_io_info_t                *io_info,
                                                           const H5D_dset_io_info_t           *di,
                                                           H5D_filtered_collective_io_info_t **chunk_list,
                                                           size_t *num_entries, int mpi_rank);
static herr_t H5D__mpio_redistribute_shared_chunks(H5D_filtered_collective_io_info_t *chunk_list,
                                                   size_t                             chunk_list_num_entries,
                                                   const H5D_io_info_t *io_info, int mpi_rank, int mpi_size,
                                                   size_t **rank_chunks_assigned_map);
static herr_t H5D__mpio_redistribute_shared_chunks_int(H5D_filtered_collective_io_info_t *chunk_list,
                                                       size_t              *num_chunks_assigned_map,
                                                       hbool_t              all_ranks_involved,
                                                       const H5D_io_info_t *io_info, int mpi_rank,
                                                       int mpi_size);
static herr_t H5D__mpio_share_chunk_modification_data(H5D_filtered_collective_io_info_t *chunk_list,
                                                      size_t *chunk_list_num_entries, H5D_io_info_t *io_info,
                                                      H5D_dset_io_info_t *dset_info, int mpi_rank,
                                                      int                                 mpi_size,
                                                      H5D_filtered_collective_io_info_t **chunk_hash_table,
                                                      unsigned char                    ***chunk_msg_bufs,
                                                      int                                *chunk_msg_bufs_len);
static herr_t H5D__mpio_collective_filtered_chunk_common_io(H5D_filtered_collective_io_info_t *chunk_list,
                                                            size_t               chunk_list_num_entries,
                                                            const H5D_io_info_t *io_info, int mpi_size);
static herr_t H5D__mpio_collective_filtered_chunk_read(H5D_filtered_collective_io_info_t *chunk_list,
                                                       size_t                    chunk_list_num_entries,
                                                       const H5D_io_info_t      *io_info,
                                                       const H5D_dset_io_info_t *di, int mpi_rank,
                                                       int mpi_size);
static herr_t H5D__mpio_collective_filtered_chunk_update(H5D_filtered_collective_io_info_t *chunk_list,
                                                         size_t chunk_list_num_entries,
                                                         H5D_filtered_collective_io_info_t *chunk_hash_table,
                                                         unsigned char                    **chunk_msg_bufs,
                                                         int chunk_msg_bufs_len, const H5D_io_info_t *io_info,
                                                         const H5D_dset_io_info_t *di, int mpi_rank,
                                                         int mpi_size);
static herr_t H5D__mpio_collective_filtered_chunk_reallocate(H5D_filtered_collective_io_info_t *chunk_list,
                                                             size_t              chunk_list_num_entries,
                                                             size_t             *num_chunks_assigned_map,
                                                             H5D_io_info_t      *io_info,
                                                             H5D_chk_idx_info_t *idx_info, int mpi_rank,
                                                             int mpi_size);
static herr_t H5D__mpio_collective_filtered_chunk_reinsert(H5D_filtered_collective_io_info_t *chunk_list,
                                                           size_t         chunk_list_num_entries,
                                                           size_t        *num_chunks_assigned_map,
                                                           H5D_io_info_t *io_info, H5D_dset_io_info_t *di,
                                                           H5D_chk_idx_info_t *idx_info, int mpi_rank,
                                                           int mpi_size);
static herr_t H5D__mpio_get_chunk_redistribute_info_types(MPI_Datatype *contig_type,
                                                          hbool_t      *contig_type_derived,
                                                          MPI_Datatype *resized_type,
                                                          hbool_t      *resized_type_derived);
static herr_t H5D__mpio_get_chunk_alloc_info_types(MPI_Datatype *contig_type, hbool_t *contig_type_derived,
                                                   MPI_Datatype *resized_type, hbool_t *resized_type_derived);
static herr_t H5D__mpio_get_chunk_insert_info_types(MPI_Datatype *contig_type, hbool_t *contig_type_derived,
                                                    MPI_Datatype *resized_type,
                                                    hbool_t      *resized_type_derived);
static herr_t H5D__mpio_collective_filtered_io_type(H5D_filtered_collective_io_info_t *chunk_list,
                                                    size_t num_entries, H5D_io_op_type_t op_type,
                                                    MPI_Datatype *new_mem_type, hbool_t *mem_type_derived,
                                                    MPI_Datatype *new_file_type, hbool_t *file_type_derived);
static int    H5D__cmp_piece_addr(const void *chunk_addr_info1, const void *chunk_addr_info2);
static int    H5D__cmp_filtered_collective_io_info_entry(const void *filtered_collective_io_info_entry1,
                                                         const void *filtered_collective_io_info_entry2);
static int    H5D__cmp_chunk_redistribute_info(const void *entry1, const void *entry2);
static int    H5D__cmp_chunk_redistribute_info_orig_owner(const void *entry1, const void *entry2);

#ifdef H5Dmpio_DEBUG
static herr_t H5D__mpio_debug_init(void);
static herr_t H5D__mpio_dump_collective_filtered_chunk_list(H5D_filtered_collective_io_info_t *chunk_list,
                                                            size_t chunk_list_num_entries, int mpi_rank);
#endif

/*********************/
/* Package Variables */
/*********************/

/*******************/
/* Local Variables */
/*******************/

/* Declare extern free list to manage the H5S_sel_iter_t struct */
H5FL_EXTERN(H5S_sel_iter_t);

#ifdef H5Dmpio_DEBUG

/* Flags to control debug actions in this file.
 * (Meant to be indexed by characters)
 *
 * These flags can be set with either (or both) the environment variable
 *      "H5D_mpio_Debug" set to a string containing one or more characters
 *      (flags) or by setting them as a string value for the
 *      "H5D_mpio_debug_key" MPI Info key.
 *
 * Supported characters in 'H5D_mpio_Debug' string:
 *      't' trace function entry and exit
 *      'f' log to file rather than debugging stream
 *      'm' show (rough) memory usage statistics
 *      'c' show critical timing information
 *
 *      To only show output from a particular MPI rank, specify its rank
 *      number as a character, e.g.:
 *
 *      '0' only show output from rank 0
 *
 *      To only show output from a particular range (up to 8 ranks supported
 *      between 0-9) of MPI ranks, specify the start and end ranks separated
 *      by a hyphen, e.g.:
 *
 *      '0-7' only show output from ranks 0 through 7
 *
 */
static int               H5D_mpio_debug_flags_s[256];
static int               H5D_mpio_debug_rank_s[8] = {-1, -1, -1, -1, -1, -1, -1, -1};
static hbool_t           H5D_mpio_debug_inited    = FALSE;
static const char *const trace_in_pre             = "-> ";
static const char *const trace_out_pre            = "<- ";
static int               debug_indent             = 0;
static FILE             *debug_stream             = NULL;

/* Determine if this rank should output debugging info */
#define H5D_MPIO_DEBUG_THIS_RANK(rank)                                                                       \
    (H5D_mpio_debug_rank_s[0] < 0 || rank == H5D_mpio_debug_rank_s[0] || rank == H5D_mpio_debug_rank_s[1] || \
     rank == H5D_mpio_debug_rank_s[2] || rank == H5D_mpio_debug_rank_s[3] ||                                 \
     rank == H5D_mpio_debug_rank_s[4] || rank == H5D_mpio_debug_rank_s[5] ||                                 \
     rank == H5D_mpio_debug_rank_s[6] || rank == H5D_mpio_debug_rank_s[7])

/* Print some debugging string */
#define H5D_MPIO_DEBUG(rank, string)                                                                         \
    do {                                                                                                     \
        if (debug_stream && H5D_MPIO_DEBUG_THIS_RANK(rank)) {                                                \
            HDfprintf(debug_stream, "%*s(Rank %d) " string "\n", debug_indent, "", rank);                    \
            HDfflush(debug_stream);                                                                          \
        }                                                                                                    \
    } while (0)

/* Print some debugging string with printf-style arguments */
#define H5D_MPIO_DEBUG_VA(rank, string, ...)                                                                 \
    do {                                                                                                     \
        if (debug_stream && H5D_MPIO_DEBUG_THIS_RANK(rank)) {                                                \
            HDfprintf(debug_stream, "%*s(Rank %d) " string "\n", debug_indent, "", rank, __VA_ARGS__);       \
            HDfflush(debug_stream);                                                                          \
        }                                                                                                    \
    } while (0)

#define H5D_MPIO_TRACE_ENTER(rank)                                                                           \
    do {                                                                                                     \
        hbool_t trace_flag = H5D_mpio_debug_flags_s[(int)'t'];                                               \
                                                                                                             \
        if (trace_flag) {                                                                                    \
            H5D_MPIO_DEBUG_VA(rank, "%s%s", trace_in_pre, __func__);                                         \
            debug_indent += (int)HDstrlen(trace_in_pre);                                                     \
        }                                                                                                    \
    } while (0)

#define H5D_MPIO_TRACE_EXIT(rank)                                                                            \
    do {                                                                                                     \
        hbool_t trace_flag = H5D_mpio_debug_flags_s[(int)'t'];                                               \
                                                                                                             \
        if (trace_flag) {                                                                                    \
            debug_indent -= (int)HDstrlen(trace_out_pre);                                                    \
            H5D_MPIO_DEBUG_VA(rank, "%s%s", trace_out_pre, __func__);                                        \
        }                                                                                                    \
    } while (0)

#define H5D_MPIO_TIME_START(rank, op_name)                                                                   \
    {                                                                                                        \
        hbool_t           time_flag  = H5D_mpio_debug_flags_s[(int)'c'];                                     \
        double            start_time = 0.0, end_time = 0.0;                                                  \
        const char *const op = op_name;                                                                      \
                                                                                                             \
        if (time_flag) {                                                                                     \
            start_time = MPI_Wtime();                                                                        \
        }

#define H5D_MPIO_TIME_STOP(rank)                                                                             \
    if (time_flag) {                                                                                         \
        end_time = MPI_Wtime();                                                                              \
        H5D_MPIO_DEBUG_VA(rank, "'%s' took %f seconds", op, (end_time - start_time));                        \
    }                                                                                                        \
    }

/*---------------------------------------------------------------------------
 * Function:    H5D__mpio_parse_debug_str
 *
 * Purpose:     Parse a string for H5Dmpio-related debugging flags
 *
 * Returns:     N/A
 *
 *---------------------------------------------------------------------------
 */
static void
H5D__mpio_parse_debug_str(const char *s)
{
    FUNC_ENTER_PACKAGE_NOERR

    HDassert(s);

    while (*s) {
        int c = (int)(*s);

        if (c >= (int)'0' && c <= (int)'9') {
            hbool_t range = FALSE;

            if (*(s + 1) && *(s + 2))
                range = (int)*(s + 1) == '-' && (int)*(s + 2) >= (int)'0' && (int)*(s + 2) <= (int)'9';

            if (range) {
                int start_rank = c - (int)'0';
                int end_rank   = (int)*(s + 2) - '0';
                int num_ranks  = end_rank - start_rank + 1;
                int i;

                if (num_ranks > 8) {
                    end_rank  = start_rank + 7;
                    num_ranks = 8;
                }

                for (i = 0; i < num_ranks; i++)
                    H5D_mpio_debug_rank_s[i] = start_rank++;

                s += 3;
            }
            else
                H5D_mpio_debug_rank_s[0] = c - (int)'0';
        }
        else
            H5D_mpio_debug_flags_s[c]++;

        s++;
    }

    FUNC_LEAVE_NOAPI_VOID
}

static herr_t
H5D__mpio_debug_init(void)
{
    const char *debug_str;
    herr_t      ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE_NOERR

    HDassert(!H5D_mpio_debug_inited);

    /* Clear the debug flag buffer */
    HDmemset(H5D_mpio_debug_flags_s, 0, sizeof(H5D_mpio_debug_flags_s));

    /* Retrieve and parse the H5Dmpio debug string */
    debug_str = HDgetenv("H5D_mpio_Debug");
    if (debug_str)
        H5D__mpio_parse_debug_str(debug_str);

    if (H5DEBUG(D))
        debug_stream = H5DEBUG(D);

    H5D_mpio_debug_inited = TRUE;

    FUNC_LEAVE_NOAPI(ret_value)
}

#endif

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_opt_possible
 *
 * Purpose:     Checks if an direct I/O transfer is possible between memory and
 *              the file.
 *
 *              This was derived from H5D__mpio_opt_possible for
 *              multi-dset work.
 *
 * Return:      Success:   Non-negative: TRUE or FALSE
 *              Failure:    Negative
 *
 *-------------------------------------------------------------------------
 */
htri_t
H5D__mpio_opt_possible(H5D_io_info_t *io_info)
{
    H5FD_mpio_xfer_t io_xfer_mode; /* MPI I/O transfer mode */
    size_t           i;
    H5D_t           *dset;
    const H5S_t     *file_space;
    const H5S_t     *mem_space;
    H5D_type_info_t *type_info;
    unsigned         local_cause[2] = {0, 0}; /* [0] Local reason(s) for breaking collective mode */
                                              /* [1] Flag if dataset is both: H5S_ALL and small */
    unsigned global_cause[2] = {0, 0};        /* Global reason(s) for breaking collective mode */
    htri_t   is_vl_storage;    /* Whether the dataset's datatype is stored in a variable-length form */
    htri_t   ret_value = TRUE; /* Return value */

    FUNC_ENTER_PACKAGE

    /* Check args */
    HDassert(io_info);

    for (i = 0; i < io_info->count; i++) {
        HDassert(io_info->dsets_info[i].file_space);
        HDassert(io_info->dsets_info[i].mem_space);
    }

    /* For independent I/O, get out quickly and don't try to form consensus */
    if (H5CX_get_io_xfer_mode(&io_xfer_mode) < 0)
        /* Set error flag, but keep going */
        local_cause[0] |= H5D_MPIO_ERROR_WHILE_CHECKING_COLLECTIVE_POSSIBLE;
    if (io_xfer_mode == H5FD_MPIO_INDEPENDENT)
        local_cause[0] |= H5D_MPIO_SET_INDEPENDENT;

    for (i = 0; i < io_info->count; i++) {
        /* Check for skipped I/O */
        if (io_info->dsets_info[i].skip_io)
            continue;

        /* Set convenience pointers */
        dset       = io_info->dsets_info[i].dset;
        file_space = io_info->dsets_info[i].file_space;
        mem_space  = io_info->dsets_info[i].mem_space;
        type_info  = &io_info->dsets_info[i].type_info;

        /* Optimized MPI types flag must be set */
        /* (based on 'HDF5_MPI_OPT_TYPES' environment variable) */
        if (!H5FD_mpi_opt_types_g)
            local_cause[0] |= H5D_MPIO_MPI_OPT_TYPES_ENV_VAR_DISABLED;

        /* Don't allow collective operations if datatype conversions need to happen */
        if (!type_info->is_conv_noop)
            local_cause[0] |= H5D_MPIO_DATATYPE_CONVERSION;

        /* Don't allow collective operations if data transform operations should occur */
        if (!type_info->is_xform_noop)
            local_cause[0] |= H5D_MPIO_DATA_TRANSFORMS;

        /* Check whether these are both simple or scalar dataspaces */
        if (!((H5S_SIMPLE == H5S_GET_EXTENT_TYPE(mem_space) ||
               H5S_SCALAR == H5S_GET_EXTENT_TYPE(mem_space)) &&
              (H5S_SIMPLE == H5S_GET_EXTENT_TYPE(file_space) ||
               H5S_SCALAR == H5S_GET_EXTENT_TYPE(file_space))))
            local_cause[0] |= H5D_MPIO_NOT_SIMPLE_OR_SCALAR_DATASPACES;

        /* Dataset storage must be contiguous or chunked */
        if (!(dset->shared->layout.type == H5D_CONTIGUOUS || dset->shared->layout.type == H5D_CHUNKED))
            local_cause[0] |= H5D_MPIO_NOT_CONTIGUOUS_OR_CHUNKED_DATASET;

        /* check if external-file storage is used */
        if (dset->shared->dcpl_cache.efl.nused > 0)
            local_cause[0] |= H5D_MPIO_NOT_CONTIGUOUS_OR_CHUNKED_DATASET;

            /* The handling of memory space is different for chunking and contiguous
             *  storage.  For contiguous storage, mem_space and file_space won't change
             *  when it it is doing disk IO.  For chunking storage, mem_space will
             *  change for different chunks. So for chunking storage, whether we can
             *  use collective IO will defer until each chunk IO is reached.
             */

#ifndef H5_HAVE_PARALLEL_FILTERED_WRITES
        /* Don't allow writes to filtered datasets if the functionality is disabled */
        if (io_info->op_type == H5D_IO_OP_WRITE && dset->shared->dcpl_cache.pline.nused > 0)
            local_cause[0] |= H5D_MPIO_PARALLEL_FILTERED_WRITES_DISABLED;
#endif

        /* Check if we are able to do a MPI_Bcast of the data from one rank
         * instead of having all the processes involved in the collective I/O call.
         */

        /* Check to see if the process is reading the entire dataset */
        if (H5S_GET_SELECT_TYPE(file_space) != H5S_SEL_ALL)
            local_cause[1] |= H5D_MPIO_RANK0_NOT_H5S_ALL;
        /* Only perform this optimization for contiguous datasets, currently */
        else if (H5D_CONTIGUOUS != dset->shared->layout.type)
            /* Flag to do a MPI_Bcast of the data from one proc instead of
             * having all the processes involved in the collective I/O.
             */
            local_cause[1] |= H5D_MPIO_RANK0_NOT_CONTIGUOUS;
        else if ((is_vl_storage = H5T_is_vl_storage(type_info->dset_type)) < 0)
            local_cause[0] |= H5D_MPIO_ERROR_WHILE_CHECKING_COLLECTIVE_POSSIBLE;
        else if (is_vl_storage)
            local_cause[1] |= H5D_MPIO_RANK0_NOT_FIXED_SIZE;
        else {
            size_t type_size; /* Size of dataset's datatype */

            /* Retrieve the size of the dataset's datatype */
            if (0 == (type_size = H5T_GET_SIZE(type_info->dset_type)))
                local_cause[0] |= H5D_MPIO_ERROR_WHILE_CHECKING_COLLECTIVE_POSSIBLE;
            else {
                hssize_t snelmts; /* [Signed] # of elements in dataset's dataspace */

                /* Retrieve the size of the dataset's datatype */
                if ((snelmts = H5S_GET_EXTENT_NPOINTS(file_space)) < 0)
                    local_cause[0] |= H5D_MPIO_ERROR_WHILE_CHECKING_COLLECTIVE_POSSIBLE;
                else {
                    hsize_t dset_size;

                    /* Determine dataset size */
                    dset_size = ((hsize_t)snelmts) * type_size;

                    /* If the size of the dataset is less than 2GB then do an MPI_Bcast
                     * of the data from one process instead of having all the processes
                     * involved in the collective I/O.
                     */
                    if (dset_size > ((hsize_t)(2.0F * H5_GB) - 1))
                        local_cause[1] |= H5D_MPIO_RANK0_GREATER_THAN_2GB;
                } /* end else */
            }     /* end else */
        }         /* end else */
    }             /* end for loop */

    /* Check for independent I/O */
    if (local_cause[0] & H5D_MPIO_SET_INDEPENDENT)
        global_cause[0] = local_cause[0];
    else {
        int mpi_code; /* MPI error code */

        /* Form consensus opinion among all processes about whether to perform
         * collective I/O
         */
        if (MPI_SUCCESS !=
            (mpi_code = MPI_Allreduce(local_cause, global_cause, 2, MPI_UNSIGNED, MPI_BOR, io_info->comm)))
            HMPI_GOTO_ERROR(FAIL, "MPI_Allreduce failed", mpi_code)
    } /* end else */

    /* Set the local & global values of no-collective-cause in the API context */
    H5CX_set_mpio_local_no_coll_cause(local_cause[0]);
    H5CX_set_mpio_global_no_coll_cause(global_cause[0]);

    /* Set read-with-rank0-and-bcast flag if possible */
    if (global_cause[0] == 0 && global_cause[1] == 0) {
        H5CX_set_mpio_rank0_bcast(TRUE);
#ifdef H5_HAVE_INSTRUMENTED_LIBRARY
        H5CX_test_set_mpio_coll_rank0_bcast(TRUE);
#endif /* H5_HAVE_INSTRUMENTED_LIBRARY */
    }  /* end if */

    /* Set the return value, based on the global cause */
    ret_value = global_cause[0] > 0 ? FALSE : TRUE;

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* H5D__mpio_opt_possible() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_get_no_coll_cause_strings
 *
 * Purpose:     When collective I/O is broken internally, it can be useful
 *              for users to see a representative string for the reason(s)
 *              why it was broken. This routine inspects the current
 *              "cause" flags from the API context and prints strings into
 *              the caller's buffers for the local and global reasons that
 *              collective I/O was broken.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5D__mpio_get_no_coll_cause_strings(char *local_cause, size_t local_cause_len, char *global_cause,
                                    size_t global_cause_len)
{
    uint32_t local_no_coll_cause;
    uint32_t global_no_coll_cause;
    size_t   local_cause_bytes_written  = 0;
    size_t   global_cause_bytes_written = 0;
    int      nbits;
    herr_t   ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert((local_cause && local_cause_len > 0) || (global_cause && global_cause_len > 0));

    /*
     * Use compile-time assertion so this routine is updated
     * when any new "no collective cause" values are added
     */
    HDcompile_assert(H5D_MPIO_NO_COLLECTIVE_MAX_CAUSE == (H5D_mpio_no_collective_cause_t)256);

    /* Initialize output buffers */
    if (local_cause)
        *local_cause = '\0';
    if (global_cause)
        *global_cause = '\0';

    /* Retrieve the local and global cause flags from the API context */
    if (H5CX_get_mpio_local_no_coll_cause(&local_no_coll_cause) < 0)
        HGOTO_ERROR(H5E_CONTEXT, H5E_CANTGET, FAIL, "unable to get local no collective cause value")
    if (H5CX_get_mpio_global_no_coll_cause(&global_no_coll_cause) < 0)
        HGOTO_ERROR(H5E_CONTEXT, H5E_CANTGET, FAIL, "unable to get global no collective cause value")

    /*
     * Append each of the "reason for breaking collective I/O"
     * error messages to the local and global cause string buffers
     */
    nbits = 8 * sizeof(local_no_coll_cause);
    for (int bit_pos = 0; bit_pos < nbits; bit_pos++) {
        H5D_mpio_no_collective_cause_t cur_cause;
        const char                    *cause_str;
        size_t                         buf_space_left;

        cur_cause = (H5D_mpio_no_collective_cause_t)(1 << bit_pos);
        if (cur_cause == H5D_MPIO_NO_COLLECTIVE_MAX_CAUSE)
            break;

        switch (cur_cause) {
            case H5D_MPIO_SET_INDEPENDENT:
                cause_str = "independent I/O was requested";
                break;
            case H5D_MPIO_DATATYPE_CONVERSION:
                cause_str = "datatype conversions were required";
                break;
            case H5D_MPIO_DATA_TRANSFORMS:
                cause_str = "data transforms needed to be applied";
                break;
            case H5D_MPIO_MPI_OPT_TYPES_ENV_VAR_DISABLED:
                cause_str = "optimized MPI types flag wasn't set";
                break;
            case H5D_MPIO_NOT_SIMPLE_OR_SCALAR_DATASPACES:
                cause_str = "one of the dataspaces was neither simple nor scalar";
                break;
            case H5D_MPIO_NOT_CONTIGUOUS_OR_CHUNKED_DATASET:
                cause_str = "dataset was not contiguous or chunked";
                break;
            case H5D_MPIO_PARALLEL_FILTERED_WRITES_DISABLED:
                cause_str = "parallel writes to filtered datasets are disabled";
                break;
            case H5D_MPIO_ERROR_WHILE_CHECKING_COLLECTIVE_POSSIBLE:
                cause_str = "an error occurred while checking if collective I/O was possible";
                break;
            case H5D_MPIO_COLLECTIVE:
            case H5D_MPIO_NO_COLLECTIVE_MAX_CAUSE:
            default:
                HDassert(0 && "invalid no collective cause reason");
                break;
        }

        /*
         * Determine if the local reasons for breaking collective I/O
         * included the current cause
         */
        if (local_cause && (cur_cause & local_no_coll_cause)) {
            buf_space_left = local_cause_len - local_cause_bytes_written;

            /*
             * Check if there were any previous error messages included. If
             * so, prepend a semicolon to separate the messages.
             */
            if (buf_space_left && local_cause_bytes_written) {
                HDstrncat(local_cause, "; ", buf_space_left);
                local_cause_bytes_written += MIN(buf_space_left, 2);
                buf_space_left -= MIN(buf_space_left, 2);
            }

            if (buf_space_left) {
                HDstrncat(local_cause, cause_str, buf_space_left);
                local_cause_bytes_written += MIN(buf_space_left, HDstrlen(cause_str));
            }
        }

        /*
         * Determine if the global reasons for breaking collective I/O
         * included the current cause
         */
        if (global_cause && (cur_cause & global_no_coll_cause)) {
            buf_space_left = global_cause_len - global_cause_bytes_written;

            /*
             * Check if there were any previous error messages included. If
             * so, prepend a semicolon to separate the messages.
             */
            if (buf_space_left && global_cause_bytes_written) {
                HDstrncat(global_cause, "; ", buf_space_left);
                global_cause_bytes_written += MIN(buf_space_left, 2);
                buf_space_left -= MIN(buf_space_left, 2);
            }

            if (buf_space_left) {
                HDstrncat(global_cause, cause_str, buf_space_left);
                global_cause_bytes_written += MIN(buf_space_left, HDstrlen(cause_str));
            }
        }
    }

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_get_no_coll_cause_strings() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_select_read
 *
 * Purpose:     MPI-IO function to read directly from app buffer to file.
 *
 *              This was referred from H5D__mpio_select_read for
 *              multi-dset work.
 *
 * Return:      non-negative on success, negative on failure.
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5D__mpio_select_read(const H5D_io_info_t *io_info, hsize_t mpi_buf_count, H5S_t H5_ATTR_UNUSED *file_space,
                      H5S_t H5_ATTR_UNUSED *mem_space)
{
    void  *rbuf      = NULL;
    herr_t ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    /* memory addr from a piece with lowest file addr */
    rbuf = io_info->base_maddr.vp;

    /*OKAY: CAST DISCARDS CONST QUALIFIER*/
    H5_CHECK_OVERFLOW(mpi_buf_count, hsize_t, size_t);
    if (H5F_shared_block_read(io_info->f_sh, H5FD_MEM_DRAW, io_info->store_faddr, (size_t)mpi_buf_count,
                              rbuf) < 0)
        HGOTO_ERROR(H5E_IO, H5E_READERROR, FAIL, "can't finish collective parallel read")

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_select_read() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_select_write
 *
 * Purpose:     MPI-IO function to write directly from app buffer to file.
 *
 *              This was referred from H5D__mpio_select_write for
 *              multi-dset work.
 *
 * Return:      non-negative on success, negative on failure.
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5D__mpio_select_write(const H5D_io_info_t *io_info, hsize_t mpi_buf_count, H5S_t H5_ATTR_UNUSED *file_space,
                       H5S_t H5_ATTR_UNUSED *mem_space)
{
    const void *wbuf      = NULL;
    herr_t      ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    /* memory addr from a piece with lowest file addr */
    wbuf = io_info->base_maddr.cvp;

    /*OKAY: CAST DISCARDS CONST QUALIFIER*/
    H5_CHECK_OVERFLOW(mpi_buf_count, hsize_t, size_t);
    if (H5F_shared_block_write(io_info->f_sh, H5FD_MEM_DRAW, io_info->store_faddr, (size_t)mpi_buf_count,
                               wbuf) < 0)
        HGOTO_ERROR(H5E_IO, H5E_WRITEERROR, FAIL, "can't finish collective parallel write")

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_select_write() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_get_sum_chunk
 *
 * Purpose:     Routine for obtaining total number of chunks to cover
 *              hyperslab selection selected by all processors.  Operates
 *              on all datasets in the operation.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_get_sum_chunk(const H5D_io_info_t *io_info, int *sum_chunkf)
{
    int    num_chunkf; /* Number of chunks to iterate over */
    size_t ori_num_chunkf;
    int    mpi_code; /* MPI return code */
    herr_t ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    /* Get the number of chunks to perform I/O on */
    num_chunkf     = 0;
    ori_num_chunkf = io_info->pieces_added;
    H5_CHECKED_ASSIGN(num_chunkf, int, ori_num_chunkf, size_t);

    /* Determine the summation of number of chunks for all processes */
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Allreduce(&num_chunkf, sum_chunkf, 1, MPI_INT, MPI_SUM, io_info->comm)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Allreduce failed", mpi_code)

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_get_sum_chunk() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_get_sum_chunk_dset
 *
 * Purpose:     Routine for obtaining total number of chunks to cover
 *              hyperslab selection selected by all processors.  Operates
 *              on a single dataset.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_get_sum_chunk_dset(const H5D_io_info_t *io_info, const H5D_dset_io_info_t *dset_info,
                             int *sum_chunkf)
{
    int    num_chunkf; /* Number of chunks to iterate over */
    size_t ori_num_chunkf;
    int    mpi_code; /* MPI return code */
    herr_t ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    /* Check for non-chunked dataset, in this case we know the number of "chunks"
     * is simply the mpi size */
    HDassert(dset_info->layout->type == H5D_CHUNKED);

    /* Get the number of chunks to perform I/O on */
    num_chunkf     = 0;
    ori_num_chunkf = H5SL_count(dset_info->layout_io_info.chunk_map->dset_sel_pieces);
    H5_CHECKED_ASSIGN(num_chunkf, int, ori_num_chunkf, size_t);

    /* Determine the summation of number of chunks for all processes */
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Allreduce(&num_chunkf, sum_chunkf, 1, MPI_INT, MPI_SUM, io_info->comm)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Allreduce failed", mpi_code)

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_get_sum_chunk_dset() */

/*-------------------------------------------------------------------------
 * Function:    H5D__piece_io
 *
 * Purpose:     Routine for
 *              1) choose an IO option:
 *                    a) One collective IO defined by one MPI derived datatype to link through all chunks
 *              or    b) multiple chunk IOs,to do MPI-IO for each chunk, the IO mode may be adjusted
 *                       due to the selection pattern for each chunk.
 *              For option a)
 *                      1. Sort the chunk address, obtain chunk info according to the sorted chunk address
 *                      2. Build up MPI derived datatype for each chunk
 *                      3. Build up the final MPI derived datatype
 *                      4. Set up collective IO property list
 *                      5. Do IO
 *              For option b)
 *                      1. Use MPI_gather and MPI_Bcast to obtain information of *collective/independent/none*
 *                         IO mode for each chunk of the selection
 *                      2. Depending on whether the IO mode is collective or independent or none,
 *                         Create either MPI derived datatype for each chunk to do collective IO or
 *                         just do independent IO or independent IO with file set view
 *                      3. Set up collective IO property list for collective mode
 *                      4. DO IO
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Muqun Yang
 *              Monday, Feb. 13th, 2006
 *
 * Modification:
 *  - Refctore to remove multi-chunk-without-opimization feature and update for
 *    multi-chunk-io accordingly
 * Programmer: Jonathan Kim
 * Date: 2012-10-10
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__piece_io(H5D_io_info_t *io_info)
{
    H5FD_mpio_chunk_opt_t chunk_opt_mode;
#ifdef H5Dmpio_DEBUG
    hbool_t log_file_flag  = FALSE;
    FILE   *debug_log_file = NULL;
#endif
    int      io_option        = H5D_MULTI_CHUNK_IO_MORE_OPT;
    hbool_t  recalc_io_option = FALSE;
    hbool_t  use_multi_dset   = FALSE;
    unsigned one_link_chunk_io_threshold; /* Threshold to use single collective I/O for all chunks */
    int      sum_chunk = -1;
    int      mpi_rank;
    int      mpi_size;
    size_t   i;
    herr_t   ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    /* Sanity checks */
    HDassert(io_info);
    HDassert(io_info->using_mpi_vfd);
    HDassert(io_info->count > 0);

    /* Obtain the current rank of the process and the number of ranks */
    if ((mpi_rank = H5F_mpi_get_rank(io_info->dsets_info[0].dset->oloc.file)) < 0)
        HGOTO_ERROR(H5E_IO, H5E_MPI, FAIL, "unable to obtain MPI rank")
    if ((mpi_size = H5F_mpi_get_size(io_info->dsets_info[0].dset->oloc.file)) < 0)
        HGOTO_ERROR(H5E_IO, H5E_MPI, FAIL, "unable to obtain MPI size")

#ifdef H5Dmpio_DEBUG
    /* Initialize file-level debugging if not initialized */
    if (!H5D_mpio_debug_inited && H5D__mpio_debug_init() < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "can't initialize H5Dmpio debugging")

    /* Open file for debugging if necessary */
    log_file_flag = H5D_mpio_debug_flags_s[(int)'f'];
    if (log_file_flag) {
        char   debug_log_filename[1024];
        time_t time_now;

        HDsnprintf(debug_log_filename, 1024, "H5Dmpio_debug.rank%d", mpi_rank);

        if (NULL == (debug_log_file = HDfopen(debug_log_filename, "a")))
            HGOTO_ERROR(H5E_IO, H5E_OPENERROR, FAIL, "couldn't open debugging log file")

        /* Print a short header for this I/O operation */
        time_now = HDtime(NULL);
        HDfprintf(debug_log_file, "##### %s", HDasctime(HDlocaltime(&time_now)));

        debug_stream = debug_log_file;
    }
#endif

    /* Check the optional property list for the collective chunk IO optimization option.
     * Only set here if it's a static option, if it needs to be calculated using the
     * number of chunks per process delay that calculation until later. */
    if (H5CX_get_mpio_chunk_opt_mode(&chunk_opt_mode) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "couldn't get chunk optimization option")

    if (H5FD_MPIO_CHUNK_ONE_IO == chunk_opt_mode)
        io_option = H5D_ONE_LINK_CHUNK_IO; /*no opt*/
    /* direct request to multi-chunk-io */
    else if (H5FD_MPIO_CHUNK_MULTI_IO == chunk_opt_mode)
        io_option = H5D_MULTI_CHUNK_IO;
    else
        recalc_io_option = TRUE;

    /* Check if we can and should use multi dataset path */
    if (io_info->count > 1 && (io_option == H5D_ONE_LINK_CHUNK_IO || recalc_io_option)) {
        /* Use multi dataset path for now */
        use_multi_dset = TRUE;

        /* Check for filtered datasets */
        for (i = 0; i < io_info->count; i++)
            if (io_info->dsets_info[i].dset->shared->dcpl_cache.pline.nused > 0) {
                use_multi_dset = FALSE;
                break;
            }

        /* Check if this I/O exceeds one linked chunk threshold */
        if (recalc_io_option && use_multi_dset) {
            /* Get the chunk optimization option threshold */
            if (H5CX_get_mpio_chunk_opt_num(&one_link_chunk_io_threshold) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                            "couldn't get chunk optimization option threshold value")

            /* If the threshold is 0, no need to check number of chunks */
            if (one_link_chunk_io_threshold > 0) {
                /* Get number of chunks for all processes */
                if (H5D__mpio_get_sum_chunk(io_info, &sum_chunk) < 0)
                    HGOTO_ERROR(H5E_DATASPACE, H5E_CANTSWAP, FAIL,
                                "unable to obtain the total chunk number of all processes");

                /* If the average number of chunk per process is less than the threshold, we will do multi
                 * chunk IO.  If this threshold is not exceeded for all datasets, no need to check it again
                 * for each individual dataset. */
                if ((unsigned)sum_chunk / (unsigned)mpi_size < one_link_chunk_io_threshold) {
                    recalc_io_option = FALSE;
                    use_multi_dset   = FALSE;
                }
            }
        }

        /* Perform multi dataset I/O if appropriate */
        if (use_multi_dset) {
#ifdef H5_HAVE_INSTRUMENTED_LIBRARY
            /*** Set collective chunk user-input optimization API. ***/
            if (H5D_ONE_LINK_CHUNK_IO == io_option) {
                if (H5CX_test_set_mpio_coll_chunk_link_hard(0) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
            } /* end if */
#endif        /* H5_HAVE_INSTRUMENTED_LIBRARY */

            /* Perform unfiltered link chunk collective IO */
            if (H5D__link_piece_collective_io(io_info, mpi_rank) < 0)
                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish linked chunk MPI-IO")
        }
    }

    if (!use_multi_dset) {
        /* Loop over datasets */
        for (i = 0; i < io_info->count; i++) {
            if (io_info->dsets_info[i].layout->type == H5D_CONTIGUOUS) {
                /* Contiguous: call H5D__inter_collective_io() directly */
                H5D_mpio_actual_io_mode_t actual_io_mode = H5D_MPIO_CONTIGUOUS_COLLECTIVE;

                io_info->store_faddr = io_info->dsets_info[i].store->contig.dset_addr;
                io_info->base_maddr  = io_info->dsets_info[i].buf;

                if (H5D__inter_collective_io(io_info, &io_info->dsets_info[i],
                                             io_info->dsets_info[i].file_space,
                                             io_info->dsets_info[i].mem_space) < 0)
                    HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish shared collective MPI-IO")

                /* Set the actual I/O mode property. internal_collective_io will not break to
                 * independent I/O, so we set it here.
                 */
                H5CX_set_mpio_actual_io_mode(actual_io_mode);
            }
            else {
                /* Chunked I/O path */
                HDassert(io_info->dsets_info[i].layout->type == H5D_CHUNKED);

                /* Recalculate io_option if necessary */
                if (recalc_io_option) {
                    /* Get the chunk optimization option threshold */
                    if (H5CX_get_mpio_chunk_opt_num(&one_link_chunk_io_threshold) < 0)
                        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                                    "couldn't get chunk optimization option threshold value")

                    /* If the threshold is 0, no need to check number of chunks */
                    if (one_link_chunk_io_threshold == 0) {
                        io_option        = H5D_ONE_LINK_CHUNK_IO_MORE_OPT;
                        recalc_io_option = FALSE;
                    }
                    else {
                        /* Get number of chunks for all processes */
                        if (H5D__mpio_get_sum_chunk_dset(io_info, &io_info->dsets_info[i], &sum_chunk) < 0)
                            HGOTO_ERROR(H5E_DATASPACE, H5E_CANTSWAP, FAIL,
                                        "unable to obtain the total chunk number of all processes");

                        /* step 1: choose an IO option */
                        /* If the average number of chunk per process is greater than a threshold, we will do
                         * one link chunked IO. */
                        if ((unsigned)sum_chunk / (unsigned)mpi_size >= one_link_chunk_io_threshold)
                            io_option = H5D_ONE_LINK_CHUNK_IO_MORE_OPT;
                        else
                            io_option = H5D_MULTI_CHUNK_IO_MORE_OPT;
                    }
                }

                /* step 2:  Go ahead to do IO.*/
                switch (io_option) {
                    case H5D_ONE_LINK_CHUNK_IO:
                    case H5D_ONE_LINK_CHUNK_IO_MORE_OPT:
                        /* Check if there are any filters in the pipeline */
                        if (io_info->dsets_info[i].dset->shared->dcpl_cache.pline.nused > 0) {
                            if (H5D__link_chunk_filtered_collective_io(io_info, &io_info->dsets_info[i],
                                                                       mpi_rank, mpi_size) < 0)
                                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL,
                                            "couldn't finish filtered linked chunk MPI-IO")
                        } /* end if */
                        else {
                            /* If there is more than one dataset we cannot make the multi dataset call here,
                             * fall back to multi chunk */
                            if (io_info->count > 1) {
                                io_option        = H5D_MULTI_CHUNK_IO_MORE_OPT;
                                recalc_io_option = TRUE;

                                if (H5D__multi_chunk_collective_io(io_info, &io_info->dsets_info[i], mpi_rank,
                                                                   mpi_size) < 0)
                                    HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL,
                                                "couldn't finish optimized multiple chunk MPI-IO")
                            }
                            else {
                                /* Perform unfiltered link chunk collective IO */
                                if (H5D__link_piece_collective_io(io_info, mpi_rank) < 0)
                                    HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL,
                                                "couldn't finish linked chunk MPI-IO")
                            }
                        }

                        break;

                    case H5D_MULTI_CHUNK_IO: /* direct request to do multi-chunk IO */
                    default:                 /* multiple chunk IO via threshold */
                        /* Check if there are any filters in the pipeline */
                        if (io_info->dsets_info[i].dset->shared->dcpl_cache.pline.nused > 0) {
                            if (H5D__multi_chunk_filtered_collective_io(io_info, &io_info->dsets_info[i],
                                                                        mpi_rank, mpi_size) < 0)
                                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL,
                                            "couldn't finish optimized multiple filtered chunk MPI-IO")
                        } /* end if */
                        else {
                            /* Perform unfiltered multi chunk collective IO */
                            if (H5D__multi_chunk_collective_io(io_info, &io_info->dsets_info[i], mpi_rank,
                                                               mpi_size) < 0)
                                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL,
                                            "couldn't finish optimized multiple chunk MPI-IO")
                        }

                        break;
                } /* end switch */

#ifdef H5_HAVE_INSTRUMENTED_LIBRARY
                {
                    /*** Set collective chunk user-input optimization APIs. ***/
                    if (H5D_ONE_LINK_CHUNK_IO == io_option) {
                        if (H5CX_test_set_mpio_coll_chunk_link_hard(0) < 0)
                            HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
                    } /* end if */
                    else if (H5D_MULTI_CHUNK_IO == io_option) {
                        if (H5CX_test_set_mpio_coll_chunk_multi_hard(0) < 0)
                            HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
                    } /* end else-if */
                    else if (H5D_ONE_LINK_CHUNK_IO_MORE_OPT == io_option) {
                        if (H5CX_test_set_mpio_coll_chunk_link_num_true(0) < 0)
                            HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
                    } /* end if */
                    else if (H5D_MULTI_CHUNK_IO_MORE_OPT == io_option) {
                        if (H5CX_test_set_mpio_coll_chunk_link_num_false(0) < 0)
                            HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
                    } /* end if */
                }
#endif /* H5_HAVE_INSTRUMENTED_LIBRARY */
            }
        }
    }

done:
#ifdef H5Dmpio_DEBUG
    /* Close debugging log file */
    if (debug_log_file) {
        HDfprintf(debug_log_file, "##############\n\n");
        if (EOF == HDfclose(debug_log_file))
            HDONE_ERROR(H5E_IO, H5E_CLOSEERROR, FAIL, "couldn't close debugging log file")
        debug_stream = H5DEBUG(D);
    }
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__piece_io */

/*-------------------------------------------------------------------------
 * Function:    H5D__collective_read
 *
 * Purpose:     Read directly from pieces (chunks/contig) in file into
 *              application memory using collective I/O.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Quincey Koziol
 *              Tuesday, March  4, 2008
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5D__collective_read(H5D_io_info_t *io_info)
{
    herr_t ret_value = SUCCEED; /* Return value */

    FUNC_ENTER_PACKAGE

    /* Call generic selection operation */
    if (H5D__piece_io(io_info) < 0)
        HGOTO_ERROR(H5E_DATASPACE, H5E_READERROR, FAIL, "read error")

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__collective_read() */

/*-------------------------------------------------------------------------
 * Function:    H5D__collective_write
 *
 * Purpose:     Write directly to pieces (chunks/contig) in file into
 *              application memory using collective I/O.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Quincey Koziol
 *              Tuesday, March  4, 2008
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5D__collective_write(H5D_io_info_t *io_info)
{
    herr_t ret_value = SUCCEED; /* Return value */

    FUNC_ENTER_PACKAGE

    /* Call generic selection operation */
    if (H5D__piece_io(io_info) < 0)
        HGOTO_ERROR(H5E_DATASPACE, H5E_WRITEERROR, FAIL, "write error")

done:
    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__collective_write() */

/*-------------------------------------------------------------------------
 * Function:    H5D__link_piece_collective_io
 *
 * Purpose:     Routine for single collective IO with one MPI derived datatype
 *              to link with all pieces (chunks + contig)
 *
 *              1. Use the piece addresses and piece info sorted in skiplist
 *              2. Build up MPI derived datatype for each chunk
 *              3. Build up the final MPI derived datatype
 *              4. Use common collective IO routine to do MPI-IO
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Muqun Yang
 *              Monday, Feb. 13th, 2006
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__link_piece_collective_io(H5D_io_info_t *io_info, int mpi_rank)
{
    MPI_Datatype  chunk_final_mtype; /* Final memory MPI datatype for all chunks with selection */
    hbool_t       chunk_final_mtype_is_derived = FALSE;
    MPI_Datatype  chunk_final_ftype; /* Final file MPI datatype for all chunks with selection */
    hbool_t       chunk_final_ftype_is_derived = FALSE;
    H5D_storage_t ctg_store; /* Storage info for "fake" contiguous dataset */
    MPI_Datatype *chunk_mtype           = NULL;
    MPI_Datatype *chunk_ftype           = NULL;
    MPI_Aint     *chunk_file_disp_array = NULL;
    MPI_Aint     *chunk_mem_disp_array  = NULL;
    hbool_t      *chunk_mft_is_derived_array =
        NULL; /* Flags to indicate each chunk's MPI file datatype is derived */
    hbool_t *chunk_mbt_is_derived_array =
        NULL;                          /* Flags to indicate each chunk's MPI memory datatype is derived */
    int *chunk_mpi_file_counts = NULL; /* Count of MPI file datatype for each chunk */
    int *chunk_mpi_mem_counts  = NULL; /* Count of MPI memory datatype for each chunk */
    int  mpi_code;                     /* MPI return code */
    H5D_mpio_actual_chunk_opt_mode_t actual_chunk_opt_mode = H5D_MPIO_LINK_CHUNK;
    H5D_mpio_actual_io_mode_t        actual_io_mode        = 0;
    size_t                           i; /* Local index variable */
    herr_t                           ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    /* set actual_io_mode */
    for (i = 0; i < io_info->count; i++) {
        HDassert(io_info->dsets_info[i].dset->shared->dcpl_cache.pline.nused == 0);
        if (io_info->dsets_info[i].layout->type == H5D_CHUNKED)
            actual_io_mode |= H5D_MPIO_CHUNK_COLLECTIVE;
        else if (io_info->dsets_info[i].layout->type == H5D_CONTIGUOUS) {
            actual_io_mode |= H5D_MPIO_CONTIGUOUS_COLLECTIVE;

            /* if only single-dset */
            if (1 == io_info->count)
                actual_chunk_opt_mode = H5D_MPIO_NO_CHUNK_OPTIMIZATION;
        }
        else
            HGOTO_ERROR(H5E_IO, H5E_UNSUPPORTED, FAIL, "unsupported storage layout")
    }

    /* Set the actual-chunk-opt-mode property. */
    H5CX_set_mpio_actual_chunk_opt(actual_chunk_opt_mode);

    /* Set the actual-io-mode property.
     * Link chunk I/O does not break to independent, so can set right away */
    H5CX_set_mpio_actual_io_mode(actual_io_mode);

    /* Code block for actual actions (Build a MPI Type, IO) */
    {
        hsize_t mpi_buf_count; /* Number of MPI types */
        size_t  num_chunk;     /* Number of chunks for this process */

        H5D_piece_info_t *piece_info;

        /* local variable for base address for buffer */
        H5_flexible_const_ptr_t base_buf_addr;
        base_buf_addr.cvp = NULL;

        /* Get the number of chunks with a selection */
        num_chunk = io_info->pieces_added;
        H5_CHECK_OVERFLOW(num_chunk, size_t, int);

#ifdef H5Dmpio_DEBUG
        H5D_MPIO_DEBUG_VA(mpi_rank, "num_chunk = %zu\n", num_chunk);
#endif

        /* Set up MPI datatype for chunks selected */
        if (num_chunk) {
            hbool_t need_sort = FALSE;

            /* Check if sel_pieces array is sorted */
            HDassert(io_info->sel_pieces[0]->faddr != HADDR_UNDEF);
            for (i = 1; i < num_chunk; i++) {
                HDassert(io_info->sel_pieces[i]->faddr != HADDR_UNDEF);

                if (io_info->sel_pieces[i]->faddr < io_info->sel_pieces[i - 1]->faddr) {
                    need_sort = TRUE;
                    break;
                }
            }

            /* Sort sel_pieces if necessary */
            if (need_sort)
                HDqsort(io_info->sel_pieces, io_info->pieces_added, sizeof(io_info->sel_pieces[0]),
                        H5D__cmp_piece_addr);

            /* Allocate chunking information */
            if (NULL == (chunk_mtype = (MPI_Datatype *)H5MM_malloc(num_chunk * sizeof(MPI_Datatype))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk memory datatype buffer")
            if (NULL == (chunk_ftype = (MPI_Datatype *)H5MM_malloc(num_chunk * sizeof(MPI_Datatype))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate chunk file datatype buffer")
            if (NULL == (chunk_file_disp_array = (MPI_Aint *)H5MM_malloc(num_chunk * sizeof(MPI_Aint))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk file displacement buffer")
            if (NULL == (chunk_mem_disp_array = (MPI_Aint *)H5MM_calloc(num_chunk * sizeof(MPI_Aint))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk memory displacement buffer")
            if (NULL == (chunk_mpi_mem_counts = (int *)H5MM_calloc(num_chunk * sizeof(int))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate chunk memory counts buffer")
            if (NULL == (chunk_mpi_file_counts = (int *)H5MM_calloc(num_chunk * sizeof(int))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate chunk file counts buffer")
            if (NULL == (chunk_mbt_is_derived_array = (hbool_t *)H5MM_calloc(num_chunk * sizeof(hbool_t))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk memory is derived datatype flags buffer")
            if (NULL == (chunk_mft_is_derived_array = (hbool_t *)H5MM_calloc(num_chunk * sizeof(hbool_t))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk file is derived datatype flags buffer")

            /* save lowest file address */
            ctg_store.contig.dset_addr = io_info->sel_pieces[0]->faddr;

            /* save base mem addr of piece for read/write */
            base_buf_addr = io_info->sel_pieces[0]->dset_info->buf;

#ifdef H5Dmpio_DEBUG
            H5D_MPIO_DEBUG(mpi_rank, "before iterate over selected pieces\n");
#endif

            /* Obtain MPI derived datatype from all individual pieces */
            /* Iterate over selected pieces for this process */
            for (i = 0; i < num_chunk; i++) {
                hsize_t *permute_map = NULL; /* array that holds the mapping from the old,
                                                out-of-order displacements to the in-order
                                                displacements of the MPI datatypes of the
                                                point selection of the file space */
                hbool_t is_permuted = FALSE;

                /* Assign convenience pointer to piece info */
                piece_info = io_info->sel_pieces[i];

                /* Obtain disk and memory MPI derived datatype */
                /* NOTE: The permute_map array can be allocated within H5S_mpio_space_type
                 *              and will be fed into the next call to H5S_mpio_space_type
                 *              where it will be freed.
                 */
                if (H5S_mpio_space_type(piece_info->fspace, piece_info->dset_info->type_info.src_type_size,
                                        &chunk_ftype[i],                  /* OUT: datatype created */
                                        &chunk_mpi_file_counts[i],        /* OUT */
                                        &(chunk_mft_is_derived_array[i]), /* OUT */
                                        TRUE,                             /* this is a file space,
                                                                             so permute the
                                                                             datatype if the point
                                                                             selections are out of
                                                                             order */
                                        &permute_map,                     /* OUT: a map to indicate the
                                                                             permutation of points
                                                                             selected in case they
                                                                             are out of order */
                                        &is_permuted /* OUT */) < 0)
                    HGOTO_ERROR(H5E_DATASPACE, H5E_BADTYPE, FAIL, "couldn't create MPI file type")

                /* Sanity check */
                if (is_permuted)
                    HDassert(permute_map);
                if (H5S_mpio_space_type(piece_info->mspace, piece_info->dset_info->type_info.dst_type_size,
                                        &chunk_mtype[i], &chunk_mpi_mem_counts[i],
                                        &(chunk_mbt_is_derived_array[i]), FALSE, /* this is a memory
                                                                                    space, so if the file
                                                                                    space is not
                                                                                    permuted, there is no
                                                                                    need to permute the
                                                                                    datatype if the point
                                                                                    selections are out of
                                                                                    order*/
                                        &permute_map,                            /* IN: the permutation map
                                                                                    generated by the
                                                                                    file_space selection
                                                                                    and applied to the
                                                                                    memory selection */
                                        &is_permuted /* IN */) < 0)
                    HGOTO_ERROR(H5E_DATASPACE, H5E_BADTYPE, FAIL, "couldn't create MPI buf type")
                /* Sanity check */
                if (is_permuted)
                    HDassert(!permute_map);

                /* Piece address relative to the first piece addr
                 * Assign piece address to MPI displacement
                 * (assume MPI_Aint big enough to hold it) */
                chunk_file_disp_array[i] = (MPI_Aint)piece_info->faddr - (MPI_Aint)ctg_store.contig.dset_addr;

                if (io_info->op_type == H5D_IO_OP_WRITE) {
                    chunk_mem_disp_array[i] =
                        (MPI_Aint)piece_info->dset_info->buf.cvp - (MPI_Aint)base_buf_addr.cvp;
                }
                else if (io_info->op_type == H5D_IO_OP_READ) {
                    chunk_mem_disp_array[i] =
                        (MPI_Aint)piece_info->dset_info->buf.vp - (MPI_Aint)base_buf_addr.vp;
                }
            } /* end for */

            /* Create final MPI derived datatype for the file */
            if (MPI_SUCCESS !=
                (mpi_code = MPI_Type_create_struct((int)num_chunk, chunk_mpi_file_counts,
                                                   chunk_file_disp_array, chunk_ftype, &chunk_final_ftype)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)

            if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(&chunk_final_ftype)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)
            chunk_final_ftype_is_derived = TRUE;

            /* Create final MPI derived datatype for memory */
            if (MPI_SUCCESS !=
                (mpi_code = MPI_Type_create_struct((int)num_chunk, chunk_mpi_mem_counts, chunk_mem_disp_array,
                                                   chunk_mtype, &chunk_final_mtype)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
            if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(&chunk_final_mtype)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)
            chunk_final_mtype_is_derived = TRUE;

            /* Free the file & memory MPI datatypes for each chunk */
            for (i = 0; i < num_chunk; i++) {
                if (chunk_mbt_is_derived_array[i])
                    if (MPI_SUCCESS != (mpi_code = MPI_Type_free(chunk_mtype + i)))
                        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

                if (chunk_mft_is_derived_array[i])
                    if (MPI_SUCCESS != (mpi_code = MPI_Type_free(chunk_ftype + i)))
                        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            } /* end for */

            /* We have a single, complicated MPI datatype for both memory & file */
            mpi_buf_count = (hsize_t)1;
        }      /* end if */
        else { /* no selection at all for this process */
            ctg_store.contig.dset_addr = 0;

            /* just provide a valid mem address. no actual IO occur */
            base_buf_addr = io_info->dsets_info[0].buf;

            /* Set the MPI datatype */
            chunk_final_ftype = MPI_BYTE;
            chunk_final_mtype = MPI_BYTE;

            /* No chunks selected for this process */
            mpi_buf_count = (hsize_t)0;
        } /* end else */

#ifdef H5Dmpio_DEBUG
        H5D_MPIO_DEBUG(mpi_rank, "before coming to final collective I/O");
#endif
        /* Set up the base storage address for this piece */
        io_info->store_faddr = ctg_store.contig.dset_addr;
        io_info->base_maddr  = base_buf_addr;

        /* Perform final collective I/O operation */
        if (H5D__final_collective_io(io_info, mpi_buf_count, chunk_final_ftype, chunk_final_mtype) < 0)
            HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish MPI-IO")
    }

done:
#ifdef H5Dmpio_DEBUG
    H5D_MPIO_DEBUG_VA(mpi_rank, "before freeing memory inside H5D_link_collective_io ret_value = %d",
                      ret_value);
#endif

    /* Release resources */
    if (chunk_mtype)
        H5MM_xfree(chunk_mtype);
    if (chunk_ftype)
        H5MM_xfree(chunk_ftype);
    if (chunk_file_disp_array)
        H5MM_xfree(chunk_file_disp_array);
    if (chunk_mem_disp_array)
        H5MM_xfree(chunk_mem_disp_array);
    if (chunk_mpi_mem_counts)
        H5MM_xfree(chunk_mpi_mem_counts);
    if (chunk_mpi_file_counts)
        H5MM_xfree(chunk_mpi_file_counts);
    if (chunk_mbt_is_derived_array)
        H5MM_xfree(chunk_mbt_is_derived_array);
    if (chunk_mft_is_derived_array)
        H5MM_xfree(chunk_mft_is_derived_array);

    /* Free the MPI buf and file types, if they were derived */
    if (chunk_final_mtype_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&chunk_final_mtype)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    if (chunk_final_ftype_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&chunk_final_ftype)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__link_piece_collective_io */

/*-------------------------------------------------------------------------
 * Function:    H5D__link_chunk_filtered_collective_io
 *
 * Purpose:     Performs collective I/O on filtered chunks by creating a
 *              single MPI derived datatype to link with all filtered
 *              chunks. The general algorithm is as follows:
 *
 *              1. Construct a list of selected chunks in the collective
 *                 I/O operation
 *              2. If the operation is a read operation
 *                 A. Ensure that the list of chunks is sorted in
 *                    monotonically non-decreasing order of chunk offset
 *                    in the file
 *                 B. Participate in a collective read of chunks from
 *                    the file
 *                 C. Loop through each selected chunk, unfiltering it and
 *                    scattering the data to the application's read buffer
 *              3. If the operation is a write operation
 *                 A. Redistribute any chunks being written by more than 1
 *                    MPI rank, such that the chunk is only owned by 1 MPI
 *                    rank. The rank writing to the chunk which currently
 *                    has the least amount of chunks assigned to it becomes
 *                    the new owner (in the case of ties, the lowest MPI
 *                    rank becomes the new owner)
 *                 B. Participate in a collective read of chunks from the
 *                    file
 *                 C. Loop through each chunk selected in the operation
 *                    and for each chunk:
 *                    I. If we actually read the chunk from the file (if
 *                       a chunk is being fully overwritten, we skip
 *                       reading it), pass the chunk through the filter
 *                       pipeline in reverse order (unfilter the chunk)
 *                    II. Update the chunk data with the modifications from
 *                        the owning MPI rank
 *                    III. Receive any modification data from other
 *                         ranks and update the chunk data with those
 *                         modifications
 *                    IV. Filter the chunk
 *                 D. Contribute the modified chunks to an array gathered
 *                    by all ranks which contains information for
 *                    re-allocating space in the file for every chunk
 *                    modified. Then, each rank collectively re-allocates
 *                    each chunk from the gathered array with their new
 *                    sizes after the filter operation
 *                 E. Proceed with the collective write operation for all
 *                    the modified chunks
 *                 F. Contribute the modified chunks to an array gathered
 *                    by all ranks which contains information for
 *                    re-inserting every chunk modified into the chunk
 *                    index. Then, each rank collectively re-inserts each
 *                    chunk from the gathered array into the chunk index
 *
 *              TODO: Note that steps D. and F. here are both collective
 *                    operations that partially share data from the
 *                    H5D_filtered_collective_io_info_t structure. To
 *                    try to conserve on memory a bit, the distributed
 *                    arrays these operations create are discarded after
 *                    each operation is performed. If memory consumption
 *                    here proves to not be an issue, the necessary data
 *                    for both operations could be combined into a single
 *                    structure so that only one collective MPI operation
 *                    is needed to carry out both operations, rather than
 *                    two.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__link_chunk_filtered_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info, int mpi_rank,
                                       int mpi_size)
{
    H5D_filtered_collective_io_info_t *chunk_list          = NULL; /* The list of chunks being read/written */
    H5D_filtered_collective_io_info_t *chunk_hash_table    = NULL;
    unsigned char                    **chunk_msg_bufs      = NULL;
    MPI_Datatype                       mem_type            = MPI_BYTE;
    MPI_Datatype                       file_type           = MPI_BYTE;
    hbool_t                            mem_type_is_derived = FALSE;
    hbool_t                            file_type_is_derived     = FALSE;
    size_t                            *rank_chunks_assigned_map = NULL;
    size_t                             chunk_list_num_entries;
    size_t                             i;
    int                                chunk_msg_bufs_len = 0;
    char   fake_buf; /* Used as a fake buffer for ranks with no chunks, thus a NULL buf pointer */
    int    mpi_code;
    herr_t ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE_TAG(dset_info->dset->oloc.addr)

    HDassert(io_info);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_DEBUG_VA(mpi_rank, "Performing Linked-chunk I/O (%s) with MPI Comm size of %d",
                      io_info->op_type == H5D_IO_OP_WRITE ? "write" : "read", mpi_size);
    H5D_MPIO_TIME_START(mpi_rank, "Linked-chunk I/O");
#endif

    /* Set the actual-chunk-opt-mode property. */
    H5CX_set_mpio_actual_chunk_opt(H5D_MPIO_LINK_CHUNK);

    /* Set the actual-io-mode property.
     * Link chunk filtered I/O does not break to independent, so can set right away
     */
    H5CX_set_mpio_actual_io_mode(H5D_MPIO_CHUNK_COLLECTIVE);

    /* Build a list of selected chunks in the collective io operation */
    if (H5D__mpio_collective_filtered_chunk_io_setup(io_info, dset_info, &chunk_list, &chunk_list_num_entries,
                                                     mpi_rank) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "couldn't construct filtered I/O info list")

    if (io_info->op_type == H5D_IO_OP_READ) { /* Filtered collective read */
        if (H5D__mpio_collective_filtered_chunk_read(chunk_list, chunk_list_num_entries, io_info, dset_info,
                                                     mpi_rank, mpi_size) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "couldn't read filtered chunks")
    }
    else { /* Filtered collective write */
        H5D_chk_idx_info_t index_info;
        hsize_t            mpi_buf_count;

        H5D_MPIO_INIT_CHUNK_IDX_INFO(index_info, dset_info->dset);

        if (mpi_size > 1) {
            /* Redistribute shared chunks being written to */
            if (H5D__mpio_redistribute_shared_chunks(chunk_list, chunk_list_num_entries, io_info, mpi_rank,
                                                     mpi_size, &rank_chunks_assigned_map) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "unable to redistribute shared chunks")

            /* Send any chunk modification messages for chunks this rank no longer owns */
            if (H5D__mpio_share_chunk_modification_data(chunk_list, &chunk_list_num_entries, io_info,
                                                        dset_info, mpi_rank, mpi_size, &chunk_hash_table,
                                                        &chunk_msg_bufs, &chunk_msg_bufs_len) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                            "unable to send chunk modification data between MPI ranks")

            /* Make sure the local chunk list was updated correctly */
            HDassert(chunk_list_num_entries == rank_chunks_assigned_map[mpi_rank]);
        }

        /* Proceed to update all the chunks this rank owns with its own
         * modification data and data from other ranks, before re-filtering
         * the chunks. As chunk reads are done collectively here, all ranks
         * must participate.
         */
        if (H5D__mpio_collective_filtered_chunk_update(chunk_list, chunk_list_num_entries, chunk_hash_table,
                                                       chunk_msg_bufs, chunk_msg_bufs_len, io_info, dset_info,
                                                       mpi_rank, mpi_size) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "couldn't update modified chunks")

        /* Free up resources used by chunk hash table now that we're done updating chunks */
        HASH_CLEAR(hh, chunk_hash_table);

        /* All ranks now collectively re-allocate file space for all chunks */
        if (H5D__mpio_collective_filtered_chunk_reallocate(chunk_list, chunk_list_num_entries,
                                                           rank_chunks_assigned_map, io_info, &index_info,
                                                           mpi_rank, mpi_size) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                        "couldn't collectively re-allocate file space for chunks")

        /* If this rank has any chunks selected, create a MPI type for collectively
         * writing out the chunks to file. Otherwise, the rank contributes to the
         * collective write with a none type.
         */
        if (H5D__mpio_collective_filtered_io_type(chunk_list, chunk_list_num_entries, io_info->op_type,
                                                  &mem_type, &mem_type_is_derived, &file_type,
                                                  &file_type_is_derived) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                        "couldn't create MPI type for writing filtered chunks")

        mpi_buf_count = (file_type_is_derived || mem_type_is_derived) ? 1 : 0;

        /* Setup contig storage info for I/O operation */
        if (chunk_list_num_entries) {
            /*
             * Override the write buffer to point to the first
             * chunk's data buffer
             */
            io_info->base_maddr.cvp = chunk_list[0].buf;

            /*
             * Setup the base storage address for this operation
             * to be the first chunk's file address
             */
            io_info->store_faddr = chunk_list[0].chunk_new.offset;
        }
        else {
            io_info->base_maddr.cvp = &fake_buf;
            io_info->store_faddr    = 0;
        }

        /* Perform I/O */
        if (H5D__final_collective_io(io_info, mpi_buf_count, file_type, mem_type) < 0)
            HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish MPI-IO")

        /* Free up resources in anticipation of following collective operation */
        for (i = 0; i < chunk_list_num_entries; i++) {
            if (chunk_list[i].buf) {
                H5MM_free(chunk_list[i].buf);
                chunk_list[i].buf = NULL;
            }
        }

        /* Participate in the collective re-insertion of all chunks modified
         * into the chunk index
         */
        if (H5D__mpio_collective_filtered_chunk_reinsert(chunk_list, chunk_list_num_entries,
                                                         rank_chunks_assigned_map, io_info, dset_info,
                                                         &index_info, mpi_rank, mpi_size) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                        "couldn't collectively re-insert modified chunks into chunk index")
    }

done:
    /* Free the MPI buf and file types, if they were derived */
    if (mem_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&mem_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    if (file_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&file_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

    if (chunk_msg_bufs) {
        for (i = 0; i < (size_t)chunk_msg_bufs_len; i++)
            H5MM_free(chunk_msg_bufs[i]);

        H5MM_free(chunk_msg_bufs);
    }

    HASH_CLEAR(hh, chunk_hash_table);

    /* Free resources used by a rank which had some selection */
    if (chunk_list) {
        for (i = 0; i < chunk_list_num_entries; i++)
            if (chunk_list[i].buf)
                H5MM_free(chunk_list[i].buf);

        H5MM_free(chunk_list);
    } /* end if */

    if (rank_chunks_assigned_map)
        H5MM_free(rank_chunks_assigned_map);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI_TAG(ret_value)
} /* end H5D__link_chunk_filtered_collective_io() */

/*-------------------------------------------------------------------------
 * Function:    H5D__multi_chunk_collective_io
 *
 * Purpose:     To do IO per chunk according to IO mode(collective/independent/none)
 *
 *              1. Use MPI_gather and MPI_Bcast to obtain IO mode in each chunk(collective/independent/none)
 *              2. Depending on whether the IO mode is collective or independent or none,
 *                 Create either MPI derived datatype for each chunk or just do independent IO
 *              3. Use common collective IO routine to do MPI-IO
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Muqun Yang
 *              Monday, Feb. 13th, 2006
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__multi_chunk_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info, int mpi_rank,
                               int mpi_size)
{
    uint8_t                   *chunk_io_option = NULL;
    haddr_t                   *chunk_addr      = NULL;
    H5D_storage_t              store; /* union of EFL and chunk pointer in file space */
    H5FD_mpio_collective_opt_t last_coll_opt_mode =
        H5FD_MPIO_COLLECTIVE_IO; /* Last parallel transfer with independent IO or collective IO with this mode
                                  */
    H5FD_mpio_collective_opt_t orig_coll_opt_mode =
        H5FD_MPIO_COLLECTIVE_IO;           /* Original parallel transfer property on entering this function */
    size_t                    total_chunk; /* Total # of chunks in dataset */
    size_t                    num_chunk;   /* Number of chunks for this process */
    H5SL_node_t              *piece_node      = NULL; /* Current node in chunk skip list */
    H5D_piece_info_t         *next_chunk_info = NULL; /* Chunk info for next selected chunk */
    size_t                    u;                      /* Local index variable */
    H5D_mpio_actual_io_mode_t actual_io_mode =
        H5D_MPIO_NO_COLLECTIVE; /* Local variable for tracking the I/O mode used. */
    herr_t ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE_TAG(dset_info->dset->oloc.addr)

    HDassert(dset_info->layout->type == H5D_CHUNKED);

    /* Get the current I/O collective opt mode so we can restore it later */
    if (H5CX_get_mpio_coll_opt(&orig_coll_opt_mode) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get MPI-I/O collective_op property")

    /* Set the actual chunk opt mode property */
    H5CX_set_mpio_actual_chunk_opt(H5D_MPIO_MULTI_CHUNK);

    /* Retrieve total # of chunks in dataset */
    H5_CHECKED_ASSIGN(total_chunk, size_t, dset_info->layout->u.chunk.nchunks, hsize_t);
    HDassert(total_chunk != 0);

    /* Allocate memories */
    chunk_io_option = (uint8_t *)H5MM_calloc(total_chunk);
    chunk_addr      = (haddr_t *)H5MM_calloc(total_chunk * sizeof(haddr_t));

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_DEBUG_VA(mpi_rank, "total_chunk %zu", total_chunk);
#endif

    /* Obtain IO option for each chunk */
    if (H5D__obtain_mpio_mode(io_info, dset_info, chunk_io_option, chunk_addr, mpi_rank, mpi_size) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTRECV, FAIL, "unable to obtain MPIO mode")

    /* Set memory buffers */
    io_info->base_maddr = dset_info->buf;

    /* Set dataset storage for I/O info */
    dset_info->store = &store;

    /* Get the number of chunks with a selection */
    num_chunk = H5SL_count(dset_info->layout_io_info.chunk_map->dset_sel_pieces);

    if (num_chunk) {
        /* Start at the beginning of the chunk map skiplist.  Since these chunks are
         * stored in index order and since we're iterating in index order we can
         * just check for each chunk being selected in order */
        if (NULL == (piece_node = H5SL_first(dset_info->layout_io_info.chunk_map->dset_sel_pieces)))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "couldn't get piece node from skip list")
        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")
    }

    /* Loop over _all_ the chunks */
    for (u = 0; u < total_chunk; u++) {
        H5D_piece_info_t *chunk_info; /* Chunk info for current chunk */
        H5S_t            *fspace;     /* Dataspace describing chunk & selection in it */
        H5S_t            *mspace; /* Dataspace describing selection in memory corresponding to this chunk */

#ifdef H5Dmpio_DEBUG
        H5D_MPIO_DEBUG_VA(mpi_rank, "mpi_rank = %d, chunk index = %zu", mpi_rank, u);
#endif

        /* Check if this chunk is the next chunk in the skip list, if there are
         * selected chunks left to process */
        HDassert(!num_chunk || next_chunk_info);
        HDassert(!num_chunk || next_chunk_info->index >= u);
        if (num_chunk && next_chunk_info->index == u) {
            /* Next chunk is this chunk */
            chunk_info = next_chunk_info;

            /* One less chunk to process */
            num_chunk--;

            /* Advance next chunk to next node in skip list, if there are more chunks selected */
            if (num_chunk) {
                if (NULL == (piece_node = H5SL_next(piece_node)))
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "chunk skip list terminated early")
                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;

        /* Collective IO for this chunk,
         * Note: even there is no selection for this process, the process still
         *      needs to contribute MPI NONE TYPE.
         */
        if (chunk_io_option[u] == H5D_CHUNK_IO_MODE_COL) {
#ifdef H5Dmpio_DEBUG
            H5D_MPIO_DEBUG_VA(mpi_rank, "inside collective chunk IO mpi_rank = %d, chunk index = %zu",
                              mpi_rank, u);
#endif

            /* Set the file & memory dataspaces */
            if (chunk_info) {
                fspace = chunk_info->fspace;
                mspace = chunk_info->mspace;

                /* Update the local variable tracking the actual io mode property.
                 *
                 * Note: H5D_MPIO_COLLECTIVE_MULTI | H5D_MPIO_INDEPENDENT = H5D_MPIO_MIXED
                 *      to ease switching between to mixed I/O without checking the current
                 *      value of the property. You can see the definition in H5Ppublic.h
                 */
                actual_io_mode = (H5D_mpio_actual_io_mode_t)(actual_io_mode | H5D_MPIO_CHUNK_COLLECTIVE);

            } /* end if */
            else {
                fspace = mspace = NULL;
            } /* end else */

            /* Switch back to collective I/O */
            if (last_coll_opt_mode != H5FD_MPIO_COLLECTIVE_IO) {
                if (H5CX_set_mpio_coll_opt(H5FD_MPIO_COLLECTIVE_IO) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "can't switch to collective I/O")
                last_coll_opt_mode = H5FD_MPIO_COLLECTIVE_IO;
            } /* end if */

            /* Initialize temporary contiguous storage address */
            io_info->store_faddr = chunk_addr[u];

            /* Perform the I/O */
            if (H5D__inter_collective_io(io_info, dset_info, fspace, mspace) < 0)
                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish shared collective MPI-IO")
        }      /* end if */
        else { /* possible independent IO for this chunk */
#ifdef H5Dmpio_DEBUG
            H5D_MPIO_DEBUG_VA(mpi_rank, "inside independent IO mpi_rank = %d, chunk index = %zu", mpi_rank,
                              u);
#endif

            HDassert(chunk_io_option[u] == 0);

            /* Set the file & memory dataspaces */
            if (chunk_info) {
                fspace = chunk_info->fspace;
                mspace = chunk_info->mspace;

                /* Update the local variable tracking the actual io mode. */
                actual_io_mode = (H5D_mpio_actual_io_mode_t)(actual_io_mode | H5D_MPIO_CHUNK_INDEPENDENT);
            } /* end if */
            else {
                fspace = mspace = NULL;
            } /* end else */

            /* Using independent I/O with file setview.*/
            if (last_coll_opt_mode != H5FD_MPIO_INDIVIDUAL_IO) {
                if (H5CX_set_mpio_coll_opt(H5FD_MPIO_INDIVIDUAL_IO) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "can't switch to individual I/O")
                last_coll_opt_mode = H5FD_MPIO_INDIVIDUAL_IO;
            } /* end if */

            /* Initialize temporary contiguous storage address */
            io_info->store_faddr = chunk_addr[u];

            /* Perform the I/O */
            if (H5D__inter_collective_io(io_info, dset_info, fspace, mspace) < 0)
                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish shared collective MPI-IO")
#ifdef H5Dmpio_DEBUG
            H5D_MPIO_DEBUG(mpi_rank, "after inter collective IO");
#endif
        } /* end else */
    }     /* end for */

    /* Write the local value of actual io mode to the API context. */
    H5CX_set_mpio_actual_io_mode(actual_io_mode);

done:
    /* Reset collective opt mode */
    if (H5CX_set_mpio_coll_opt(orig_coll_opt_mode) < 0)
        HDONE_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "can't reset MPI-I/O collective_op property")

    /* Free memory */
    if (chunk_io_option)
        H5MM_xfree(chunk_io_option);
    if (chunk_addr)
        H5MM_xfree(chunk_addr);

    FUNC_LEAVE_NOAPI_TAG(ret_value)
} /* end H5D__multi_chunk_collective_io */

/*-------------------------------------------------------------------------
 * Function:    H5D__multi_chunk_filtered_collective_io
 *
 * Purpose:     Performs collective I/O on filtered chunks iteratively to
 *              save on memory and potentially get better performance
 *              depending on the average number of chunks per rank. While
 *              linked-chunk I/O will construct and work on a list of all
 *              of the chunks selected in the I/O operation at once, this
 *              function works iteratively on a set of chunks at a time; at
 *              most one chunk per rank per iteration.  The general
 *              algorithm is as follows:
 *
 *              1. Construct a list of selected chunks in the collective
 *                 I/O operation
 *              2. If the operation is a read operation, loop an amount of
 *                 times equal to the maximum number of chunks selected on
 *                 any particular rank and on each iteration:
 *                 A. Participate in a collective read of chunks from
 *                    the file (ranks that run out of chunks still need
 *                    to participate)
 *                 B. Unfilter the chunk that was read (if any)
 *                 C. Scatter the read chunk's data to the application's
 *                    read buffer
 *              3. If the operation is a write operation, redistribute any
 *                 chunks being written to by more than 1 MPI rank, such
 *                 that the chunk is only owned by 1 MPI rank. The rank
 *                 writing to the chunk which currently has the least
 *                 amount of chunks assigned to it becomes the new owner
 *                 (in the case of ties, the lowest MPI rank becomes the
 *                 new owner). Then, loop an amount of times equal to the
 *                 maximum number of chunks selected on any particular
 *                 rank and on each iteration:
 *                 A. Participate in a collective read of chunks from
 *                    the file (ranks that run out of chunks still need
 *                    to participate)
 *                    I. If we actually read a chunk from the file (if
 *                       a chunk is being fully overwritten, we skip
 *                       reading it), pass the chunk through the filter
 *                       pipeline in reverse order (unfilter the chunk)
 *                 B. Update the chunk data with the modifications from
 *                    the owning rank
 *                 C. Receive any modification data from other ranks and
 *                    update the chunk data with those modifications
 *                 D. Filter the chunk
 *                 E. Contribute the chunk to an array gathered by
 *                    all ranks which contains information for
 *                    re-allocating space in the file for every chunk
 *                    modified in this iteration (up to one chunk per
 *                    rank; some ranks may not have a selection/may have
 *                    less chunks to work on than other ranks). Then,
 *                    each rank collectively re-allocates each chunk
 *                    from the gathered array with their new sizes
 *                    after the filter operation
 *                 F. Proceed with the collective write operation
 *                    for the chunks modified on this iteration
 *                 G. Contribute the chunk to an array gathered by
 *                    all ranks which contains information for
 *                    re-inserting every chunk modified on this
 *                    iteration into the chunk index. Then, each rank
 *                    collectively re-inserts each chunk from the
 *                    gathered array into the chunk index
 *
 *              TODO: Note that steps E. and G. here are both collective
 *                    operations that partially share data from the
 *                    H5D_filtered_collective_io_info_t structure. To
 *                    try to conserve on memory a bit, the distributed
 *                    arrays these operations create are discarded after
 *                    each operation is performed. If memory consumption
 *                    here proves to not be an issue, the necessary data
 *                    for both operations could be combined into a single
 *                    structure so that only one collective MPI operation
 *                    is needed to carry out both operations, rather than
 *                    two.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__multi_chunk_filtered_collective_io(H5D_io_info_t *io_info, H5D_dset_io_info_t *dset_info, int mpi_rank,
                                        int mpi_size)
{
    H5D_filtered_collective_io_info_t *chunk_list       = NULL; /* The list of chunks being read/written */
    H5D_filtered_collective_io_info_t *chunk_hash_table = NULL;
    unsigned char                    **chunk_msg_bufs   = NULL;
    H5D_io_info_t                      ctg_io_info; /* Contiguous I/O info object */
    MPI_Datatype                       mem_type             = MPI_BYTE;
    MPI_Datatype                       file_type            = MPI_BYTE;
    hbool_t                            mem_type_is_derived  = FALSE;
    hbool_t                            file_type_is_derived = FALSE;
    hbool_t                            have_chunk_to_process;
    size_t                             chunk_list_num_entries;
    size_t                             i;
    size_t                             max_num_chunks;
    int                                chunk_msg_bufs_len = 0;
    int                                mpi_code;
    herr_t                             ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE_TAG(dset_info->dset->oloc.addr)

    HDassert(io_info);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_DEBUG_VA(mpi_rank, "Performing Multi-chunk I/O (%s) with MPI Comm size of %d",
                      io_info->op_type == H5D_IO_OP_WRITE ? "write" : "read", mpi_size);
    H5D_MPIO_TIME_START(mpi_rank, "Multi-chunk I/O");
#endif

    /* Set the actual chunk opt mode property */
    H5CX_set_mpio_actual_chunk_opt(H5D_MPIO_MULTI_CHUNK);

    /* Set the actual_io_mode property.
     * Multi chunk I/O does not break to independent, so can set right away
     */
    H5CX_set_mpio_actual_io_mode(H5D_MPIO_CHUNK_COLLECTIVE);

    /* Build a list of selected chunks in the collective IO operation */
    if (H5D__mpio_collective_filtered_chunk_io_setup(io_info, dset_info, &chunk_list, &chunk_list_num_entries,
                                                     mpi_rank) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "couldn't construct filtered I/O info list")

    /* Retrieve the maximum number of chunks selected for any rank */
    if (MPI_SUCCESS != (mpi_code = MPI_Allreduce(&chunk_list_num_entries, &max_num_chunks, 1,
                                                 MPI_UNSIGNED_LONG_LONG, MPI_MAX, io_info->comm)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Allreduce failed", mpi_code)

    /* If no one has anything selected at all, end the operation */
    if (0 == max_num_chunks)
        HGOTO_DONE(SUCCEED);

    /* Set up contiguous I/O info object */
    H5MM_memcpy(&ctg_io_info, io_info, sizeof(ctg_io_info));

    if (io_info->op_type == H5D_IO_OP_READ) { /* Filtered collective read */
        for (i = 0; i < max_num_chunks; i++) {
            /* Check if this rank has a chunk to work on for this iteration */
            have_chunk_to_process = (i < chunk_list_num_entries);

            if (H5D__mpio_collective_filtered_chunk_read(have_chunk_to_process ? &chunk_list[i] : NULL,
                                                         have_chunk_to_process ? 1 : 0, io_info, dset_info,
                                                         mpi_rank, mpi_size) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "couldn't read filtered chunks")

            if (have_chunk_to_process && chunk_list[i].buf) {
                H5MM_free(chunk_list[i].buf);
                chunk_list[i].buf = NULL;
            }
        }
    }
    else { /* Filtered collective write */
        H5D_chk_idx_info_t index_info;
        hsize_t            mpi_buf_count;

        /* Construct chunked index info */
        H5D_MPIO_INIT_CHUNK_IDX_INFO(index_info, dset_info->dset);

        if (mpi_size > 1) {
            /* Redistribute shared chunks being written to */
            if (H5D__mpio_redistribute_shared_chunks(chunk_list, chunk_list_num_entries, io_info, mpi_rank,
                                                     mpi_size, NULL) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "unable to redistribute shared chunks")

            /* Send any chunk modification messages for chunks this rank no longer owns */
            if (H5D__mpio_share_chunk_modification_data(chunk_list, &chunk_list_num_entries, io_info,
                                                        dset_info, mpi_rank, mpi_size, &chunk_hash_table,
                                                        &chunk_msg_bufs, &chunk_msg_bufs_len) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                            "unable to send chunk modification data between MPI ranks")
        }

        /* Iterate over the max number of chunks among all ranks, as this rank could
         * have no chunks left to work on, but it still needs to participate in the
         * collective re-allocation and re-insertion of chunks modified by other ranks.
         */
        for (i = 0; i < max_num_chunks; i++) {
            /* Check if this rank has a chunk to work on for this iteration */
            have_chunk_to_process = (i < chunk_list_num_entries) && (mpi_rank == chunk_list[i].new_owner);

            /* Proceed to update the chunk this rank owns (if any left) with its
             * own modification data and data from other ranks, before re-filtering
             * the chunks. As chunk reads are done collectively here, all ranks
             * must participate.
             */
            if (H5D__mpio_collective_filtered_chunk_update(have_chunk_to_process ? &chunk_list[i] : NULL,
                                                           have_chunk_to_process ? 1 : 0, chunk_hash_table,
                                                           chunk_msg_bufs, chunk_msg_bufs_len, io_info,
                                                           dset_info, mpi_rank, mpi_size) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "couldn't update modified chunks")

            /* All ranks now collectively re-allocate file space for all chunks */
            if (H5D__mpio_collective_filtered_chunk_reallocate(have_chunk_to_process ? &chunk_list[i] : NULL,
                                                               have_chunk_to_process ? 1 : 0, NULL, io_info,
                                                               &index_info, mpi_rank, mpi_size) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                            "couldn't collectively re-allocate file space for chunks")

            /*
             * If this rank has a chunk to work on, create a MPI type
             * for writing out the chunk. Otherwise, the rank will
             * use MPI_BYTE for the file and memory type and specify
             * a count of 0.
             */
            if (H5D__mpio_collective_filtered_io_type(
                    have_chunk_to_process ? &chunk_list[i] : NULL, have_chunk_to_process ? 1 : 0,
                    io_info->op_type, &mem_type, &mem_type_is_derived, &file_type, &file_type_is_derived) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                            "couldn't create MPI type for writing filtered chunks")

            mpi_buf_count = (file_type_is_derived || mem_type_is_derived) ? 1 : 0;

            /* Override the write buffer to point to the chunk data buffer */
            if (have_chunk_to_process) {
                /*
                 * Override the write buffer to point to the
                 * chunk's data buffer
                 */
                ctg_io_info.base_maddr.cvp = chunk_list[i].buf;

                /*
                 * Setup the base storage address for this
                 * operation to be the chunk's file address
                 */
                ctg_io_info.store_faddr = chunk_list[i].chunk_new.offset;
            }
            else {
                ctg_io_info.store_faddr = 0;
                ctg_io_info.base_maddr  = dset_info->buf;
            }

            /* Perform the I/O */
            if (H5D__final_collective_io(&ctg_io_info, mpi_buf_count, file_type, mem_type) < 0)
                HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish MPI-IO")

            /* Free up resources in anticipation of following collective operation */
            if (have_chunk_to_process && chunk_list[i].buf) {
                H5MM_free(chunk_list[i].buf);
                chunk_list[i].buf = NULL;
            }

            /* Participate in the collective re-insertion of all chunks modified
             * in this iteration into the chunk index
             */
            if (H5D__mpio_collective_filtered_chunk_reinsert(have_chunk_to_process ? &chunk_list[i] : NULL,
                                                             have_chunk_to_process ? 1 : 0, NULL, io_info,
                                                             dset_info, &index_info, mpi_rank, mpi_size) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                            "couldn't collectively re-insert modified chunks into chunk index")

            /* Free the MPI types, if they were derived */
            if (mem_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&mem_type)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            mem_type_is_derived = FALSE;
            if (file_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&file_type)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            file_type_is_derived = FALSE;
        } /* end for */
    }

done:
    /* Free the MPI buf and file types, if they were derived */
    if (mem_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&mem_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    if (file_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&file_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

    if (chunk_msg_bufs) {
        for (i = 0; i < (size_t)chunk_msg_bufs_len; i++)
            H5MM_free(chunk_msg_bufs[i]);

        H5MM_free(chunk_msg_bufs);
    }

    HASH_CLEAR(hh, chunk_hash_table);

    /* Free resources used by a rank which had some selection */
    if (chunk_list) {
        for (i = 0; i < chunk_list_num_entries; i++)
            if (chunk_list[i].buf)
                H5MM_free(chunk_list[i].buf);

        H5MM_free(chunk_list);
    } /* end if */

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI_TAG(ret_value)
} /* end H5D__multi_chunk_filtered_collective_io() */

/*-------------------------------------------------------------------------
 * Function:    H5D__inter_collective_io
 *
 * Purpose:     Routine for the shared part of collective IO between multiple chunk
 *              collective IO and contiguous collective IO
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Muqun Yang
 *              Monday, Feb. 13th, 2006
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__inter_collective_io(H5D_io_info_t *io_info, const H5D_dset_io_info_t *di, H5S_t *file_space,
                         H5S_t *mem_space)
{
    int          mpi_buf_count; /* # of MPI types */
    hbool_t      mbt_is_derived = FALSE;
    hbool_t      mft_is_derived = FALSE;
    MPI_Datatype mpi_file_type, mpi_buf_type;
    int          mpi_code; /* MPI return code */
#ifdef H5Dmpio_DEBUG
    int mpi_rank;
#endif
    herr_t ret_value = SUCCEED; /* return value */

    FUNC_ENTER_PACKAGE

#ifdef H5Dmpio_DEBUG
    mpi_rank = H5F_mpi_get_rank(di->dset->oloc.file);
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Inter collective I/O");
    if (mpi_rank < 0)
        HGOTO_ERROR(H5E_IO, H5E_MPI, FAIL, "unable to obtain MPI rank")
#endif

    HDassert(io_info);

    if ((file_space != NULL) && (mem_space != NULL)) {
        int      mpi_file_count;     /* Number of file "objects" to transfer */
        hsize_t *permute_map = NULL; /* array that holds the mapping from the old,
                                        out-of-order displacements to the in-order
                                        displacements of the MPI datatypes of the
                                        point selection of the file space */
        hbool_t is_permuted = FALSE;

        HDassert(di);

        /* Obtain disk and memory MPI derived datatype */
        /* NOTE: The permute_map array can be allocated within H5S_mpio_space_type
         *              and will be fed into the next call to H5S_mpio_space_type
         *              where it will be freed.
         */
        if (H5S_mpio_space_type(file_space, di->type_info.src_type_size, &mpi_file_type, &mpi_file_count,
                                &mft_is_derived, /* OUT: datatype created */
                                TRUE,            /* this is a file space, so
                                                    permute the datatype if the
                                                    point selection is out of
                                                    order */
                                &permute_map,    /* OUT: a map to indicate
                                                    the permutation of
                                                    points selected in
                                                    case they are out of
                                                    order */
                                &is_permuted /* OUT */) < 0)
            HGOTO_ERROR(H5E_DATASPACE, H5E_BADTYPE, FAIL, "couldn't create MPI file type")
        /* Sanity check */
        if (is_permuted)
            HDassert(permute_map);
        if (H5S_mpio_space_type(mem_space, di->type_info.src_type_size, &mpi_buf_type, &mpi_buf_count,
                                &mbt_is_derived, /* OUT: datatype created */
                                FALSE,           /* this is a memory space, so if
                                                    the file space is not
                                                    permuted, there is no need to
                                                    permute the datatype if the
                                                    point selections are out of
                                                    order*/
                                &permute_map     /* IN: the permutation map
                                                    generated by the
                                                    file_space selection
                                                    and applied to the
                                                    memory selection */
                                ,
                                &is_permuted /* IN */) < 0)
            HGOTO_ERROR(H5E_DATASPACE, H5E_BADTYPE, FAIL, "couldn't create MPI buffer type")
        /* Sanity check */
        if (is_permuted)
            HDassert(!permute_map);
    } /* end if */
    else {
        /* For non-selection, participate with a none MPI derived datatype, the count is 0.  */
        mpi_buf_type   = MPI_BYTE;
        mpi_file_type  = MPI_BYTE;
        mpi_buf_count  = 0;
        mbt_is_derived = FALSE;
        mft_is_derived = FALSE;
    } /* end else */

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_DEBUG(mpi_rank, "before final collective I/O");
#endif

    /* Perform final collective I/O operation */
    if (H5D__final_collective_io(io_info, (hsize_t)mpi_buf_count, mpi_file_type, mpi_buf_type) < 0)
        HGOTO_ERROR(H5E_IO, H5E_CANTGET, FAIL, "couldn't finish collective MPI-IO")

done:
    /* Free the MPI buf and file types, if they were derived */
    if (mbt_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&mpi_buf_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    if (mft_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&mpi_file_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_DEBUG_VA(mpi_rank, "before leaving inter_collective_io ret_value = %d", ret_value);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__inter_collective_io() */

/*-------------------------------------------------------------------------
 * Function:    H5D__final_collective_io
 *
 * Purpose:     Routine for the common part of collective IO with different storages.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Muqun Yang
 *              Monday, Feb. 13th, 2006
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__final_collective_io(H5D_io_info_t *io_info, hsize_t mpi_buf_count, MPI_Datatype mpi_file_type,
                         MPI_Datatype mpi_buf_type)
{
#ifdef H5Dmpio_DEBUG
    int mpi_rank;
#endif
    herr_t ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

#ifdef H5Dmpio_DEBUG
    mpi_rank = H5F_mpi_get_rank(io_info->dsets_info[0].dset->oloc.file);
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Final collective I/O");
    if (mpi_rank < 0)
        HGOTO_ERROR(H5E_IO, H5E_MPI, FAIL, "unable to obtain MPI rank")
#endif

    /* Pass buf type, file type to the file driver.  */
    if (H5CX_set_mpi_coll_datatypes(mpi_buf_type, mpi_file_type) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "can't set MPI-I/O collective I/O datatypes")

    if (io_info->op_type == H5D_IO_OP_WRITE) {
        if ((io_info->md_io_ops.single_write_md)(io_info, mpi_buf_count, NULL, NULL) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "optimized write failed")
    } /* end if */
    else {
        if ((io_info->md_io_ops.single_read_md)(io_info, mpi_buf_count, NULL, NULL) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "optimized read failed")
    } /* end else */

done:
#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_DEBUG_VA(mpi_rank, "ret_value before leaving final_collective_io=%d", ret_value);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__final_collective_io */

/*-------------------------------------------------------------------------
 * Function:    H5D__cmp_piece_addr
 *
 * Purpose:     Routine to compare piece addresses
 *
 * Description: Callback for qsort() to compare piece addresses
 *
 * Return:      -1, 0, 1
 *
 *-------------------------------------------------------------------------
 */
static int
H5D__cmp_piece_addr(const void *piece_info1, const void *piece_info2)
{
    haddr_t addr1;
    haddr_t addr2;

    FUNC_ENTER_PACKAGE_NOERR

    addr1 = (*((const H5D_piece_info_t *const *)piece_info1))->faddr;
    addr2 = (*((const H5D_piece_info_t *const *)piece_info2))->faddr;

    FUNC_LEAVE_NOAPI(H5F_addr_cmp(addr1, addr2))
} /* end H5D__cmp_chunk_addr() */

/*-------------------------------------------------------------------------
 * Function:    H5D__cmp_filtered_collective_io_info_entry
 *
 * Purpose:     Routine to compare filtered collective chunk io info
 *              entries
 *
 * Description: Callback for qsort() to compare filtered collective chunk
 *              io info entries
 *
 * Return:      -1, 0, 1
 *
 *-------------------------------------------------------------------------
 */
static int
H5D__cmp_filtered_collective_io_info_entry(const void *filtered_collective_io_info_entry1,
                                           const void *filtered_collective_io_info_entry2)
{
    const H5D_filtered_collective_io_info_t *entry1;
    const H5D_filtered_collective_io_info_t *entry2;
    haddr_t                                  addr1 = HADDR_UNDEF;
    haddr_t                                  addr2 = HADDR_UNDEF;
    int                                      ret_value;

    FUNC_ENTER_PACKAGE_NOERR

    entry1 = (const H5D_filtered_collective_io_info_t *)filtered_collective_io_info_entry1;
    entry2 = (const H5D_filtered_collective_io_info_t *)filtered_collective_io_info_entry2;

    addr1 = entry1->chunk_new.offset;
    addr2 = entry2->chunk_new.offset;

    /*
     * If both chunk addresses are defined, H5F_addr_cmp is safe to use.
     * Otherwise, if both addresses aren't defined, compared chunk
     * entries based on their chunk index. Finally, if only one chunk
     * address is defined, return the appropriate value based on which
     * is defined.
     */
    if (H5F_addr_defined(addr1) && H5F_addr_defined(addr2)) {
        ret_value = H5F_addr_cmp(addr1, addr2);
    }
    else if (!H5F_addr_defined(addr1) && !H5F_addr_defined(addr2)) {
        hsize_t chunk_idx1 = entry1->index_info.chunk_idx;
        hsize_t chunk_idx2 = entry2->index_info.chunk_idx;

        ret_value = (chunk_idx1 > chunk_idx2) - (chunk_idx1 < chunk_idx2);
    }
    else
        ret_value = H5F_addr_defined(addr1) ? 1 : -1;

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__cmp_filtered_collective_io_info_entry() */

/*-------------------------------------------------------------------------
 * Function:    H5D__cmp_chunk_redistribute_info
 *
 * Purpose:     Routine to compare two H5D_chunk_redistribute_info_t
 *              structures
 *
 * Description: Callback for qsort() to compare two
 *              H5D_chunk_redistribute_info_t structures
 *
 * Return:      -1, 0, 1
 *
 *-------------------------------------------------------------------------
 */
static int
H5D__cmp_chunk_redistribute_info(const void *_entry1, const void *_entry2)
{
    const H5D_chunk_redistribute_info_t *entry1;
    const H5D_chunk_redistribute_info_t *entry2;
    hsize_t                              chunk_index1;
    hsize_t                              chunk_index2;
    int                                  ret_value;

    FUNC_ENTER_PACKAGE_NOERR

    entry1 = (const H5D_chunk_redistribute_info_t *)_entry1;
    entry2 = (const H5D_chunk_redistribute_info_t *)_entry2;

    chunk_index1 = entry1->chunk_idx;
    chunk_index2 = entry2->chunk_idx;

    if (chunk_index1 == chunk_index2) {
        int orig_owner1 = entry1->orig_owner;
        int orig_owner2 = entry2->orig_owner;

        ret_value = (orig_owner1 > orig_owner2) - (orig_owner1 < orig_owner2);
    }
    else
        ret_value = (chunk_index1 > chunk_index2) - (chunk_index1 < chunk_index2);

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__cmp_chunk_redistribute_info() */

/*-------------------------------------------------------------------------
 * Function:    H5D__cmp_chunk_redistribute_info_orig_owner
 *
 * Purpose:     Routine to compare the original owning MPI rank for two
 *              H5D_chunk_redistribute_info_t structures
 *
 * Description: Callback for qsort() to compare the original owning MPI
 *              rank for two H5D_chunk_redistribute_info_t
 *              structures
 *
 * Return:      -1, 0, 1
 *
 *-------------------------------------------------------------------------
 */
static int
H5D__cmp_chunk_redistribute_info_orig_owner(const void *_entry1, const void *_entry2)
{
    const H5D_chunk_redistribute_info_t *entry1;
    const H5D_chunk_redistribute_info_t *entry2;
    int                                  owner1 = -1;
    int                                  owner2 = -1;
    int                                  ret_value;

    FUNC_ENTER_PACKAGE_NOERR

    entry1 = (const H5D_chunk_redistribute_info_t *)_entry1;
    entry2 = (const H5D_chunk_redistribute_info_t *)_entry2;

    owner1 = entry1->orig_owner;
    owner2 = entry2->orig_owner;

    if (owner1 == owner2) {
        haddr_t addr1 = entry1->chunk_block.offset;
        haddr_t addr2 = entry2->chunk_block.offset;

        /*
         * If both chunk addresses are defined, H5F_addr_cmp is safe to use.
         * Otherwise, if both addresses aren't defined, compared chunk
         * entries based on their chunk index. Finally, if only one chunk
         * address is defined, return the appropriate value based on which
         * is defined.
         */
        if (H5F_addr_defined(addr1) && H5F_addr_defined(addr2)) {
            ret_value = H5F_addr_cmp(addr1, addr2);
        }
        else if (!H5F_addr_defined(addr1) && !H5F_addr_defined(addr2)) {
            hsize_t chunk_idx1 = entry1->chunk_idx;
            hsize_t chunk_idx2 = entry2->chunk_idx;

            ret_value = (chunk_idx1 > chunk_idx2) - (chunk_idx1 < chunk_idx2);
        }
        else
            ret_value = H5F_addr_defined(addr1) ? 1 : -1;
    }
    else
        ret_value = (owner1 > owner2) - (owner1 < owner2);

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__cmp_chunk_redistribute_info_orig_owner() */

/*-------------------------------------------------------------------------
 * Function:    H5D__obtain_mpio_mode
 *
 * Purpose:     Routine to obtain each io mode(collective,independent or none) for each chunk;
 *              Each chunk address is also obtained.
 *
 * Description:
 *
 *              1) Each process provides two piece of information for all chunks having selection
 *                 a) chunk index
 *                 b) whether this chunk is regular(for MPI derived datatype not working case)
 *
 *              2) Gather all the information to the root process
 *
 *              3) Root process will do the following:
 *                 a) Obtain chunk addresses for all chunks in this dataspace
 *                 b) With the consideration of the user option, calculate IO mode for each chunk
 *                 c) Build MPI derived datatype to combine "chunk address" and "assign_io" information
 *                      in order to do MPI Bcast only once
 *                 d) MPI Bcast the IO mode and chunk address information for each chunk.
 *              4) Each process then retrieves IO mode and chunk address information to assign_io_mode and
 *chunk_addr.
 *
 * Parameters:
 *
 *              Input: H5D_io_info_t* io_info,
 *                      H5D_dset_io_info_t *di,(dataset info struct)
 *              Output: uint8_t assign_io_mode[], : IO mode, collective, independent or none
 *                      haddr_t chunk_addr[],     : chunk address array for each chunk
 *
 * Return:      Non-negative on success/Negative on failure
 *
 * Programmer:  Muqun Yang
 *              Monday, Feb. 13th, 2006
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__obtain_mpio_mode(H5D_io_info_t *io_info, H5D_dset_io_info_t *di, uint8_t assign_io_mode[],
                      haddr_t chunk_addr[], int mpi_rank, int mpi_size)
{
    size_t                  total_chunks;
    unsigned                percent_nproc_per_chunk, threshold_nproc_per_chunk;
    uint8_t                *io_mode_info      = NULL;
    uint8_t                *recv_io_mode_info = NULL;
    uint8_t                *mergebuf          = NULL;
    uint8_t                *tempbuf;
    H5SL_node_t            *chunk_node;
    H5D_piece_info_t       *chunk_info;
    H5P_coll_md_read_flag_t md_reads_file_flag;
    hbool_t                 md_reads_context_flag;
    hbool_t                 restore_md_reads_state = FALSE;
    MPI_Comm                comm;
    int                     root;
    size_t                  ic;
    int                     mpi_code;
    herr_t                  ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(di->layout->type == H5D_CHUNKED);

    /* Assign the rank 0 to the root */
    root = 0;
    comm = io_info->comm;

    /* Setup parameters */
    H5_CHECKED_ASSIGN(total_chunks, size_t, di->layout->u.chunk.nchunks, hsize_t);
    if (H5CX_get_mpio_chunk_opt_ratio(&percent_nproc_per_chunk) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "couldn't get percent nproc per chunk")
    /* if ratio is 0, perform collective io */
    if (0 == percent_nproc_per_chunk) {
        if (H5D__chunk_addrmap(di->dset, chunk_addr) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get chunk address");
        for (ic = 0; ic < total_chunks; ic++)
            assign_io_mode[ic] = H5D_CHUNK_IO_MODE_COL;

        HGOTO_DONE(SUCCEED)
    } /* end if */

    threshold_nproc_per_chunk = (unsigned)mpi_size * percent_nproc_per_chunk / 100;

    /* Allocate memory */
    if (NULL == (io_mode_info = (uint8_t *)H5MM_calloc(total_chunks)))
        HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate I/O mode info buffer")
    if (NULL == (mergebuf = (uint8_t *)H5MM_malloc((sizeof(haddr_t) + 1) * total_chunks)))
        HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate mergebuf buffer")
    tempbuf = mergebuf + total_chunks;
    if (mpi_rank == root)
        if (NULL == (recv_io_mode_info = (uint8_t *)H5MM_malloc(total_chunks * (size_t)mpi_size)))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate recv I/O mode info buffer")

    /* Obtain the regularity and selection information for all chunks in this process. */
    chunk_node = H5SL_first(di->layout_io_info.chunk_map->dset_sel_pieces);
    while (chunk_node) {
        chunk_info = (H5D_piece_info_t *)H5SL_item(chunk_node);

        io_mode_info[chunk_info->index] = H5D_CHUNK_SELECT_REG; /* this chunk is selected and is "regular" */
        chunk_node                      = H5SL_next(chunk_node);
    } /* end while */

    /* Gather all the information */
    H5_CHECK_OVERFLOW(total_chunks, size_t, int)
    if (MPI_SUCCESS != (mpi_code = MPI_Gather(io_mode_info, (int)total_chunks, MPI_BYTE, recv_io_mode_info,
                                              (int)total_chunks, MPI_BYTE, root, comm)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Gather failed", mpi_code)

    /* Calculate the mode for IO(collective, independent or none) at root process */
    if (mpi_rank == root) {
        size_t    nproc;
        unsigned *nproc_per_chunk;

        /*
         * If enabled, disable collective metadata reads here.
         * Since the chunk address mapping is done on rank 0
         * only here, it will cause problems if collective
         * metadata reads are enabled.
         */
        if (H5F_get_coll_metadata_reads(di->dset->oloc.file)) {
            md_reads_file_flag    = H5P_FORCE_FALSE;
            md_reads_context_flag = FALSE;
            H5F_set_coll_metadata_reads(di->dset->oloc.file, &md_reads_file_flag, &md_reads_context_flag);
            restore_md_reads_state = TRUE;
        }

        /* pre-computing: calculate number of processes and
            regularity of the selection occupied in each chunk */
        if (NULL == (nproc_per_chunk = (unsigned *)H5MM_calloc(total_chunks * sizeof(unsigned))))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate nproc_per_chunk buffer")

        /* calculating the chunk address */
        if (H5D__chunk_addrmap(di->dset, chunk_addr) < 0) {
            H5MM_free(nproc_per_chunk);
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get chunk address")
        } /* end if */

        /* checking for number of process per chunk and regularity of the selection*/
        for (nproc = 0; nproc < (size_t)mpi_size; nproc++) {
            uint8_t *tmp_recv_io_mode_info = recv_io_mode_info + (nproc * total_chunks);

            /* Calculate the number of process per chunk and adding irregular selection option */
            for (ic = 0; ic < total_chunks; ic++, tmp_recv_io_mode_info++) {
                if (*tmp_recv_io_mode_info != 0) {
                    nproc_per_chunk[ic]++;
                } /* end if */
            }     /* end for */
        }         /* end for */

        /* Calculating MPIO mode for each chunk (collective, independent, none) */
        for (ic = 0; ic < total_chunks; ic++) {
            if (nproc_per_chunk[ic] > MAX(1, threshold_nproc_per_chunk)) {
                assign_io_mode[ic] = H5D_CHUNK_IO_MODE_COL;
            } /* end if */
        }     /* end for */

        /* merge buffer io_mode info and chunk addr into one */
        H5MM_memcpy(mergebuf, assign_io_mode, total_chunks);
        H5MM_memcpy(tempbuf, chunk_addr, sizeof(haddr_t) * total_chunks);

        H5MM_free(nproc_per_chunk);
    } /* end if */

    /* Broadcasting the MPI_IO option info. and chunk address info. */
    if ((sizeof(haddr_t) + 1) * total_chunks > INT_MAX)
        HGOTO_ERROR(H5E_DATASET, H5E_BADVALUE, FAIL, "result overflow")
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Bcast(mergebuf, (int)((sizeof(haddr_t) + 1) * total_chunks), MPI_BYTE, root, comm)))
        HMPI_GOTO_ERROR(FAIL, "MPI_BCast failed", mpi_code)

    H5MM_memcpy(assign_io_mode, mergebuf, total_chunks);
    H5MM_memcpy(chunk_addr, tempbuf, sizeof(haddr_t) * total_chunks);

#ifdef H5_HAVE_INSTRUMENTED_LIBRARY
    {
        hbool_t coll_op = FALSE;

        for (ic = 0; ic < total_chunks; ic++)
            if (assign_io_mode[ic] == H5D_CHUNK_IO_MODE_COL) {
                if (H5CX_test_set_mpio_coll_chunk_multi_ratio_coll(0) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
                coll_op = TRUE;
                break;
            } /* end if */

        if (!coll_op)
            if (H5CX_test_set_mpio_coll_chunk_multi_ratio_ind(0) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTSET, FAIL, "unable to set property value")
    }
#endif

done:
    /* Re-enable collective metadata reads if we disabled them */
    if (restore_md_reads_state)
        H5F_set_coll_metadata_reads(di->dset->oloc.file, &md_reads_file_flag, &md_reads_context_flag);

    if (io_mode_info)
        H5MM_free(io_mode_info);
    if (mergebuf)
        H5MM_free(mergebuf);
    if (recv_io_mode_info) {
        HDassert(mpi_rank == root);
        H5MM_free(recv_io_mode_info);
    } /* end if */

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__obtain_mpio_mode() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_chunk_io_setup
 *
 * Purpose:     Constructs a list of entries which contain the necessary
 *              information for inter-process communication when performing
 *              collective io on filtered chunks. This list is used by
 *              each MPI rank when performing I/O on locally selected
 *              chunks and also in operations that must be collectively
 *              done on every chunk, such as chunk re-allocation, insertion
 *              of chunks into the chunk index, etc.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_chunk_io_setup(const H5D_io_info_t *io_info, const H5D_dset_io_info_t *di,
                                             H5D_filtered_collective_io_info_t **chunk_list,
                                             size_t *num_entries, int mpi_rank)
{
    H5D_filtered_collective_io_info_t *local_info_array = NULL;
    H5D_chunk_ud_t                     udata;
    hbool_t                            filter_partial_edge_chunks;
    size_t                             num_chunks_selected;
    size_t                             i;
    herr_t                             ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(io_info);
    HDassert(di);
    HDassert(chunk_list);
    HDassert(num_entries);
#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Filtered Collective I/O Setup");
#endif

    HDassert(di->layout->type == H5D_CHUNKED);

    /* Each rank builds a local list of the chunks they have selected */
    if ((num_chunks_selected = H5SL_count(di->layout_io_info.chunk_map->dset_sel_pieces))) {
        H5D_piece_info_t *chunk_info;
        H5SL_node_t      *chunk_node;
        hsize_t           select_npoints;
        hbool_t           need_sort = FALSE;

        /* Determine whether partial edge chunks should be filtered */
        filter_partial_edge_chunks =
            !(di->dset->shared->layout.u.chunk.flags & H5O_LAYOUT_CHUNK_DONT_FILTER_PARTIAL_BOUND_CHUNKS);

        if (NULL == (local_info_array = H5MM_malloc(num_chunks_selected * sizeof(*local_info_array))))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate local io info array buffer")

        chunk_node = H5SL_first(di->layout_io_info.chunk_map->dset_sel_pieces);
        for (i = 0; chunk_node; i++) {
            chunk_info = (H5D_piece_info_t *)H5SL_item(chunk_node);

            /* Obtain this chunk's address */
            if (H5D__chunk_lookup(di->dset, chunk_info->scaled, &udata) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "error looking up chunk address")

            /* Initialize rank-local chunk info */
            local_info_array[i].chunk_info     = chunk_info;
            local_info_array[i].chunk_buf_size = 0;
            local_info_array[i].num_writers    = 0;
            local_info_array[i].orig_owner     = mpi_rank;
            local_info_array[i].new_owner      = mpi_rank;
            local_info_array[i].buf            = NULL;

            select_npoints              = H5S_GET_SELECT_NPOINTS(chunk_info->fspace);
            local_info_array[i].io_size = (size_t)select_npoints * di->type_info.dst_type_size;

            /*
             * Determine whether this chunk will need to be read from the file. If this is
             * a read operation, the chunk will be read. If this is a write operation, we
             * generally need to read a filtered chunk from the file before modifying it,
             * unless the chunk is being fully overwritten.
             *
             * TODO: Currently the full overwrite status of a chunk is only obtained on a
             * per-rank basis. This means that if the total selection in the chunk, as
             * determined by the combination of selections of all of the ranks interested in
             * the chunk, covers the entire chunk, the performance optimization of not reading
             * the chunk from the file is still valid, but is not applied in the current
             * implementation.
             *
             * To implement this case, a few approaches were considered:
             *
             *  - Keep a running total (distributed to each rank) of the number of chunk
             *    elements selected during chunk redistribution and compare that to the total
             *    number of elements in the chunk once redistribution is finished
             *
             *  - Process all incoming chunk messages before doing I/O (these are currently
             *    processed AFTER doing I/O), combine the owning rank's selection in a chunk
             *    with the selections received from other ranks and check to see whether that
             *    combined selection covers the entire chunk
             *
             * The first approach will be dangerous if the application performs an overlapping
             * write to a chunk, as the number of selected elements can equal or exceed the
             * number of elements in the chunk without the whole chunk selection being covered.
             * While it might be considered erroneous for an application to do an overlapping
             * write, we don't explicitly disallow it.
             *
             * The second approach contains a bit of complexity in that part of the chunk
             * messages will be needed before doing I/O and part will be needed after doing I/O.
             * Since modification data from chunk messages can't be applied until after any I/O
             * is performed (otherwise, we'll overwrite any applied modification data), chunk
             * messages are currently entirely processed after I/O. However, in order to determine
             * if a chunk is being fully overwritten, we need the dataspace portion of the chunk
             * messages before doing I/O. The naive way to do this is to process chunk messages
             * twice, using just the relevant information from the message before and after I/O.
             * The better way would be to avoid processing chunk messages twice by extracting (and
             * keeping around) the dataspace portion of the message before I/O and processing the
             * rest of the chunk message after I/O. Note that the dataspace portion of each chunk
             * message is used to correctly apply chunk modification data from the message, so
             * must be kept around both before and after I/O in this case.
             */
            if (io_info->op_type == H5D_IO_OP_READ)
                local_info_array[i].need_read = TRUE;
            else {
                local_info_array[i].need_read =
                    local_info_array[i].io_size < (size_t)di->dset->shared->layout.u.chunk.size;
            }

            local_info_array[i].skip_filter_pline = FALSE;
            if (!filter_partial_edge_chunks) {
                /*
                 * If this is a partial edge chunk and the "don't filter partial edge
                 * chunks" flag is set, make sure not to apply filters to the chunk.
                 */
                if (H5D__chunk_is_partial_edge_chunk(di->dset->shared->ndims,
                                                     di->dset->shared->layout.u.chunk.dim, chunk_info->scaled,
                                                     di->dset->shared->curr_dims))
                    local_info_array[i].skip_filter_pline = TRUE;
            }

            /* Initialize the chunk's shared info */
            local_info_array[i].chunk_current = udata.chunk_block;
            local_info_array[i].chunk_new     = udata.chunk_block;

            /*
             * Check if the list is not in ascending order of offset in the file
             * or has unallocated chunks. In either case, the list should get
             * sorted.
             */
            if (i) {
                haddr_t curr_chunk_offset = local_info_array[i].chunk_current.offset;
                haddr_t prev_chunk_offset = local_info_array[i - 1].chunk_current.offset;

                if (!H5F_addr_defined(prev_chunk_offset) || !H5F_addr_defined(curr_chunk_offset) ||
                    (curr_chunk_offset < prev_chunk_offset))
                    need_sort = TRUE;
            }

            /*
             * Extensible arrays may calculate a chunk's index a little differently
             * than normal when the dataset's unlimited dimension is not the
             * slowest-changing dimension, so set the index here based on what the
             * extensible array code calculated instead of what was calculated
             * in the chunk file mapping.
             */
            if (di->dset->shared->layout.u.chunk.idx_type == H5D_CHUNK_IDX_EARRAY)
                local_info_array[i].index_info.chunk_idx = udata.chunk_idx;
            else
                local_info_array[i].index_info.chunk_idx = chunk_info->index;

            local_info_array[i].index_info.filter_mask = udata.filter_mask;
            local_info_array[i].index_info.need_insert = FALSE;

            chunk_node = H5SL_next(chunk_node);
        }

        /* Ensure the chunk list is sorted in ascending order of offset in the file */
        if (need_sort)
            HDqsort(local_info_array, num_chunks_selected, sizeof(H5D_filtered_collective_io_info_t),
                    H5D__cmp_filtered_collective_io_info_entry);

#ifdef H5Dmpio_DEBUG
        H5D__mpio_dump_collective_filtered_chunk_list(local_info_array, num_chunks_selected, mpi_rank);
#endif
    }
    else if (H5F_get_coll_metadata_reads(di->dset->oloc.file)) {
        hsize_t scaled[H5O_LAYOUT_NDIMS] = {0};

        /*
         * If this rank has no selection in the dataset and collective
         * metadata reads are enabled, do a fake lookup of a chunk to
         * ensure that this rank has the chunk index opened. Otherwise,
         * only the ranks that had a selection will have opened the
         * chunk index and they will have done so independently. Therefore,
         * when ranks with no selection participate in later collective
         * metadata reads, they will try to open the chunk index collectively
         * and issues will occur since other ranks won't participate.
         *
         * In the future, we should consider having a chunk index "open"
         * callback that can be used to ensure collectivity between ranks
         * in a more natural way, but this hack should suffice for now.
         */
        if (H5D__chunk_lookup(di->dset, scaled, &udata) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "error looking up chunk address")
    }

    *chunk_list  = local_info_array;
    *num_entries = num_chunks_selected;

done:
#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_collective_filtered_chunk_io_setup() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_redistribute_shared_chunks
 *
 * Purpose:     When performing a parallel write on a chunked Dataset with
 *              filters applied, we must ensure that any particular chunk
 *              is only written to by a single MPI rank in order to avoid
 *              potential data races on the chunk. This function is used to
 *              redistribute (by assigning ownership to a single rank) any
 *              chunks which are selected by more than one MPI rank.
 *
 *              An initial Allgather is performed to determine how many
 *              chunks each rank has selected in the write operation and
 *              then that number is compared against a threshold value to
 *              determine whether chunk redistribution should be done on
 *              MPI rank 0 only, or on all MPI ranks.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_redistribute_shared_chunks(H5D_filtered_collective_io_info_t *chunk_list,
                                     size_t chunk_list_num_entries, const H5D_io_info_t *io_info,
                                     int mpi_rank, int mpi_size, size_t **rank_chunks_assigned_map)
{
    hbool_t redistribute_on_all_ranks;
    size_t *num_chunks_map       = NULL;
    size_t  coll_chunk_list_size = 0;
    size_t  i;
    int     mpi_code;
    herr_t  ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == chunk_list_num_entries);
    HDassert(io_info);
    HDassert(mpi_size > 1); /* No chunk sharing is possible for MPI Comm size of 1 */

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Redistribute shared chunks");
#endif

    /*
     * Allocate an array for each rank to keep track of the number of
     * chunks assigned to any other rank in order to cut down on future
     * MPI communication.
     */
    if (NULL == (num_chunks_map = H5MM_malloc((size_t)mpi_size * sizeof(*num_chunks_map))))
        HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL, "couldn't allocate assigned chunks array")

    /* Perform initial Allgather to determine the collective chunk list size */
    if (MPI_SUCCESS != (mpi_code = MPI_Allgather(&chunk_list_num_entries, 1, H5_SIZE_T_AS_MPI_TYPE,
                                                 num_chunks_map, 1, H5_SIZE_T_AS_MPI_TYPE, io_info->comm)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Allgather failed", mpi_code)

    for (i = 0; i < (size_t)mpi_size; i++)
        coll_chunk_list_size += num_chunks_map[i];

    /*
     * Determine whether we should perform chunk redistribution on all
     * ranks or just rank 0. For a relatively small number of chunks,
     * we redistribute on all ranks to cut down on MPI communication
     * overhead. For a larger number of chunks, we redistribute on
     * rank 0 only to cut down on memory usage.
     */
    redistribute_on_all_ranks = coll_chunk_list_size < H5D_CHUNK_REDISTRIBUTE_THRES;

    if (H5D__mpio_redistribute_shared_chunks_int(chunk_list, num_chunks_map, redistribute_on_all_ranks,
                                                 io_info, mpi_rank, mpi_size) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTREDISTRIBUTE, FAIL, "can't redistribute shared chunks")

    /*
     * If the caller provided a pointer for the mapping from
     * rank value -> number of chunks assigned, return that
     * mapping here.
     */
    if (rank_chunks_assigned_map) {
        /*
         * If we performed chunk redistribution on rank 0 only, distribute
         * the rank value -> number of chunks assigned mapping back to all
         * ranks.
         */
        if (!redistribute_on_all_ranks) {
            if (MPI_SUCCESS !=
                (mpi_code = MPI_Bcast(num_chunks_map, mpi_size, H5_SIZE_T_AS_MPI_TYPE, 0, io_info->comm)))
                HMPI_GOTO_ERROR(FAIL, "couldn't broadcast chunk mapping to other ranks", mpi_code)
        }

        *rank_chunks_assigned_map = num_chunks_map;
    }

done:
    if (!rank_chunks_assigned_map || (ret_value < 0)) {
        num_chunks_map = H5MM_xfree(num_chunks_map);
    }

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_redistribute_shared_chunks() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_redistribute_shared_chunks_int
 *
 * Purpose:     Routine to perform redistribution of shared chunks during
 *              parallel writes to datasets with filters applied.
 *
 *              If `all_ranks_involved` is TRUE, chunk redistribution
 *              occurs on all MPI ranks. This is usually done when there
 *              is a relatively small number of chunks involved in order to
 *              cut down on MPI communication overhead while increasing
 *              total memory usage a bit.
 *
 *              If `all_ranks_involved` is FALSE, only rank 0 will perform
 *              chunk redistribution. This is usually done when there is
 *              a relatively large number of chunks involved in order to
 *              cut down on total memory usage at the cost of increased
 *              overhead from MPI communication.
 *
 *              This implementation is as follows:
 *
 *              - All MPI ranks send their list of selected chunks to the
 *                ranks involved in chunk redistribution. Then, the
 *                involved ranks sort this new list in order of chunk
 *                index.
 *
 *              - The involved ranks scan the list looking for matching
 *                runs of chunk index values (corresponding to a shared
 *                chunk which has been selected by more than one rank in
 *                the I/O operation) and for each shared chunk,
 *                redistribute the chunk to the MPI rank writing to the
 *                chunk which currently has the least amount of chunks
 *                assigned to it. This is done by modifying the "new_owner"
 *                field in each of the list entries corresponding to that
 *                chunk. The involved ranks then re-sort the list in order
 *                of original chunk owner so that each rank's section of
 *                contributed chunks is contiguous in the collective chunk
 *                list.
 *
 *              - If chunk redistribution occurred on all ranks, each rank
 *                scans through the collective chunk list to find their
 *                contributed section of chunks and uses that to update
 *                their local chunk list with the newly-updated "new_owner"
 *                and "num_writers" fields. If chunk redistribution
 *                occurred only on rank 0, an MPI_Scatterv operation will
 *                be used to scatter the segments of the collective chunk
 *                list from rank 0 back to the corresponding ranks.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_redistribute_shared_chunks_int(H5D_filtered_collective_io_info_t *chunk_list,
                                         size_t *num_chunks_assigned_map, hbool_t all_ranks_involved,
                                         const H5D_io_info_t *io_info, int mpi_rank, int mpi_size)
{
    MPI_Datatype struct_type;
    MPI_Datatype packed_type;
    hbool_t      struct_type_derived = FALSE;
    hbool_t      packed_type_derived = FALSE;
    size_t       i;
    size_t       coll_chunk_list_num_entries = 0;
    void        *coll_chunk_list             = NULL;
    int         *counts_disps_array          = NULL;
    int         *counts_ptr                  = NULL;
    int         *displacements_ptr           = NULL;
    int          num_chunks_int;
    int          mpi_code;
    herr_t       ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(num_chunks_assigned_map);
    HDassert(chunk_list || 0 == num_chunks_assigned_map[mpi_rank]);
    HDassert(io_info);
    HDassert(mpi_size > 1);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Redistribute shared chunks (internal)");
#endif

    /*
     * Make sure it's safe to cast this rank's number
     * of chunks to be sent into an int for MPI
     */
    H5_CHECKED_ASSIGN(num_chunks_int, int, num_chunks_assigned_map[mpi_rank], size_t);

    /*
     * Phase 1 - Participate in collective gathering of every rank's
     * list of chunks to the ranks which are performing the redistribution
     * operation.
     */

    if (all_ranks_involved || (mpi_rank == 0)) {
        /*
         * Allocate array to store the receive counts of each rank, as well as
         * the displacements into the final array where each rank will place
         * their data. The first half of the array contains the receive counts
         * (in rank order), while the latter half contains the displacements
         * (also in rank order).
         */
        if (NULL == (counts_disps_array = H5MM_malloc(2 * (size_t)mpi_size * sizeof(*counts_disps_array)))) {
            /* Push an error, but still participate in collective gather operation */
            HDONE_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL,
                        "couldn't allocate receive counts and displacements array")
        }
        else {
            /* Set the receive counts from the assigned chunks map */
            counts_ptr = counts_disps_array;

            for (i = 0; i < (size_t)mpi_size; i++)
                H5_CHECKED_ASSIGN(counts_ptr[i], int, num_chunks_assigned_map[i], size_t);

            /* Set the displacements into the receive buffer for the gather operation */
            displacements_ptr = &counts_disps_array[mpi_size];

            *displacements_ptr = 0;
            for (i = 1; i < (size_t)mpi_size; i++)
                displacements_ptr[i] = displacements_ptr[i - 1] + counts_ptr[i - 1];
        }
    }

    /*
     * Construct MPI derived types for extracting information
     * necessary for MPI communication
     */
    if (H5D__mpio_get_chunk_redistribute_info_types(&packed_type, &packed_type_derived, &struct_type,
                                                    &struct_type_derived) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                    "can't create derived datatypes for chunk redistribution info")

    /* Perform gather operation */
    if (H5_mpio_gatherv_alloc(chunk_list, num_chunks_int, struct_type, counts_ptr, displacements_ptr,
                              packed_type, all_ranks_involved, 0, io_info->comm, mpi_rank, mpi_size,
                              &coll_chunk_list, &coll_chunk_list_num_entries) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGATHER, FAIL,
                    "can't gather chunk redistribution info to involved ranks")

    /*
     * If all ranks are redistributing shared chunks, we no
     * longer need the receive counts and displacements array
     */
    if (all_ranks_involved) {
        counts_disps_array = H5MM_xfree(counts_disps_array);
    }

    /*
     * Phase 2 - Involved ranks now redistribute any shared chunks to new
     * owners as necessary.
     */

    if (all_ranks_involved || (mpi_rank == 0)) {
        H5D_chunk_redistribute_info_t *chunk_entry;
        hsize_t                        curr_chunk_idx;
        size_t                         set_begin_index;
        int                            num_writers;
        int                            new_chunk_owner;

        /* Clear the mapping from rank value -> number of assigned chunks */
        HDmemset(num_chunks_assigned_map, 0, (size_t)mpi_size * sizeof(*num_chunks_assigned_map));

        /* Sort collective chunk list according to chunk index */
        HDqsort(coll_chunk_list, coll_chunk_list_num_entries, sizeof(H5D_chunk_redistribute_info_t),
                H5D__cmp_chunk_redistribute_info);

        /*
         * Process all chunks in the collective chunk list.
         * Note that the loop counter is incremented by both
         * the outer loop (while processing each entry in
         * the collective chunk list) and the inner loop
         * (while processing duplicate entries for shared
         * chunks).
         */
        chunk_entry = &((H5D_chunk_redistribute_info_t *)coll_chunk_list)[0];
        for (i = 0; i < coll_chunk_list_num_entries;) {
            /* Set chunk's initial new owner to its original owner */
            new_chunk_owner = chunk_entry->orig_owner;

            /*
             * Set the current chunk index so we know when we've processed
             * all duplicate entries for a particular shared chunk
             */
            curr_chunk_idx = chunk_entry->chunk_idx;

            /* Reset the initial number of writers to this chunk */
            num_writers = 0;

            /* Set index for the beginning of this section of duplicate chunk entries */
            set_begin_index = i;

            /*
             * Process each chunk entry in the set for the current
             * (possibly shared) chunk and increment the loop counter
             * while doing so.
             */
            do {
                /*
                 * The new owner of the chunk is determined by the rank
                 * writing to the chunk which currently has the least amount
                 * of chunks assigned to it
                 */
                if (num_chunks_assigned_map[chunk_entry->orig_owner] <
                    num_chunks_assigned_map[new_chunk_owner])
                    new_chunk_owner = chunk_entry->orig_owner;

                /* Update the number of writers to this particular chunk */
                num_writers++;

                chunk_entry++;
            } while (++i < coll_chunk_list_num_entries && chunk_entry->chunk_idx == curr_chunk_idx);

            /* We should never have more writers to a chunk than the number of MPI ranks */
            HDassert(num_writers <= mpi_size);

            /* Set all processed chunk entries' "new_owner" and "num_writers" fields */
            for (; set_begin_index < i; set_begin_index++) {
                H5D_chunk_redistribute_info_t *entry;

                entry = &((H5D_chunk_redistribute_info_t *)coll_chunk_list)[set_begin_index];

                entry->new_owner   = new_chunk_owner;
                entry->num_writers = num_writers;
            }

            /* Update the number of chunks assigned to the MPI rank that now owns this chunk */
            num_chunks_assigned_map[new_chunk_owner]++;
        }

        /*
         * Re-sort the collective chunk list in order of original chunk owner
         * so that each rank's section of contributed chunks is contiguous in
         * the collective chunk list.
         *
         * NOTE: this re-sort is frail in that it needs to sort the collective
         *       chunk list so that each rank's section of contributed chunks
         *       is in the exact order it was contributed in, or things will
         *       be scrambled when each rank's local chunk list is updated.
         *       Therefore, the sorting algorithm here is tied to the one
         *       used during the I/O setup operation. Specifically, chunks
         *       are first sorted by ascending order of offset in the file and
         *       then by chunk index. In the future, a better redistribution
         *       algorithm may be devised that doesn't rely on frail sorting,
         *       but the current implementation is a quick and naive approach.
         */
        HDqsort(coll_chunk_list, coll_chunk_list_num_entries, sizeof(H5D_chunk_redistribute_info_t),
                H5D__cmp_chunk_redistribute_info_orig_owner);
    }

    if (all_ranks_involved) {
        /*
         * If redistribution occurred on all ranks, search for the section
         * in the collective chunk list corresponding to this rank's locally
         * selected chunks and update the local list after redistribution.
         */
        for (i = 0; i < coll_chunk_list_num_entries; i++)
            if (mpi_rank == ((H5D_chunk_redistribute_info_t *)coll_chunk_list)[i].orig_owner)
                break;

        for (size_t j = 0; j < (size_t)num_chunks_int; j++) {
            H5D_chunk_redistribute_info_t *coll_entry;

            coll_entry = &((H5D_chunk_redistribute_info_t *)coll_chunk_list)[i++];

            chunk_list[j].new_owner   = coll_entry->new_owner;
            chunk_list[j].num_writers = coll_entry->num_writers;
        }
    }
    else {
        /*
         * If redistribution occurred only on rank 0, scatter the segments
         * of the collective chunk list back to each rank so that their
         * local chunk lists get updated
         */
        if (MPI_SUCCESS !=
            (mpi_code = MPI_Scatterv(coll_chunk_list, counts_ptr, displacements_ptr, packed_type, chunk_list,
                                     num_chunks_int, struct_type, 0, io_info->comm)))
            HMPI_GOTO_ERROR(FAIL, "unable to scatter shared chunks info buffer", mpi_code)
    }

#ifdef H5Dmpio_DEBUG
    H5D__mpio_dump_collective_filtered_chunk_list(chunk_list, num_chunks_assigned_map[mpi_rank], mpi_rank);
#endif

done:
    H5MM_free(coll_chunk_list);

    if (struct_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&struct_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }
    if (packed_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&packed_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }

    H5MM_free(counts_disps_array);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_redistribute_shared_chunks_int() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_share_chunk_modification_data
 *
 * Purpose:     When performing a parallel write on a chunked dataset with
 *              filters applied, we must first ensure that any particular
 *              chunk is only written to by a single MPI rank in order to
 *              avoid potential data races on the chunk. Once dataset
 *              chunks have been redistributed in a suitable manner, each
 *              MPI rank must send its chunk data to other ranks for each
 *              chunk it no longer owns.
 *
 *              The current implementation here follows the Nonblocking
 *              Consensus algorithm described in:
 *              http://unixer.de/publications/img/hoefler-dsde-protocols.pdf
 *
 *              First, each MPI rank scans through its list of selected
 *              chunks and does the following for each chunk:
 *
 *               * If a chunk in the MPI rank's chunk list is still owned
 *                 by that rank, the rank checks how many messages are
 *                 incoming for that chunk and adds that to its running
 *                 total. Then, the rank updates its local chunk list so
 *                 that any previous chunk entries for chunks that are no
 *                 longer owned by the rank get overwritten by chunk
 *                 entries for chunks the rank still owns. Since the data
 *                 for the chunks no longer owned will have already been
 *                 sent, those chunks can effectively be discarded.
 *               * If a chunk in the MPI rank's chunk list is no longer
 *                 owned by that rank, the rank sends the data it wishes to
 *                 update the chunk with to the MPI rank that now has
 *                 ownership of that chunk. To do this, it encodes the
 *                 chunk's index, its selection in the chunk and its
 *                 modification data into a buffer and then posts a
 *                 non-blocking MPI_Issend to the owning rank.
 *
 *              Once this step is complete, all MPI ranks allocate arrays
 *              to hold chunk message receive buffers and MPI request
 *              objects for each non-blocking receive they will post for
 *              incoming chunk modification messages. Then, all MPI ranks
 *              enter a loop that alternates between non-blocking
 *              MPI_Iprobe calls to probe for incoming messages and
 *              MPI_Testall calls to see if all send requests have
 *              completed. As chunk modification messages arrive,
 *              non-blocking MPI_Irecv calls will be posted for each
 *              message.
 *
 *              Once all send requests have completed, an MPI_Ibarrier is
 *              posted and the loop then alternates between MPI_Iprobe
 *              calls and MPI_Test calls to check if all ranks have reached
 *              the non-blocking barrier. Once all ranks have reached the
 *              barrier, processing can move on to updating the selected
 *              chunks that are owned in the operation.
 *
 *              Any chunk messages that were received from other ranks
 *              will be returned through the `chunk_msg_bufs` array and
 *              `chunk_msg_bufs_len` will be set appropriately.
 *
 *              NOTE: The use of non-blocking sends and receives of chunk
 *                    data here may contribute to large amounts of memory
 *                    usage/MPI request overhead if the number of shared
 *                    chunks is high. If this becomes a problem, it may be
 *                    useful to split the message receiving loop away so
 *                    that chunk modification messages can be received and
 *                    processed immediately (MPI_Recv) using a single chunk
 *                    message buffer. However, it's possible this may
 *                    degrade performance since the chunk message sends
 *                    are synchronous (MPI_Issend) in the Nonblocking
 *                    Consensus algorithm.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_share_chunk_modification_data(H5D_filtered_collective_io_info_t *chunk_list,
                                        size_t *chunk_list_num_entries, H5D_io_info_t *io_info,
                                        H5D_dset_io_info_t *dset_info, int mpi_rank,
                                        int H5_ATTR_NDEBUG_UNUSED           mpi_size,
                                        H5D_filtered_collective_io_info_t **chunk_hash_table,
                                        unsigned char ***chunk_msg_bufs, int *chunk_msg_bufs_len)
{
#if H5_CHECK_MPI_VERSION(3, 0)
    H5D_filtered_collective_io_info_t *chunk_table       = NULL;
    H5S_sel_iter_t                    *mem_iter          = NULL;
    unsigned char                    **msg_send_bufs     = NULL;
    unsigned char                    **msg_recv_bufs     = NULL;
    MPI_Request                       *send_requests     = NULL;
    MPI_Request                       *recv_requests     = NULL;
    MPI_Request                        ibarrier          = MPI_REQUEST_NULL;
    hbool_t                            mem_iter_init     = FALSE;
    hbool_t                            ibarrier_posted   = FALSE;
    size_t                             send_bufs_nalloc  = 0;
    size_t                             num_send_requests = 0;
    size_t                             num_recv_requests = 0;
    size_t                             num_msgs_incoming = 0;
    size_t                             last_assigned_idx;
    size_t                             i;
    int                                mpi_code;
    herr_t                             ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list_num_entries);
    HDassert(chunk_list || 0 == *chunk_list_num_entries);
    HDassert(io_info);
    HDassert(dset_info);
    HDassert(mpi_size > 1);
    HDassert(chunk_msg_bufs);
    HDassert(chunk_msg_bufs_len);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Share chunk modification data");
#endif

    /* Set to latest format for encoding dataspace */
    H5CX_set_libver_bounds(NULL);

    if (*chunk_list_num_entries) {
        /* Allocate a selection iterator for iterating over chunk dataspaces */
        if (NULL == (mem_iter = H5FL_MALLOC(H5S_sel_iter_t)))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate dataspace selection iterator")

        /*
         * Allocate send buffer and MPI_Request arrays for non-blocking
         * sends of outgoing chunk messages
         */
        send_bufs_nalloc = H5D_CHUNK_NUM_SEND_MSGS_INIT;
        if (NULL == (msg_send_bufs = H5MM_malloc(send_bufs_nalloc * sizeof(*msg_send_bufs))))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                        "couldn't allocate chunk modification message buffer array")

        if (NULL == (send_requests = H5MM_malloc(send_bufs_nalloc * sizeof(*send_requests))))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate send requests array")
    }

    /*
     * For each chunk this rank owns, add to the total number of
     * incoming MPI messages, then update the local chunk list to
     * overwrite any previous chunks no longer owned by this rank.
     * Since the data for those chunks will have already been sent,
     * this rank should no longer be interested in them and they
     * can effectively be discarded. This bookkeeping also makes
     * the code for the collective file space re-allocation and
     * chunk re-insertion operations a bit simpler.
     *
     * For each chunk this rank doesn't own, use non-blocking
     * synchronous sends to send the data this rank is writing to
     * the rank that does own the chunk.
     */
    for (i = 0, last_assigned_idx = 0; i < *chunk_list_num_entries; i++) {
        H5D_filtered_collective_io_info_t *chunk_entry = &chunk_list[i];

        if (mpi_rank == chunk_entry->new_owner) {
            num_msgs_incoming += (size_t)(chunk_entry->num_writers - 1);

            /*
             * Overwrite chunk entries this rank doesn't own with entries that it
             * does own, since it has sent the necessary data and is no longer
             * interested in the chunks it doesn't own.
             */
            chunk_list[last_assigned_idx] = chunk_list[i];

            /*
             * Since, at large scale, a chunk's index value may be larger than
             * the maximum value that can be stored in an int, we cannot rely
             * on using a chunk's index value as the tag for the MPI messages
             * sent/received for a chunk. Therefore, add this chunk to a hash
             * table with the chunk's index as a key so that we can quickly find
             * the chunk when processing chunk messages that were received. The
             * message itself will contain the chunk's index so we can update
             * the correct chunk with the received data.
             */
            HASH_ADD(hh, chunk_table, index_info.chunk_idx, sizeof(hsize_t), &chunk_list[last_assigned_idx]);

            last_assigned_idx++;
        }
        else {
            H5D_piece_info_t *chunk_info = chunk_entry->chunk_info;
            unsigned char    *mod_data_p = NULL;
            hsize_t           iter_nelmts;
            size_t            mod_data_size = 0;
            size_t            space_size    = 0;

            /* Add the size of the chunk index to the encoded size */
            mod_data_size += sizeof(hsize_t);

            /* Determine size of serialized chunk file dataspace */
            if (H5S_encode(chunk_info->fspace, &mod_data_p, &space_size) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "unable to get encoded dataspace size")
            mod_data_size += space_size;

            /* Determine size of data being written */
            iter_nelmts = H5S_GET_SELECT_NPOINTS(chunk_info->mspace);
            H5_CHECK_OVERFLOW(iter_nelmts, hsize_t, size_t);

            mod_data_size += (size_t)iter_nelmts * dset_info->type_info.src_type_size;

            if (NULL == (msg_send_bufs[num_send_requests] = H5MM_malloc(mod_data_size)))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk modification message buffer")

            mod_data_p = msg_send_bufs[num_send_requests];

            /* Store the chunk's index into the buffer */
            HDmemcpy(mod_data_p, &chunk_entry->index_info.chunk_idx, sizeof(hsize_t));
            mod_data_p += sizeof(hsize_t);

            /* Serialize the chunk's file dataspace into the buffer */
            if (H5S_encode(chunk_info->fspace, &mod_data_p, &mod_data_size) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTENCODE, FAIL, "unable to encode dataspace")

            /* Initialize iterator for memory selection */
            if (H5S_select_iter_init(mem_iter, chunk_info->mspace, dset_info->type_info.src_type_size,
                                     H5S_SEL_ITER_SHARE_WITH_DATASPACE) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL,
                            "unable to initialize memory selection information")
            mem_iter_init = TRUE;

            /* Collect the modification data into the buffer */
            if (0 == H5D__gather_mem(dset_info->buf.cvp, mem_iter, (size_t)iter_nelmts, mod_data_p))
                HGOTO_ERROR(H5E_IO, H5E_CANTGATHER, FAIL, "couldn't gather from write buffer")

            /*
             * Ensure that the size of the chunk data being sent can be
             * safely cast to an int for MPI. Note that this should
             * generally be OK for now (unless a rank is sending a
             * whole 32-bit-sized chunk of data + its encoded selection),
             * but if we allow larger than 32-bit-sized chunks in the
             * future, this may become a problem and derived datatypes
             * will need to be used.
             */
            H5_CHECK_OVERFLOW(mod_data_size, size_t, int)

            /* Send modification data to new owner */
            if (MPI_SUCCESS !=
                (mpi_code = MPI_Issend(msg_send_bufs[num_send_requests], (int)mod_data_size, MPI_BYTE,
                                       chunk_entry->new_owner, H5D_CHUNK_MOD_DATA_TAG, io_info->comm,
                                       &send_requests[num_send_requests])))
                HMPI_GOTO_ERROR(FAIL, "MPI_Issend failed", mpi_code)

            num_send_requests++;

            /* Resize send buffer and send request arrays if necessary */
            if (num_send_requests == send_bufs_nalloc) {
                void *tmp_alloc;

                send_bufs_nalloc = (size_t)((double)send_bufs_nalloc * 1.5);

                if (NULL ==
                    (tmp_alloc = H5MM_realloc(msg_send_bufs, send_bufs_nalloc * sizeof(*msg_send_bufs))))
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                                "couldn't resize chunk modification message buffer array")
                msg_send_bufs = tmp_alloc;

                if (NULL ==
                    (tmp_alloc = H5MM_realloc(send_requests, send_bufs_nalloc * sizeof(*send_requests))))
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't resize send requests array")
                send_requests = tmp_alloc;
            }

            if (H5S_SELECT_ITER_RELEASE(mem_iter) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "couldn't release memory selection iterator")
            mem_iter_init = FALSE;
        }
    }

    /* Check if the number of send or receive requests will overflow an int (MPI requirement) */
    if (num_send_requests > INT_MAX || num_msgs_incoming > INT_MAX)
        HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL,
                    "too many shared chunks in parallel filtered write operation")

    H5_CHECK_OVERFLOW(num_send_requests, size_t, int)
    H5_CHECK_OVERFLOW(num_msgs_incoming, size_t, int)

    /*
     * Allocate receive buffer and MPI_Request arrays for non-blocking
     * receives of incoming chunk messages
     */
    if (num_msgs_incoming) {
        if (NULL == (msg_recv_bufs = H5MM_malloc(num_msgs_incoming * sizeof(*msg_recv_bufs))))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                        "couldn't allocate chunk modification message buffer array")

        if (NULL == (recv_requests = H5MM_malloc(num_msgs_incoming * sizeof(*recv_requests))))
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate receive requests array")
    }

    /* Process any incoming messages until everyone is done */
    do {
        MPI_Status status;
        int        msg_flag;

        /* Probe for an incoming message from any rank */
        if (MPI_SUCCESS != (mpi_code = MPI_Iprobe(MPI_ANY_SOURCE, H5D_CHUNK_MOD_DATA_TAG, io_info->comm,
                                                  &msg_flag, &status)))
            HMPI_GOTO_ERROR(FAIL, "MPI_Iprobe failed", mpi_code)

        /*
         * If a message was found, allocate a buffer for the message and
         * post a non-blocking receive to receive it
         */
        if (msg_flag) {
#if H5_CHECK_MPI_VERSION(3, 0)
            MPI_Count msg_size = 0;

            if (MPI_SUCCESS != (mpi_code = MPI_Get_elements_x(&status, MPI_BYTE, &msg_size)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Get_elements_x failed", mpi_code)

            H5_CHECK_OVERFLOW(msg_size, MPI_Count, int)
#else
            int msg_size = 0;

            if (MPI_SUCCESS != (mpi_code = MPI_Get_elements(&status, MPI_BYTE, &msg_size)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Get_elements failed", mpi_code)
#endif

            if (msg_size <= 0)
                HGOTO_ERROR(H5E_DATASET, H5E_BADVALUE, FAIL, "invalid chunk modification message size")

            HDassert((num_recv_requests + 1) <= num_msgs_incoming);
            if (NULL ==
                (msg_recv_bufs[num_recv_requests] = H5MM_malloc((size_t)msg_size * sizeof(unsigned char))))
                HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL,
                            "couldn't allocate chunk modification message receive buffer")

            if (MPI_SUCCESS != (mpi_code = MPI_Irecv(msg_recv_bufs[num_recv_requests], (int)msg_size,
                                                     MPI_BYTE, status.MPI_SOURCE, H5D_CHUNK_MOD_DATA_TAG,
                                                     io_info->comm, &recv_requests[num_recv_requests])))
                HMPI_GOTO_ERROR(FAIL, "MPI_Irecv failed", mpi_code)

            num_recv_requests++;
        }

        if (ibarrier_posted) {
            int ibarrier_completed;

            if (MPI_SUCCESS != (mpi_code = MPI_Test(&ibarrier, &ibarrier_completed, MPI_STATUS_IGNORE)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Test failed", mpi_code)

            if (ibarrier_completed)
                break;
        }
        else {
            int all_sends_completed;

            /* Determine if all send requests have completed
             *
             * gcc 11 complains about passing MPI_STATUSES_IGNORE as an MPI_Status
             * array. See the discussion here:
             *
             * https://github.com/pmodels/mpich/issues/5687
             */
            H5_GCC_DIAG_OFF("stringop-overflow")
            if (MPI_SUCCESS != (mpi_code = MPI_Testall((int)num_send_requests, send_requests,
                                                       &all_sends_completed, MPI_STATUSES_IGNORE)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Testall failed", mpi_code)
            H5_GCC_DIAG_ON("stringop-overflow")

            if (all_sends_completed) {
                /* Post non-blocking barrier */
                if (MPI_SUCCESS != (mpi_code = MPI_Ibarrier(io_info->comm, &ibarrier)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Ibarrier failed", mpi_code)
                ibarrier_posted = TRUE;

                /*
                 * Now that all send requests have completed, free up the
                 * send buffers used in the non-blocking operations
                 */
                if (msg_send_bufs) {
                    for (i = 0; i < num_send_requests; i++) {
                        if (msg_send_bufs[i])
                            H5MM_free(msg_send_bufs[i]);
                    }

                    msg_send_bufs = H5MM_xfree(msg_send_bufs);
                }
            }
        }
    } while (1);

    /*
     * Ensure all receive requests have completed before moving on.
     * For linked-chunk I/O, more overlap with computation could
     * theoretically be achieved by returning the receive requests
     * array and postponing this wait until during chunk updating
     * when the data is really needed. However, multi-chunk I/O
     * only updates a chunk at a time and the messages may not come
     * in the order that chunks are processed. So, the safest way to
     * support both I/O modes is to simply make sure all messages
     * are available.
     *
     * gcc 11 complains about passing MPI_STATUSES_IGNORE as an MPI_Status
     * array. See the discussion here:
     *
     * https://github.com/pmodels/mpich/issues/5687
     */
    H5_GCC_DIAG_OFF("stringop-overflow")
    if (MPI_SUCCESS != (mpi_code = MPI_Waitall((int)num_recv_requests, recv_requests, MPI_STATUSES_IGNORE)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Waitall failed", mpi_code)
    H5_GCC_DIAG_ON("stringop-overflow")

    /* Set the new number of locally-selected chunks */
    *chunk_list_num_entries = last_assigned_idx;

    /* Return chunk message buffers if any were received */
    *chunk_hash_table   = chunk_table;
    *chunk_msg_bufs     = msg_recv_bufs;
    *chunk_msg_bufs_len = (int)num_recv_requests;

done:
    if (ret_value < 0) {
        /* If this rank failed, make sure to participate in collective barrier */
        if (!ibarrier_posted) {
            if (MPI_SUCCESS != (mpi_code = MPI_Ibarrier(io_info->comm, &ibarrier)))
                HMPI_GOTO_ERROR(FAIL, "MPI_Ibarrier failed", mpi_code)
        }

        if (num_send_requests) {
            for (i = 0; i < num_send_requests; i++) {
                MPI_Cancel(&send_requests[i]);
            }
        }

        if (recv_requests) {
            for (i = 0; i < num_recv_requests; i++) {
                MPI_Cancel(&recv_requests[i]);
            }
        }

        if (msg_recv_bufs) {
            for (i = 0; i < num_recv_requests; i++) {
                H5MM_free(msg_recv_bufs[i]);
            }

            H5MM_free(msg_recv_bufs);
        }

        HASH_CLEAR(hh, chunk_table);
    }

    if (recv_requests)
        H5MM_free(recv_requests);
    if (send_requests)
        H5MM_free(send_requests);

    if (msg_send_bufs) {
        for (i = 0; i < num_send_requests; i++) {
            if (msg_send_bufs[i])
                H5MM_free(msg_send_bufs[i]);
        }

        H5MM_free(msg_send_bufs);
    }

    if (mem_iter) {
        if (mem_iter_init && H5S_SELECT_ITER_RELEASE(mem_iter) < 0)
            HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "couldn't release dataspace selection iterator")
        mem_iter = H5FL_FREE(H5S_sel_iter_t, mem_iter);
    }

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
#else
    FUNC_ENTER_PACKAGE
    HERROR(
        H5E_DATASET, H5E_WRITEERROR,
        "unable to send chunk modification data between MPI ranks - MPI version < 3 (MPI_Ibarrier missing)")
    FUNC_LEAVE_NOAPI(FAIL)
#endif
} /* end H5D__mpio_share_chunk_modification_data() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_chunk_common_io
 *
 * Purpose:     This routine performs the common part of collective I/O
 *              when reading or writing filtered chunks collectively.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_chunk_common_io(H5D_filtered_collective_io_info_t *chunk_list,
                                              size_t chunk_list_num_entries, const H5D_io_info_t *io_info,
                                              int mpi_size)
{
    H5D_io_info_t coll_io_info;
    MPI_Datatype  file_type            = MPI_DATATYPE_NULL;
    MPI_Datatype  mem_type             = MPI_DATATYPE_NULL;
    hbool_t       mem_type_is_derived  = FALSE;
    hbool_t       file_type_is_derived = FALSE;
    hsize_t       mpi_buf_count;
    haddr_t       base_read_offset = HADDR_UNDEF;
    size_t        num_chunks;
    size_t        i;
    char          fake_buf; /* Used as a fake buffer for ranks with no chunks, thus a NULL buf pointer */
    int           mpi_code;
    herr_t        ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == chunk_list_num_entries);
    HDassert(io_info);

    /* Initialize temporary I/O info */
    coll_io_info = *io_info;

    /*
     * Construct MPI derived datatype for collective I/O on chunks
     */
    if (H5D__mpio_collective_filtered_io_type(chunk_list, chunk_list_num_entries, io_info->op_type, &mem_type,
                                              &mem_type_is_derived, &file_type, &file_type_is_derived) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_BADTYPE, FAIL, "couldn't create MPI I/O type for chunk I/O")

    /*
     * For reads, determine how many chunks are actually being read.
     * Note that if this is a read during a write operation
     * (read chunk -> unfilter -> modify -> write back), some
     * chunks may not need to be read if they're being fully
     * overwritten during a write operation.
     */
    if (io_info->op_type == H5D_IO_OP_READ) {
        for (i = 0, num_chunks = 0; i < chunk_list_num_entries; i++) {
            HDassert(chunk_list[i].buf);

            if (chunk_list[i].need_read) {
                if (!H5F_addr_defined(base_read_offset))
                    base_read_offset = chunk_list[i].chunk_current.offset;

                num_chunks++;
            }
        }
    }
    else
        num_chunks = chunk_list_num_entries;

    /*
     * If this rank doesn't have a selection, it can
     * skip I/O if the MPI communicator size is 1.
     *
     * Otherwise, this rank has to participate in
     * collective I/O, but probably has a NULL buf
     * pointer, so override to a fake buffer since our
     * write/read function expects one.
     */
    if (num_chunks == 0) {
        if (mpi_size == 1)
            HGOTO_DONE(SUCCEED)
        else {
            if (io_info->op_type == H5D_IO_OP_WRITE)
                coll_io_info.base_maddr.cvp = &fake_buf;
            else
                coll_io_info.base_maddr.vp = &fake_buf;
        }
    }

    /*
     * Setup for I/O operation
     */

    mpi_buf_count = (num_chunks) ? 1 : 0;

    if (num_chunks) {
        /*
         * Setup the base storage address for this operation
         * to be the first chunk's file address
         */
        if (io_info->op_type == H5D_IO_OP_WRITE)
            coll_io_info.store_faddr = chunk_list[0].chunk_new.offset;
        else
            coll_io_info.store_faddr = base_read_offset;
    }
    else
        coll_io_info.store_faddr = 0;

    /* Perform I/O */
    if (H5D__final_collective_io(&coll_io_info, mpi_buf_count, file_type, mem_type) < 0)
        HGOTO_ERROR(H5E_IO, H5E_READERROR, FAIL, "couldn't finish MPI I/O")

done:
    /* Free the MPI buf and file types, if they were derived */
    if (mem_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&mem_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    if (file_type_is_derived && MPI_SUCCESS != (mpi_code = MPI_Type_free(&file_type)))
        HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_collective_filtered_chunk_common_io() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_chunk_read
 *
 * Purpose:     This routine coordinates a collective read across all ranks
 *              of the chunks they have selected. Each rank will then go
 *              and
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_chunk_read(H5D_filtered_collective_io_info_t *chunk_list,
                                         size_t chunk_list_num_entries, const H5D_io_info_t *io_info,
                                         const H5D_dset_io_info_t *di, int mpi_rank, int mpi_size)
{
    H5D_fill_buf_info_t fb_info;
    H5D_piece_info_t   *chunk_info = NULL;
    H5D_io_info_t       coll_io_info;
    H5Z_EDC_t           err_detect; /* Error detection info */
    H5Z_cb_t            filter_cb;  /* I/O filter callback function */
    hsize_t             file_chunk_size = 0;
    hsize_t             iter_nelmts; /* Number of points to iterate over for the chunk IO operation */
    hbool_t             should_fill  = FALSE;
    hbool_t             fb_info_init = FALSE;
    hbool_t             index_empty  = FALSE;
    size_t              i;
    H5S_t              *fill_space    = NULL;
    void               *base_read_buf = NULL;
    herr_t              ret_value     = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == chunk_list_num_entries);
    HDassert(io_info);
    HDassert(di);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Filtered collective chunk read");
#else
    (void)mpi_rank;
#endif

    /* Initialize temporary I/O info */
    coll_io_info               = *io_info;
    coll_io_info.base_maddr.vp = NULL;

    if (chunk_list_num_entries) {
        /* Retrieve filter settings from API context */
        if (H5CX_get_err_detect(&err_detect) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get error detection info")
        if (H5CX_get_filter_cb(&filter_cb) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get I/O filter callback function")

        /* Set size of full chunks in dataset */
        file_chunk_size = di->dset->shared->layout.u.chunk.size;

        /* Determine if fill values should be "read" for unallocated chunks */
        should_fill = (di->dset->shared->dcpl_cache.fill.fill_time == H5D_FILL_TIME_ALLOC) ||
                      ((di->dset->shared->dcpl_cache.fill.fill_time == H5D_FILL_TIME_IFSET) &&
                       di->dset->shared->dcpl_cache.fill.fill_defined);
    }

    /*
     * Allocate memory buffers for all chunks being read. Chunk data buffers are of
     * the largest size between the chunk's current filtered size and the chunk's true
     * size, as calculated by the number of elements in the chunk's file space extent
     * multiplied by the datatype size. This tries to ensure that:
     *
     *  * If we're reading the chunk and the filter normally reduces the chunk size,
     *    the unfiltering operation won't need to grow the buffer.
     *  * If we're reading the chunk and the filter normally grows the chunk size,
     *    we make sure to read into a buffer of size equal to the filtered chunk's
     *    size; reading into a (smaller) buffer of size equal to the unfiltered
     *    chunk size would of course be bad.
     */
    for (i = 0; i < chunk_list_num_entries; i++) {
        HDassert(chunk_list[i].need_read);

        chunk_list[i].chunk_buf_size = MAX(chunk_list[i].chunk_current.length, file_chunk_size);

        if (NULL == (chunk_list[i].buf = H5MM_malloc(chunk_list[i].chunk_buf_size))) {
            /* Push an error, but participate in collective read */
            HDONE_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate chunk data buffer")
            break;
        }

        /*
         * Check if chunk is currently allocated. If not, don't try to
         * read it from the file. Instead, just fill the chunk buffer
         * with the fill value if necessary.
         */
        if (H5F_addr_defined(chunk_list[i].chunk_current.offset)) {
            /* Set first read buffer */
            if (!base_read_buf)
                base_read_buf = chunk_list[i].buf;

            /* Set chunk's new length for eventual filter pipeline calls */
            if (chunk_list[i].skip_filter_pline)
                chunk_list[i].chunk_new.length = file_chunk_size;
            else
                chunk_list[i].chunk_new.length = chunk_list[i].chunk_current.length;
        }
        else {
            chunk_list[i].need_read = FALSE;

            /* Set chunk's new length for eventual filter pipeline calls */
            chunk_list[i].chunk_new.length = file_chunk_size;

            if (should_fill) {
                /* Initialize fill value buffer if not already initialized */
                if (!fb_info_init) {
                    hsize_t chunk_dims[H5S_MAX_RANK];

                    HDassert(di->dset->shared->ndims == di->dset->shared->layout.u.chunk.ndims - 1);
                    for (size_t j = 0; j < di->dset->shared->layout.u.chunk.ndims - 1; j++)
                        chunk_dims[j] = (hsize_t)di->dset->shared->layout.u.chunk.dim[j];

                    /* Get a dataspace for filling chunk memory buffers */
                    if (NULL == (fill_space = H5S_create_simple(di->dset->shared->layout.u.chunk.ndims - 1,
                                                                chunk_dims, NULL)))
                        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to create chunk fill dataspace")

                    /* Initialize fill value buffer */
                    if (H5D__fill_init(
                            &fb_info, NULL, (H5MM_allocate_t)H5D__chunk_mem_alloc,
                            (void *)&di->dset->shared->dcpl_cache.pline, (H5MM_free_t)H5D__chunk_mem_free,
                            (void *)&di->dset->shared->dcpl_cache.pline, &di->dset->shared->dcpl_cache.fill,
                            di->dset->shared->type, di->dset->shared->type_id, 0, file_chunk_size) < 0)
                        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "can't initialize fill value buffer")

                    fb_info_init = TRUE;
                }

                /* Write fill value to memory buffer */
                HDassert(fb_info.fill_buf);
                if (H5D__fill(fb_info.fill_buf, di->dset->shared->type, chunk_list[i].buf,
                              di->type_info.mem_type, fill_space) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "couldn't fill chunk buffer with fill value")
            }
        }
    }

    /*
     * If dataset is incrementally allocated and hasn't been written to
     * yet, the chunk index should be empty. In this case, a collective
     * read of chunks is essentially a no-op, so avoid it here.
     */
    index_empty = FALSE;
    if (di->dset->shared->dcpl_cache.fill.alloc_time == H5D_ALLOC_TIME_INCR)
        if (H5D__chunk_index_empty(di->dset, &index_empty) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "couldn't determine if chunk index is empty")

    if (!index_empty) {
        /*
         * Override the read buffer to point to the address of
         * the first chunk data buffer being read into
         */
        if (base_read_buf)
            coll_io_info.base_maddr.vp = base_read_buf;

        /* Perform collective chunk read */
        if (H5D__mpio_collective_filtered_chunk_common_io(chunk_list, chunk_list_num_entries, &coll_io_info,
                                                          mpi_size) < 0)
            HGOTO_ERROR(H5E_IO, H5E_READERROR, FAIL, "couldn't finish collective filtered chunk read")
    }

    /*
     * Iterate through all the read chunks, unfiltering them and scattering their
     * data out to the application's read buffer.
     */
    for (i = 0; i < chunk_list_num_entries; i++) {
        chunk_info = chunk_list[i].chunk_info;

        /* Unfilter the chunk, unless we didn't read it from the file */
        if (chunk_list[i].need_read && !chunk_list[i].skip_filter_pline) {
            if (H5Z_pipeline(&di->dset->shared->dcpl_cache.pline, H5Z_FLAG_REVERSE,
                             &(chunk_list[i].index_info.filter_mask), err_detect, filter_cb,
                             (size_t *)&chunk_list[i].chunk_new.length, &chunk_list[i].chunk_buf_size,
                             &chunk_list[i].buf) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTFILTER, FAIL, "couldn't unfilter chunk for modifying")
        }

        /* Scatter the chunk data to the read buffer */
        iter_nelmts = H5S_GET_SELECT_NPOINTS(chunk_info->fspace);

        if (H5D_select_io_mem(di->buf.vp, chunk_info->mspace, chunk_list[i].buf, chunk_info->fspace,
                              di->type_info.src_type_size, (size_t)iter_nelmts) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_READERROR, FAIL, "couldn't copy chunk data to read buffer")
    }

done:
    /* Free all resources used by entries in the chunk list */
    for (i = 0; i < chunk_list_num_entries; i++) {
        if (chunk_list[i].buf) {
            H5MM_free(chunk_list[i].buf);
            chunk_list[i].buf = NULL;
        }
    }

    /* Release the fill buffer info, if it's been initialized */
    if (fb_info_init && H5D__fill_term(&fb_info) < 0)
        HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "Can't release fill buffer info")
    if (fill_space && (H5S_close(fill_space) < 0))
        HDONE_ERROR(H5E_DATASET, H5E_CLOSEERROR, FAIL, "can't close fill space")

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_collective_filtered_chunk_read() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_chunk_update
 *
 * Purpose:     When performing a parallel write on a chunked dataset with
 *              filters applied, all ranks must update their owned chunks
 *              with their own modification data and data from other ranks.
 *              This routine is responsible for coordinating that process.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_chunk_update(H5D_filtered_collective_io_info_t *chunk_list,
                                           size_t                             chunk_list_num_entries,
                                           H5D_filtered_collective_io_info_t *chunk_hash_table,
                                           unsigned char **chunk_msg_bufs, int chunk_msg_bufs_len,
                                           const H5D_io_info_t *io_info, const H5D_dset_io_info_t *di,
                                           int H5_ATTR_NDEBUG_UNUSED mpi_rank, int mpi_size)
{
    const H5D_type_info_t *type_info = NULL;
    H5D_fill_buf_info_t    fb_info;
    H5D_piece_info_t      *chunk_info = NULL;
    H5S_sel_iter_t        *sel_iter   = NULL; /* Dataspace selection iterator for H5D__scatter_mem */
    H5D_io_info_t          coll_io_info;
    H5Z_EDC_t              err_detect; /* Error detection info */
    H5Z_cb_t               filter_cb;  /* I/O filter callback function */
    hsize_t                file_chunk_size = 0;
    hsize_t                iter_nelmts; /* Number of points to iterate over for the chunk IO operation */
    hbool_t                should_fill   = FALSE;
    hbool_t                fb_info_init  = FALSE;
    hbool_t                sel_iter_init = FALSE;
    hbool_t                index_empty   = FALSE;
    size_t                 i;
    H5S_t                 *dataspace     = NULL;
    H5S_t                 *fill_space    = NULL;
    void                  *base_read_buf = NULL;
    herr_t                 ret_value     = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == chunk_list_num_entries);
    HDassert((chunk_msg_bufs && chunk_hash_table) || 0 == chunk_msg_bufs_len);
    HDassert(io_info);
    HDassert(di);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Filtered collective chunk update");
#endif

    /* Set convenience pointers */
    type_info = &(di->type_info);
    HDassert(type_info);

    if (chunk_list_num_entries) {
        /* Retrieve filter settings from API context */
        if (H5CX_get_err_detect(&err_detect) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get error detection info")
        if (H5CX_get_filter_cb(&filter_cb) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't get I/O filter callback function")

        /* Set size of full chunks in dataset */
        file_chunk_size = di->dset->shared->layout.u.chunk.size;

        /* Determine if fill values should be written to chunks */
        should_fill = (di->dset->shared->dcpl_cache.fill.fill_time == H5D_FILL_TIME_ALLOC) ||
                      ((di->dset->shared->dcpl_cache.fill.fill_time == H5D_FILL_TIME_IFSET) &&
                       di->dset->shared->dcpl_cache.fill.fill_defined);
    }

    /*
     * Allocate memory buffers for all owned chunks. Chunk data buffers are of the
     * largest size between the chunk's current filtered size and the chunk's true
     * size, as calculated by the number of elements in the chunk's file space extent
     * multiplied by the datatype size. This tries to ensure that:
     *
     *  * If we're fully overwriting the chunk and the filter normally reduces the
     *    chunk size, we simply have the exact buffer size required to hold the
     *    unfiltered chunk data.
     *  * If we're fully overwriting the chunk and the filter normally grows the
     *    chunk size (e.g., fletcher32 filter), the final filtering operation
     *    (hopefully) won't need to grow the buffer.
     *  * If we're reading the chunk and the filter normally reduces the chunk size,
     *    the unfiltering operation won't need to grow the buffer.
     *  * If we're reading the chunk and the filter normally grows the chunk size,
     *    we make sure to read into a buffer of size equal to the filtered chunk's
     *    size; reading into a (smaller) buffer of size equal to the unfiltered
     *    chunk size would of course be bad.
     */
    for (i = 0; i < chunk_list_num_entries; i++) {
        HDassert(mpi_rank == chunk_list[i].new_owner);

        chunk_list[i].chunk_buf_size = MAX(chunk_list[i].chunk_current.length, file_chunk_size);

        /*
         * If this chunk hasn't been allocated yet and we aren't writing
         * out fill values to it, make sure to 0-fill its memory buffer
         * so we don't use uninitialized memory.
         */
        if (!H5F_addr_defined(chunk_list[i].chunk_current.offset) && !should_fill)
            chunk_list[i].buf = H5MM_calloc(chunk_list[i].chunk_buf_size);
        else
            chunk_list[i].buf = H5MM_malloc(chunk_list[i].chunk_buf_size);

        if (NULL == chunk_list[i].buf) {
            /* Push an error, but participate in collective read */
            HDONE_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate chunk data buffer")
            break;
        }

        /* Set chunk's new length for eventual filter pipeline calls */
        if (chunk_list[i].need_read) {
            /*
             * Check if chunk is currently allocated. If not, don't try to
             * read it from the file. Instead, just fill the chunk buffer
             * with the fill value if fill values are to be written.
             */
            if (H5F_addr_defined(chunk_list[i].chunk_current.offset)) {
                /* Set first read buffer */
                if (!base_read_buf)
                    base_read_buf = chunk_list[i].buf;

                /* Set chunk's new length for eventual filter pipeline calls */
                if (chunk_list[i].skip_filter_pline)
                    chunk_list[i].chunk_new.length = file_chunk_size;
                else
                    chunk_list[i].chunk_new.length = chunk_list[i].chunk_current.length;
            }
            else {
                chunk_list[i].need_read = FALSE;

                /* Set chunk's new length for eventual filter pipeline calls */
                chunk_list[i].chunk_new.length = file_chunk_size;

                if (should_fill) {
                    /* Initialize fill value buffer if not already initialized */
                    if (!fb_info_init) {
                        hsize_t chunk_dims[H5S_MAX_RANK];

                        HDassert(di->dset->shared->ndims == di->dset->shared->layout.u.chunk.ndims - 1);
                        for (size_t j = 0; j < di->dset->shared->layout.u.chunk.ndims - 1; j++)
                            chunk_dims[j] = (hsize_t)di->dset->shared->layout.u.chunk.dim[j];

                        /* Get a dataspace for filling chunk memory buffers */
                        if (NULL == (fill_space = H5S_create_simple(
                                         di->dset->shared->layout.u.chunk.ndims - 1, chunk_dims, NULL)))
                            HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL,
                                        "unable to create chunk fill dataspace")

                        /* Initialize fill value buffer */
                        if (H5D__fill_init(&fb_info, NULL, (H5MM_allocate_t)H5D__chunk_mem_alloc,
                                           (void *)&di->dset->shared->dcpl_cache.pline,
                                           (H5MM_free_t)H5D__chunk_mem_free,
                                           (void *)&di->dset->shared->dcpl_cache.pline,
                                           &di->dset->shared->dcpl_cache.fill, di->dset->shared->type,
                                           di->dset->shared->type_id, 0, file_chunk_size) < 0)
                            HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "can't initialize fill value buffer")

                        fb_info_init = TRUE;
                    }

                    /* Write fill value to memory buffer */
                    HDassert(fb_info.fill_buf);
                    if (H5D__fill(fb_info.fill_buf, di->dset->shared->type, chunk_list[i].buf,
                                  type_info->mem_type, fill_space) < 0)
                        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL,
                                    "couldn't fill chunk buffer with fill value")
                }
            }
        }
        else
            chunk_list[i].chunk_new.length = file_chunk_size;
    }

    /*
     * If dataset is incrementally allocated and hasn't been written to
     * yet, the chunk index should be empty. In this case, a collective
     * read of chunks is essentially a no-op, so avoid it here.
     */
    index_empty = FALSE;
    if (di->dset->shared->dcpl_cache.fill.alloc_time == H5D_ALLOC_TIME_INCR)
        if (H5D__chunk_index_empty(di->dset, &index_empty) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "couldn't determine if chunk index is empty")

    if (!index_empty) {
        /*
         * Setup for I/O operation
         */

        /* Initialize temporary I/O info */
        coll_io_info         = *io_info;
        coll_io_info.op_type = H5D_IO_OP_READ;

        /* Override the read buffer to point to the address of the first
         * chunk data buffer being read into
         */
        if (base_read_buf) {
            coll_io_info.base_maddr.vp = base_read_buf;
        }

        /* Read all chunks that need to be read from the file */
        if (H5D__mpio_collective_filtered_chunk_common_io(chunk_list, chunk_list_num_entries, &coll_io_info,
                                                          mpi_size) < 0)
            HGOTO_ERROR(H5E_IO, H5E_READERROR, FAIL, "couldn't finish collective filtered chunk read")
    }

    /*
     * Now that all owned chunks have been read, update the chunks
     * with modification data from the owning rank and other ranks.
     */

    /* Process all chunks with data from the owning rank first */
    for (i = 0; i < chunk_list_num_entries; i++) {
        HDassert(mpi_rank == chunk_list[i].new_owner);

        chunk_info = chunk_list[i].chunk_info;

        /*
         * If this chunk wasn't being fully overwritten, we read it from
         * the file, so we need to unfilter it
         */
        if (chunk_list[i].need_read && !chunk_list[i].skip_filter_pline) {
            if (H5Z_pipeline(&di->dset->shared->dcpl_cache.pline, H5Z_FLAG_REVERSE,
                             &(chunk_list[i].index_info.filter_mask), err_detect, filter_cb,
                             (size_t *)&chunk_list[i].chunk_new.length, &chunk_list[i].chunk_buf_size,
                             &chunk_list[i].buf) < 0)
                HGOTO_ERROR(H5E_DATASET, H5E_CANTFILTER, FAIL, "couldn't unfilter chunk for modifying")
        }

        iter_nelmts = H5S_GET_SELECT_NPOINTS(chunk_info->mspace);

        if (H5D_select_io_mem(chunk_list[i].buf, chunk_info->fspace, di->buf.cvp, chunk_info->mspace,
                              type_info->dst_type_size, (size_t)iter_nelmts) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "couldn't copy chunk data to write buffer")
    }

    /* Allocate iterator for memory selection */
    if (NULL == (sel_iter = H5FL_MALLOC(H5S_sel_iter_t)))
        HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "couldn't allocate memory iterator")

    /* Now process all received chunk message buffers */
    for (i = 0; i < (size_t)chunk_msg_bufs_len; i++) {
        H5D_filtered_collective_io_info_t *chunk_entry = NULL;
        const unsigned char               *msg_ptr     = chunk_msg_bufs[i];
        hsize_t                            chunk_idx;

        if (msg_ptr) {
            /* Retrieve the chunk's index value */
            HDmemcpy(&chunk_idx, msg_ptr, sizeof(hsize_t));
            msg_ptr += sizeof(hsize_t);

            /* Find the chunk entry according to its chunk index */
            HASH_FIND(hh, chunk_hash_table, &chunk_idx, sizeof(hsize_t), chunk_entry);
            HDassert(chunk_entry);
            HDassert(mpi_rank == chunk_entry->new_owner);

            /*
             * Only process the chunk if its data buffer is allocated.
             * In the case of multi-chunk I/O, we're only working on
             * a chunk at a time, so we need to skip over messages
             * that aren't for the chunk we're currently working on.
             */
            if (!chunk_entry->buf)
                continue;
            else {
                /* Decode the chunk file dataspace from the message */
                if (NULL == (dataspace = H5S_decode(&msg_ptr)))
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTDECODE, FAIL, "unable to decode dataspace")

                if (H5S_select_iter_init(sel_iter, dataspace, type_info->dst_type_size,
                                         H5S_SEL_ITER_SHARE_WITH_DATASPACE) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL,
                                "unable to initialize memory selection information")
                sel_iter_init = TRUE;

                iter_nelmts = H5S_GET_SELECT_NPOINTS(dataspace);

                /* Update the chunk data with the received modification data */
                if (H5D__scatter_mem(msg_ptr, sel_iter, (size_t)iter_nelmts, chunk_entry->buf) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "couldn't scatter to write buffer")

                if (H5S_SELECT_ITER_RELEASE(sel_iter) < 0)
                    HGOTO_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "couldn't release selection iterator")
                sel_iter_init = FALSE;

                if (dataspace) {
                    if (H5S_close(dataspace) < 0)
                        HGOTO_ERROR(H5E_DATASPACE, H5E_CANTFREE, FAIL, "can't close dataspace")
                    dataspace = NULL;
                }

                H5MM_free(chunk_msg_bufs[i]);
                chunk_msg_bufs[i] = NULL;
            }
        }
    }

    /* Finally, filter all the chunks */
    for (i = 0; i < chunk_list_num_entries; i++) {
        if (!chunk_list[i].skip_filter_pline) {
            if (H5Z_pipeline(&di->dset->shared->dcpl_cache.pline, 0, &(chunk_list[i].index_info.filter_mask),
                             err_detect, filter_cb, (size_t *)&chunk_list[i].chunk_new.length,
                             &chunk_list[i].chunk_buf_size, &chunk_list[i].buf) < 0)
                HGOTO_ERROR(H5E_PLINE, H5E_CANTFILTER, FAIL, "output pipeline failed")
        }

#if H5_SIZEOF_SIZE_T > 4
        /* Check for the chunk expanding too much to encode in a 32-bit value */
        if (chunk_list[i].chunk_new.length > ((size_t)0xffffffff))
            HGOTO_ERROR(H5E_DATASET, H5E_BADRANGE, FAIL, "chunk too large for 32-bit length")
#endif
    }

done:
    if (sel_iter) {
        if (sel_iter_init && H5S_SELECT_ITER_RELEASE(sel_iter) < 0)
            HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "couldn't release selection iterator")
        sel_iter = H5FL_FREE(H5S_sel_iter_t, sel_iter);
    }
    if (dataspace && (H5S_close(dataspace) < 0))
        HDONE_ERROR(H5E_DATASPACE, H5E_CANTFREE, FAIL, "can't close dataspace")
    if (fill_space && (H5S_close(fill_space) < 0))
        HDONE_ERROR(H5E_DATASET, H5E_CLOSEERROR, FAIL, "can't close fill space")

    /* Release the fill buffer info, if it's been initialized */
    if (fb_info_init && H5D__fill_term(&fb_info) < 0)
        HDONE_ERROR(H5E_DATASET, H5E_CANTFREE, FAIL, "Can't release fill buffer info")

    /* On failure, try to free all resources used by entries in the chunk list */
    if (ret_value < 0) {
        for (i = 0; i < chunk_list_num_entries; i++) {
            if (chunk_list[i].buf) {
                H5MM_free(chunk_list[i].buf);
                chunk_list[i].buf = NULL;
            }
        }
    }

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_collective_filtered_chunk_update() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_chunk_reallocate
 *
 * Purpose:     When performing a parallel write on a chunked dataset with
 *              filters applied, all ranks must eventually get together and
 *              perform a collective reallocation of space in the file for
 *              all chunks that were modified on all ranks. This routine is
 *              responsible for coordinating that process.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_chunk_reallocate(H5D_filtered_collective_io_info_t *chunk_list,
                                               size_t chunk_list_num_entries, size_t *num_chunks_assigned_map,
                                               H5D_io_info_t *io_info, H5D_chk_idx_info_t *idx_info,
                                               int mpi_rank, int mpi_size)
{
    H5D_chunk_alloc_info_t *collective_list = NULL;
    MPI_Datatype            send_type;
    MPI_Datatype            recv_type;
    hbool_t                 send_type_derived          = FALSE;
    hbool_t                 recv_type_derived          = FALSE;
    hbool_t                 need_sort                  = FALSE;
    size_t                  collective_num_entries     = 0;
    size_t                  num_local_chunks_processed = 0;
    size_t                  i;
    void                   *gathered_array     = NULL;
    int                    *counts_disps_array = NULL;
    int                    *counts_ptr         = NULL;
    int                    *displacements_ptr  = NULL;
    int                     mpi_code;
    herr_t                  ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == chunk_list_num_entries);
    HDassert(io_info);
    HDassert(idx_info);
    HDassert(idx_info->storage->idx_type != H5D_CHUNK_IDX_NONE);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Reallocation of chunk file space");
#endif

    /*
     * Make sure it's safe to cast this rank's number
     * of chunks to be sent into an int for MPI
     */
    H5_CHECK_OVERFLOW(chunk_list_num_entries, size_t, int);

    /* Create derived datatypes for the chunk file space info needed */
    if (H5D__mpio_get_chunk_alloc_info_types(&recv_type, &recv_type_derived, &send_type, &send_type_derived) <
        0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                    "can't create derived datatypes for chunk file space info")

    /*
     * Gather the new chunk sizes to all ranks for a collective reallocation
     * of the chunks in the file.
     */
    if (num_chunks_assigned_map) {
        /*
         * If a mapping between rank value -> number of assigned chunks has
         * been provided (usually during linked-chunk I/O), we can use this
         * to optimize MPI overhead a bit since MPI ranks won't need to
         * first inform each other about how many chunks they're contributing.
         */
        if (NULL == (counts_disps_array = H5MM_malloc(2 * (size_t)mpi_size * sizeof(*counts_disps_array)))) {
            /* Push an error, but still participate in collective gather operation */
            HDONE_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL,
                        "couldn't allocate receive counts and displacements array")
        }
        else {
            /* Set the receive counts from the assigned chunks map */
            counts_ptr = counts_disps_array;

            for (i = 0; i < (size_t)mpi_size; i++)
                H5_CHECKED_ASSIGN(counts_ptr[i], int, num_chunks_assigned_map[i], size_t);

            /* Set the displacements into the receive buffer for the gather operation */
            displacements_ptr = &counts_disps_array[mpi_size];

            *displacements_ptr = 0;
            for (i = 1; i < (size_t)mpi_size; i++)
                displacements_ptr[i] = displacements_ptr[i - 1] + counts_ptr[i - 1];
        }

        /* Perform gather operation */
        if (H5_mpio_gatherv_alloc(chunk_list, (int)chunk_list_num_entries, send_type, counts_ptr,
                                  displacements_ptr, recv_type, TRUE, 0, io_info->comm, mpi_rank, mpi_size,
                                  &gathered_array, &collective_num_entries) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGATHER, FAIL, "can't gather chunk file space info to/from ranks")
    }
    else {
        /*
         * If no mapping between rank value -> number of assigned chunks has
         * been provided (usually during multi-chunk I/O), all MPI ranks will
         * need to first inform other ranks about how many chunks they're
         * contributing before performing the actual gather operation. Use
         * the 'simple' MPI_Allgatherv wrapper for this.
         */
        if (H5_mpio_gatherv_alloc_simple(chunk_list, (int)chunk_list_num_entries, send_type, recv_type, TRUE,
                                         0, io_info->comm, mpi_rank, mpi_size, &gathered_array,
                                         &collective_num_entries) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGATHER, FAIL, "can't gather chunk file space info to/from ranks")
    }

    /* Collectively re-allocate the modified chunks (from each rank) in the file */
    collective_list = (H5D_chunk_alloc_info_t *)gathered_array;
    for (i = 0, num_local_chunks_processed = 0; i < collective_num_entries; i++) {
        H5D_chunk_alloc_info_t *coll_entry = &collective_list[i];
        hbool_t                 need_insert;
        hbool_t                 update_local_chunk;

        if (H5D__chunk_file_alloc(idx_info, &coll_entry->chunk_current, &coll_entry->chunk_new, &need_insert,
                                  NULL) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTALLOC, FAIL, "unable to allocate chunk")

        /*
         * If we just re-allocated a chunk that is local to this
         * rank, make sure to update the chunk entry in the local
         * chunk list
         */
        update_local_chunk =
            (num_local_chunks_processed < chunk_list_num_entries) &&
            (coll_entry->chunk_idx == chunk_list[num_local_chunks_processed].index_info.chunk_idx);

        if (update_local_chunk) {
            H5D_filtered_collective_io_info_t *local_chunk;

            local_chunk = &chunk_list[num_local_chunks_processed];

            /* Sanity check that this chunk is actually local */
            HDassert(mpi_rank == local_chunk->orig_owner);
            HDassert(mpi_rank == local_chunk->new_owner);

            local_chunk->chunk_new              = coll_entry->chunk_new;
            local_chunk->index_info.need_insert = need_insert;

            /*
             * Since chunk reallocation can move chunks around, check if
             * the local chunk list is still in ascending offset of order
             * in the file
             */
            if (num_local_chunks_processed) {
                haddr_t curr_chunk_offset = local_chunk->chunk_new.offset;
                haddr_t prev_chunk_offset = chunk_list[num_local_chunks_processed - 1].chunk_new.offset;

                HDassert(H5F_addr_defined(prev_chunk_offset) && H5F_addr_defined(curr_chunk_offset));
                if (curr_chunk_offset < prev_chunk_offset)
                    need_sort = TRUE;
            }

            num_local_chunks_processed++;
        }
    }

    HDassert(chunk_list_num_entries == num_local_chunks_processed);

    /*
     * Ensure this rank's local chunk list is sorted in
     * ascending order of offset in the file
     */
    if (need_sort)
        HDqsort(chunk_list, chunk_list_num_entries, sizeof(H5D_filtered_collective_io_info_t),
                H5D__cmp_filtered_collective_io_info_entry);

done:
    H5MM_free(gathered_array);
    H5MM_free(counts_disps_array);

    if (send_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&send_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }
    if (recv_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&recv_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* H5D__mpio_collective_filtered_chunk_reallocate() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_chunk_reinsert
 *
 * Purpose:     When performing a parallel write on a chunked dataset with
 *              filters applied, all ranks must eventually get together and
 *              perform a collective reinsertion into the dataset's chunk
 *              index of chunks that were modified. This routine is
 *              responsible for coordinating that process.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_chunk_reinsert(H5D_filtered_collective_io_info_t *chunk_list,
                                             size_t chunk_list_num_entries, size_t *num_chunks_assigned_map,
                                             H5D_io_info_t *io_info, H5D_dset_io_info_t *di,
                                             H5D_chk_idx_info_t *idx_info, int mpi_rank, int mpi_size)
{
    H5D_chunk_ud_t chunk_ud;
    MPI_Datatype   send_type;
    MPI_Datatype   recv_type;
    hbool_t        send_type_derived = FALSE;
    hbool_t        recv_type_derived = FALSE;
    hsize_t        scaled_coords[H5O_LAYOUT_NDIMS];
    size_t         collective_num_entries = 0;
    size_t         i;
    void          *gathered_array     = NULL;
    int           *counts_disps_array = NULL;
    int           *counts_ptr         = NULL;
    int           *displacements_ptr  = NULL;
    int            mpi_code;
    herr_t         ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == chunk_list_num_entries);
    HDassert(io_info);
    HDassert(di);
    HDassert(idx_info);

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TRACE_ENTER(mpi_rank);
    H5D_MPIO_TIME_START(mpi_rank, "Reinsertion of modified chunks into chunk index");
#endif

    /* Only re-insert chunks if index has an insert method */
    if (!idx_info->storage->ops->insert)
        HGOTO_DONE(SUCCEED);

    /*
     * Make sure it's safe to cast this rank's number
     * of chunks to be sent into an int for MPI
     */
    H5_CHECK_OVERFLOW(chunk_list_num_entries, size_t, int);

    /* Create derived datatypes for the chunk re-insertion info needed */
    if (H5D__mpio_get_chunk_insert_info_types(&recv_type, &recv_type_derived, &send_type,
                                              &send_type_derived) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL,
                    "can't create derived datatypes for chunk re-insertion info")

    /*
     * Gather information to all ranks for a collective re-insertion
     * of the modified chunks into the chunk index
     */
    if (num_chunks_assigned_map) {
        /*
         * If a mapping between rank value -> number of assigned chunks has
         * been provided (usually during linked-chunk I/O), we can use this
         * to optimize MPI overhead a bit since MPI ranks won't need to
         * first inform each other about how many chunks they're contributing.
         */
        if (NULL == (counts_disps_array = H5MM_malloc(2 * (size_t)mpi_size * sizeof(*counts_disps_array)))) {
            /* Push an error, but still participate in collective gather operation */
            HDONE_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL,
                        "couldn't allocate receive counts and displacements array")
        }
        else {
            /* Set the receive counts from the assigned chunks map */
            counts_ptr = counts_disps_array;

            for (i = 0; i < (size_t)mpi_size; i++)
                H5_CHECKED_ASSIGN(counts_ptr[i], int, num_chunks_assigned_map[i], size_t);

            /* Set the displacements into the receive buffer for the gather operation */
            displacements_ptr = &counts_disps_array[mpi_size];

            *displacements_ptr = 0;
            for (i = 1; i < (size_t)mpi_size; i++)
                displacements_ptr[i] = displacements_ptr[i - 1] + counts_ptr[i - 1];
        }

        /* Perform gather operation */
        if (H5_mpio_gatherv_alloc(chunk_list, (int)chunk_list_num_entries, send_type, counts_ptr,
                                  displacements_ptr, recv_type, TRUE, 0, io_info->comm, mpi_rank, mpi_size,
                                  &gathered_array, &collective_num_entries) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGATHER, FAIL,
                        "can't gather chunk index re-insertion info to/from ranks")
    }
    else {
        /*
         * If no mapping between rank value -> number of assigned chunks has
         * been provided (usually during multi-chunk I/O), all MPI ranks will
         * need to first inform other ranks about how many chunks they're
         * contributing before performing the actual gather operation. Use
         * the 'simple' MPI_Allgatherv wrapper for this.
         */
        if (H5_mpio_gatherv_alloc_simple(chunk_list, (int)chunk_list_num_entries, send_type, recv_type, TRUE,
                                         0, io_info->comm, mpi_rank, mpi_size, &gathered_array,
                                         &collective_num_entries) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTGATHER, FAIL,
                        "can't gather chunk index re-insertion info to/from ranks")
    }

    /* Initialize static chunk udata fields from chunk index info */
    H5D_MPIO_INIT_CHUNK_UD_INFO(chunk_ud, idx_info);

    for (i = 0; i < collective_num_entries; i++) {
        H5D_chunk_insert_info_t *coll_entry = &((H5D_chunk_insert_info_t *)gathered_array)[i];

        /*
         * We only need to reinsert this chunk if we had to actually
         * allocate or reallocate space in the file for it
         */
        if (!coll_entry->index_info.need_insert)
            continue;

        chunk_ud.chunk_block   = coll_entry->chunk_block;
        chunk_ud.chunk_idx     = coll_entry->index_info.chunk_idx;
        chunk_ud.filter_mask   = coll_entry->index_info.filter_mask;
        chunk_ud.common.scaled = scaled_coords;

        /* Calculate scaled coordinates for the chunk */
        if (idx_info->layout->idx_type == H5D_CHUNK_IDX_EARRAY && idx_info->layout->u.earray.unlim_dim > 0) {
            /*
             * Extensible arrays where the unlimited dimension is not
             * the slowest-changing dimension "swizzle" the coordinates
             * to move the unlimited dimension value to offset 0. Therefore,
             * we use the "swizzled" down chunks to calculate the "swizzled"
             * scaled coordinates and then we undo the "swizzle" operation.
             *
             * TODO: In the future, this is something that should be handled
             *       by the particular chunk index rather than manually
             *       here. Likely, the chunk index ops should get a new
             *       callback that accepts a chunk index and provides the
             *       caller with the scaled coordinates for that chunk.
             */
            H5VM_array_calc_pre(chunk_ud.chunk_idx, di->dset->shared->ndims,
                                idx_info->layout->u.earray.swizzled_down_chunks, scaled_coords);

            H5VM_unswizzle_coords(hsize_t, scaled_coords, idx_info->layout->u.earray.unlim_dim);
        }
        else {
            H5VM_array_calc_pre(chunk_ud.chunk_idx, di->dset->shared->ndims,
                                di->dset->shared->layout.u.chunk.down_chunks, scaled_coords);
        }

        scaled_coords[di->dset->shared->ndims] = 0;

#ifndef NDEBUG
        /*
         * If a matching local chunk entry is found, the
         * `chunk_info` structure (which contains the chunk's
         * pre-computed scaled coordinates) will be valid
         * for this rank. Compare those coordinates against
         * the calculated coordinates above to make sure
         * they match.
         */
        for (size_t dbg_idx = 0; dbg_idx < chunk_list_num_entries; dbg_idx++) {
            if (coll_entry->index_info.chunk_idx == chunk_list[dbg_idx].index_info.chunk_idx) {
                hbool_t coords_match = !HDmemcmp(scaled_coords, chunk_list[dbg_idx].chunk_info->scaled,
                                                 di->dset->shared->ndims * sizeof(hsize_t));

                HDassert(coords_match && "Calculated scaled coordinates for chunk didn't match "
                                         "chunk's actual scaled coordinates!");
                break;
            }
        }
#endif

        if ((idx_info->storage->ops->insert)(idx_info, &chunk_ud, di->dset) < 0)
            HGOTO_ERROR(H5E_DATASET, H5E_CANTINSERT, FAIL, "unable to insert chunk address into index")
    }

done:
    H5MM_free(gathered_array);
    H5MM_free(counts_disps_array);

    if (send_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&send_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }
    if (recv_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&recv_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }

#ifdef H5Dmpio_DEBUG
    H5D_MPIO_TIME_STOP(mpi_rank);
    H5D_MPIO_TRACE_EXIT(mpi_rank);
#endif

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_collective_filtered_chunk_reinsert() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_get_chunk_redistribute_info_types
 *
 * Purpose:     Constructs MPI derived datatypes for communicating the
 *              info from a H5D_filtered_collective_io_info_t structure
 *              that is necessary for redistributing shared chunks during a
 *              collective write of filtered chunks.
 *
 *              The datatype returned through `contig_type` has an extent
 *              equal to the size of an H5D_chunk_redistribute_info_t
 *              structure and is suitable for communicating that structure
 *              type.
 *
 *              The datatype returned through `resized_type` has an extent
 *              equal to the size of an H5D_filtered_collective_io_info_t
 *              structure. This makes it suitable for sending an array of
 *              those structures, while extracting out just the info
 *              necessary for the chunk redistribution operation during
 *              communication.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_get_chunk_redistribute_info_types(MPI_Datatype *contig_type, hbool_t *contig_type_derived,
                                            MPI_Datatype *resized_type, hbool_t *resized_type_derived)
{
    MPI_Datatype struct_type              = MPI_DATATYPE_NULL;
    hbool_t      struct_type_derived      = FALSE;
    MPI_Datatype chunk_block_type         = MPI_DATATYPE_NULL;
    hbool_t      chunk_block_type_derived = FALSE;
    MPI_Datatype types[5];
    MPI_Aint     displacements[5];
    int          block_lengths[5];
    int          field_count;
    int          mpi_code;
    herr_t       ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(contig_type);
    HDassert(contig_type_derived);
    HDassert(resized_type);
    HDassert(resized_type_derived);

    *contig_type_derived  = FALSE;
    *resized_type_derived = FALSE;

    /* Create struct type for the inner H5F_block_t structure */
    if (H5F_mpi_get_file_block_type(FALSE, &chunk_block_type, &chunk_block_type_derived) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't create derived type for chunk file description")

    field_count = 5;
    HDassert(field_count == (sizeof(types) / sizeof(MPI_Datatype)));

    /*
     * Create structure type to pack chunk H5F_block_t structure
     * next to chunk_idx, orig_owner, new_owner and num_writers
     * fields
     */
    block_lengths[0] = 1;
    block_lengths[1] = 1;
    block_lengths[2] = 1;
    block_lengths[3] = 1;
    block_lengths[4] = 1;
    displacements[0] = offsetof(H5D_chunk_redistribute_info_t, chunk_block);
    displacements[1] = offsetof(H5D_chunk_redistribute_info_t, chunk_idx);
    displacements[2] = offsetof(H5D_chunk_redistribute_info_t, orig_owner);
    displacements[3] = offsetof(H5D_chunk_redistribute_info_t, new_owner);
    displacements[4] = offsetof(H5D_chunk_redistribute_info_t, num_writers);
    types[0]         = chunk_block_type;
    types[1]         = HSIZE_AS_MPI_TYPE;
    types[2]         = MPI_INT;
    types[3]         = MPI_INT;
    types[4]         = MPI_INT;
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Type_create_struct(field_count, block_lengths, displacements, types, contig_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
    *contig_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(contig_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

    /* Create struct type to extract the chunk_current, chunk_idx, orig_owner,
     * new_owner and num_writers fields from a H5D_filtered_collective_io_info_t
     * structure
     */
    block_lengths[0] = 1;
    block_lengths[1] = 1;
    block_lengths[2] = 1;
    block_lengths[3] = 1;
    block_lengths[4] = 1;
    displacements[0] = offsetof(H5D_filtered_collective_io_info_t, chunk_current);
    displacements[1] = offsetof(H5D_filtered_collective_io_info_t, index_info.chunk_idx);
    displacements[2] = offsetof(H5D_filtered_collective_io_info_t, orig_owner);
    displacements[3] = offsetof(H5D_filtered_collective_io_info_t, new_owner);
    displacements[4] = offsetof(H5D_filtered_collective_io_info_t, num_writers);
    types[0]         = chunk_block_type;
    types[1]         = HSIZE_AS_MPI_TYPE;
    types[2]         = MPI_INT;
    types[3]         = MPI_INT;
    types[4]         = MPI_INT;
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Type_create_struct(field_count, block_lengths, displacements, types, &struct_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
    struct_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_create_resized(
                            struct_type, 0, sizeof(H5D_filtered_collective_io_info_t), resized_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_resized failed", mpi_code)
    *resized_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(resized_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

done:
    if (struct_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&struct_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }
    if (chunk_block_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&chunk_block_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }

    if (ret_value < 0) {
        if (*resized_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(resized_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *resized_type_derived = FALSE;
        }
        if (*contig_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(contig_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *contig_type_derived = FALSE;
        }
    }

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_get_chunk_redistribute_info_types() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_get_chunk_alloc_info_types
 *
 * Purpose:     Constructs MPI derived datatypes for communicating the info
 *              from a H5D_filtered_collective_io_info_t structure that is
 *              necessary for re-allocating file space during a collective
 *              write of filtered chunks.
 *
 *              The datatype returned through `contig_type` has an extent
 *              equal to the size of an H5D_chunk_alloc_info_t structure
 *              and is suitable for communicating that structure type.
 *
 *              The datatype returned through `resized_type` has an extent
 *              equal to the size of an H5D_filtered_collective_io_info_t
 *              structure. This makes it suitable for sending an array of
 *              those structures, while extracting out just the info
 *              necessary for the chunk file space reallocation operation
 *              during communication.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_get_chunk_alloc_info_types(MPI_Datatype *contig_type, hbool_t *contig_type_derived,
                                     MPI_Datatype *resized_type, hbool_t *resized_type_derived)
{
    MPI_Datatype struct_type              = MPI_DATATYPE_NULL;
    hbool_t      struct_type_derived      = FALSE;
    MPI_Datatype chunk_block_type         = MPI_DATATYPE_NULL;
    hbool_t      chunk_block_type_derived = FALSE;
    MPI_Datatype types[3];
    MPI_Aint     displacements[3];
    int          block_lengths[3];
    int          field_count;
    int          mpi_code;
    herr_t       ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(contig_type);
    HDassert(contig_type_derived);
    HDassert(resized_type);
    HDassert(resized_type_derived);

    *contig_type_derived  = FALSE;
    *resized_type_derived = FALSE;

    /* Create struct type for the inner H5F_block_t structure */
    if (H5F_mpi_get_file_block_type(FALSE, &chunk_block_type, &chunk_block_type_derived) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't create derived type for chunk file description")

    field_count = 3;
    HDassert(field_count == (sizeof(types) / sizeof(MPI_Datatype)));

    /*
     * Create structure type to pack both chunk H5F_block_t structures
     * next to chunk_idx field
     */
    block_lengths[0] = 1;
    block_lengths[1] = 1;
    block_lengths[2] = 1;
    displacements[0] = offsetof(H5D_chunk_alloc_info_t, chunk_current);
    displacements[1] = offsetof(H5D_chunk_alloc_info_t, chunk_new);
    displacements[2] = offsetof(H5D_chunk_alloc_info_t, chunk_idx);
    types[0]         = chunk_block_type;
    types[1]         = chunk_block_type;
    types[2]         = HSIZE_AS_MPI_TYPE;
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Type_create_struct(field_count, block_lengths, displacements, types, contig_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
    *contig_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(contig_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

    /*
     * Create struct type to extract the chunk_current, chunk_new and chunk_idx
     * fields from a H5D_filtered_collective_io_info_t structure
     */
    block_lengths[0] = 1;
    block_lengths[1] = 1;
    block_lengths[2] = 1;
    displacements[0] = offsetof(H5D_filtered_collective_io_info_t, chunk_current);
    displacements[1] = offsetof(H5D_filtered_collective_io_info_t, chunk_new);
    displacements[2] = offsetof(H5D_filtered_collective_io_info_t, index_info.chunk_idx);
    types[0]         = chunk_block_type;
    types[1]         = chunk_block_type;
    types[2]         = HSIZE_AS_MPI_TYPE;
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Type_create_struct(field_count, block_lengths, displacements, types, &struct_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
    struct_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_create_resized(
                            struct_type, 0, sizeof(H5D_filtered_collective_io_info_t), resized_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_resized failed", mpi_code)
    *resized_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(resized_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

done:
    if (struct_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&struct_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }
    if (chunk_block_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&chunk_block_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }

    if (ret_value < 0) {
        if (*resized_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(resized_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *resized_type_derived = FALSE;
        }
        if (*contig_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(contig_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *contig_type_derived = FALSE;
        }
    }

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_get_chunk_alloc_info_types() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_get_chunk_insert_info_types
 *
 * Purpose:     Constructs MPI derived datatypes for communicating the
 *              information necessary when reinserting chunks into a
 *              dataset's chunk index. This includes the chunk's new offset
 *              and size (H5F_block_t) and the inner `index_info` structure
 *              of a H5D_filtered_collective_io_info_t structure.
 *
 *              The datatype returned through `contig_type` has an extent
 *              equal to the size of an H5D_chunk_insert_info_t structure
 *              and is suitable for communicating that structure type.
 *
 *              The datatype returned through `resized_type` has an extent
 *              equal to the size of the encompassing
 *              H5D_filtered_collective_io_info_t structure. This makes it
 *              suitable for sending an array of
 *              H5D_filtered_collective_io_info_t structures, while
 *              extracting out just the information needed during
 *              communication.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_get_chunk_insert_info_types(MPI_Datatype *contig_type, hbool_t *contig_type_derived,
                                      MPI_Datatype *resized_type, hbool_t *resized_type_derived)
{
    MPI_Datatype struct_type              = MPI_DATATYPE_NULL;
    hbool_t      struct_type_derived      = FALSE;
    MPI_Datatype chunk_block_type         = MPI_DATATYPE_NULL;
    hbool_t      chunk_block_type_derived = FALSE;
    MPI_Aint     contig_type_extent;
    MPI_Datatype types[4];
    MPI_Aint     displacements[4];
    int          block_lengths[4];
    int          field_count;
    int          mpi_code;
    herr_t       ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(contig_type);
    HDassert(contig_type_derived);
    HDassert(resized_type);
    HDassert(resized_type_derived);

    *contig_type_derived  = FALSE;
    *resized_type_derived = FALSE;

    /* Create struct type for an H5F_block_t structure */
    if (H5F_mpi_get_file_block_type(FALSE, &chunk_block_type, &chunk_block_type_derived) < 0)
        HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't create derived type for chunk file description")

    field_count = 4;
    HDassert(field_count == (sizeof(types) / sizeof(MPI_Datatype)));

    /*
     * Create struct type to pack information into memory as follows:
     *
     * Chunk's new Offset/Size (H5F_block_t) ->
     * Chunk Index Info (H5D_chunk_index_info_t)
     */
    block_lengths[0] = 1;
    block_lengths[1] = 1;
    block_lengths[2] = 1;
    block_lengths[3] = 1;
    displacements[0] = offsetof(H5D_chunk_insert_info_t, chunk_block);
    displacements[1] = offsetof(H5D_chunk_insert_info_t, index_info.chunk_idx);
    displacements[2] = offsetof(H5D_chunk_insert_info_t, index_info.filter_mask);
    displacements[3] = offsetof(H5D_chunk_insert_info_t, index_info.need_insert);
    types[0]         = chunk_block_type;
    types[1]         = HSIZE_AS_MPI_TYPE;
    types[2]         = MPI_UNSIGNED;
    types[3]         = MPI_C_BOOL;
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Type_create_struct(field_count, block_lengths, displacements, types, &struct_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
    struct_type_derived = TRUE;

    contig_type_extent = (MPI_Aint)(sizeof(H5F_block_t) + sizeof(H5D_chunk_index_info_t));

    if (MPI_SUCCESS != (mpi_code = MPI_Type_create_resized(struct_type, 0, contig_type_extent, contig_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_resized failed", mpi_code)
    *contig_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(contig_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

    struct_type_derived = FALSE;
    if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&struct_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_free failed", mpi_code)

    /*
     * Create struct type to correctly extract all needed
     * information from a H5D_filtered_collective_io_info_t
     * structure.
     */
    displacements[0] = offsetof(H5D_filtered_collective_io_info_t, chunk_new);
    displacements[1] = offsetof(H5D_filtered_collective_io_info_t, index_info.chunk_idx);
    displacements[2] = offsetof(H5D_filtered_collective_io_info_t, index_info.filter_mask);
    displacements[3] = offsetof(H5D_filtered_collective_io_info_t, index_info.need_insert);
    if (MPI_SUCCESS !=
        (mpi_code = MPI_Type_create_struct(field_count, block_lengths, displacements, types, &struct_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_struct failed", mpi_code)
    struct_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_create_resized(
                            struct_type, 0, sizeof(H5D_filtered_collective_io_info_t), resized_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_resized failed", mpi_code)
    *resized_type_derived = TRUE;

    if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(resized_type)))
        HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

done:
    if (struct_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&struct_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }
    if (chunk_block_type_derived) {
        if (MPI_SUCCESS != (mpi_code = MPI_Type_free(&chunk_block_type)))
            HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
    }

    if (ret_value < 0) {
        if (*resized_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(resized_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *resized_type_derived = FALSE;
        }
        if (*contig_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(contig_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *contig_type_derived = FALSE;
        }
    }

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_get_chunk_insert_info_types() */

/*-------------------------------------------------------------------------
 * Function:    H5D__mpio_collective_filtered_io_type
 *
 * Purpose:     Constructs a MPI derived datatype for both the memory and
 *              the file for a collective I/O operation on filtered chunks.
 *              The datatype contains the chunk offsets and lengths in the
 *              file and the locations of the chunk data buffers to read
 *              into/write from.
 *
 * Return:      Non-negative on success/Negative on failure
 *
 *-------------------------------------------------------------------------
 */
static herr_t
H5D__mpio_collective_filtered_io_type(H5D_filtered_collective_io_info_t *chunk_list, size_t num_entries,
                                      H5D_io_op_type_t op_type, MPI_Datatype *new_mem_type,
                                      hbool_t *mem_type_derived, MPI_Datatype *new_file_type,
                                      hbool_t *file_type_derived)
{
    MPI_Aint *io_buf_array      = NULL; /* Relative displacements of filtered chunk data buffers */
    MPI_Aint *file_offset_array = NULL; /* Chunk offsets in the file */
    int      *length_array      = NULL; /* Filtered Chunk lengths */
    int       mpi_code;
    herr_t    ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE

    HDassert(chunk_list || 0 == num_entries);
    HDassert(new_mem_type);
    HDassert(mem_type_derived);
    HDassert(new_file_type);
    HDassert(file_type_derived);

    *mem_type_derived  = FALSE;
    *file_type_derived = FALSE;
    *new_mem_type      = MPI_BYTE;
    *new_file_type     = MPI_BYTE;

    if (num_entries > 0) {
        H5F_block_t *chunk_block;
        size_t       last_valid_idx = 0;
        size_t       i;
        int          chunk_count;

        /*
         * Determine number of chunks for I/O operation and
         * setup for derived datatype creation if I/O operation
         * includes multiple chunks
         */
        if (num_entries == 1) {
            /* Set last valid index to 0 for contiguous datatype creation */
            last_valid_idx = 0;

            if (op_type == H5D_IO_OP_WRITE)
                chunk_count = 1;
            else
                chunk_count = chunk_list[0].need_read ? 1 : 0;
        }
        else {
            MPI_Aint chunk_buf;
            MPI_Aint base_buf;
            haddr_t  base_offset = HADDR_UNDEF;

            H5_CHECK_OVERFLOW(num_entries, size_t, int);

            /* Allocate arrays */
            if (NULL == (length_array = H5MM_malloc((size_t)num_entries * sizeof(int))))
                HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL,
                            "memory allocation failed for filtered collective I/O length array")
            if (NULL == (io_buf_array = H5MM_malloc((size_t)num_entries * sizeof(MPI_Aint))))
                HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL,
                            "memory allocation failed for filtered collective I/O buf length array")
            if (NULL == (file_offset_array = H5MM_malloc((size_t)num_entries * sizeof(MPI_Aint))))
                HGOTO_ERROR(H5E_RESOURCE, H5E_CANTALLOC, FAIL,
                            "memory allocation failed for filtered collective I/O offset array")

            /*
             * If doing a write, we can set the base chunk offset
             * and base chunk data buffer right away.
             *
             * If doing a read, some chunks may be skipped over
             * for reading if they aren't yet allocated in the
             * file. Therefore, we have to find the first chunk
             * actually being read in order to set the base chunk
             * offset and base chunk data buffer.
             */
            if (op_type == H5D_IO_OP_WRITE) {
#if H5_CHECK_MPI_VERSION(3, 0)
                if (MPI_SUCCESS != (mpi_code = MPI_Get_address(chunk_list[0].buf, &base_buf)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Get_address failed", mpi_code)
#else
                base_buf = (MPI_Aint)chunk_list[0].buf;
#endif

                base_offset = chunk_list[0].chunk_new.offset;
            }

            for (i = 0, chunk_count = 0; i < num_entries; i++) {
                if (op_type == H5D_IO_OP_READ) {
                    /*
                     * If this chunk isn't being read, don't add it
                     * to the MPI type we're building up for I/O
                     */
                    if (!chunk_list[i].need_read)
                        continue;

                    /*
                     * If this chunk is being read, go ahead and
                     * set the base chunk offset and base chunk
                     * data buffer if we haven't already
                     */
                    if (!H5F_addr_defined(base_offset)) {
#if H5_CHECK_MPI_VERSION(3, 0)
                        if (MPI_SUCCESS != (mpi_code = MPI_Get_address(chunk_list[i].buf, &base_buf)))
                            HMPI_GOTO_ERROR(FAIL, "MPI_Get_address failed", mpi_code)
#else
                        base_buf = (MPI_Aint)chunk_list[i].buf;
#endif

                        base_offset = chunk_list[i].chunk_current.offset;
                    }
                }

                /* Set convenience pointer for current chunk block */
                chunk_block =
                    (op_type == H5D_IO_OP_READ) ? &chunk_list[i].chunk_current : &chunk_list[i].chunk_new;

                /*
                 * Set the current chunk entry's offset in the file, relative to
                 * the first chunk entry
                 */
                HDassert(H5F_addr_defined(chunk_block->offset));
                file_offset_array[chunk_count] = (MPI_Aint)(chunk_block->offset - base_offset);

                /*
                 * Ensure the chunk list is sorted in ascending ordering of
                 * offset in the file
                 */
                if (chunk_count)
                    HDassert(file_offset_array[chunk_count] > file_offset_array[chunk_count - 1]);

                /* Set the current chunk entry's size for the I/O operation */
                H5_CHECK_OVERFLOW(chunk_block->length, hsize_t, int);
                length_array[chunk_count] = (int)chunk_block->length;

                /*
                 * Set the displacement of the chunk entry's chunk data buffer,
                 * relative to the first entry's data buffer
                 */
#if H5_CHECK_MPI_VERSION(3, 1)
                if (MPI_SUCCESS != (mpi_code = MPI_Get_address(chunk_list[i].buf, &chunk_buf)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Get_address failed", mpi_code)

                io_buf_array[chunk_count] = MPI_Aint_diff(chunk_buf, base_buf);
#else
                chunk_buf                 = (MPI_Aint)chunk_list[i].buf;
                io_buf_array[chunk_count] = chunk_buf - base_buf;
#endif

                /*
                 * Set last valid index in case only a single chunk will
                 * be involved in the I/O operation
                 */
                last_valid_idx = i;

                chunk_count++;
            } /* end for */
        }

        /*
         * Create derived datatypes for the chunk list if this
         * rank has any chunks to work on
         */
        if (chunk_count > 0) {
            if (chunk_count == 1) {
                int chunk_len;

                /* Single chunk - use a contiguous type for both memory and file */

                /* Ensure that we can cast chunk size to an int for MPI */
                chunk_block = (op_type == H5D_IO_OP_READ) ? &chunk_list[last_valid_idx].chunk_current
                                                          : &chunk_list[last_valid_idx].chunk_new;
                H5_CHECKED_ASSIGN(chunk_len, int, chunk_block->length, hsize_t);

                if (MPI_SUCCESS != (mpi_code = MPI_Type_contiguous(chunk_len, MPI_BYTE, new_file_type)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Type_contiguous failed", mpi_code)
                *new_mem_type = *new_file_type;

                /*
                 * Since we use the same datatype for both memory and file, only
                 * mark the file type as derived so the caller doesn't try to
                 * free the same type twice
                 */
                *mem_type_derived  = FALSE;
                *file_type_derived = TRUE;

                if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(new_file_type)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)
            }
            else {
                HDassert(file_offset_array);
                HDassert(length_array);
                HDassert(io_buf_array);

                /* Multiple chunks - use an hindexed type for both memory and file */

                /* Create memory MPI type */
                if (MPI_SUCCESS != (mpi_code = MPI_Type_create_hindexed(
                                        chunk_count, length_array, io_buf_array, MPI_BYTE, new_mem_type)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_hindexed failed", mpi_code)
                *mem_type_derived = TRUE;

                if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(new_mem_type)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)

                /* Create file MPI type */
                if (MPI_SUCCESS !=
                    (mpi_code = MPI_Type_create_hindexed(chunk_count, length_array, file_offset_array,
                                                         MPI_BYTE, new_file_type)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Type_create_hindexed failed", mpi_code)
                *file_type_derived = TRUE;

                if (MPI_SUCCESS != (mpi_code = MPI_Type_commit(new_file_type)))
                    HMPI_GOTO_ERROR(FAIL, "MPI_Type_commit failed", mpi_code)
            }
        }
    } /* end if */

done:
    if (file_offset_array)
        H5MM_free(file_offset_array);
    if (io_buf_array)
        H5MM_free(io_buf_array);
    if (length_array)
        H5MM_free(length_array);

    if (ret_value < 0) {
        if (*file_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(new_file_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *file_type_derived = FALSE;
        }
        if (*mem_type_derived) {
            if (MPI_SUCCESS != (mpi_code = MPI_Type_free(new_mem_type)))
                HMPI_DONE_ERROR(FAIL, "MPI_Type_free failed", mpi_code)
            *mem_type_derived = FALSE;
        }
    }

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_collective_filtered_io_type() */

#ifdef H5Dmpio_DEBUG

static herr_t
H5D__mpio_dump_collective_filtered_chunk_list(H5D_filtered_collective_io_info_t *chunk_list,
                                              size_t chunk_list_num_entries, int mpi_rank)
{
    H5D_filtered_collective_io_info_t *chunk_entry;
    size_t                             i;
    herr_t                             ret_value = SUCCEED;

    FUNC_ENTER_PACKAGE_NOERR

    H5D_MPIO_DEBUG(mpi_rank, "CHUNK LIST: [");
    for (i = 0; i < chunk_list_num_entries; i++) {
        unsigned chunk_rank;

        chunk_entry = &chunk_list[i];

        HDassert(chunk_entry->chunk_info);
        chunk_rank = (unsigned)H5S_GET_EXTENT_NDIMS(chunk_entry->chunk_info->fspace);

        H5D_MPIO_DEBUG(mpi_rank, " {");
        H5D_MPIO_DEBUG_VA(mpi_rank, "   - Entry %zu -", i);

        H5D_MPIO_DEBUG(mpi_rank, "   - Chunk Fspace Info -");
        H5D_MPIO_DEBUG_VA(mpi_rank,
                          "     Chunk Current Info: { Offset: %" PRIuHADDR ", Length: %" PRIuHADDR " }",
                          chunk_entry->chunk_current.offset, chunk_entry->chunk_current.length);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Chunk New Info: { Offset: %" PRIuHADDR ", Length: %" PRIuHADDR " }",
                          chunk_entry->chunk_new.offset, chunk_entry->chunk_new.length);

        H5D_MPIO_DEBUG(mpi_rank, "   - Chunk Insert Info -");
        H5D_MPIO_DEBUG_VA(mpi_rank,
                          "     Chunk Scaled Coords (4-d): { %" PRIuHSIZE ", %" PRIuHSIZE ", %" PRIuHSIZE
                          ", %" PRIuHSIZE " }",
                          chunk_rank < 1 ? 0 : chunk_entry->chunk_info->scaled[0],
                          chunk_rank < 2 ? 0 : chunk_entry->chunk_info->scaled[1],
                          chunk_rank < 3 ? 0 : chunk_entry->chunk_info->scaled[2],
                          chunk_rank < 4 ? 0 : chunk_entry->chunk_info->scaled[3]);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Chunk Index: %" PRIuHSIZE, chunk_entry->index_info.chunk_idx);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Filter Mask: %u", chunk_entry->index_info.filter_mask);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Need Insert: %s",
                          chunk_entry->index_info.need_insert ? "YES" : "NO");

        H5D_MPIO_DEBUG(mpi_rank, "   - Other Info -");
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Chunk Info Ptr: %p", (void *)chunk_entry->chunk_info);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Need Read: %s", chunk_entry->need_read ? "YES" : "NO");
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Chunk I/O Size: %zu", chunk_entry->io_size);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Chunk Buffer Size: %zu", chunk_entry->chunk_buf_size);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Original Owner: %d", chunk_entry->orig_owner);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     New Owner: %d", chunk_entry->new_owner);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     # of Writers: %d", chunk_entry->num_writers);
        H5D_MPIO_DEBUG_VA(mpi_rank, "     Chunk Data Buffer Ptr: %p", (void *)chunk_entry->buf);

        H5D_MPIO_DEBUG(mpi_rank, " }");
    }
    H5D_MPIO_DEBUG(mpi_rank, "]");

    FUNC_LEAVE_NOAPI(ret_value)
} /* end H5D__mpio_dump_collective_filtered_chunk_list() */

#endif

#endif /* H5_HAVE_PARALLEL */