path: root/src/H5VLiod_util.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 files COPYING and Copyright.html.  COPYING can be found at the root   *
 * of the source code distribution tree; Copyright.html can be found at the  *
 * root level of an installed copy of the electronic HDF5 document set and   *
 * is linked from the top-level documents page.  It can also be found at     *
 * http://hdfgroup.org/HDF5/doc/Copyright.html.  If you do not have          *
 * access to either file, you may request a copy from help@hdfgroup.org.     *
 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#define H5G_PACKAGE		/*suppress error about including H5Gpkg   */

#include "H5Gpkg.h"		/* Groups		  		*/
#include "H5Sprivate.h"		/* Dataspaces		  		*/
#include "H5WBprivate.h"        /* Wrapped Buffers                      */
#include "H5VLiod_server.h"

#ifdef H5_HAVE_EFF

static H5I_type_t H5VL__iod_get_h5_obj_type(iod_obj_id_t oid, iod_handle_t coh, 
                                            iod_trans_id_t rtid, uint32_t cs_scope);

/*
 * Programmer:  Mohamad Chaarawi <chaarawi@hdfgroup.gov>
 *              February, 2013
 *
 * Purpose:	The IOD plugin server utility routines.
 */


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_server_traverse
 *
 * Purpose: 
 *     Function to Traverse the path in IOD KV objects given the
 *     starting location ID, object handle, and path name. This walks
 *     through the IOD KV objects to get to the last component in the
 *     path. The last component is not opened. Usually this is called when
 *     creating an object. The Function returns the iod ID and object
 *     handle of the before to last component, and the last component
 *     string, if the user requests it.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_server_traverse(iod_handle_t coh, iod_obj_id_t loc_id, iod_handles_t loc_handle, 
                         const char *path, iod_trans_id_t wtid, iod_trans_id_t rtid, 
                         hbool_t create_interm_grps, uint32_t cs_scope,
                         /* out */char **last_comp, /* out */iod_obj_id_t *iod_id, 
                         /* out */iod_handles_t *iod_oh)
{
    char comp_buf[1024];     /* Temporary buffer for path components */
    char *comp;              /* Pointer to buffer for path components */
    H5WB_t *wb = NULL;       /* Wrapped buffer for temporary buffer */
    size_t nchars;	     /* component name length	*/
    iod_handles_t cur_oh;
    iod_handle_t prev_oh;
    iod_obj_id_t cur_id;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    /* Creating intermediate groups is not supported for now */
    assert(FALSE == create_interm_grps);

    cur_oh.rd_oh.cookie = loc_handle.rd_oh.cookie;
    cur_oh.wr_oh.cookie = loc_handle.wr_oh.cookie;
    cur_id = loc_id;

    /* open the current group */
    if(cur_oh.rd_oh.cookie == IOD_OH_UNDEFINED) {
        if (iod_obj_open_read(coh, loc_id, wtid, NULL, &cur_oh.rd_oh, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't open current group");
    }

    /* Wrap the local buffer for serialized header info */
    if(NULL == (wb = H5WB_wrap(comp_buf, sizeof(comp_buf))))
        HGOTO_ERROR_FF(FAIL, "can't wrap buffer")
    /* Get a pointer to a buffer that's large enough  */
    if(NULL == (comp = (char *)H5WB_actual(wb, (HDstrlen(path) + 1))))
        HGOTO_ERROR_FF(FAIL, "can't get actual buffer")

    /* Traverse the path */
    while((path = H5G__component(path, &nchars)) && *path) {
        const char *s;                  /* Temporary string pointer */
        iod_size_t kv_size;
        H5VL_iod_link_t value;

        /* Copy the component path into a null-terminated buffer. */
	HDmemcpy(comp, path, nchars);
	comp[nchars] = '\0';

	/*
	 * The special path `.' is a no-op.
	 */
	if('.' == comp[0] && !comp[1]) {
	    path += nchars;
	    continue;
	} /* end if */

        /* Check if this is the last component of the path */
        if(!((s = H5G__component(path + nchars, NULL)) && *s)) {
            if(last_comp) {
                *last_comp = HDstrdup(comp);
            }
            break;
        }

        kv_size = sizeof(H5VL_iod_link_t);

        prev_oh.cookie = cur_oh.rd_oh.cookie;

        /* lookup next object in the current group */
        ret = H5VL_iod_get_metadata(cur_oh.rd_oh, rtid, H5VL_IOD_LINK, 
                                    comp, cs_scope, NULL, &value);
        if(SUCCEED != ret)
            HGOTO_ERROR_FF(ret, "failed to get link value");

        /* if this a soft link, traverse the link value if the ID is undefined */
        if(H5L_TYPE_SOFT == value.link_type) {
            if('/' == *value.u.symbolic_name) {
                cur_id = ROOT_ID;
            }

            /* Traverse Path and open the target object */
            ret = H5VL_iod_server_open_path(coh, cur_id, cur_oh, value.u.symbolic_name, 
                                            rtid, cs_scope, &cur_id, &cur_oh);
            if(SUCCEED != ret)
                HGOTO_ERROR_FF(ret, "failed to traverse object path");

            free(value.u.symbolic_name);
        }
        else
            cur_id = value.u.iod_id;

        /* Close previous read handle unless it is the original one */
        if(loc_handle.rd_oh.cookie != prev_oh.cookie) {
            ret = iod_obj_close(prev_oh, NULL, NULL);
            if(ret < 0)
                HGOTO_ERROR_FF(ret, "failed to close IOD object");
        }

        /* open the current group */
        ret = iod_obj_open_read(coh, cur_id, rtid, NULL, &cur_oh.rd_oh, NULL);
        if(ret < 0)
            HGOTO_ERROR_FF(ret, "failed to open IOD object");

	/* Advance to next component in string */
	path += nchars;
    } /* end while */

    /* Release temporary component buffer */
    if(wb && H5WB_unwrap(wb) < 0)
        HGOTO_ERROR_FF(FAIL, "can't release wrapped buffer");

    *iod_id = cur_id;
    (*iod_oh).rd_oh.cookie = cur_oh.rd_oh.cookie;

    if(cur_id != loc_id ||
       loc_handle.wr_oh.cookie == IOD_OH_UNDEFINED) {
        /* open a write handle on the ID. */
        ret = iod_obj_open_write(coh, cur_id, wtid, NULL, &cur_oh.wr_oh, NULL);
        if(ret < 0)
            HGOTO_ERROR_FF(ret, "can't open IOD object");
    }
        
    (*iod_oh).wr_oh.cookie = cur_oh.wr_oh.cookie;

done:
    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_server_open_path
 *
 * Purpose: Function to Traverse the path in IOD KV objects given the
 * starting location ID, object handle, and path name. This walks
 * through the IOD KV objects to get to the last component in the path
 * and opens the last component for read only, i.e. does not open the
 * wr_oh, but will set it to undefined. Usually this is called when
 * opening an object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_server_open_path(iod_handle_t coh, iod_obj_id_t loc_id, iod_handles_t loc_handle, 
                          const char *path, iod_trans_id_t rtid, uint32_t cs_scope,
                          /*out*/iod_obj_id_t *iod_id, /*out*/iod_handles_t *iod_oh)
{
    char comp_buf[1024];     /* Temporary buffer for path components */
    char *comp;              /* Pointer to buffer for path components */
    H5WB_t *wb = NULL;       /* Wrapped buffer for temporary buffer */
    size_t nchars;	     /* component name length	*/
    iod_handles_t cur_oh;
    iod_handle_t prev_oh;
    iod_obj_id_t cur_id;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    cur_oh.rd_oh.cookie = loc_handle.rd_oh.cookie;
    cur_oh.wr_oh.cookie = loc_handle.wr_oh.cookie;
    cur_id = loc_id;

    if(cur_oh.rd_oh.cookie == IOD_OH_UNDEFINED) {
        /* open the current group */
        if (iod_obj_open_read(coh, loc_id, rtid, NULL, &cur_oh.rd_oh, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't open start location");
    }

    /* Wrap the local buffer for serialized header info */
    if(NULL == (wb = H5WB_wrap(comp_buf, sizeof(comp_buf))))
        HGOTO_ERROR_FF(FAIL, "can't wrap buffer")
    /* Get a pointer to a buffer that's large enough  */
    if(NULL == (comp = (char *)H5WB_actual(wb, (HDstrlen(path) + 1))))
        HGOTO_ERROR_FF(FAIL, "can't get actual buffer")

    /* Traverse the path */
    while((path = H5G__component(path, &nchars)) && *path) {
        iod_size_t kv_size;
        H5VL_iod_link_t value;

        /* Copy the component path into a null-terminated buffer. */
	HDmemcpy(comp, path, nchars);
	comp[nchars] = '\0';

	/*
	 * The special path `.' is a no-op.
	 */
	if('.' == comp[0] && !comp[1]) {
	    path += nchars;
	    continue;
	} /* end if */

        kv_size = sizeof(H5VL_iod_link_t);

        prev_oh.cookie = cur_oh.rd_oh.cookie;

        /* lookup next object in the current group */
        ret = H5VL_iod_get_metadata(cur_oh.rd_oh, rtid, H5VL_IOD_LINK, 
                                    comp, cs_scope, NULL, &value); 
        if(SUCCEED != ret) {
            /* Close previous handle unless it is the original one */
            if(loc_handle.rd_oh.cookie != prev_oh.cookie && 
               iod_obj_close(prev_oh, NULL, NULL) < 0)
                HGOTO_ERROR_FF(ret, "can't close current object handle");
            HGOTO_ERROR_FF(ret, "failed to retrieve link value");
        }

        /* if this a soft link, traverse the link value if the ID is undefined */
        if(H5L_TYPE_SOFT == value.link_type) {
            if('/' == *value.u.symbolic_name) {
                cur_id = ROOT_ID;
            }

            /* Traverse Path and open the target object */
            ret = H5VL_iod_server_open_path(coh, cur_id, cur_oh, value.u.symbolic_name, 
                                            rtid, cs_scope, &cur_id, &cur_oh);
            if(SUCCEED != ret)
                HGOTO_ERROR_FF(ret, "failed to traverse object path");

            free(value.u.symbolic_name);
        }
        else
            cur_id = value.u.iod_id;

        /* Close previous handle unless it is the original one */
        if(loc_handle.rd_oh.cookie != prev_oh.cookie && 
           iod_obj_close(prev_oh, NULL, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't close current object handle");

        /* open the current group */
        ret = iod_obj_open_read(coh, cur_id, rtid, NULL, &cur_oh.rd_oh, NULL);
        if(ret < 0)
            HGOTO_ERROR_FF(ret, "failed to open IOD object");

	/* Advance to next component in string */
	path += nchars;
    } /* end while */

    /* Release temporary component buffer */
    if(wb && H5WB_unwrap(wb) < 0)
        HGOTO_ERROR_FF(FAIL, "can't release wrapped buffer");

    *iod_id = cur_id;
    (*iod_oh).rd_oh.cookie = cur_oh.rd_oh.cookie;
    (*iod_oh).wr_oh.cookie = IOD_OH_UNDEFINED;

done:
    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_get_file_desc
 *
 * Purpose: 
 *     Function to generate IOD hyperslab objects from HDF5
 *     dataspace selections. If hslabs is NULL, a count of the number of
 *     hslabs needed is returned in count.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_get_file_desc(hid_t space_id, hssize_t *count, iod_hyperslab_t *hslabs)
{
    hssize_t num_descriptors = 0, n;
    int ndims = 0, i;
    herr_t ret_value = SUCCEED;

    /* get the rank of this dataspace */
    if((ndims = H5Sget_simple_extent_ndims(space_id)) < 0)
        HGOTO_ERROR_FF(FAIL, "unable to get dataspace dimesnsion");

    switch(H5Sget_select_type(space_id)) {
    case H5S_SEL_NONE:
        /* nothing selected */
        num_descriptors = 0;
        HGOTO_DONE_FF(SUCCEED);
    case H5S_SEL_ALL:
        /* The entire dataspace is selected, 1 large iod hyperslab is needed */
        num_descriptors = 1;

        if(NULL != hslabs) {
            hsize_t dims[H5S_MAX_RANK];

            /* get the dimensions sizes of the dataspace */
            if(H5Sget_simple_extent_dims(space_id, dims, NULL) < 0)
                HGOTO_ERROR_FF(FAIL, "unable to get dataspace dimesnsion sizes");
            /* populate the hyperslab */
            for(i=0 ; i<ndims ; i++) {
                hslabs[0].start[i] = 0;
                hslabs[0].block[i] = dims[i];
                hslabs[0].stride[i] = dims[i];
                hslabs[0].count[i] = 1;
            }
        }
        break;
    case H5S_SEL_POINTS:
        {
            /* we need a hyperslab element for each point */
            if((num_descriptors = H5Sget_select_elem_npoints(space_id)) < 0)
                HGOTO_ERROR_FF(FAIL, "invalid point selection");

            if(NULL != hslabs) {
                hsize_t *points = NULL;
                size_t point_count = 0;

                point_count = (hsize_t)num_descriptors * (unsigned int)ndims * sizeof(hsize_t);

                if(NULL == (points = (hsize_t *)malloc(point_count)))
                    HGOTO_ERROR_FF(FAIL, "can't allocate array for points coords");

                if(H5Sget_select_elem_pointlist(space_id, (hsize_t)0, 
                                                (hsize_t)num_descriptors, points) < 0)
                    HGOTO_ERROR_FF(FAIL, "Failed to retrieve point coordinates");

                /* populate the hyperslab */
                for(n=0 ; n<num_descriptors ; n++) {
                    hsize_t *cur_ptr = points; /* temp pointer into points array */

                    /* adjust the current pointer to the current point */
                    cur_ptr += n*ndims;
                    for(i=0 ; i<ndims ; i++) {
                        hslabs[n].start[i] = *(cur_ptr+i);
                        hslabs[n].count[i] = 1;
                        hslabs[n].block[i] = 1;
                        hslabs[n].stride[i] = hslabs[n].block[i];
                    }
                }

                free(points);
            }
            break;
        }
    case H5S_SEL_HYPERSLABS:
        {
            /* if the selection is a regular hyperslab
               selection, only 1 iod hyperslab object is
               needed */
            if(TRUE == H5Sselect_is_regular(space_id)) {
                num_descriptors = 1;

                if(NULL != hslabs) {
                    if(H5Sget_reg_hyperslab_params(space_id, 
                                                   hslabs[0].start, 
                                                   hslabs[0].stride,
                                                   hslabs[0].count,
                                                   hslabs[0].block) < 0)
                        HGOTO_ERROR_FF(FAIL, "Failed to retrieve hyperslab selection");
                    for(i=0 ; i<ndims ; i++) {
                        hslabs[0].stride[i] = hslabs[0].block[i] * hslabs[0].stride[i];
                    }
                }
            }
            /* Otherwise populate the hslabs by getting every block */
            else {
                if((num_descriptors = H5Sget_select_hyper_nblocks(space_id)) < 0)
                    HGOTO_ERROR_FF(FAIL, "invalid hyperslab selection");

                if(NULL != hslabs) {
                    hsize_t *blocks = NULL;
                    size_t block_count = 0;
                    hsize_t *cur_ptr;

                    block_count = (unsigned int)ndims * (hsize_t)num_descriptors * sizeof(hsize_t) * 2;

                    if(NULL == (blocks = (hsize_t *)malloc(block_count)))
                        HGOTO_ERROR_FF(FAIL, "can't allocate array for points coords");

                    if(H5Sget_select_hyper_blocklist(space_id, (hsize_t)0, 
                                                     (hsize_t)num_descriptors, blocks) < 0)
                        HGOTO_ERROR_FF(FAIL, "Failed to retrieve point coordinates");


                    cur_ptr = blocks; /* temp pointer into blocks array */
                    /* populate the hyperslab */
                    for(n=0 ; n<num_descriptors ; n++) {
                        /* adjust the current pointer to the current block */
                        cur_ptr += n*ndims*2;
                        for(i=0 ; i<ndims ; i++) {
                            hslabs[n].start[i] = *(cur_ptr+i);
                            hslabs[n].count[i] = 1;
                            hslabs[n].block[i] = *(cur_ptr+ndims+i) - hslabs[n].start[i] + 1;
                            hslabs[n].stride[i] = hslabs[n].block[i];
                        }
                    }

                    free(blocks);
                }
            }
            break;
        }
    case H5S_SEL_ERROR:
    case H5S_SEL_N:
    default:
        HGOTO_ERROR_FF(FAIL, "Invalid Selection type");
    }

    *count = num_descriptors;

done:
    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_plist
 *
 * Purpose:     Function to insert a creation property list in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_plist(iod_handle_t oh, iod_trans_id_t tid, hid_t plist_id,
                      uint32_t cs_scope, iod_hint_list_t *hints, iod_event_t *event)
{
    char *key = NULL;
    void *value = NULL;
    iod_kv_t kv;
    size_t buf_size;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    /* insert group creation properties in Metadata KV */
    key = strdup(H5VL_IOD_KEY_OBJ_CPL);

    /* determine the buffer size needed to store the encoded plist */ 
    if(H5Pencode(plist_id,  NULL, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode plist");
    if(NULL == (value = malloc (buf_size)))
        HGOTO_ERROR_FF(FAIL, "can't allocate plist buffer");
    /* encode plist */ 
    if(H5Pencode(plist_id, value, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode plist");

    kv.key = (void *)key;
    kv.key_len = (iod_size_t)strlen(key) + 1;
    kv.value = value;
    kv.value_len = (iod_size_t)buf_size;

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "PLIST Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for plist");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for plist");
    }

done:
    if(key) {
        free(key); 
        key = NULL;
    }
    if(value) {
        free(value); 
        value = NULL;
    }

    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_link_count
 *
 * Purpose:     Function to insert the link count in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_link_count(iod_handle_t oh, iod_trans_id_t tid, uint64_t count,
                           uint32_t cs_scope, iod_hint_list_t *hints, iod_event_t *event)
{
    char *key = NULL;
    iod_kv_t kv;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    key = strdup(H5VL_IOD_KEY_OBJ_LINK_COUNT);

    kv.key = (void *)key;
    kv.key_len = (iod_size_t)strlen(key) + 1;
    kv.value = &count;
    kv.value_len = sizeof(uint64_t);

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Link Count Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for link count");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for link count");
    }

done:
    if(key) {
        free(key); 
        key = NULL;
    }
    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_object_type
 *
 * Purpose:     Function to insert the object type in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_object_type(iod_handle_t oh, iod_trans_id_t tid, H5I_type_t obj_type,
                            uint32_t cs_scope, iod_hint_list_t *hints, iod_event_t *event)
{
    char *key = NULL;
    iod_kv_t kv;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    key = strdup(H5VL_IOD_KEY_OBJ_TYPE);

    kv.key = (void *)key;
    kv.key_len = (iod_size_t)strlen(key) + 1;
    kv.value = &obj_type;
    kv.value_len = sizeof(int32_t);

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Object Type Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for object type");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for object type");
    }

done:
    if(key) {
        free(key); 
        key = NULL;
    }
    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_datatype
 *
 * Purpose:     Function to insert a datatype in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_datatype(iod_handle_t oh, iod_trans_id_t tid, hid_t type_id,
                         uint32_t cs_scope, iod_hint_list_t *hints, iod_event_t *event)
{
    char *key = NULL;
    void *value = NULL;
    iod_kv_t kv;
    size_t buf_size;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    /* insert group creation properties in Metadata KV */
    key = strdup(H5VL_IOD_KEY_OBJ_DATATYPE);

    /* determine the buffer size needed to store the encoded type */ 
    if(H5Tencode(type_id,  NULL, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode type");
    if(NULL == (value = malloc (buf_size)))
        HGOTO_ERROR_FF(FAIL, "can't allocate type buffer");
    /* encode type */ 
    if(H5Tencode(type_id, value, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode type");

    kv.key = (void *)key;
    kv.key_len = (iod_size_t)strlen(key) + 1;
    kv.value = value;
    kv.value_len = (iod_size_t)buf_size;

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Datatype Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for datatype");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for datatype");
    }

done:
    if(key) {
        free(key); 
        key = NULL;
    }
    if(value) {
        free(value); 
        value = NULL;
    }

    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_datatype_with_key
 *
 * Purpose:     Function to insert a datatype in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_datatype_with_key(iod_handle_t oh, iod_trans_id_t tid, hid_t type_id, 
                                  const char *key, uint32_t cs_scope, iod_hint_list_t *hints, 
                                  iod_event_t *event)
{
    void *value = NULL;
    iod_kv_t kv;
    size_t buf_size;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    /* determine the buffer size needed to store the encoded type */ 
    if(H5Tencode(type_id,  NULL, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode type");
    if(NULL == (value = malloc (buf_size)))
        HGOTO_ERROR_FF(FAIL, "can't allocate type buffer");
    /* encode type */ 
    if(H5Tencode(type_id, value, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode type");

    kv.key = (void *)key;
    kv.key_len = (iod_size_t)strlen(key) + 1;
    kv.value = value;
    kv.value_len = (iod_size_t)buf_size;

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Map Datatype Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for datatype");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for datatype");
    }

done:
    if(value) {
        free(value); 
        value = NULL;
    }

    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_dataspace
 *
 * Purpose:     Function to insert a dataspace in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_dataspace(iod_handle_t oh, iod_trans_id_t tid, hid_t space_id,
                          uint32_t cs_scope, iod_hint_list_t *hints, iod_event_t *event)
{
    char *key = NULL;
    void *value = NULL;
    iod_kv_t kv;
    size_t buf_size;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    /* insert group creation properties in Metadata KV */
    key = strdup(H5VL_IOD_KEY_OBJ_DATASPACE);

    /* determine the buffer size needed to store the encoded space */ 
    if(H5Sencode(space_id,  NULL, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode space");
    if(NULL == (value = malloc (buf_size)))
        HGOTO_ERROR_FF(FAIL, "can't allocate space buffer");
    /* encode space */ 
    if(H5Sencode(space_id, value, &buf_size) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to encode space");

    kv.key = (void *)key;
    kv.key_len = (iod_size_t)strlen(key) + 1;
    kv.value = value;
    kv.value_len = (iod_size_t)buf_size;

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Dataspace Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for dataspace");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for dataspace");
    }

done:
    if(key) {
        free(key); 
        key = NULL;
    }
    if(value) {
        free(value); 
        value = NULL;
    }

    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_insert_new_link
 *
 * Purpose:     Function to insert a link to another object in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_insert_new_link(iod_handle_t oh, iod_trans_id_t tid, const char *link_name,
                         H5L_type_t link_type, const void *link_val, uint32_t cs_scope, 
                         iod_hint_list_t *hints, iod_event_t *event)
{
    iod_kv_t kv;
    void  *value = NULL;
    uint8_t *val_ptr = NULL;
    size_t value_len;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    switch(link_type) {
        case H5L_TYPE_HARD:
            {
                value_len = sizeof(H5L_type_t) + sizeof(iod_obj_id_t);
                value = malloc(value_len);

                val_ptr = (uint8_t *)value;

                memcpy(val_ptr, &link_type, sizeof(H5L_type_t));
                memcpy(val_ptr+sizeof(H5L_type_t), link_val, sizeof(iod_obj_id_t));

                break;
            }
        case H5L_TYPE_SOFT:
            {
                value_len = sizeof(H5L_type_t) + strlen((const char *)link_val) + 1;
                value = malloc(value_len);

                val_ptr = (uint8_t *)value;

                memcpy(val_ptr, &link_type, sizeof(H5L_type_t));
                strcpy((char *)(val_ptr+sizeof(H5L_type_t)), (const char *)link_val);

                break;
            }
        case H5L_TYPE_ERROR:
        case H5L_TYPE_EXTERNAL:
        case H5L_TYPE_MAX:
        default:
            HGOTO_ERROR_FF(FAIL, "unsupported link type");
    }

    kv.key = (void *)link_name;
    kv.key_len = (iod_size_t)strlen(link_name) + 1;
    kv.value = value;
    kv.value_len = value_len;

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(kv.key, kv.key_len);
        cs[1] = H5_checksum_crc64(kv.value, kv.value_len);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Link Type Checksums key = %016lX  value = %016lX\n", cs[0], cs[1]);
#endif

        ret = iod_kv_set(oh, tid, hints, &kv, cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for link");
    }
    else {
        ret = iod_kv_set(oh, tid, hints, &kv, NULL, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "can't set KV pair for link");
    }

done:

    if(value)
        free(value);

    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_get_metadata
 *
 * Purpose:     Function to retrieve a particular metadata value from an
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_get_metadata(iod_handle_t oh, iod_trans_id_t tid, H5VL_iod_metadata_t md_type,
                      const char *key, uint32_t cs_scope, iod_event_t *event, void *md_value)
{
    iod_size_t key_size = strlen(key) + 1;
    iod_size_t val_size = 0;
    void *value = NULL;
    iod_checksum_t *iod_cs = NULL;
    iod_ret_t ret;
    herr_t ret_value = SUCCEED;

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_cs = (iod_checksum_t *)malloc(sizeof(iod_checksum_t) * 2);
    }

    switch(md_type) {
    case H5VL_IOD_PLIST:
        {
            hid_t plist_id;

            ret = iod_kv_get_value(oh, tid, key, key_size, NULL, &val_size, NULL, event);
            if(ret != 0)
                HGOTO_ERROR_FF(ret, "plist lookup failed");

            if(NULL == (value = malloc((size_t)val_size)))
                HGOTO_ERROR_FF(FAIL, "can't allocate value buffer");

            ret = iod_kv_get_value(oh, tid, key, key_size, (char *)value, &val_size, iod_cs, event);
            if(ret != 0)
                HGOTO_ERROR_FF(ret, "plist lookup failed");

            if((plist_id = H5Pdecode(value)) < 0)
                HGOTO_ERROR_FF(FAIL, "failed to decode cpl");

            *((hid_t *)md_value) = plist_id;
            break;
        }
    case H5VL_IOD_LINK_COUNT:
        val_size = sizeof(uint64_t);
        if(NULL == (value = malloc((size_t)val_size)))
            HGOTO_ERROR_FF(FAIL, "can't allocate value buffer");

        ret = iod_kv_get_value(oh, tid, key, key_size, (char *)value, &val_size, iod_cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "link_count lookup failed");

        memcpy(md_value, value, val_size);
        break;
    case H5VL_IOD_DATATYPE:
        {
            hid_t type_id;

            ret = iod_kv_get_value(oh, tid, key, key_size, NULL, &val_size, NULL, event);
            if(ret != 0)
                HGOTO_ERROR_FF(ret, "datatype lookup failed");

            if(NULL == (value = malloc((size_t)val_size)))
                HGOTO_ERROR_FF(FAIL, "can't allocate value buffer");

            ret = iod_kv_get_value(oh, tid, key, key_size, (char *)value, &val_size, iod_cs, event);
            if(ret != 0)
                HGOTO_ERROR_FF(ret, "datatype lookup failed");

            if((type_id = H5Tdecode(value)) < 0)
                HGOTO_ERROR_FF(FAIL, "failed to decode datatype");

            *((hid_t *)md_value) = type_id;
            break;
        }
    case H5VL_IOD_DATASPACE:
        {
            hid_t space_id;

            ret = iod_kv_get_value(oh, tid, key, key_size, NULL, &val_size, NULL, event);
            if(ret != 0)
                HGOTO_ERROR_FF(ret, "dataspace lookup failed");

            if(NULL == (value = malloc((size_t)val_size)))
                HGOTO_ERROR_FF(FAIL, "can't allocate value buffer");

            ret = iod_kv_get_value(oh, tid, key, key_size, (char *)value, &val_size, iod_cs, event);
            if(ret != 0)
                HGOTO_ERROR_FF(ret, "dataspace lookup failed");

            if((space_id = H5Sdecode(value)) < 0)
                HGOTO_ERROR_FF(FAIL, "failed to decode dataspace");

            *((hid_t *)md_value) = space_id;
            break;
        }
    case H5VL_IOD_OBJECT_TYPE:
        val_size = sizeof(int32_t);
        if(NULL == (value = malloc((size_t)val_size)))
            HGOTO_ERROR_FF(FAIL, "can't allocate value buffer");

        ret = iod_kv_get_value(oh, tid, key, key_size, (char *)value, &val_size, iod_cs, event);
        if(ret != 0)
            HGOTO_ERROR_FF(ret, "object type lookup failed");

        memcpy(md_value, value, val_size);
        break;
    case H5VL_IOD_LINK:
        {
            H5VL_iod_link_t *iod_link = (H5VL_iod_link_t *)md_value;
            uint8_t *val_ptr;

            ret = iod_kv_get_value(oh, tid, key, key_size, NULL, &val_size, NULL, event);
            if(ret != 0) {
                fprintf(stderr, "failed to get link %s\n", key);
                HGOTO_ERROR_FF(ret, "link lookup failed");
            }

            if(NULL == (value = malloc((size_t)val_size)))
                HGOTO_ERROR_FF(FAIL, "can't allocate value buffer");

            ret = iod_kv_get_value(oh, tid, key, key_size, (char *)value, &val_size, iod_cs, event);
            if(ret != 0){
                fprintf(stderr, "failed to get link %s\n", key);
                HGOTO_ERROR_FF(ret, "link lookup failed");
            }

            val_ptr = (uint8_t *)value;

            iod_link->link_type = *((H5L_type_t *)val_ptr);
            val_ptr += sizeof(H5L_type_t);

            switch(iod_link->link_type) {
                case H5L_TYPE_HARD:
                    iod_link->u.iod_id = *((iod_obj_id_t *)val_ptr);
                    break;
                case H5L_TYPE_SOFT:
                    iod_link->u.symbolic_name = strdup((char *)val_ptr);
                    break;
                case H5L_TYPE_ERROR:
                case H5L_TYPE_EXTERNAL:
                case H5L_TYPE_MAX:
                default:
                    HGOTO_ERROR_FF(FAIL, "unsupported link type");
            }
            break;
        }
    default:
        HGOTO_ERROR_FF(FAIL, "invalide metadata type");
    }

    if(cs_scope & H5_CHECKSUM_IOD) {
        iod_checksum_t cs[2];

        cs[0] = H5_checksum_crc64(key, key_size);
        cs[1] = H5_checksum_crc64(value, val_size);

#if H5_EFF_DEBUG 
        fprintf(stderr, "Key CS iod = %016lX computed = %016lX\n", iod_cs[0], cs[0]);
        fprintf(stderr, "Value CS iod = %016lX computed = %016lX\n", iod_cs[1], cs[1]);
#endif
        if(iod_cs[0] != cs[0] || iod_cs[1] != cs[1])
            HGOTO_ERROR_FF(FAIL, "Corruption detected when reading metadata from IOD");
    }

done:
    if(value) {
        free(value); 
        value = NULL;
    }

    if(iod_cs) {
        free(iod_cs);
        iod_cs = NULL;
    }

    return ret_value;
}


/*-------------------------------------------------------------------------
 * Function:    H5VL__iod_server_adjust_buffer
 *
 * Checks datatypes to see if type conversion is required, if
 * yes, the buffer is resized accordingly.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 * Programmer:  Mohamad Chaarawi
 *              August, 2013
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5VL__iod_server_adjust_buffer(hid_t mem_type_id, hid_t dset_type_id, size_t nelmts, 
                               hid_t UNUSED dxpl_id, na_bool_t is_coresident, size_t size, void **buf, 
                               hbool_t *is_vl_data, size_t *_buf_size)
{
    herr_t ret_value = SUCCEED;

    if(H5VL__iod_server_type_is_vl(dset_type_id, is_vl_data) < 0)
        HGOTO_ERROR_FF(FAIL, "failed to check dataype");

    if(!(*is_vl_data)) {
        hsize_t buf_size = 0;
        size_t mem_type_size, dset_type_size;

        /* retrieve source and destination datatype sizes for data conversion */
        mem_type_size = H5Tget_size(mem_type_id);
        dset_type_size = H5Tget_size(dset_type_id);

        /* adjust buffer size for data conversion */
        if(mem_type_size < dset_type_size) {
            buf_size = dset_type_size * nelmts;

            /* if we are coresident, and buffer extension is
               required, make a new buffer so we don't mess
               with the user buffer. */
            if(is_coresident) {
                void *new_buf = NULL;

                if(NULL == (new_buf = malloc((size_t)buf_size)))
                    HGOTO_ERROR_FF(FAIL, "Can't malloc new buffer for DT conversion");
                memcpy(new_buf, *buf, size);
                *buf = new_buf;
            }
            else {
                if(NULL == (*buf = realloc(*buf, (size_t)buf_size)))
                    HGOTO_ERROR_FF(FAIL, "Can't adjust buffer for DT conversion");
            }
#if H5_EFF_DEBUG
            fprintf(stderr, "Adjusted Buffer size for dt conversion from %zu to %lld\n", 
                    size, buf_size);
#endif
        }
        else {
            buf_size = mem_type_size * nelmts;
            if(buf_size != size)
                HGOTO_ERROR_FF(FAIL, "Incoming data size is not equal to expected size");
        }

        *_buf_size = buf_size;
    }
    else {
        *_buf_size = size;
    }

done:
    return ret_value;
} /* end H5VL__iod_server_adjust_buffer */


/*-------------------------------------------------------------------------
 * Function:    H5VL__iod_server_type_is_vl
 *
 *              Checks datatypes to see if it's Variable length.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 * Programmer:  Mohamad Chaarawi
 *              August, 2013
 *
 *-------------------------------------------------------------------------
 */
herr_t
H5VL__iod_server_type_is_vl(hid_t type_id, hbool_t *is_vl_data)
{
    herr_t ret_value = SUCCEED;

    switch(H5Tget_class(type_id)) {
        case H5T_STRING:
            if(H5Tis_variable_str(type_id)) {
                *is_vl_data = TRUE;
                break;
            }
        case H5T_INTEGER:
        case H5T_FLOAT:
        case H5T_TIME:
        case H5T_BITFIELD:
        case H5T_OPAQUE:
        case H5T_ENUM:
        case H5T_ARRAY:
        case H5T_NO_CLASS:
        case H5T_REFERENCE:
        case H5T_NCLASSES:
        case H5T_COMPOUND:
            *is_vl_data = FALSE;
            break;
        case H5T_VLEN:
            *is_vl_data = TRUE;
            break;
        default:
            HGOTO_ERROR_FF(FAIL, "unsupported datatype");
    }

done:
    return ret_value;
} /* end H5VL__iod_server_type_is_vl */


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_verify_scratch_pad
 *
 * Purpose:     Function to insert the link count in an 
 *              IOD KV object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_verify_scratch_pad(scratch_pad *sp, iod_checksum_t iod_cs)
{
    iod_checksum_t computed_cs = 0;
    herr_t ret_value = SUCCEED;

    computed_cs = H5_checksum_crc64(sp, sizeof(scratch_pad));

    if(computed_cs != iod_cs) {
        fprintf(stderr, "Scratch pad integrity check failed. IOD cs = %"PRIu64", Computed cs = %"PRIu64"\n",
                iod_cs, computed_cs);
        ret_value = FAIL;
    }

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

herr_t
H5VL_iod_verify_kv_pair(void *key, iod_size_t key_size, void *value, iod_size_t val_size, 
                        iod_checksum_t *iod_cs)
{
    iod_checksum_t cs[2];
    herr_t ret_value = SUCCEED;

    cs[0] = H5_checksum_crc64(key, key_size);
    cs[1] = H5_checksum_crc64(value, val_size);

#if H5_EFF_DEBUG 
    fprintf(stderr, "Key CS iod = %016lX computed = %016lX\n", iod_cs[0], cs[0]);
    fprintf(stderr, "Value CS iod = %016lX computed = %016lX\n", iod_cs[1], cs[1]);
#endif

    if(iod_cs[0] != cs[0] && iod_cs[1] != cs[1])
        HGOTO_ERROR_FF(FAIL, "Corruption detected in IOD KV pair");

done:
    return ret_value;
} /* H5VL_iod_verify_kv_pair */


/*-------------------------------------------------------------------------
 * Function:	H5VL__iod_get_h5_obj_type
 *
 * Purpose:     Function to retrieve the HDF5 object type of an IOD object.
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
static H5I_type_t 
H5VL__iod_get_h5_obj_type(iod_obj_id_t oid, iod_handle_t coh, iod_trans_id_t rtid, uint32_t cs_scope)
{
    iod_handle_t mdkv_oh, oh;
    H5I_type_t obj_type;
    iod_obj_type_t iod_type;
    herr_t ret_value = -1;

    iod_type = IOD_OBJID_GETTYPE(oid);

    if(IOD_OBJ_ARRAY == iod_type)
        obj_type = H5I_DATASET;
    else if(IOD_OBJ_BLOB == iod_type)
        obj_type = H5I_DATATYPE;
    else {
        scratch_pad sp;
        iod_checksum_t sp_cs = 0;

        if (iod_obj_open_read(coh, oid, rtid, NULL /*hints*/, &oh, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't open object");

        /* get scratch pad of the object */
        if(iod_obj_get_scratch(oh, rtid, &sp, &sp_cs, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't get scratch pad for object");
        if(sp_cs && (cs_scope & H5_CHECKSUM_IOD)) {
            /* verify scratch pad integrity */
            if(H5VL_iod_verify_scratch_pad(&sp, sp_cs) < 0)
                HGOTO_ERROR_FF(FAIL, "Scratch Pad failed integrity check");
        }

        /* open the metadata KV */
        if (iod_obj_open_read(coh, sp[0], rtid, NULL /*hints*/, &mdkv_oh, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't open MDKV");

        if(H5VL_iod_get_metadata(mdkv_oh, rtid, H5VL_IOD_OBJECT_TYPE, H5VL_IOD_KEY_OBJ_TYPE,
                                 cs_scope, NULL, &obj_type) < 0)
            HGOTO_ERROR_FF(FAIL, "failed to retrieve link count");

        if(iod_obj_close(mdkv_oh, NULL, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't close current object handle");
        if(iod_obj_close(oh, NULL, NULL) < 0)
            HGOTO_ERROR_FF(FAIL, "can't close current object handle");
    }

    ret_value = obj_type;

done:
    return ret_value;
} /* end H5VL__iod_get_h5_obj_type() */


/*-------------------------------------------------------------------------
 * Function:	H5VL_iod_server_iterate
 *
 * Purpose: 
 *
 * Return:	Success:	SUCCEED 
 *		Failure:	Negative
 *
 *-------------------------------------------------------------------------
 */
herr_t 
H5VL_iod_server_iterate(iod_handle_t coh, iod_obj_id_t obj_id, iod_trans_id_t rtid,
                        H5I_type_t obj_type, const char *link_name, const char *attr_name, 
                        uint32_t cs_scope, H5VL_iterate_op_t op, void *op_data)
{
    iod_handle_t obj_oh;
    herr_t ret;
    herr_t ret_value = SUCCEED;

    ret = (*op)(coh, obj_id, rtid, obj_type, link_name, attr_name, cs_scope, op_data);
    if(ret != 0)
        HGOTO_ERROR_FF(FAIL, "can't apply iterate callback on current object");

    if (iod_obj_open_read(coh, obj_id, rtid, NULL, &obj_oh, NULL) < 0)
        HGOTO_ERROR_FF(FAIL, "can't open current group");

    /* Get the object type, if it is not a group do not check for links */
    if(H5I_GROUP == obj_type || H5I_FILE == obj_type) {
        int num_entries = 0;

        /* Get the object ID and iterate into every member in the group */
        ret = iod_kv_get_num(obj_oh, rtid, &num_entries, NULL);
        if(ret != 0)
            HGOTO_ERROR_FF(FAIL, "can't get number of KV entries");

        if(0 != num_entries) {
            iod_kv_params_t *kvs = NULL;
            iod_kv_t *kv = NULL;
            iod_checksum_t *oid_cs = NULL;
            iod_ret_t *oid_ret = NULL;
            int i;

            kvs = (iod_kv_params_t *)malloc(sizeof(iod_kv_params_t) * (size_t)num_entries);
            kv = (iod_kv_t *)malloc(sizeof(iod_kv_t) * (size_t)num_entries);
            oid_cs = (iod_checksum_t *)malloc(sizeof(iod_checksum_t) * (size_t)num_entries);
            oid_ret = (iod_ret_t *)malloc(sizeof(iod_ret_t) * (size_t)num_entries);

            for(i=0 ; i<num_entries ; i++) {
                kv[i].key = malloc(IOD_KV_KEY_MAXLEN);
                kv[i].key_len = IOD_KV_KEY_MAXLEN;
                kvs[i].kv = &kv[i];
                kvs[i].cs = &oid_cs[i];
                kvs[i].ret = &oid_ret[i];
            }

            ret = iod_kv_list_key(obj_oh, rtid, NULL, 0, &num_entries, kvs, NULL);
            if(ret != 0)
                HGOTO_ERROR_FF(FAIL, "can't get list of keys");

            for(i=0 ; i<num_entries ; i++) {
                H5I_type_t otype;
                iod_obj_id_t oid;
                H5VL_iod_link_t value;

                /* lookup object in the current group */
                ret = H5VL_iod_get_metadata(obj_oh, rtid, H5VL_IOD_LINK, 
                                            (char *)(kv[i].key), cs_scope, NULL, &value);
                if(SUCCEED != ret)
                    HGOTO_ERROR_FF(ret, "failed to retrieve link value");

                if(H5L_TYPE_SOFT == value.link_type) {
                    continue;
                }
                else
                    oid = value.u.iod_id;

#if H5_EFF_DEBUG
                fprintf(stderr, "Iterating into %s OID %"PRIx64"\n", ((char *)kv[i].key), oid);
#endif

                /* Get the object type. */
                if((otype = H5VL__iod_get_h5_obj_type(oid, coh, rtid, cs_scope)) < 0)
                    HGOTO_ERROR_FF(FAIL, "can't get object type");

                if(H5VL_iod_server_iterate(coh, oid, rtid, otype, ((char *)kv[i].key), 
                                           NULL, cs_scope, op, op_data) < 0)
                    HGOTO_ERROR_FF(FAIL, "can't iterate into current object");
            }

            for(i=0 ; i<num_entries ; i++) {
                free(kv[i].key);
            }

            free(kv);
            free(oid_cs);
            free(oid_ret);
            free(kvs);
        }
    }

    ret_value = ret;

done:
    iod_obj_close(obj_oh, NULL, NULL);
    return ret_value;
} /* H5VL_iod_server_iterate */

void
print_iod_obj_map(iod_obj_map_t *obj_map)
{
    int i;
    uint32_t u;

    fprintf(stderr, "MAP: oid: %"PRIx64"   type: %d\n", obj_map->oid, obj_map->type);
    fprintf(stderr, "n_bb_loc %d:\n", obj_map->n_bb_loc);
    for (i = 0; i < obj_map->n_bb_loc ; i++) {
        iod_bb_loc_info_t *info = obj_map->bb_loc_infos[i];
        int j;

        fprintf(stderr, "Shadow path: %s  nrank = %d:\n", info->shadow_path, info->nrank);
        for(j=0 ; j<info->nrank; j++)
            fprintf(stderr, "%d ", info->direct_ranks[j]);
        fprintf(stderr, "\n");
    }
    fprintf(stderr, "n_central_loc %d:\n", obj_map->n_central_loc);
    for (i = 0; i < obj_map->n_central_loc ; i++) {
        iod_central_loc_info_t *info = obj_map->central_loc_infos[i];
        int j;

        fprintf(stderr, "Shard ID: %u  nrank = %d:\n", info->shard_id, info->nrank);
        for(j=0 ; j<info->nrank; j++)
            fprintf(stderr, "%d ", info->nearest_ranks[j]);
        fprintf(stderr, "\n");
    }
    switch(obj_map->type) {
    case IOD_OBJ_BLOB:
        {
            fprintf(stderr, "nranges = %d\n", obj_map->u_map.blob_map.n_range);
            for(u = 0; u < obj_map->u_map.blob_map.n_range ; u++) {
                fprintf(stderr, 
                        "offset: %"PRIu64"  length: %zu  location: %d  index: %d  nearest rank: %d\n",
                        obj_map->u_map.blob_map.blob_range[u].offset, 
                        obj_map->u_map.blob_map.blob_range[u].len,
                        obj_map->u_map.blob_map.blob_range[u].loc_type, 
                        obj_map->u_map.blob_map.blob_range[u].loc_index,
                        obj_map->u_map.blob_map.blob_range[u].nearest_rank);
            }
            break;
        }
    case IOD_OBJ_ARRAY:
        {
            fprintf(stderr, "nranges = %d\n", obj_map->u_map.array_map.n_range);
            for(u = 0; u < obj_map->u_map.array_map.n_range ; u++) {
                fprintf(stderr, 
                        "location: %d  index: %d  nearest rank: %d\n",
                        obj_map->u_map.array_map.array_range[u].loc_type,
                        obj_map->u_map.array_map.array_range[u].loc_index,
                        obj_map->u_map.array_map.array_range[u].nearest_rank);
                fprintf(stderr, "start [%"PRIu64"] end [%"PRIu64"]\n",
                        obj_map->u_map.array_map.array_range[u].start_cell[0],
                        obj_map->u_map.array_map.array_range[u].end_cell[0]);
            }
            break;
        }
        break;
    case IOD_OBJ_KV:
    case IOD_OBJ_INVALID:
    case IOD_OBJ_ANY:
    default:
        break;
    }
}
#if 0
herr_t
H5VL_iod_map_type_convert(hid_t src_id, hid_t dst_id, void *buf, size_t buf_size)
{
    H5T_class_t class;
    herr_t ret_value = SUCCEED;

    class = H5Tget_class(src_id);
    if(H5T_VLEN == class || (H5T_STRING == class && H5Tis_variable_str(src_id)))
        HGOTO_ERROR_FF(FAIL, "Can't convert VL or string types");

    class = H5Tget_class(dset_id);
    if(H5T_VLEN == class || (H5T_STRING == class && H5Tis_variable_str(dst_id)))
        HGOTO_ERROR_FF(FAIL, "Can't convert VL or string types");

    /* Check (and do) Type conversion on the Key */
    src_size = H5Tget_size(src_id);
    dst_size = H5Tget_size(dst_id);

    /* adjust buffer size for datatype conversion */
    if(src_size < dst_size) {
        new_size = dst_size;
        if(NULL == (*buf = realloc(*buf, new_size)))
            HGOTO_ERROR_FF(FAIL, "Can't adjust buffer for DT conversion");
    }
    else {
        new_size = src_size;
    }

    if(NULL == (key_buf = malloc((size_t)key_size)))
        HGOTO_ERROR_FF(FAIL, "can't allocate buffer");
    memcpy(key_buf, key.buf, src_size);

    if(H5Tconvert(src_id, dst_id, 1, key_buf, NULL, dxpl_id) < 0)
        HGOTO_ERROR_FF(FAIL, "data type conversion failed")

done:
    return ret_value;
}
#endif
#endif /* H5_HAVE_EFF */