[svn-r23963] Working version -- Short-ciruiting reads with bug fixes.

2 files changed, 105 insertions, 48 deletions

diff --git a/src/H5VLiod_compactor.c b/src/H5VLiod_compactor.c
index cfefb7a..a50417f 100644
--- a/src/H5VLiod_compactor.c
+++ b/src/H5VLiod_compactor.c
@@ -94,7 +94,7 @@ static int H5VL_check_overlapped_offsets(hsize_t start_i, hsize_t start_j,
 
 static int H5VL_get_read_spread (hsize_t start_offset, hsize_t end_offset,
 				 hsize_t *start_offsets, hsize_t *end_offsets,
-				 int **requests, int *nreqs);
+				 int nentries, int *requests, int *nreqs);
 
 /*---------------------------------------------------------------------*/
 
@@ -188,10 +188,8 @@ int H5VL_iod_create_request_list (compactor *queue, request_list_t **list,
   int j = 0, request_id = 0, i, current_dset_id = 0, lreq = 0, flag = 0;
   size_t ii;
   uint64_t axe_id;
-
-#if H5_DO_NATIVE
   char dname[256], dname1[256];
-#endif
+
   
 
 
@@ -559,33 +557,16 @@ int H5VL_iod_create_request_list (compactor *queue, request_list_t **list,
 static
 int H5VL_get_read_spread (hsize_t start_offset, hsize_t end_offset,
 			  hsize_t *start_offsets, hsize_t *end_offsets,
-			  int **requests, int *nreqs){
+			  int nentries, int *request_list, int *nreqs){
   
   int ret_value = CP_SUCCESS;
-  int i, reqs, *request_list;
-  hsize_t tmp_offset = 0;
-
+  int i, reqs = 0;
   
-  FUNC_ENTER_NOAPI(NULL);
+
   
-  reqs = 0;
-  request_list = (int *) malloc (nentries * sizeof(int));
-  if (NULL == request_list){
-    HGOTO_ERROR(H5E_HEAP, H5E_CANTALLOC, CP_FAIL,"Memory allocation error for request_list");
-  }
   
-  start_offsets = (hsize_t *) malloc (nentries * sizeof(hsize_t));
-  if (NULL == start_offsets){
-    HGOTO_ERROR(H5E_HEAP, H5E_CANTALLOC, CP_FAIL,"Memory allocation error for start_offsets");
-  }
-
-  end_offsets = (hsize_t *) malloc (nentries * sizeof(hsize_t));
-  if (NULL == end_offsets){
-    HGOTO_ERROR(H5E_HEAP, H5E_CANTALLOC, CP_FAIL,"Memory allocation error for start_offsets");
-  }
-
-
   for ( i = 0; i < nentries; i++){
+
     if ((start_offset >= start_offsets[i]) &&
 	(end_offset <= end_offsets[i])){
       request_list[reqs] = i;
@@ -604,9 +585,12 @@ int H5VL_get_read_spread (hsize_t start_offset, hsize_t end_offset,
     }
   }
   
-  if (start_offset != end_offset){
+#if DEBUG_COMPACTOR
+  fprintf (stderr, "Rstart: %lli, Rend: %lli\n",
+	   start_offset, end_offset);
+#endif
+  if (start_offset == end_offset){
     ret_value = 1;
-    *requests = request_list;
     *nreqs = reqs;
   }
   else{
@@ -615,9 +599,7 @@ int H5VL_get_read_spread (hsize_t start_offset, hsize_t end_offset,
   
   
   
- done:
-  FUNC_LEAVE_NOAPI(ret_value);
-  
+  return ret_value;
 }
 
 
@@ -641,32 +623,90 @@ int  H5VL_iod_short_circuit_reads (request_list_t *wlist, int nentries,
 
   int i, j,  nreqs = 0;
   hsize_t curr_start_offset, curr_end_offset;
-  int *requests = NULL;
-
+  hsize_t *start_offsets = NULL, *end_offsets = NULL;
+  int *request_list = NULL;
+  int ret_value = CP_SUCCESS;
+  
+  FUNC_ENTER_NOAPI(NULL)
+    
+  start_offsets = (hsize_t *) malloc (nentries * sizeof(hsize_t));
+  if (NULL == start_offsets){
+    HGOTO_ERROR(H5E_HEAP, H5E_CANTALLOC, CP_FAIL,"Memory allocation error for start_offsets");
+  }
+  
+  end_offsets = (hsize_t *) malloc (nentries * sizeof(hsize_t));
+  if (NULL == end_offsets){
+    HGOTO_ERROR(H5E_HEAP, H5E_CANTALLOC, CP_FAIL,"Memory allocation error for start_offsets");
+  }
 
+  for (i = 0; i < nentries; i++){
+    start_offsets[i] = wlist[i].fblocks[0].offset;
+    end_offsets[i] = wlist[i].fblocks[wlist[i].num_fblocks - 1].len +
+      wlist[i].fblocks[wlist[i].num_fblocks - 1].len - 1;
+    
+#if DEBUG_COMPACTOR
+    fprintf (stderr, "start : %lli, end: %lli \n", 
+	     start_offsets[i], end_offsets[i]);
+#endif
+  }
 
+  
 
   for (i = 0; i < nrentries; i++){
+
+
+    request_list = (int *) malloc (nentries * sizeof(int));
+    if (NULL == request_list){
+      HGOTO_ERROR(H5E_HEAP, H5E_CANTALLOC, CP_FAIL,"Memory allocation error for request_list");
+    }
+    
+
     
-    curr_start_offset = rlist[0].fblocks[0].offset;
-    curr_end_offset = rlist[num_fblocks - 1].offset + 
-      rlist[num_fblocks - 1].len - 1;
+    curr_start_offset = rlist[i].fblocks[0].offset;
+    curr_end_offset = rlist[i].fblocks[rlist[i].num_fblocks - 1].offset + 
+      rlist[i].fblocks[rlist[i].num_fblocks - 1].len - 1;
+    
+#if DEBUG_COMPACTOR
+    fprintf (stderr,"curr_start: %lli, curr_end: %lli\n", 
+	     curr_start_offset,
+	     curr_end_offset);
+#endif
 
     if (H5VL_get_read_spread (curr_start_offset,curr_end_offset,
 			      start_offsets, end_offsets,
-			      &requests, &nreqs) ){
-     
+			      nentries, request_list, &nreqs) ){
+      
+#if DEBUG_COMPACTOR
+      fprintf (stderr, "nreqs: %d\n",
+	       nreqs);
+#endif
       for (j = 0; j < nreqs; j++){
+	
 	memcpy(rlist[i].mem_buf,
-	       wlist[request[j]].mem_buf,
-	       wlist[request[j]].mem_len);
+	       wlist[request_list[j]].mem_buf,
+	       wlist[request_list[j]].mem_length);
+
       }
       rlist[i].merged = SS;
     }
+    
+    if (NULL != request_list){
+      free(request_list);
+      request_list = NULL;
+    }
+  }
 
+  if (NULL != start_offsets){
+    free(start_offsets);
+    start_offsets = NULL;
+  }
+  if (NULL != end_offsets){
+    free(end_offsets);
+    end_offsets = NULL;
   }
-  
 
+ done:
+  FUNC_LEAVE_NOAPI(ret_value);
 
 }
 
diff --git a/src/H5VLiod_dset.c b/src/H5VLiod_dset.c
index 67068c2..4db26fe 100644
--- a/src/H5VLiod_dset.c
+++ b/src/H5VLiod_dset.c
@@ -708,8 +708,10 @@ done:
      
 
       for ( i = 0; i < ndatasets; i ++){
+
 	H5VL_iod_compact_requests (wlist, &nentries,dlist[i].num_requests,
 				  dlist[i].requests);    
+
 	if (CP_SUCCESS != H5VL_iod_server_compactor_write (wlist, nentries)){
 #if DEBUG_COMPACTOR
 	  fprintf (stderr,"COMPACTOR CB: compactor write failed \n");
@@ -718,6 +720,10 @@ done:
 	}	
       }
       
+      H5VL_iod_short_circuit_reads(wlist, nentries,
+				   rlist, nrentries);
+      
+
       for ( i = 0; i < nrdatasets; i++){
 
 	if (CP_SUCCESS != H5VL_iod_server_compactor_read (rlist, nrentries)){
@@ -813,7 +819,7 @@ int H5VL_iod_server_compactor_read (void *_list, int num_requests)
   uint32_t cs = 0; 
   na_addr_t dest;
   hbool_t flag;
- 
+  hid_t dxpl_id ;
 
 
   FUNC_ENTER_NOAPI_NOINIT
@@ -844,7 +850,7 @@ int H5VL_iod_server_compactor_read (void *_list, int num_requests)
     dst_size = H5Tget_size(dst_id);
     src_size = H5Tget_size(src_id);
     buf = list[request_counter].mem_buf;
-    
+    dxpl_id = input->dxpl_id;
     /* open the dataset if we don't have the handle yet */
     if(iod_oh.cookie == (int)IOD_OH_UNDEFINED) {
         if (iod_obj_open_write(coh, iod_id, NULL /*hints*/, &iod_oh, NULL) < 0)
@@ -857,7 +863,7 @@ int H5VL_iod_server_compactor_read (void *_list, int num_requests)
     
     /* get the number of points selected */
     nelmts = (size_t)H5Sget_select_npoints(space_id);
-    
+    flag = FALSE;
     if (list[request_counter].merged != SS){
 
       /* get the rank of the dataspace */
@@ -949,20 +955,31 @@ int H5VL_iod_server_compactor_read (void *_list, int num_requests)
 	  HGOTO_ERROR(H5E_SYM, H5E_READERROR, FAIL, "can't read from array object");
       }
       
-      flag = FALSE;
-      ptr = (int *)buf;
+
+
 #if H5_DO_NATIVE
-      ret_value = H5Dread(iod_oh.cookie, src_id, H5S_ALL, space_id, H5P_DEFAULT, buf);
+      ret_value = H5Dread(iod_oh.cookie, src_id, H5S_ALL, space_id, dxpl_id, buf);
 #else /* fake data */
       for(i=0;i<64;++i)
 	ptr[i] = i;
 #endif
+    }
+    /* do data conversion */
+    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);
-    
     dest = HG_Handler_get_addr (op_data->hg_handle);
     
     /* Write bulk data here and wait for the data to be there  */