[svn-r23890] Added routine to write from the compactor callback using the constructed

request list. This routines also closes the handles instructing the operation
is completed. 

6 files changed, 907 insertions, 720 deletions

diff --git a/src/H5VLiod_compactor.c b/src/H5VLiod_compactor.c
index 9ec9f51..86c017d 100644
--- a/src/H5VLiod_compactor.c
+++ b/src/H5VLiod_compactor.c
@@ -400,7 +400,7 @@ int H5VL_iod_create_request_list (compactor *queue, request_list_t **list,
 	}
 
 	newlist[request_id].request_id = request_id;
-	newlist[request_id].merged = 0;
+	newlist[request_id].merged = NOT_MERGED;
 	newlist[request_id].num_fblocks = num_entries;
 
 	newlist[request_id].num_mblocks = num_entries; 
@@ -409,9 +409,9 @@ int H5VL_iod_create_request_list (compactor *queue, request_list_t **list,
 	newlist[request_id].selection_id = space_id;
 	/*Incase its not merged, to call the I/O operation 
           directly with selection and memory descriptor*/
-	newlist[request_id].mem_buffer = (char *)buf; 
+	newlist[request_id].mem_buf = (void *)buf; 
 	newlist[request_id].mem_length = buf_size;
-
+	newlist[request_id].op_data = op_data;
 	local_fcont_ptr = newlist[request_id].fblocks;
 	local_mcont_ptr = newlist[request_id].mblocks;
 
@@ -652,7 +652,7 @@ int H5VL_iod_select_overlap (hid_t dataspace_1,
  * Return:	Success:	CP_SUCCESS 
  *		Failure:	Negative
  *
-o * Programmer:  Vishwanth Venkatesan
+ * Programmer:  Vishwanth Venkatesan
  *              June, 2013
  *
  *-------------------------------------------------------------------------
@@ -707,8 +707,8 @@ int H5VL_iod_compact_requests (request_list_t *list, int *total_requests,
 				      &res_dataspace)){
     lselected_req[0] = 0;
     lselected_req[1] = 1;
-    list[request_list[0]].merged = 1;
-    list[request_list[1]].merged = 1;
+    list[request_list[0]].merged = USED_IN_MERGING;
+    list[request_list[1]].merged = USED_IN_MERGING;
     num_selected += 2;
     if (!merge_flag){
       merge_flag = 1;
@@ -718,11 +718,11 @@ int H5VL_iod_compact_requests (request_list_t *list, int *total_requests,
   else{
     if (current_space == res_dataspace){
       lselected_req[0] = 0;
-      list[request_list[0]].merged = 1;
+      list[request_list[0]].merged = USED_IN_MERGING;
     }
     else{
       lselected_req[0] = 1;
-      list[request_list[1]].merged = 1;
+      list[request_list[1]].merged = USED_IN_MERGING;
     }
     num_selected += 1;
   }
@@ -754,7 +754,7 @@ int H5VL_iod_compact_requests (request_list_t *list, int *total_requests,
 	  merge_flag = 0;
 	  /* We need  to create a request list entry for the merged selection */
 	  list[original_requests].request_id = list[original_requests - 1].request_id + 1;
-	  list[original_requests].merged = 2;
+	  list[original_requests].merged = MERGED ;
 	  list[original_requests].selection_id = last_merged;
 	  original_requests++;	  
 	  if (CP_SUCCESS != H5VL_iod_construct_merged_request(list,
@@ -804,7 +804,7 @@ int H5VL_iod_compact_requests (request_list_t *list, int *total_requests,
   if(merge_flag){
     list[original_requests].request_id = list[original_requests - 1].request_id + 1;
     list[original_requests].dataset_id = list[request_list[0]].dataset_id;
-    list[original_requests].merged = 2;
+    list[original_requests].merged = MERGED;
     list[original_requests].selection_id = last_merged;
     original_requests++;	  
     if (CP_SUCCESS != H5VL_iod_construct_merged_request(list,
@@ -844,6 +844,22 @@ int H5VL_iod_compact_requests (request_list_t *list, int *total_requests,
   
 }/*end H5VL_iod_compact_requests*/
 
+
+/*-------------------------------------------------------------------------
+ * Function:	H5VL_iod_construct_merged_request
+ *
+ * Purpose:	Whenever there is a merged request, we need to construct the
+ *              entry with appropriate memory description 
+ *
+ * Return:	Success:	CP_SUCCESS 
+ *		Failure:	Negative
+ *
+ * Programmer:  Vishwanth Venkatesan
+ *              July, 2013
+ *
+ *-------------------------------------------------------------------------
+ */
+
 static 
 int H5VL_iod_construct_merged_request (request_list_t *list,
 				       size_t m_elmnt_size,
@@ -1044,6 +1060,8 @@ int H5VL_iod_construct_merged_request (request_list_t *list,
   FUNC_LEAVE_NOAPI(ret_value);  
   
  }/*end  H5VL_iod_construct_merged_request*/
+
+
 /*-------------------------------------------------------------------------
  * Function:	H5VL_iod_request_exist
  *
diff --git a/src/H5VLiod_compactor.h b/src/H5VLiod_compactor.h
index 8b511ab..e478a49 100644
--- a/src/H5VLiod_compactor.h
+++ b/src/H5VLiod_compactor.h
@@ -57,11 +57,14 @@ typedef struct {
   block_container_t *mblocks;  /* Memory offset/len  */
   size_t num_fblocks;          /* Number of File blocks */   
   size_t num_mblocks;          /* Number of Memory blocks */ 
+  /*Convenience values!*/ 
   size_t elementsize;          /* Size of each element in the dataset */
   hid_t dataset_id;            /* The ID of the dataset */
   hid_t selection_id;          /* The ID of the dataspace  */
-  char *mem_buffer;            /* The Memory buffer address (contiguous) */
-  size_t mem_length;           /* Length of the Memory buffer */
+  /*------------------------------------------------------------------*/
+  void *mem_buf;
+  size_t mem_length;
+  op_data_t *op_data;          /* The datastructure that containe all the information*/
 } request_list_t;
 
 
diff --git a/src/H5VLiod_compactor_queue.h b/src/H5VLiod_compactor_queue.h
index 5a87055..dc63df6 100644
--- a/src/H5VLiod_compactor_queue.h
+++ b/src/H5VLiod_compactor_queue.h
@@ -19,6 +19,9 @@
  
 #define READ 100
 #define WRITE 150
+#define MERGED 297
+#define NOT_MERGED 298
+#define USED_IN_MERGING 299
 
 #define CP_SUCCESS 0
 #define CP_FAIL -1
diff --git a/src/H5VLiod_dset.c b/src/H5VLiod_dset.c
index 67f239a..dc56cd0 100644
--- a/src/H5VLiod_dset.c
+++ b/src/H5VLiod_dset.c
@@ -1,759 +1,920 @@
-/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
- * 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.     *
- * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
-
-#include "H5VLiod_server.h"
-#include "H5VLiod_compactor.h"
-#include "H5VLiod_compactor_queue.h"
-
-#ifdef H5_HAVE_EFF
-
-/*
- * Programmer:  Mohamad Chaarawi <chaarawi@hdfgroup.gov>
- *              June, 2013
- *
- * Purpose:	The IOD plugin server side dataset routines.
- */
-
-
-/*-------------------------------------------------------------------------
- * Function:	H5VL_iod_server_dset_create_cb
- *
- * Purpose:	Creates a dset as a iod object.
- *
- * Return:	Success:	SUCCEED 
- *		Failure:	Negative
- *
- * Programmer:  Mohamad Chaarawi
- *              February, 2013
- *
- *-------------------------------------------------------------------------
- */
-void
-H5VL_iod_server_dset_create_cb(AXE_engine_t UNUSED axe_engine, 
-                               size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
-                               size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
-                               void *_op_data)
-{
-    op_data_t *op_data = (op_data_t *)_op_data;
-    dset_create_in_t *input = (dset_create_in_t *)op_data->input;
-    dset_create_out_t output;
-    iod_handle_t coh = input->coh;
-    iod_handle_t loc_handle = input->loc_oh;
-    iod_obj_id_t loc_id = input->loc_id; /* The ID of the current location object */
-    iod_obj_id_t dset_id = input->dset_id; /* The ID of the dataset that needs to be created */
-    iod_handle_t dset_oh, cur_oh, mdkv_oh;
-    iod_obj_id_t cur_id, mdkv_id;
-    const char *name = input->name;
-    char *last_comp; /* the name of the dataset obtained from the last component in the path */
-    iod_kv_t kv;
-    iod_array_struct_t array;
-    iod_size_t *max_dims;
-    size_t buf_size;
-    scratch_pad_t sp;
-    iod_ret_t ret;
-    hbool_t collective = FALSE; /* MSC - change when we allow for collective */
-    herr_t ret_value = SUCCEED;
+ /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
+  * 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.     *
+  * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
+
+ #include "H5VLiod_server.h"
+ #include "H5VLiod_compactor_queue.h"
+ #include "H5VLiod_compactor.h"
+
+ #ifdef H5_HAVE_EFF
+
+ /*
+  * Programmer:  Mohamad Chaarawi <chaarawi@hdfgroup.gov>
+  *              June, 2013
+  *
+  * Purpose:	The IOD plugin server side dataset routines.
+  */
+
+ 
+ /*-------------------------------------------------------------------------
+  * Function:	H5VL_iod_server_dset_create_cb
+  *
+  * Purpose:	Creates a dset as a iod object.
+  *
+  * Return:	Success:	SUCCEED 
+  *		Failure:	Negative
+  *
+  * Programmer:  Mohamad Chaarawi
+  *              February, 2013
+  *
+  *-------------------------------------------------------------------------
+  */
+ void
+ H5VL_iod_server_dset_create_cb(AXE_engine_t UNUSED axe_engine, 
+				size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
+				size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
+				void *_op_data)
+ {
+     op_data_t *op_data = (op_data_t *)_op_data;
+     dset_create_in_t *input = (dset_create_in_t *)op_data->input;
+     dset_create_out_t output;
+     iod_handle_t coh = input->coh;
+     iod_handle_t loc_handle = input->loc_oh;
+     iod_obj_id_t loc_id = input->loc_id; /* The ID of the current location object */
+     iod_obj_id_t dset_id = input->dset_id; /* The ID of the dataset that needs to be created */
+     iod_handle_t dset_oh, cur_oh, mdkv_oh;
+     iod_obj_id_t cur_id, mdkv_id;
+     const char *name = input->name;
+     char *last_comp; /* the name of the dataset obtained from the last component in the path */
+     iod_kv_t kv;
+     iod_array_struct_t array;
+     iod_size_t *max_dims;
+     size_t buf_size;
+     scratch_pad_t sp;
+     iod_ret_t ret;
+     hbool_t collective = FALSE; /* MSC - change when we allow for collective */
+     herr_t ret_value = SUCCEED;
+
+     FUNC_ENTER_NOAPI_NOINIT
+
+     /* the traversal will retrieve the location where the dataset needs
+	to be created. The traversal will fail if an intermediate group
+	does not exist. */
+     if(H5VL_iod_server_traverse(coh, loc_id, loc_handle, name, FALSE, 
+				 &last_comp, &cur_id, &cur_oh) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't traverse path");
+
+     /* Set the IOD array creation parameters */
+     array.cell_size = H5Tget_size(input->type_id);
+     array.num_dims = H5Sget_simple_extent_ndims(input->space_id);
+     if(NULL == (array.current_dims = (iod_size_t *)malloc (sizeof(iod_size_t) * array.num_dims)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dimention size array");
+     if(NULL == (max_dims = (iod_size_t *)malloc (sizeof(iod_size_t) * array.num_dims)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dimention size array");
+     if(H5Sget_simple_extent_dims(input->space_id, array.current_dims, max_dims) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTGET, FAIL, "can't get dimentions' sizes");
+     array.firstdim_max = max_dims[0];
+     array.chunk_dims = NULL;
+     array.dims_seq = NULL;
+
+     /* MSC - NEED TO FIX THAT */
+ #if 0
+     if(layout.type == H5D_CHUNKED) {
+	 if(NULL == (array.chunk_dims = malloc (sizeof(iod_size_t) * layout.u.chunk.ndims)))
+	     HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate chunk dimention size array");
+	 array.chunk_dims;
+     }
+ #endif
+
+ #if H5VL_IOD_DEBUG 
+     fprintf(stderr, "now creating the dataset %s cellsize %d num dimenstions %d\n",
+	     last_comp, array.cell_size, array.num_dims);
+ #endif
+
+     /* create the dataset */
+     ret = iod_obj_create(coh, IOD_TID_UNKNOWN, NULL/*hints*/, IOD_OBJ_ARRAY, NULL, &array,
+			  &dset_id, NULL /*event*/);
+     if(collective && (0 == ret || EEXISTS == ret)) {
+	 /* Dataset has been created by another process, open it */
+	 if (iod_obj_open_write(coh, dset_id, NULL, &dset_oh, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open Dataset");
+     }
+     else if(!collective && 0 != ret) {
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't create Dataset");
+     }
+
+     /* for the process that succeeded in creating the dataset, update
+	the parent KV, create scratch pad */
+     if(0 == ret) {
+	 /* create the metadata KV object for the dataset */
+	 if(iod_obj_create(coh, IOD_TID_UNKNOWN, NULL, IOD_OBJ_KV, 
+			   NULL, NULL, &mdkv_id, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't create metadata KV object");
+
+	 /* set values for the scratch pad object */
+	 sp.mdkv_id = mdkv_id;
+	 sp.attr_id = IOD_ID_UNDEFINED;
+	 sp.filler1_id = IOD_ID_UNDEFINED;
+	 sp.filler2_id = IOD_ID_UNDEFINED;
+
+	 /* set scratch pad in dataset */
+	 if (iod_obj_set_scratch(dset_oh, IOD_TID_UNKNOWN, &sp, NULL, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set scratch pad");
+
+	 /* Store Metadata in scratch pad */
+	 if (iod_obj_open_write(coh, mdkv_id, NULL, &mdkv_oh, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't create scratch pad");
+
+	 /* MSC - TODO store things */
+ #if 0
+	 /* insert layout metadata into scratch pad */
+	 kv.key = HDstrdup("dataset_dcpl");
+	 /* determine the buffer size needed to store the encoded dcpl of the dataset */ 
+	 if(H5Pencode(input->dcpl_id,  NULL, &buf_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset dcpl");
+	 if(NULL == (kv.value = malloc (buf_size)))
+	     HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dcpl buffer");
+	 /* encode dcpl of the dataset */ 
+	 if(H5Pencode(input->dcpl_id, kv.value, &buf_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset dcpl");
+	 kv.value_len = (iod_size_t)buf_size;
+	 /* insert kv pair into scratch pad */
+	 if (iod_kv_set(mdkv_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
+	 HDfree(kv.key);
+	 free(kv.value);
+
+	 /* insert datatyoe metadata into scratch pad */
+	 kv.key = HDstrdup("dataset_dtype");
+	 /* determine the buffer size needed to store the encoded type of the dataset */ 
+	 if(H5Tencode(input->type_id, NULL, &buf_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset type");
+	 if(NULL == (kv.value = malloc (buf_size)))
+	     HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate type buffer");
+	 /* encode datatype of the dataset */ 
+	 if(H5Tencode(input->type_id, kv.value, &buf_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset type");
+	 kv.value_len = (iod_size_t)buf_size;
+	 /* insert kv pair into scratch pad */
+	 if (iod_kv_set(mdkv_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
+	 HDfree(kv.key);
+	 free(kv.value);
+
+	 kv.key = HDstrdup("dataset_dspace");
+	 /* determine the buffer size needed to store the encoded space of the dataset */ 
+	 if(H5Sencode(input->space_id, NULL, &buf_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset space");
+	 if(NULL == (kv.value = malloc (buf_size)))
+	     HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate space buffer");
+	 /* encode dataspace of the dataset */ 
+	 if(H5Sencode(input->space_id, kv.value, &buf_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset space");
+	 kv.value_len = (iod_size_t)buf_size;
+	 /* insert kv pair into scratch pad */
+	 if (iod_kv_set(mdkv_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
+	 HDfree(kv.key);
+	 free(kv.value);
+ #endif
+	 /* close the Metadata KV object */
+	 if(iod_obj_close(mdkv_oh, NULL, NULL))
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close object");
+
+	 kv.key = HDstrdup(last_comp);
+	 kv.value = &dset_id;
+	 kv.value_len = sizeof(iod_obj_id_t);
+	 /* insert new dataset in kv store of current group */
+	 if (iod_kv_set(cur_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
+	 HDfree(kv.key);
+     }
+
+     /* close parent group if it is not the location we started the
+	traversal into */
+     if(loc_handle.cookie != cur_oh.cookie) {
+	 iod_obj_close(cur_oh, NULL, NULL);
+     }
+
+ #if H5_DO_NATIVE
+     cur_oh.cookie = H5Dcreate2(loc_handle.cookie, last_comp, input->type_id, 
+				input->space_id, input->lcpl_id, 
+				input->dcpl_id, input->dapl_id);
+     HDassert(cur_oh.cookie);
+ #endif
+
+     output.iod_oh = cur_oh;
+
+ #if H5VL_IOD_DEBUG 
+     fprintf(stderr, "Done with dset create, sending response to client\n");
+ #endif
+
+     HG_Handler_start_output(op_data->hg_handle, &output);
 
-    FUNC_ENTER_NOAPI_NOINIT
-
-    /* the traversal will retrieve the location where the dataset needs
-       to be created. The traversal will fail if an intermediate group
-       does not exist. */
-    if(H5VL_iod_server_traverse(coh, loc_id, loc_handle, name, FALSE, 
-                                &last_comp, &cur_id, &cur_oh) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't traverse path");
-
-    /* Set the IOD array creation parameters */
-    array.cell_size = H5Tget_size(input->type_id);
-    array.num_dims = H5Sget_simple_extent_ndims(input->space_id);
-    if(NULL == (array.current_dims = (iod_size_t *)malloc (sizeof(iod_size_t) * array.num_dims)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dimention size array");
-    if(NULL == (max_dims = (iod_size_t *)malloc (sizeof(iod_size_t) * array.num_dims)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dimention size array");
-    if(H5Sget_simple_extent_dims(input->space_id, array.current_dims, max_dims) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTGET, FAIL, "can't get dimentions' sizes");
-    array.firstdim_max = max_dims[0];
-    array.chunk_dims = NULL;
-    array.dims_seq = NULL;
-
-    /* MSC - NEED TO FIX THAT */
-#if 0
-    if(layout.type == H5D_CHUNKED) {
-        if(NULL == (array.chunk_dims = malloc (sizeof(iod_size_t) * layout.u.chunk.ndims)))
-            HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate chunk dimention size array");
-        array.chunk_dims;
-    }
-#endif
+ done:
+     /* return an UNDEFINED oh to the client if the operation failed */
+     if(ret_value < 0) {
+	 output.iod_oh.cookie = IOD_OH_UNDEFINED;
+	 HG_Handler_start_output(op_data->hg_handle, &output);
+     }
+
+     if(max_dims) free(max_dims);
+     if(array.current_dims) free(array.current_dims);
+     last_comp = (char *)H5MM_xfree(last_comp);
+     input = (dset_create_in_t *)H5MM_xfree(input);
+     op_data = (op_data_t *)H5MM_xfree(op_data);
+
+     FUNC_LEAVE_NOAPI_VOID
+ } /* end H5VL_iod_server_dset_create_cb() */
+
+ 
+ /*-------------------------------------------------------------------------
+  * Function:	H5VL_iod_server_dset_open_cb
+  *
+  * Purpose:	Opens a dataset as a iod object.
+  *
+  * Return:	Success:	SUCCEED 
+  *		Failure:	Negative
+  *
+  * Programmer:  Mohamad Chaarawi
+  *              February, 2013
+  *
+  *-------------------------------------------------------------------------
+  */
+ void
+ H5VL_iod_server_dset_open_cb(AXE_engine_t UNUSED axe_engine, 
+			      size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
+			      size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
+			      void *_op_data)
+ {
+     op_data_t *op_data = (op_data_t *)_op_data;
+     dset_open_in_t *input = (dset_open_in_t *)op_data->input;
+     dset_open_out_t output;
+     iod_handle_t coh = input->coh;
+     iod_handle_t loc_handle = input->loc_oh;
+     iod_obj_id_t loc_id = input->loc_id;
+     iod_handle_t cur_oh, mdkv_oh;
+     iod_obj_id_t cur_id;
+     iod_obj_id_t dset_id;
+     char *name = input->name;
+     char *last_comp;
+     scratch_pad_t sp;
+     iod_size_t kv_size = sizeof(iod_obj_id_t);
+     herr_t ret_value = SUCCEED;
+
+     FUNC_ENTER_NOAPI_NOINIT
+
+ #if H5VL_IOD_DEBUG 
+     fprintf(stderr, "Start dataset Open %s\n", name);
+ #endif
+
+     /* the traversal will retrieve the location where the dataset needs
+	to be opened. The traversal will fail if an intermediate group
+	does not exist. */
+     if(H5VL_iod_server_traverse(coh, loc_id, loc_handle, name, FALSE, 
+				 &last_comp, &cur_id, &cur_oh) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't traverse path");
+
+     if(iod_kv_get_value(cur_oh, IOD_TID_UNKNOWN, last_comp, &dset_id, 
+			 kv_size , NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTGET, FAIL, "can't retrieve Array ID from parent KV store");
+
+     /* close parent group and its scratch pad if it is not the
+	location we started the traversal into */
+     if(loc_handle.cookie != cur_oh.cookie) {
+	 iod_obj_close(cur_oh, NULL, NULL);
+     }
+
+     /* open the dataset */
+     if (iod_obj_open_write(coh, dset_id, NULL /*hints*/, &cur_oh, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open dataset");
+
+     /* get scratch pad of the dataset */
+     if(iod_obj_get_scratch(cur_oh, IOD_TID_UNKNOWN, &sp, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_FILE, H5E_CANTINIT, FAIL, "can't get scratch pad for object");
+
+     /* open the metadata scratch pad */
+     if (iod_obj_open_write(coh, sp.mdkv_id, NULL /*hints*/, &mdkv_oh, NULL) < 0)
+	 HGOTO_ERROR(H5E_FILE, H5E_CANTINIT, FAIL, "can't open scratch pad");
+
+ #if 0
+     /* MSC - retrieve all metadata from scratch pad */
+     if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dcpl", NULL, 
+			 &output.dcpl_size, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dcpl lookup failed");
+     if(NULL == (output.dcpl = H5MM_malloc (output.dcpl_size)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dcpl buffer");
+     if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dcpl", output.dcpl, 
+			 &output.dcpl_size, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dcpl lookup failed");
+
+     if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dtype", NULL, 
+			 &output.dtype_size, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dtype lookup failed");
+     if(NULL == (output.dtype = H5MM_malloc (output.dtype_size)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dtype buffer");
+     if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dtype", output.dtype, 
+			 &output.dtype_size, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dtype lookup failed");
+
+     if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dspace", NULL, 
+			 &output.dspace_size, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dspace lookup failed");
+     if(NULL == (output.dspace = H5MM_malloc (output.dspace_size)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dspace buffer");
+     if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dspace", output.dspace, 
+			 &output.dspace_size, NULL, NULL) < 0)
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dspace lookup failed");
+ #endif
+
+     /* close the metadata scratch pad */
+     if(iod_obj_close(mdkv_oh, NULL, NULL))
+	 HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close object");
+
+     {
+	 hsize_t dims[1];
+	 //hid_t space_id, type_id;
+
+ #if H5_DO_NATIVE
+	 printf("dataset name %s    location %d\n", name, loc_handle.cookie);
+	 cur_oh.cookie = H5Dopen(loc_handle.cookie, name, input->dapl_id);
+	 HDassert(cur_oh.cookie);
+	 output.space_id = H5Dget_space(cur_oh.cookie);
+	 output.type_id = H5Dget_type(cur_oh.cookie);
+	 output.dcpl_id = H5P_DATASET_CREATE_DEFAULT;
+ #else
+	 /* fake a dataspace, type, and dcpl */
+	 dims [0] = 60;
+	 output.space_id = H5Screate_simple(1, dims, NULL);
+	 output.type_id = H5Tcopy(H5T_NATIVE_INT);
+	 output.dcpl_id = H5P_DATASET_CREATE_DEFAULT;
+	 cur_oh.cookie = 1;
+ #endif
+
+ #if 0
+	 output.dcpl_size = 0;
+	 output.dcpl = NULL;
+
+	 /* get Type size to encode */
+	 if(H5Tencode(type_id, NULL, &output.dtype_size) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset type");
+	 if(NULL == (output.dtype = H5MM_malloc (output.dtype_size)))
+	     HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate datatype buffer");
+	 if(H5Tencode(type_id, output.dtype, &output.dtype_size) < 0)
+	     HGOTO_ERROR(H5E_DATATYPE, H5E_CANTENCODE, FAIL, "can't encode datatype");
+
+	 /* get Dataspace size to encode */
+	 if(H5Sencode(space_id, NULL, &output.dspace_size)<0)
+	     HGOTO_ERROR(H5E_DATASPACE, H5E_CANTENCODE, FAIL, "can't encode dataspace");
+	 if(NULL == (output.dspace = H5MM_malloc (output.dspace_size)))
+	     HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate datatype buffer");
+	 if(H5Sencode(space_id, output.dspace, &output.dspace_size) < 0)
+	     HGOTO_ERROR(H5E_DATASPACE, H5E_CANTENCODE, FAIL, "can't encode dataspace");
+
+	 H5Sclose(space_id);
+ #endif
+
+     }
+
+     dset_id = 1;
+     output.iod_id = dset_id;
+     output.iod_oh.cookie = cur_oh.cookie;
+
+ #if H5VL_IOD_DEBUG 
+     fprintf(stderr, "Done with dset open, sending response to client\n");
+ #endif
+
+     HG_Handler_start_output(op_data->hg_handle, &output);
 
-#if H5VL_IOD_DEBUG 
-    fprintf(stderr, "now creating the dataset %s cellsize %d num dimenstions %d\n",
-            last_comp, array.cell_size, array.num_dims);
-#endif
+ done:
+     if(ret_value < 0) {
+	 output.iod_oh.cookie = IOD_OH_UNDEFINED;
+	 output.iod_id = IOD_ID_UNDEFINED;
+	 HG_Handler_start_output(op_data->hg_handle, &output);
+     }
+
+     H5Tclose(output.type_id);
+     H5Sclose(output.space_id);
+
+     input = (dset_open_in_t *)H5MM_xfree(input);
+     op_data = (op_data_t *)H5MM_xfree(op_data);
+     last_comp = (char *)H5MM_xfree(last_comp);
+
+     FUNC_LEAVE_NOAPI_VOID
+ } /* end H5VL_iod_server_dset_open_cb() */
+
+ 
+ /*-------------------------------------------------------------------------
+  * Function:	H5VL_iod_server_dset_read_cb
+  *
+  * Purpose:	Reads from IOD into the function shipper BDS handle.
+  *
+  * Return:	Success:	SUCCEED 
+  *		Failure:	Negative
+  *
+  * Programmer:  Mohamad Chaarawi
+  *              January, 2013
+  *
+  *-------------------------------------------------------------------------
+  */
+ void
+ H5VL_iod_server_dset_read_cb(AXE_engine_t UNUSED axe_engine, 
+			      size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
+			      size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
+			      void *_op_data)
+ {
+     op_data_t *op_data = (op_data_t *)_op_data;
+     dset_io_in_t *input = (dset_io_in_t *)op_data->input;
+     dset_read_out_t output;
+     iod_handle_t coh = input->coh;
+     iod_handle_t iod_oh = input->iod_oh;
+     iod_obj_id_t iod_id = input->iod_id;    
+     hg_bulk_t bulk_handle = input->bulk_handle;
+     hid_t space_id = input->space_id;
+     hid_t dxpl_id = input->dxpl_id;
+     hid_t src_id = input->dset_type_id;
+     hid_t dst_id = input->mem_type_id;
+     hg_bulk_block_t bulk_block_handle;
+     hg_bulk_request_t bulk_request;
+     iod_mem_desc_t mem_desc;
+     iod_array_iodesc_t file_desc;
+     iod_hyperslab_t *hslabs = NULL;
+     size_t size, buf_size, src_size, dst_size;
+     void *buf;
+     uint8_t *buf_ptr;
+     hssize_t num_descriptors = 0, n;
+     int ndims, i;
+     uint32_t cs = 0;
+     size_t nelmts;
+     na_addr_t dest = HG_Handler_get_addr(op_data->hg_handle);
+     hbool_t opened_locally = FALSE;
+     herr_t ret_value = SUCCEED;
+
+     FUNC_ENTER_NOAPI_NOINIT
+
+     /* open the dataset if we don't have the handle yet */
+     if(iod_oh.cookie == IOD_OH_UNDEFINED) {
+	 if (iod_obj_open_write(coh, iod_id, NULL /*hints*/, &iod_oh, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open current group");
+	 opened_locally = TRUE;
+     }
+
+     size = HG_Bulk_handle_get_size(bulk_handle);
+
+     nelmts = (size_t)H5Sget_select_npoints(space_id);
+
+     src_size = H5Tget_size(src_id);
+     dst_size = H5Tget_size(dst_id);
+
+     /* adjust buffer size for datatype conversion */
+     if(src_size > dst_size) {
+	 buf_size = src_size * nelmts;
+ #if H5VL_IOD_DEBUG 
+	 fprintf(stderr, "Adjusted Buffer size because of datatype conversion from %d to %d: ", 
+		 size, buf_size);        
+ #endif
+     }
+     else {
+	 buf_size = dst_size * nelmts;
+	 assert(buf_size == size);
+     }
+
+     if(NULL == (buf = malloc(buf_size)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate read buffer");
+
+     /* get the rank of the dataspace */
+     if((ndims = H5Sget_simple_extent_ndims(space_id)) < 0)
+	 HGOTO_ERROR(H5E_INTERNAL, H5E_CANTGET, FAIL, "unable to get dataspace dimesnsion");
+
+     /* get the number of decriptors required, i.e. the numbers of iod
+	I/O operations needed */
+     if(H5VL_iod_get_file_desc(space_id, &num_descriptors, NULL) < 0)
+	 HGOTO_ERROR(H5E_DATASPACE, H5E_CANTGET, FAIL, "unable to generate IOD file descriptor from dataspace selection");
+
+     /* allocate the IOD hyperslab descriptors needed */
+     if(NULL == (hslabs = (iod_hyperslab_t *)malloc
+		 (sizeof(iod_hyperslab_t) * (size_t)num_descriptors)))
+	 HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate iod array descriptors");
+
+     for(n=0 ; n<num_descriptors ; n++) {
+	 hslabs[n].start = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+	 hslabs[n].stride = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+	 hslabs[n].block = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+	 hslabs[n].count = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+     }
+
+     /* generate the descriptors after allocating the array */
+     if(H5VL_iod_get_file_desc(space_id, &num_descriptors, hslabs) < 0)
+	 HGOTO_ERROR(H5E_DATASPACE, H5E_CANTGET, FAIL, "unable to generate IOD file descriptor from dataspace selection");
+
+     buf_ptr = (uint8_t *)buf;
+
+     /* read each descriptore from the IOD container */
+     for(n=0 ; n<num_descriptors ; n++) {
+	 hsize_t num_bytes = 0;
+	 hsize_t num_elems = 0;
+
+	 /* determine how many bytes the current descriptor holds */
+	 for(i=0 ; i<ndims ; i++)
+	     num_elems *= (hslabs[n].count[i] * hslabs[n].block[i]);
+	 num_bytes = num_elems * src_size;
+
+ #if 0
+	 /* set the memory descriptor */
+	 mem_desc.nfrag = 1;
+	 mem_desc.frag->addr = (void *)buf_ptr;
+	 mem_desc.frag->len = (iod_size_t)num_bytes;
+ #endif
+
+	 buf_ptr += num_bytes;
+
+	 /* set the file descriptor */
+	 file_desc = hslabs[n];
+
+ #if H5VL_IOD_DEBUG 
+	 for(i=0 ; i<ndims ; i++) {
+	     fprintf(stderr, "Dim %d:  start %zu   stride %zu   block %zu   count %zu\n", 
+		     i, (size_t)file_desc.start[i], (size_t)file_desc.stride[i], 
+		     (size_t)file_desc.block[i], (size_t)file_desc.count[i]);
+	 }
+ #endif
+
+	 /* read from array object */
+	 if(iod_array_read(iod_oh, IOD_TID_UNKNOWN, NULL, &mem_desc, &file_desc, NULL, NULL) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
+     }
+
+     {
+	 hbool_t flag = FALSE;
+	 int *ptr = (int *)buf;
+
+ #if H5_DO_NATIVE
+	 ret_value = H5Dread(iod_oh.cookie, src_id, H5S_ALL, space_id, dxpl_id, buf);
+ #else /* fake data */
+	 for(i=0;i<60;++i)
+	     ptr[i] = i;
+ #endif
+	 if(H5Tconvert(src_id, dst_id, nelmts, buf, NULL, dxpl_id) < 0)
+	     HGOTO_ERROR(H5E_DATATYPE, H5E_CANTINIT, FAIL, "data type conversion failed");
+
+	 /* calculate a checksum for the data to be sent */
+	 cs = H5checksum(buf, size, NULL);
+
+	 /* MSC - check if client requested to corrupt data */
+	 if(dxpl_id != H5P_DEFAULT && H5Pget_dxpl_inject_corruption(dxpl_id, &flag) < 0)
+	     HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read property list");
+	 if(flag) {
+	     fprintf(stderr, "Injecting a bad data value to cause corruption \n");
+	     ptr[0] = 10;
+	 }
+     }
+
+     /* Create a new block handle to write the data */
+     HG_Bulk_block_handle_create(buf, size, HG_BULK_READ_ONLY, &bulk_block_handle);
+
+     /* Write bulk data here and wait for the data to be there  */
+     if(HG_SUCCESS != HG_Bulk_write_all(dest, bulk_handle, bulk_block_handle, &bulk_request))
+	 HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
+     /* wait for it to complete */
+     if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_BULK_MAX_IDLE_TIME, HG_BULK_STATUS_IGNORE))
+	 HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
 
-    /* create the dataset */
-    ret = iod_obj_create(coh, IOD_TID_UNKNOWN, NULL/*hints*/, IOD_OBJ_ARRAY, NULL, &array,
-                         &dset_id, NULL /*event*/);
-    if(collective && (0 == ret || EEXISTS == ret)) {
-        /* Dataset has been created by another process, open it */
-        if (iod_obj_open_write(coh, dset_id, NULL, &dset_oh, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open Dataset");
-    }
-    else if(!collective && 0 != ret) {
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't create Dataset");
-    }
+ done:
 
-    /* for the process that succeeded in creating the dataset, update
-       the parent KV, create scratch pad */
-    if(0 == ret) {
-        /* create the metadata KV object for the dataset */
-        if(iod_obj_create(coh, IOD_TID_UNKNOWN, NULL, IOD_OBJ_KV, 
-                          NULL, NULL, &mdkv_id, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't create metadata KV object");
-
-        /* set values for the scratch pad object */
-        sp.mdkv_id = mdkv_id;
-        sp.attr_id = IOD_ID_UNDEFINED;
-        sp.filler1_id = IOD_ID_UNDEFINED;
-        sp.filler2_id = IOD_ID_UNDEFINED;
-
-        /* set scratch pad in dataset */
-        if (iod_obj_set_scratch(dset_oh, IOD_TID_UNKNOWN, &sp, NULL, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set scratch pad");
-
-        /* Store Metadata in scratch pad */
-        if (iod_obj_open_write(coh, mdkv_id, NULL, &mdkv_oh, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't create scratch pad");
-
-        /* MSC - TODO store things */
-#if 0
-        /* insert layout metadata into scratch pad */
-        kv.key = HDstrdup("dataset_dcpl");
-        /* determine the buffer size needed to store the encoded dcpl of the dataset */ 
-        if(H5Pencode(input->dcpl_id,  NULL, &buf_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset dcpl");
-        if(NULL == (kv.value = malloc (buf_size)))
-            HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dcpl buffer");
-        /* encode dcpl of the dataset */ 
-        if(H5Pencode(input->dcpl_id, kv.value, &buf_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset dcpl");
-        kv.value_len = (iod_size_t)buf_size;
-        /* insert kv pair into scratch pad */
-        if (iod_kv_set(mdkv_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
-        HDfree(kv.key);
-        free(kv.value);
-
-        /* insert datatyoe metadata into scratch pad */
-        kv.key = HDstrdup("dataset_dtype");
-        /* determine the buffer size needed to store the encoded type of the dataset */ 
-        if(H5Tencode(input->type_id, NULL, &buf_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset type");
-        if(NULL == (kv.value = malloc (buf_size)))
-            HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate type buffer");
-        /* encode datatype of the dataset */ 
-        if(H5Tencode(input->type_id, kv.value, &buf_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset type");
-        kv.value_len = (iod_size_t)buf_size;
-        /* insert kv pair into scratch pad */
-        if (iod_kv_set(mdkv_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
-        HDfree(kv.key);
-        free(kv.value);
-
-        kv.key = HDstrdup("dataset_dspace");
-        /* determine the buffer size needed to store the encoded space of the dataset */ 
-        if(H5Sencode(input->space_id, NULL, &buf_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset space");
-        if(NULL == (kv.value = malloc (buf_size)))
-            HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate space buffer");
-        /* encode dataspace of the dataset */ 
-        if(H5Sencode(input->space_id, kv.value, &buf_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset space");
-        kv.value_len = (iod_size_t)buf_size;
-        /* insert kv pair into scratch pad */
-        if (iod_kv_set(mdkv_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
-        HDfree(kv.key);
-        free(kv.value);
+     output.ret = ret_value;
+     output.cs = cs;
+
+ #if H5VL_IOD_DEBUG 
+     fprintf(stderr, "Done with dset read, checksum %u, sending response to client\n", cs);
+ #endif
+
+     if(HG_SUCCESS != HG_Handler_start_output(op_data->hg_handle, &output))
+	 HDONE_ERROR(H5E_SYM, H5E_WRITEERROR, FAIL, "can't send result of write to client");
+     if(HG_SUCCESS != HG_Bulk_block_handle_free(bulk_block_handle))
+	 HDONE_ERROR(H5E_SYM, H5E_WRITEERROR, FAIL, "can't free bds block handle");
+
+     input = (dset_io_in_t *)H5MM_xfree(input);
+     op_data = (op_data_t *)H5MM_xfree(op_data);
+     free(buf);
+
+     /* free allocated descriptors */
+     for(n=0 ; n<num_descriptors ; n++) {
+	 free(hslabs[n].start);
+	 free(hslabs[n].stride);
+	 free(hslabs[n].block);
+	 free(hslabs[n].count);
+     }
+     if(hslabs)
+	 free(hslabs);
+
+     /* close the dataset if we opened it in this routine */
+     if(opened_locally) {
+	 if(iod_obj_close(iod_oh, NULL, NULL))
+	     HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close Array object");
+     }
+     FUNC_LEAVE_NOAPI_VOID
+ } /* end H5VL_iod_server_dset_read_cb() */
+
+ /*-------------------------------------------------------------------------
+  * Function:	H5VL_iod_server_dset_compactor_cb
+  *
+  * Purpose:     Compacts the requests and calls the appropriate read/write.
+  *
+  * Return:	Success:	SUCCEED 
+  *		Failure:	Negative
+  *
+  * Programmer:  Vishwanath Venkatesan
+  *              July, 2013
+  *
+  *-------------------------------------------------------------------------
+  */
+
+ void
+ H5VL_iod_server_dset_compactor_cb (AXE_engine_t UNUSED axe_engine, 
+				    size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
+				    size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
+				    void *_queue)
+ {
+
+   compactor *cqueue = (compactor *)_queue;
+   int n_requests, i = 0;
+   op_data_t *op_data = NULL;
+   herr_t ret_value = SUCCEED;
+   request_list_t *wlist=NULL;
+   dataset_container_t *dlist=NULL;
+   int nentries = 0, ndatasets = 0;
+
+   FUNC_ENTER_NOAPI_NOINIT
+
+ #if DEBUG_COMPACTOR
+     fprintf(stderr, "Enters Call BACK!\n");
+     fprintf (stderr, "Number of requests : %d from call back in queue : %p\n", 
+	      H5VL_iod_get_number_of_requests(cqueue), (void *)cqueue);
+     fflush(stderr);
+ #endif
+
+     pthread_mutex_lock(&lock);
+     compactor_queue_flag = 0;
+     curr_queue = NULL;
+     pthread_mutex_unlock(&lock);  
+
+
+
+     ret_value = H5VL_iod_create_request_list (cqueue,
+					       &wlist,
+					       &nentries,
+					       &dlist,
+					       &ndatasets,
+					       WRITE);
+
+ #if DEBUG_COMPACTOR
+     if (ret_value != CP_SUCCESS){
+       fprintf(stderr,"ERROR !! Compactor create request list failed with error %d  \n",
+	       ret_value);
+     }
 #endif
-        /* close the Metadata KV object */
-        if(iod_obj_close(mdkv_oh, NULL, NULL))
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close object");
-
-        kv.key = HDstrdup(last_comp);
-        kv.value = &dset_id;
-        kv.value_len = sizeof(iod_obj_id_t);
-        /* insert new dataset in kv store of current group */
-        if (iod_kv_set(cur_oh, IOD_TID_UNKNOWN, NULL, &kv, NULL, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't set KV pair in parent");
-        HDfree(kv.key);
-    }
 
-    /* close parent group if it is not the location we started the
-       traversal into */
-    if(loc_handle.cookie != cur_oh.cookie) {
-        iod_obj_close(cur_oh, NULL, NULL);
-    }
-
-#if H5_DO_NATIVE
-    cur_oh.cookie = H5Dcreate2(loc_handle.cookie, last_comp, input->type_id, 
-                               input->space_id, input->lcpl_id, 
-                               input->dcpl_id, input->dapl_id);
-    HDassert(cur_oh.cookie);
-#endif
-
-    output.iod_oh = cur_oh;
-
-#if H5VL_IOD_DEBUG 
-    fprintf(stderr, "Done with dset create, sending response to client\n");
+#if 0     
+     for ( i = 0; i < ndatasets; i ++){
+       H5VL_iod_compact_requests (wlist, &nentries,dlist[i].num_requests,
+				  dlist[i].requests);    
+     }
+#endif  
+     if (CP_SUCCESS != H5VL_iod_server_compactor_write (wlist, nentries)){
+#if DEBUG_COMPACTOR
+       fprintf (stderr,"COMPACTOR CB: compactor write failed \n");
 #endif
+       HGOTO_ERROR(H5E_SYM, H5E_WRITEERROR, CP_FAIL, "Lower lever write failed\n");
+     }	
+     
+     if (NULL != wlist){
+       free(wlist);
+       wlist = NULL;
+     }
+     if (NULL != dlist){
+       free(dlist);
+       dlist = NULL;
+     }
+     
+     if (CP_SUCCESS != H5VL_iod_destroy_compactor_queue(cqueue)){
+       HGOTO_ERROR(H5E_HEAP, H5E_NOSPACE, CP_FAIL, "Cannot free NULL queue\n");
+     }
+     
+ done:
+    FUNC_LEAVE_NOAPI_VOID   
+} /*end H5VL_iod_server_dset_compactor_cb */
 
-    HG_Handler_start_output(op_data->hg_handle, &output);
-
-done:
-    /* return an UNDEFINED oh to the client if the operation failed */
-    if(ret_value < 0) {
-        output.iod_oh.cookie = IOD_OH_UNDEFINED;
-        HG_Handler_start_output(op_data->hg_handle, &output);
-    }
-
-    if(max_dims) free(max_dims);
-    if(array.current_dims) free(array.current_dims);
-    last_comp = (char *)H5MM_xfree(last_comp);
-    input = (dset_create_in_t *)H5MM_xfree(input);
-    op_data = (op_data_t *)H5MM_xfree(op_data);
-
-    FUNC_LEAVE_NOAPI_VOID
-} /* end H5VL_iod_server_dset_create_cb() */
 
-
-/*-------------------------------------------------------------------------
- * Function:	H5VL_iod_server_dset_open_cb
+/*------------------------------------------------------------------------
+ * Function:	H5VL_iod_server_compactor_write
  *
- * Purpose:	Opens a dataset as a iod object.
+ * Purpose:     Compactor lower lever write operations
  *
  * Return:	Success:	SUCCEED 
  *		Failure:	Negative
  *
- * Programmer:  Mohamad Chaarawi
- *              February, 2013
+ * Programmer:  Vishwanath Venkatesan
+ *              July, 2013
  *
  *-------------------------------------------------------------------------
  */
-void
-H5VL_iod_server_dset_open_cb(AXE_engine_t UNUSED axe_engine, 
-                             size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
-                             size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
-                             void *_op_data)
-{
-    op_data_t *op_data = (op_data_t *)_op_data;
-    dset_open_in_t *input = (dset_open_in_t *)op_data->input;
-    dset_open_out_t output;
-    iod_handle_t coh = input->coh;
-    iod_handle_t loc_handle = input->loc_oh;
-    iod_obj_id_t loc_id = input->loc_id;
-    iod_handle_t cur_oh, mdkv_oh;
-    iod_obj_id_t cur_id;
-    iod_obj_id_t dset_id;
-    char *name = input->name;
-    char *last_comp;
-    scratch_pad_t sp;
-    iod_size_t kv_size = sizeof(iod_obj_id_t);
-    herr_t ret_value = SUCCEED;
 
-    FUNC_ENTER_NOAPI_NOINIT
 
-#if H5VL_IOD_DEBUG 
-    fprintf(stderr, "Start dataset Open %s\n", name);
-#endif
-
-    /* the traversal will retrieve the location where the dataset needs
-       to be opened. The traversal will fail if an intermediate group
-       does not exist. */
-    if(H5VL_iod_server_traverse(coh, loc_id, loc_handle, name, FALSE, 
-                                &last_comp, &cur_id, &cur_oh) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't traverse path");
-
-    if(iod_kv_get_value(cur_oh, IOD_TID_UNKNOWN, last_comp, &dset_id, 
-                        kv_size , NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTGET, FAIL, "can't retrieve Array ID from parent KV store");
-
-    /* close parent group and its scratch pad if it is not the
-       location we started the traversal into */
-    if(loc_handle.cookie != cur_oh.cookie) {
-        iod_obj_close(cur_oh, NULL, NULL);
-    }
-
-    /* open the dataset */
-    if (iod_obj_open_write(coh, dset_id, NULL /*hints*/, &cur_oh, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open dataset");
-
-    /* get scratch pad of the dataset */
-    if(iod_obj_get_scratch(cur_oh, IOD_TID_UNKNOWN, &sp, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_FILE, H5E_CANTINIT, FAIL, "can't get scratch pad for object");
-
-    /* open the metadata scratch pad */
-    if (iod_obj_open_write(coh, sp.mdkv_id, NULL /*hints*/, &mdkv_oh, NULL) < 0)
-        HGOTO_ERROR(H5E_FILE, H5E_CANTINIT, FAIL, "can't open scratch pad");
-
-#if 0
-    /* MSC - retrieve all metadata from scratch pad */
-    if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dcpl", NULL, 
-                        &output.dcpl_size, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dcpl lookup failed");
-    if(NULL == (output.dcpl = H5MM_malloc (output.dcpl_size)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dcpl buffer");
-    if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dcpl", output.dcpl, 
-                        &output.dcpl_size, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dcpl lookup failed");
-
-    if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dtype", NULL, 
-                        &output.dtype_size, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dtype lookup failed");
-    if(NULL == (output.dtype = H5MM_malloc (output.dtype_size)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dtype buffer");
-    if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dtype", output.dtype, 
-                        &output.dtype_size, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dtype lookup failed");
-
-    if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dspace", NULL, 
-                        &output.dspace_size, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dspace lookup failed");
-    if(NULL == (output.dspace = H5MM_malloc (output.dspace_size)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate dspace buffer");
-    if(iod_kv_get_value(mdkv_oh, IOD_TID_UNKNOWN, "dataset_dspace", output.dspace, 
-                        &output.dspace_size, NULL, NULL) < 0)
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "dataset dspace lookup failed");
-#endif
-
-    /* close the metadata scratch pad */
-    if(iod_obj_close(mdkv_oh, NULL, NULL))
-        HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close object");
+int 
+H5VL_iod_server_compactor_write (void *_list, int num_requests)
+{
 
-    {
-        hsize_t dims[1];
-        //hid_t space_id, type_id;
+  
+  int ret_value = CP_SUCCESS;
+  int i, *ptr = NULL, request_counter= 0;
+  int ndims;
+  request_list_t *list = (request_list_t *)_list;
+  op_data_t *op_data;
+  dset_io_in_t *input;
+  iod_handle_t coh;
+  iod_handle_t iod_oh;
+  iod_obj_id_t iod_id;    
+  hg_bulk_t bulk_handle;
+  hid_t space_id, dst_id;
+  size_t size, dst_size;
+  void *buf;
+  uint8_t *buf_ptr;
+  hssize_t num_descriptors = 0, n =0;
+  hbool_t opened_locally = FALSE;
+  iod_hyperslab_t *hslabs = NULL;
+  iod_mem_desc_t mem_desc;
+  iod_array_iodesc_t file_desc;
+  hsize_t num_bytes = 0;
+  hsize_t num_elems = 0;	
 
-#if H5_DO_NATIVE
-        printf("dataset name %s    location %d\n", name, loc_handle.cookie);
-        cur_oh.cookie = H5Dopen(loc_handle.cookie, name, input->dapl_id);
-        HDassert(cur_oh.cookie);
-        output.space_id = H5Dget_space(cur_oh.cookie);
-        output.type_id = H5Dget_type(cur_oh.cookie);
-        output.dcpl_id = H5P_DATASET_CREATE_DEFAULT;
-#else
-        /* fake a dataspace, type, and dcpl */
-        dims [0] = 60;
-        output.space_id = H5Screate_simple(1, dims, NULL);
-        output.type_id = H5Tcopy(H5T_NATIVE_INT);
-        output.dcpl_id = H5P_DATASET_CREATE_DEFAULT;
-        cur_oh.cookie = 1;
+  FUNC_ENTER_NOAPI_NOINIT
+   
+  if (num_requests <= 0){
+#if DEBUG_COMPACTOR
+    fprintf (stderr,"COMPACTOR_WRITE: Request < 0 We should not be here!\n");
 #endif
-
-#if 0
-        output.dcpl_size = 0;
-        output.dcpl = NULL;
-
-        /* get Type size to encode */
-        if(H5Tencode(type_id, NULL, &output.dtype_size) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTENCODE, FAIL, "failed to encode dataset type");
-        if(NULL == (output.dtype = H5MM_malloc (output.dtype_size)))
-            HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate datatype buffer");
-        if(H5Tencode(type_id, output.dtype, &output.dtype_size) < 0)
-            HGOTO_ERROR(H5E_DATATYPE, H5E_CANTENCODE, FAIL, "can't encode datatype");
-
-        /* get Dataspace size to encode */
-        if(H5Sencode(space_id, NULL, &output.dspace_size)<0)
-            HGOTO_ERROR(H5E_DATASPACE, H5E_CANTENCODE, FAIL, "can't encode dataspace");
-        if(NULL == (output.dspace = H5MM_malloc (output.dspace_size)))
-            HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate datatype buffer");
-        if(H5Sencode(space_id, output.dspace, &output.dspace_size) < 0)
-            HGOTO_ERROR(H5E_DATASPACE, H5E_CANTENCODE, FAIL, "can't encode dataspace");
-
-        H5Sclose(space_id);
+    ret_value = CP_FAIL;
+    goto done;
+  }
+  
+#if DEBUG_COMPACTOR
+  fprintf (stderr,"Entering COMPACTOR WRITE with requests %d\n", num_requests);
 #endif
+  for (request_counter = 0; request_counter < num_requests; request_counter++){
+    
+    op_data = (op_data_t *)list[request_counter].op_data;
+    input = (dset_io_in_t *)op_data->input;
+    iod_oh = input->iod_oh;
+    iod_id = input->iod_id;
+    bulk_handle = input->bulk_handle;
+    space_id = list[request_counter].selection_id;
+    dst_id = input->dset_type_id;
+    dst_size = H5Tget_size(dst_id);
+    
+    if((ndims = H5Sget_simple_extent_ndims(space_id)) < 0)
+      HGOTO_ERROR(H5E_INTERNAL, H5E_CANTGET, FAIL, "unable to get dataspace dimesnsion");
+    
+    fprintf (stderr,"space_id: %d, selection_id: %d, ndims : %d\n",
+	     space_id, list[request_counter].selection_id, ndims);
 
+    if(iod_oh.cookie == IOD_OH_UNDEFINED) {
+      if (iod_obj_open_write(coh, iod_id, NULL /*hints*/, &iod_oh, NULL) < 0)
+	HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open current group");
+      opened_locally = TRUE;
     }
-
-    dset_id = 1;
-    output.iod_id = dset_id;
-    output.iod_oh.cookie = cur_oh.cookie;
-
-#if H5VL_IOD_DEBUG 
-    fprintf(stderr, "Done with dset open, sending response to client\n");
-#endif
-
-    HG_Handler_start_output(op_data->hg_handle, &output);
-
-done:
-    if(ret_value < 0) {
-        output.iod_oh.cookie = IOD_OH_UNDEFINED;
-        output.iod_id = IOD_ID_UNDEFINED;
-        HG_Handler_start_output(op_data->hg_handle, &output);
+    
+    if (list[request_counter].merged == NOT_MERGED){
+      /*Even in the case its not merged the buffer was already extracted*/
+      fprintf (stderr,"COMPACTOR WRITE: Request %d has not been merged \n", request_counter);
+      size  = list[request_counter].mem_length;
+      buf = (void *)list[request_counter].mem_buf;
     }
 
-    H5Tclose(output.type_id);
-    H5Sclose(output.space_id);
-
-    input = (dset_open_in_t *)H5MM_xfree(input);
-    op_data = (op_data_t *)H5MM_xfree(op_data);
-    last_comp = (char *)H5MM_xfree(last_comp);
-
-    FUNC_LEAVE_NOAPI_VOID
-} /* end H5VL_iod_server_dset_open_cb() */
-
-
-/*-------------------------------------------------------------------------
- * Function:	H5VL_iod_server_dset_read_cb
- *
- * Purpose:	Reads from IOD into the function shipper BDS handle.
- *
- * Return:	Success:	SUCCEED 
- *		Failure:	Negative
- *
- * Programmer:  Mohamad Chaarawi
- *              January, 2013
- *
- *-------------------------------------------------------------------------
- */
-void
-H5VL_iod_server_dset_read_cb(AXE_engine_t UNUSED axe_engine, 
-                             size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
-                             size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
-                             void *_op_data)
-{
-    op_data_t *op_data = (op_data_t *)_op_data;
-    dset_io_in_t *input = (dset_io_in_t *)op_data->input;
-    dset_read_out_t output;
-    iod_handle_t coh = input->coh;
-    iod_handle_t iod_oh = input->iod_oh;
-    iod_obj_id_t iod_id = input->iod_id;    
-    hg_bulk_t bulk_handle = input->bulk_handle;
-    hid_t space_id = input->space_id;
-    hid_t dxpl_id = input->dxpl_id;
-    hid_t src_id = input->dset_type_id;
-    hid_t dst_id = input->mem_type_id;
-    hg_bulk_block_t bulk_block_handle;
-    hg_bulk_request_t bulk_request;
-    iod_mem_desc_t mem_desc;
-    iod_array_iodesc_t file_desc;
-    iod_hyperslab_t *hslabs = NULL;
-    size_t size, buf_size, src_size, dst_size;
-    void *buf;
-    uint8_t *buf_ptr;
-    hssize_t num_descriptors = 0, n;
-    int ndims, i;
-    uint32_t cs = 0;
-    size_t nelmts;
-    na_addr_t dest = HG_Handler_get_addr(op_data->hg_handle);
-    hbool_t opened_locally = FALSE;
-    herr_t ret_value = SUCCEED;
-
-    FUNC_ENTER_NOAPI_NOINIT
-
-    /* open the dataset if we don't have the handle yet */
-    if(iod_oh.cookie == IOD_OH_UNDEFINED) {
-        if (iod_obj_open_write(coh, iod_id, NULL /*hints*/, &iod_oh, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't open current group");
-        opened_locally = TRUE;
+#if 1
+    ptr = (int *) buf;
+   
+    fprintf(stderr,"COMPACTOR WRITE: Received a buffer or size %zd with values :",
+	    size);
+    for(i=0;i<size/4;++i){
+      fprintf(stderr, "COMPACTOR WRITE: buf[%d] : %d\n",
+	      i, ptr[i]);
     }
-
-    size = HG_Bulk_handle_get_size(bulk_handle);
-
-    nelmts = (size_t)H5Sget_select_npoints(space_id);
-
-    src_size = H5Tget_size(src_id);
-    dst_size = H5Tget_size(dst_id);
-
-    /* adjust buffer size for datatype conversion */
-    if(src_size > dst_size) {
-        buf_size = src_size * nelmts;
-#if H5VL_IOD_DEBUG 
-        fprintf(stderr, "Adjusted Buffer size because of datatype conversion from %d to %d: ", 
-                size, buf_size);        
+    fprintf(stderr, "\n");
 #endif
-    }
-    else {
-        buf_size = dst_size * nelmts;
-        assert(buf_size == size);
-    }
-
-    if(NULL == (buf = malloc(buf_size)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate read buffer");
 
     /* get the rank of the dataspace */
-    if((ndims = H5Sget_simple_extent_ndims(space_id)) < 0)
-        HGOTO_ERROR(H5E_INTERNAL, H5E_CANTGET, FAIL, "unable to get dataspace dimesnsion");
 
+  
     /* get the number of decriptors required, i.e. the numbers of iod
-       I/O operations needed */
+	 I/O operations needed */
     if(H5VL_iod_get_file_desc(space_id, &num_descriptors, NULL) < 0)
-        HGOTO_ERROR(H5E_DATASPACE, H5E_CANTGET, FAIL, "unable to generate IOD file descriptor from dataspace selection");
-
+      HGOTO_ERROR(H5E_DATASPACE, H5E_CANTGET, FAIL, "unable to generate IOD file descriptor from dataspace selection");
+      
+    fprintf (stderr,"num_descriptors : %llu, ndims: %d\n", 
+	     num_descriptors,
+	     ndims);
     /* allocate the IOD hyperslab descriptors needed */
     if(NULL == (hslabs = (iod_hyperslab_t *)malloc
-                (sizeof(iod_hyperslab_t) * (size_t)num_descriptors)))
-        HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate iod array descriptors");
-
-    for(n=0 ; n<num_descriptors ; n++) {
-        hslabs[n].start = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
-        hslabs[n].stride = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
-        hslabs[n].block = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
-        hslabs[n].count = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
-    }
-
-    /* generate the descriptors after allocating the array */
-    if(H5VL_iod_get_file_desc(space_id, &num_descriptors, hslabs) < 0)
+		(sizeof(iod_hyperslab_t) * (size_t)num_descriptors)))
+      HGOTO_ERROR(H5E_SYM, H5E_NOSPACE, FAIL, "can't allocate iod array descriptors");
+    
+      for (n = 0; n < num_descriptors; n++){
+	hslabs[n].start = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+	hslabs[n].stride = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+	hslabs[n].block = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+	hslabs[n].count = (iod_size_t *)malloc(sizeof(iod_size_t) * ndims);
+      }
+      
+      /* generate the descriptors after allocating the array */
+      if(H5VL_iod_get_file_desc(space_id, &num_descriptors, hslabs) < 0)
         HGOTO_ERROR(H5E_DATASPACE, H5E_CANTGET, FAIL, "unable to generate IOD file descriptor from dataspace selection");
-
-    buf_ptr = (uint8_t *)buf;
-
-    /* read each descriptore from the IOD container */
-    for(n=0 ; n<num_descriptors ; n++) {
-        hsize_t num_bytes = 0;
-        hsize_t num_elems = 0;
-
-        /* determine how many bytes the current descriptor holds */
-        for(i=0 ; i<ndims ; i++)
-            num_elems *= (hslabs[n].count[i] * hslabs[n].block[i]);
-        num_bytes = num_elems * src_size;
-
+      
+      buf_ptr = (uint8_t *)buf;
+      for(n=0 ; n<num_descriptors ; n++) {
+
+	fprintf (stderr, "Enters for num_descriptors : %llu, n : %llu\n", num_descriptors, n);
+	num_bytes = 0;
+	num_elems = 0;
+
+	
+	/* determine how many bytes the current descriptor holds */
+        for(i=0 ; i<ndims ; i++){
+	  fprintf (stderr, "COMPACTOR WRITE: n: %llu, i: %d\n", n, i);
+	  num_elems *= (hslabs[n].count[i] * hslabs[n].block[i]);
+	}
+        num_bytes = num_elems * dst_size;
 #if 0
         /* set the memory descriptor */
-        mem_desc.nfrag = 1;
-        mem_desc.frag->addr = (void *)buf_ptr;
-        mem_desc.frag->len = (iod_size_t)num_bytes;
+	mem_desc = (iod_mem_desc_t *) malloc (sizeof (iod_mem_desc_t));
+	mem_desc->nfrag = 1;
+        mem_desc->frag[0].addr = (void *)buf_ptr;
+        mem_desc->frag[0].len = (iod_size_t)num_bytes;
 #endif
-
-        buf_ptr += num_bytes;
-
-        /* set the file descriptor */
-        file_desc = hslabs[n];
+	buf_ptr += num_bytes;
+	file_desc = hslabs[n];
 
 #if H5VL_IOD_DEBUG 
         for(i=0 ; i<ndims ; i++) {
-            fprintf(stderr, "Dim %d:  start %zu   stride %zu   block %zu   count %zu\n", 
-                    i, (size_t)file_desc.start[i], (size_t)file_desc.stride[i], 
-                    (size_t)file_desc.block[i], (size_t)file_desc.count[i]);
-        }
-#endif
-
-        /* read from array object */
-        if(iod_array_read(iod_oh, IOD_TID_UNKNOWN, NULL, &mem_desc, &file_desc, NULL, NULL) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
-    }
-    
-    {
-        hbool_t flag = FALSE;
-        int *ptr = (int *)buf;
-
-#if H5_DO_NATIVE
-        ret_value = H5Dread(iod_oh.cookie, src_id, H5S_ALL, space_id, dxpl_id, buf);
-#else /* fake data */
-        for(i=0;i<60;++i)
-            ptr[i] = i;
+	  fprintf(stderr, "Dim %d:  start %zu   stride %zu   block %zu   count %zu\n", 
+		  i, (size_t)file_desc.start[i], (size_t)file_desc.stride[i], 
+		  (size_t)file_desc.block[i], (size_t)file_desc.count[i]);
+	}
 #endif
-        if(H5Tconvert(src_id, dst_id, nelmts, buf, NULL, dxpl_id) < 0)
-            HGOTO_ERROR(H5E_DATATYPE, H5E_CANTINIT, FAIL, "data type conversion failed");
-
-        /* calculate a checksum for the data to be sent */
-        cs = H5checksum(buf, size, NULL);
-
-        /* MSC - check if client requested to corrupt data */
-        if(dxpl_id != H5P_DEFAULT && H5Pget_dxpl_inject_corruption(dxpl_id, &flag) < 0)
-            HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read property list");
-        if(flag) {
-            fprintf(stderr, "Injecting a bad data value to cause corruption \n");
-            ptr[0] = 10;
-        }
-    }
-
-    /* Create a new block handle to write the data */
-    HG_Bulk_block_handle_create(buf, size, HG_BULK_READ_ONLY, &bulk_block_handle);
-
-    /* Write bulk data here and wait for the data to be there  */
-    if(HG_SUCCESS != HG_Bulk_write_all(dest, bulk_handle, bulk_block_handle, &bulk_request))
-        HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
-    /* wait for it to complete */
-    if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_BULK_MAX_IDLE_TIME, HG_BULK_STATUS_IGNORE))
-        HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
-
-done:
-
-    output.ret = ret_value;
-    output.cs = cs;
+	/* write from array object */
+	/* TODO: VV Change checksum once that is fixed*/
+	if(iod_array_write(iod_oh, IOD_TID_UNKNOWN, NULL, NULL /*This is where the memory descriptor goes*/,
+			   &file_desc, NULL, NULL) < 0)
+	  HGOTO_ERROR(H5E_DATASET, H5E_WRITEERROR, FAIL, "can't write to array object");
+      }     
 
 #if H5VL_IOD_DEBUG 
-    fprintf(stderr, "Done with dset read, checksum %u, sending response to client\n", cs);
+      fprintf(stderr, "Done with dset write, sending %d response to client\n", ret_value);
 #endif
+      free(buf);
+      
+      /* free allocated descriptors */
+      for(n=0 ; n<num_descriptors ; n++) {
+	free(hslabs[n].start);
+	free(hslabs[n].stride);
+	free(hslabs[n].block);
+	free(hslabs[n].count);
+      }
+      
+      if(hslabs)
+	free(hslabs);
 
-    if(HG_SUCCESS != HG_Handler_start_output(op_data->hg_handle, &output))
-        HDONE_ERROR(H5E_SYM, H5E_WRITEERROR, FAIL, "can't send result of write to client");
-    if(HG_SUCCESS != HG_Bulk_block_handle_free(bulk_block_handle))
-        HDONE_ERROR(H5E_SYM, H5E_WRITEERROR, FAIL, "can't free bds block handle");
-
+      if(opened_locally) {
+	if(iod_obj_close(iod_oh, NULL, NULL))
+	  HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close Array object");
+      }
+  }
+  for (i = 0; i< num_requests; i++){
+    op_data = (op_data_t *)list[i].op_data;
+    input = (dset_io_in_t *)op_data->input;
+    if(HG_SUCCESS != HG_Handler_start_output(op_data->hg_handle, &ret_value))
+      HDONE_ERROR(H5E_SYM, H5E_WRITEERROR, FAIL, "can't send result of write to client");
     input = (dset_io_in_t *)H5MM_xfree(input);
     op_data = (op_data_t *)H5MM_xfree(op_data);
-    free(buf);
-
-    /* free allocated descriptors */
-    for(n=0 ; n<num_descriptors ; n++) {
-        free(hslabs[n].start);
-        free(hslabs[n].stride);
-        free(hslabs[n].block);
-        free(hslabs[n].count);
-    }
-    if(hslabs)
-        free(hslabs);
-
-    /* close the dataset if we opened it in this routine */
-    if(opened_locally) {
-        if(iod_obj_close(iod_oh, NULL, NULL))
-            HGOTO_ERROR(H5E_SYM, H5E_CANTINIT, FAIL, "can't close Array object");
-    }
-    FUNC_LEAVE_NOAPI_VOID
-} /* end H5VL_iod_server_dset_read_cb() */
-
-/*-------------------------------------------------------------------------
- * Function:	H5VL_iod_server_dset_compactor_cb
- *
- * Purpose:     Compacts the requests and calls the appropriate read/write.
- *
- * Return:	Success:	SUCCEED 
- *		Failure:	Negative
- *
- * Programmer:  Vishwanath Venkatesan
- *              July, 2013
- *
- *-------------------------------------------------------------------------
- */
-
-void
-H5VL_iod_server_dset_compactor_cb (AXE_engine_t UNUSED axe_engine, 
-				   size_t UNUSED num_n_parents, AXE_task_t UNUSED n_parents[], 
-				   size_t UNUSED num_s_parents, AXE_task_t UNUSED s_parents[], 
-				   void *_queue)
-{
-
-  compactor *cqueue = (compactor *)_queue;
-  int n_requests, i = 0;
-  op_data_t *op_data = NULL;
-  herr_t ret_value = SUCCEED;
-  request_list_t *wlist=NULL;
-  dataset_container_t *dlist=NULL;
-  int nentries = 0, ndatasets = 0;
-
-  FUNC_ENTER_NOAPI_NOINIT
-
-#if DEBUG_COMPACTOR
-    fprintf(stderr, "Enters Call BACK!\n");
-    fprintf (stderr, "Number of requests : %d from call back in queue : %p\n", 
-	     H5VL_iod_get_number_of_requests(cqueue), (void *)cqueue);
-    fflush(stderr);
-#endif
-
-    pthread_mutex_lock(&lock);
-    compactor_queue_flag = 0;
-    curr_queue = NULL;
-    pthread_mutex_unlock(&lock);  
-    
-
-
-    ret_value = H5VL_iod_create_request_list (cqueue,
-					      &wlist,
-					      &nentries,
-					      &dlist,
-					      &ndatasets,
-					      WRITE);
-
-#if DEBUG_COMPACTOR
-    if (ret_value != CP_SUCCESS){
-      fprintf(stderr,"ERROR !! Compactor create request list failed with error %d  \n",
-	      ret_value);
-    }
-#endif
-
-   
-#if DEBUG_COMPACTOR
-    fprintf(stderr, "=========================================================================\n");
-    fprintf(stderr,"nentires : %d############################################\n", nentries);
-    fprintf(stderr,"Compactor Request List \n");
-    fprintf(stderr,"id -- dataset -- dataspace -- fileblocks -- memblocks\n");
-    for ( i = 0; i < nentries; i++){
-      fprintf(stderr, "%d -- %d -- %d -- %zd -- %zd\n",
-	      wlist[i].request_id,
-	      wlist[i].dataset_id,
-	      wlist[i].selection_id,
-	      wlist[i].num_fblocks,
-	      wlist[i].num_mblocks);
-    }
-    fprintf(stderr, "=========================================================================\n");
-#endif
-
-#if 1
-    for ( i = 0; i < ndatasets; i ++){
-      H5VL_iod_compact_requests (wlist, &nentries,dlist[i].num_requests,
-				 dlist[i].requests);
-
-    }
-#endif
-
-    n_requests = H5VL_iod_get_number_of_requests(cqueue);
-    i = 0;
-    while ( i < n_requests){
-      compactor_entry t_entry;
-      /*      H5VL_iod_remove_request_from_compactor (cqueue, &t_entry);*/
-      H5VL_iod_get_request_at (cqueue, &t_entry, i);
-      if ( t_entry.type_request == WRITE){
-#if DEBUG_COMPACTOR
-	fprintf (stderr, "Request: %d is a WRITE request\n ", t_entry.request_id);
-	fprintf (stderr, "With op_data : %p\n ",
-		 (void *)t_entry.input_structure);
-	fflush(stderr);
-#endif
-	op_data = NULL;
-	op_data = t_entry.input_structure;
-	H5VL_iod_server_dset_write_cb (NULL,NULL,NULL,NULL,NULL,op_data);
-      }
-      else{
-	fprintf (stderr, "Request: %d is a READ request\n ", t_entry.request_id);
-	/*  Have to do the same for READ requests!*/
-      }
-      i++;
-    }
-    
-    if (NULL != wlist){
-      free(wlist);
-      wlist = NULL;
-    }
-    if (NULL != dlist){
-      free(dlist);
-      dlist = NULL;
-    }
-    if (CP_SUCCESS != H5VL_iod_destroy_compactor_queue(cqueue)){
-      HGOTO_ERROR(H5E_HEAP, H5E_NOSPACE, CP_FAIL, "Cannot free NULL queue\n");
-    }
-
+  }
+  
  done:
-    FUNC_LEAVE_NOAPI_VOID   
-} /*end H5VL_iod_server_dset_compactor_cb */
-
-
-
-
-
+  
+  FUNC_LEAVE_NOAPI(ret_value);
+  
+}/*H5VL_iod_server_compactor_write*/
+	
+	
 /*-------------------------------------------------------------------------
  * Function:	H5VL_iod_server_dset_write_cb
  *
diff --git a/src/H5VLiod_server.c b/src/H5VLiod_server.c
index 3ce68c7..c9d28e6 100644
--- a/src/H5VLiod_server.c
+++ b/src/H5VLiod_server.c
@@ -19,12 +19,12 @@
 #include "H5Sprivate.h"		/* Dataspaces		  		*/
 #include "H5WBprivate.h"        /* Wrapped Buffers                      */
 #include "H5VLiod_server.h"
-#include "H5VLiod_compactor_queue.h"
-#include "H5VLiod_compactor.h"             /* Compactor Routine       */  
-
 
 #ifdef H5_HAVE_EFF
 
+#include "H5VLiod_compactor_queue.h"
+#include "H5VLiod_compactor.h"
+
 /*
  * Programmer:  Mohamad Chaarawi <chaarawi@hdfgroup.gov>
  *              February, 2013
diff --git a/src/H5VLiod_server.h b/src/H5VLiod_server.h
index 50340d1..4407665 100644
--- a/src/H5VLiod_server.h
+++ b/src/H5VLiod_server.h
@@ -98,6 +98,8 @@ H5_DLL void H5VL_iod_server_dset_compactor_cb(AXE_engine_t axe_engine,
 					      size_t num_s_parents, AXE_task_t s_parents[], 
 					      void *queue);
 
+H5_DLL int H5VL_iod_server_compactor_write (void *list, int num_requests);
+
 H5_DLL int H5VL_iod_reconstruct_parents (AXE_engine_t axe_engine,
 					 axe_ids_t *old_parents,
 					 axe_ids_t **new_parents);