diff options
| -rw-r--r-- | examples/h5ff_client_dset.c | 18 | ||||
| -rw-r--r-- | examples/h5ff_client_prefetch.c | 2 | ||||
| -rw-r--r-- | src/H5VLiod.c | 5 | ||||
| -rw-r--r-- | src/H5VLiod_attr.c | 118 | ||||
| -rw-r--r-- | src/H5VLiod_client.c | 2 | ||||
| -rw-r--r-- | src/H5VLiod_dset.c | 120 | ||||
| -rw-r--r-- | src/H5VLiod_file.c | 4 | ||||
| -rw-r--r-- | src/H5VLiod_map.c | 158 | ||||
| -rw-r--r-- | src/H5VLiod_server.c | 3 | ||||
| -rw-r--r-- | src/H5VLiod_server.h | 1 | ||||
| -rw-r--r-- | src/H5VLiod_util.c | 132 |
11 files changed, 343 insertions, 220 deletions
diff --git a/examples/h5ff_client_dset.c b/examples/h5ff_client_dset.c index b5dfb89..15d0e00 100644 --- a/examples/h5ff_client_dset.c +++ b/examples/h5ff_client_dset.c @@ -52,7 +52,7 @@ int main(int argc, char **argv) { uint64_t version; uint64_t trans_num; - int32_t *wdata1 = NULL, *wdata2 = NULL; + int32_t *wdata1 = NULL; int16_t *wdata3 = NULL; int32_t *ex_wdata = NULL, *ex_rdata = NULL; int32_t *rdata1 = NULL, *rdata2 = NULL; @@ -96,17 +96,15 @@ int main(int argc, char **argv) { /* allocate and initialize arrays for dataset I/O */ wdata1 = malloc (sizeof(int32_t)*nelem); - wdata2 = malloc (sizeof(int32_t)*nelem); wdata3 = malloc (sizeof(int16_t)*nelem); rdata1 = malloc (sizeof(int32_t)*nelem); rdata2 = malloc (sizeof(int32_t)*nelem); - rdata3 = malloc (sizeof(int16_t)*nelem); + rdata3 = malloc (sizeof(int32_t)*nelem); for(i=0;i<nelem;++i) { rdata1[i] = 0; rdata2[i] = 0; rdata3[i] = 0; wdata1[i]=i; - wdata2[i]=i; wdata3[i]=i; } @@ -384,20 +382,19 @@ int main(int argc, char **argv) { H5Pset_dxpl_inject_corruption(dxpl_id, 1); /* Give a location to the DXPL to store the checksum once the read has completed */ H5Pset_dxpl_checksum_ptr(dxpl_id, &read2_cs); - ret = H5Dread_ff(did1, dtid, H5S_ALL, H5S_ALL, dxpl_id, rdata2, rid2, e_stack); assert(ret == 0); + H5Pclose(dxpl_id); - /* tell HDF5 to disable all data integrity checks for this write */ + dxpl_id = H5Pcreate (H5P_DATASET_XFER); + /* tell HDF5 to disable all data integrity checks for this read */ cs_scope = 0; ret = H5Pset_rawdata_integrity_scope(dxpl_id, cs_scope); assert(ret == 0); - /* Raw data read on D3. This is asynchronous. Note that the type is different than the dataset type. */ ret = H5Dread_ff(did3, H5T_STD_I16BE, H5S_ALL, H5S_ALL, dxpl_id, rdata3, rid2, e_stack); assert(ret == 0); - H5Pclose(dxpl_id); /* Raw data read on D1. This is asynchronous. The read is done into a @@ -541,7 +538,7 @@ int main(int argc, char **argv) { ret = H5Gclose_ff(gid1, e_stack); assert(ret == 0); - H5Fclose_ff(file_id, 0, H5_EVENT_STACK_NULL); + H5Fclose_ff(file_id, 1, H5_EVENT_STACK_NULL); H5ESget_count(e_stack, &num_events); @@ -587,7 +584,7 @@ int main(int argc, char **argv) { assert(read1_cs == array_cs); assert(read2_cs != array_cs); - printf("Read Data3 (32 LE converted to 16 bit BE order): "); + printf("Read Data3 with datatype conversion (expect to see same as others since it is converted twice): \n"); for(i=0;i<nelem;++i) printf("%d ",rdata3[i]); printf("\n"); @@ -596,7 +593,6 @@ int main(int argc, char **argv) { assert(ret == 0); free(wdata1); - free(wdata2); free(wdata3); free(rdata1); free(rdata2); diff --git a/examples/h5ff_client_prefetch.c b/examples/h5ff_client_prefetch.c index cc63552..0a1def9 100644 --- a/examples/h5ff_client_prefetch.c +++ b/examples/h5ff_client_prefetch.c @@ -280,7 +280,9 @@ int main(int argc, char **argv) { value = -1;
ret = H5Mget_ff(map, H5T_STD_I32LE, &key, H5T_STD_I32LE, &value,
dxpl_id, rid2, H5_EVENT_STACK_NULL);
+ assert(0 == ret);
printf("Value recieved = %d\n", value);
+ assert(1000 == value);
H5Pclose(dxpl_id);
diff --git a/src/H5VLiod.c b/src/H5VLiod.c index df44932..41fe2bc 100644 --- a/src/H5VLiod.c +++ b/src/H5VLiod.c @@ -740,7 +740,10 @@ EFF_init(MPI_Comm comm, MPI_Info UNUSED info) } /* Wait for it to compete */ - HG_Wait(hg_req, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE); + if(HG_FAIL == HG_Wait(hg_req, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) { + fprintf(stderr, "Failed to initialize Stack\n"); + return FAIL; + } /* Free Mercury request */ if(HG_Request_free(hg_req) != HG_SUCCESS) diff --git a/src/H5VLiod_attr.c b/src/H5VLiod_attr.c index bfe35ac..b431c13 100644 --- a/src/H5VLiod_attr.c +++ b/src/H5VLiod_attr.c @@ -514,20 +514,27 @@ H5VL_iod_server_attr_read_cb(AXE_engine_t UNUSED axe_engine, size = HG_Bulk_handle_get_size(bulk_handle); - if(NULL == (buf = malloc(size))) - HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); - - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &buf, &size, - HG_BULK_READWRITE, &bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); - -#if 0 - /* get mercury buffer where data is */ - if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, size, - HG_BULK_READ_ONLY, 1, &buf, &size, NULL)) - HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#endif + if(is_coresident) { + size_t bulk_size = 0; + + bulk_block_handle = bulk_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, size, + HG_BULK_READWRITE, 1, &buf, &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); + + assert(size == bulk_size); + } + else { + if(NULL == (buf = malloc(size))) + HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &buf, &size, + HG_BULK_READWRITE, &bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + } + /* Get dataspace if it is not available */ if(H5I_UNINIT == space_id) { /* open the metadata scratch pad of the attribute */ @@ -606,14 +613,16 @@ H5VL_iod_server_attr_read_cb(AXE_engine_t UNUSED axe_engine, HGOTO_ERROR_FF(ret, "can't read from array object"); } - /* Push data to the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, bulk_handle, 0, - bulk_block_handle, 0, size, &bulk_request)) - HGOTO_ERROR_FF(FAIL, "Transfer data failed"); + if(!is_coresident) { + /* Push data to the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, bulk_handle, 0, + bulk_block_handle, 0, size, &bulk_request)) + HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + } done: #if H5_EFF_DEBUG @@ -623,11 +632,14 @@ done: if(HG_SUCCESS != HG_Handler_start_output(op_data->hg_handle, &ret_value)) HDONE_ERROR_FF(FAIL, "can't send result of write to client"); - if(buf) { + if(!is_coresident) { /* free block handle */ if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); - buf = NULL; + if(buf) { + free(buf); + buf = NULL; + } } HG_Handler_free_input(op_data->hg_handle, input); @@ -686,7 +698,7 @@ H5VL_iod_server_attr_write_cb(AXE_engine_t UNUSED axe_engine, iod_handle_t mdkv_oh; hg_bulk_t bulk_block_handle; /* HG block handle */ hg_bulk_request_t bulk_request; /* HG request */ - iod_mem_desc_t *mem_desc; /* memory descriptor used for writing array */ + iod_mem_desc_t *mem_desc = NULL; /* memory descriptor used for writing array */ iod_array_iodesc_t file_desc; /* file descriptor used to write array */ iod_hyperslab_t hslabs; /* IOD hyperslab generated from HDF5 filespace */ size_t size; /* size of outgoing bulk data */ @@ -717,29 +729,35 @@ H5VL_iod_server_attr_write_cb(AXE_engine_t UNUSED axe_engine, /* Read bulk data here and wait for the data to be here */ size = HG_Bulk_handle_get_size(bulk_handle); - if(NULL == (buf = malloc(size))) - HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + if(is_coresident) { + size_t bulk_size = 0; - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &buf, &size, - HG_BULK_READWRITE, &bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + bulk_block_handle = bulk_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, size, + HG_BULK_READWRITE, 1, &buf, &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); - /* Pull data from the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PULL, source, bulk_handle, 0, - bulk_block_handle, 0, size, &bulk_request)) - HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); - -#if 0 - /* get mercury buffer where data is */ - if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, size, - HG_BULK_READWRITE, 1, &buf, &size, NULL)) - HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#endif + assert(size == bulk_size); + } + else { + if(NULL == (buf = malloc(size))) + HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &buf, &size, + HG_BULK_READWRITE, &bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + + /* Pull data from the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PULL, source, bulk_handle, 0, + bulk_block_handle, 0, size, &bulk_request)) + HGOTO_ERROR_FF(FAIL, "Transfer data failed"); + + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + } /* Get dataspace if it is not available */ if(H5I_UNINIT == space_id) { @@ -822,11 +840,15 @@ done: if(HG_SUCCESS != HG_Handler_start_output(op_data->hg_handle, &ret_value)) HDONE_ERROR_FF(FAIL, "can't send result of write to client"); - if(buf) { + if(!is_coresident) { /* free block handle */ if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); - buf = NULL; + + if(buf) { + free(buf); + buf = NULL; + } } HG_Handler_free_input(op_data->hg_handle, input); @@ -834,8 +856,6 @@ done: input = (attr_io_in_t *)H5MM_xfree(input); op_data = (op_data_t *)H5MM_xfree(op_data); - free(buf); - /* free allocated descriptors */ free(hslabs.start); free(hslabs.stride); diff --git a/src/H5VLiod_client.c b/src/H5VLiod_client.c index be8ae9a..fce72c1 100644 --- a/src/H5VLiod_client.c +++ b/src/H5VLiod_client.c @@ -2871,7 +2871,7 @@ H5VL_iod_map_dtype_info(hid_t type_id, /*out*/ hbool_t *is_vl, /*out*/size_t *si done: FUNC_LEAVE_NOAPI(ret_value) -} /* end H5VL_iod_map_get_size */ +} /* end H5VL_iod_map_dtype_info */ #if 0 static herr_t diff --git a/src/H5VLiod_dset.c b/src/H5VLiod_dset.c index 8c00039..8942f0a 100644 --- a/src/H5VLiod_dset.c +++ b/src/H5VLiod_dset.c @@ -587,6 +587,10 @@ H5VL_iod_server_dset_read_cb(AXE_engine_t axe_engine, hbool_t opened_locally = FALSE; /* flag to indicate whether we opened the dset here or if it was already open */ herr_t ret_value = SUCCEED; + /* MSC - for now do memcpy if segment count > 1 */ + if(is_coresident && 1 != HG_Bulk_handle_get_segment_count(bulk_handle)) + is_coresident = NA_FALSE; + /* open the dataset if we don't have the handle yet */ if(iod_oh.rd_oh.cookie == IOD_OH_UNDEFINED) { ret = iod_obj_open_read(coh, iod_id, rtid, NULL, &iod_oh.rd_oh, NULL); @@ -610,14 +614,27 @@ H5VL_iod_server_dset_read_cb(AXE_engine_t axe_engine, /* retrieve size of bulk data asked for to be read */ size = HG_Bulk_handle_get_size(bulk_handle); - /* allocate buffer to hold data */ - if(NULL == (buf = malloc(size))) - HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + if(is_coresident) { + size_t bulk_size = 0; + + bulk_block_handle = bulk_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, size, + HG_BULK_READWRITE, 1, &buf, &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &buf, &size, - HG_BULK_READWRITE, &bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + assert(size == bulk_size); + } + else { + /* allocate buffer to hold data */ + if(NULL == (buf = malloc(size))) + HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &buf, &size, + HG_BULK_READWRITE, &bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + } /* get the number of points selected */ nelmts = (size_t)H5Sget_select_npoints(space_id); @@ -625,14 +642,20 @@ H5VL_iod_server_dset_read_cb(AXE_engine_t axe_engine, /* Adjust buffer is type conversion is needed. If the data elements are of variable length, just return that they are in is_vl_data for special processing */ + if(H5VL__iod_server_type_is_vl(src_id, &is_vl_data) < 0) + HGOTO_ERROR_FF(FAIL, "failed to check dataype"); + + /* if(H5VL__iod_server_adjust_buffer(dst_id, src_id, nelmts, dxpl_id, is_coresident, size, &buf, &is_vl_data, &buf_size) < 0) - HGOTO_ERROR_FF(FAIL, "failed to setup write operation"); - + HGOTO_ERROR_FF(FAIL, "failed to setup read operation"); + */ if(!is_vl_data) { size_t elmt_size; iod_trans_id_t read_tid; + buf_size = H5Tget_size(src_id) * nelmts; + /* get replica ID from dxpl */ if(H5Pget_read_replica(dxpl_id, &read_tid) < 0) HGOTO_ERROR_FF(FAIL, "can't get replica ID from dxpl"); @@ -684,18 +707,20 @@ H5VL_iod_server_dset_read_cb(AXE_engine_t axe_engine, HGOTO_ERROR_FF(ret, "can't read from array object"); } - /* Push data to the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, bulk_handle, 0, - bulk_block_handle, 0, size, &bulk_request)) - HGOTO_ERROR_FF(FAIL, "Transfer data failed"); + if(!is_coresident) { + /* Push data to the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, bulk_handle, 0, + bulk_block_handle, 0, size, &bulk_request)) + HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); - /* free block handle */ - if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); + /* free block handle */ + if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); + } #if H5_EFF_DEBUG fprintf(stderr, "Done with dset read, checksum %016lX, sending response to client\n", cs); @@ -717,7 +742,7 @@ done: input = (dset_io_in_t *)H5MM_xfree(input); op_data = (op_data_t *)H5MM_xfree(op_data); - if(buf) { + if(!is_coresident && buf) { free(buf); buf = NULL; } @@ -808,14 +833,27 @@ H5VL_iod_server_dset_get_vl_size_cb(AXE_engine_t UNUSED axe_engine, unsigned u; H5VL_iod_type_info_t type_info; - /* allocate buffer to hold data */ - if(NULL == (temp_buf = malloc(buf_size))) - HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + if(is_coresident) { + size_t bulk_size = 0; - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &temp_buf, &buf_size, - HG_BULK_READWRITE, &bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + bulk_block_handle = bulk_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, buf_size, + HG_BULK_READWRITE, 1, &temp_buf, &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); + + assert(buf_size == bulk_size); + } + else { + /* allocate buffer to hold data */ + if(NULL == (temp_buf = malloc(buf_size))) + HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &temp_buf, &buf_size, + HG_BULK_READWRITE, &bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + } buf_ptr = (uint8_t *)buf; temp_ptr = (uint8_t *)temp_buf; @@ -850,22 +888,24 @@ H5VL_iod_server_dset_get_vl_size_cb(AXE_engine_t UNUSED axe_engine, H5VL_iod_type_info_reset(&type_info); - /* Push data to the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, bulk_handle, 0, - bulk_block_handle, 0, buf_size, &bulk_request)) - HGOTO_ERROR_FF(FAIL, "Transfer data failed"); + if(!is_coresident) { + /* Push data to the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, bulk_handle, 0, + bulk_block_handle, 0, buf_size, &bulk_request)) + HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); - /* free block handle */ - if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); + /* free block handle */ + if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); - if(temp_buf) { - free(temp_buf); - temp_buf = NULL; + if(temp_buf) { + free(temp_buf); + temp_buf = NULL; + } } } diff --git a/src/H5VLiod_file.c b/src/H5VLiod_file.c index 2349e36..ae4ee8e 100644 --- a/src/H5VLiod_file.c +++ b/src/H5VLiod_file.c @@ -289,6 +289,10 @@ done: con_open_hint = NULL; } + if(obj_create_hint) { + free(obj_create_hint); + obj_create_hint = NULL; + } } /* end H5VL_iod_server_file_create_cb() */ diff --git a/src/H5VLiod_map.c b/src/H5VLiod_map.c index 5f6256a..d957f06 100644 --- a/src/H5VLiod_map.c +++ b/src/H5VLiod_map.c @@ -483,30 +483,36 @@ H5VL_iod_server_map_set_cb(AXE_engine_t UNUSED axe_engine, /* retrieve size of incoming bulk data */ val_size = HG_Bulk_handle_get_size(value_handle); - /* allocate buffer to hold data */ - if(NULL == (val_buf = malloc(val_size))) - HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + if(is_coresident) { + size_t bulk_size = 0; - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &val_buf, &val_size, - HG_BULK_READWRITE, &bulk_block_handle)) + bulk_block_handle = value_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, val_size, + HG_BULK_READWRITE, 1, &val_buf, &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); + + assert(val_size == bulk_size); + } + else { + /* allocate buffer to hold data */ + if(NULL == (val_buf = malloc(val_size))) + HGOTO_ERROR_FF(FAIL, "can't allocate read buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &val_buf, &val_size, + HG_BULK_READWRITE, &bulk_block_handle)) HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); - /* Pull data from the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PULL, source, value_handle, 0, - bulk_block_handle, 0, val_size, &bulk_request)) + /* Pull data from the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PULL, source, value_handle, 0, + bulk_block_handle, 0, val_size, &bulk_request)) HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); - -#if 0 - /* get mercury buffer where data is */ - if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, val_size, - HG_BULK_READWRITE, 1, &val_buf, &val_size, NULL)) - HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#endif + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + } /* get the scope for data integrity checks for raw data */ if(H5Pget_rawdata_integrity_scope(dxpl_id, &raw_cs_scope) < 0) @@ -605,11 +611,15 @@ done: input = (map_set_in_t *)H5MM_xfree(input); op_data = (op_data_t *)H5MM_xfree(op_data); - if(val_buf) { + if(!is_coresident) { /* free block handle */ if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); - val_buf = NULL; + + if(val_buf) { + free(val_buf); + val_buf = NULL; + } } /* close the map if we opened it in this routine */ @@ -728,20 +738,27 @@ H5VL_iod_server_map_get_cb(AXE_engine_t UNUSED axe_engine, fprintf(stderr, "val size = %zu\n", src_size); #endif if(client_val_buf_size) { - if(NULL == (val_buf = malloc((size_t)src_size))) - HGOTO_ERROR_FF(FAIL, "can't allocate buffer"); + if(is_coresident) { + size_t bulk_size = 0; - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &val_buf, &src_size, - HG_BULK_READWRITE, &bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + bulk_block_handle = value_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, src_size, + HG_BULK_READWRITE, 1, &val_buf, + &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#if 0 - /* get mercury buffer where data is */ - if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, src_size, - HG_BULK_READ_ONLY, 1, &val_buf, &src_size, NULL)) - HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#endif + assert(src_size == bulk_size); + } + else { + if(NULL == (val_buf = malloc((size_t)src_size))) + HGOTO_ERROR_FF(FAIL, "can't allocate buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &val_buf, &src_size, + HG_BULK_READWRITE, &bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + } ret = iod_kv_get_value(iod_oh, rtid, key.buf, (iod_size_t)key.buf_size, val_buf, &src_size, kv_cs, NULL); @@ -766,34 +783,43 @@ H5VL_iod_server_map_get_cb(AXE_engine_t UNUSED axe_engine, } #endif - /* Push data to the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, value_handle, 0, - bulk_block_handle, 0, src_size, &bulk_request)) - HGOTO_ERROR_FF(FAIL, "Transfer data failed"); + if(!is_coresident) { + /* Push data to the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, value_handle, 0, + bulk_block_handle, 0, src_size, &bulk_request)) + HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + } } } else { /* retrieve size of bulk data asked for to be read */ src_size = HG_Bulk_handle_get_size(value_handle); - if(NULL == (val_buf = malloc((size_t)src_size))) - HGOTO_ERROR_FF(FAIL, "can't allocate buffer"); + if(is_coresident) { + size_t bulk_size = 0; - /* Create bulk handle */ - if(HG_SUCCESS != HG_Bulk_handle_create(1, &val_buf, &src_size, - HG_BULK_READWRITE, &bulk_block_handle)) - HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + bulk_block_handle = value_handle; + /* get mercury buffer where data is */ + if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, src_size, + HG_BULK_READWRITE, 1, &val_buf, + &bulk_size, NULL)) + HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#if 0 - /* get mercury buffer where data is */ - if(HG_SUCCESS != HG_Bulk_handle_access(bulk_block_handle, 0, src_size, - HG_BULK_READ_ONLY, 1, &val_buf, &src_size, NULL)) - HGOTO_ERROR_FF(FAIL, "Could not access handle"); -#endif + assert(src_size == bulk_size); + } + else { + if(NULL == (val_buf = malloc((size_t)src_size))) + HGOTO_ERROR_FF(FAIL, "can't allocate buffer"); + + /* Create bulk handle */ + if(HG_SUCCESS != HG_Bulk_handle_create(1, &val_buf, &src_size, + HG_BULK_READWRITE, &bulk_block_handle)) + HGOTO_ERROR_FF(FAIL, "can't create bulk handle"); + } ret = iod_kv_get_value(iod_oh, rtid, key.buf, (iod_size_t)key.buf_size, val_buf, &src_size, kv_cs, NULL); @@ -813,10 +839,7 @@ H5VL_iod_server_map_get_cb(AXE_engine_t UNUSED axe_engine, output.val_cs = kv_cs[1]; } - /* adjust buffers for datatype conversion */ - if(H5VL__iod_server_adjust_buffer(val_memtype_id, val_maptype_id, 1, dxpl_id, is_coresident, - (size_t)src_size, &val_buf, &val_is_vl, &val_size) < 0) - HGOTO_ERROR_FF(FAIL, "data type conversion failed"); + val_size = H5Tget_size(val_maptype_id); /* do data conversion */ if(H5Tconvert(val_maptype_id, val_memtype_id, 1, val_buf, NULL, dxpl_id) < 0) @@ -835,14 +858,16 @@ H5VL_iod_server_map_get_cb(AXE_engine_t UNUSED axe_engine, output.val_size = val_size; output.ret = ret_value; - /* Push data to the client */ - if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, value_handle, 0, - bulk_block_handle, 0, src_size, &bulk_request)) - HGOTO_ERROR_FF(FAIL, "Transfer data failed"); + if(!is_coresident) { + /* Push data to the client */ + if(HG_SUCCESS != HG_Bulk_transfer(HG_BULK_PUSH, dest, value_handle, 0, + bulk_block_handle, 0, src_size, &bulk_request)) + HGOTO_ERROR_FF(FAIL, "Transfer data failed"); - /* Wait for bulk data read to complete */ - if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) - HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + /* Wait for bulk data read to complete */ + if(HG_SUCCESS != HG_Bulk_wait(bulk_request, HG_MAX_IDLE_TIME, HG_STATUS_IGNORE)) + HGOTO_ERROR_FF(FAIL, "can't wait for bulk data operation"); + } } #if H5_EFF_DEBUG @@ -862,11 +887,14 @@ done: HDONE_ERROR_FF(FAIL, "can't send result of map get"); } - if(val_buf) { + if(!is_coresident) { /* free block handle */ if(HG_SUCCESS != HG_Bulk_handle_free(bulk_block_handle)) HGOTO_ERROR_FF(FAIL, "can't free bds block handle"); - val_buf = NULL; + if(val_buf) { + free(val_buf); + val_buf = NULL; + } } HG_Handler_free_input(op_data->hg_handle, input); diff --git a/src/H5VLiod_server.c b/src/H5VLiod_server.c index ce6a84b..ba88d6a 100644 --- a/src/H5VLiod_server.c +++ b/src/H5VLiod_server.c @@ -315,6 +315,8 @@ H5VL_iod_server_eff_init(hg_handle_t handle) done: HG_Handler_start_output(handle, &ret_value); + HG_Handler_free(handle); + return ret_value; } /* end H5VL_iod_server_eff_init() */ @@ -352,6 +354,7 @@ H5VL_iod_server_eff_finalize(hg_handle_t handle) done: HG_Handler_start_output(handle, &ret_value); + HG_Handler_free(handle); return ret_value; } /* end H5VL_iod_server_eff_finalize() */ diff --git a/src/H5VLiod_server.h b/src/H5VLiod_server.h index afadffb..b3ca6d0 100644 --- a/src/H5VLiod_server.h +++ b/src/H5VLiod_server.h @@ -407,6 +407,7 @@ H5_DLL herr_t H5VL__iod_server_adjust_buffer(hid_t from_type_id, hid_t to_type_i hid_t dxpl_id, na_bool_t is_coresident, size_t size, void **buf, hbool_t *is_vl_data, size_t *_buf_size); +H5_DLL herr_t H5VL__iod_server_type_is_vl(hid_t type_id, hbool_t *is_vl_data); H5_DLL herr_t H5VL_iod_verify_scratch_pad(scratch_pad *sp, iod_checksum_t iod_cs); H5_DLL 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); diff --git a/src/H5VLiod_util.c b/src/H5VLiod_util.c index b5060c2..29b4e3b 100644 --- a/src/H5VLiod_util.c +++ b/src/H5VLiod_util.c @@ -1146,7 +1146,82 @@ H5VL__iod_server_adjust_buffer(hid_t mem_type_id, hid_t dset_type_id, size_t nel { herr_t ret_value = SUCCEED; - switch(H5Tget_class(dset_type_id)) { + 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: @@ -1155,62 +1230,13 @@ H5VL__iod_server_adjust_buffer(hid_t mem_type_id, hid_t dset_type_id, size_t nel case H5T_ENUM: case H5T_ARRAY: case H5T_NO_CLASS: - case H5T_STRING: - if(H5Tis_variable_str(dset_type_id)) { - *is_vl_data = TRUE; - *_buf_size = size; - break; - } case H5T_REFERENCE: case H5T_NCLASSES: case H5T_COMPOUND: - { - hsize_t buf_size = 0; - size_t mem_type_size, dset_type_size; - - *is_vl_data = FALSE; - - /* 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, *_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; - - break; - } + *is_vl_data = FALSE; + break; case H5T_VLEN: *is_vl_data = TRUE; - *_buf_size = size; break; default: HGOTO_ERROR_FF(FAIL, "unsupported datatype"); @@ -1218,7 +1244,7 @@ H5VL__iod_server_adjust_buffer(hid_t mem_type_id, hid_t dset_type_id, size_t nel done: return ret_value; -} /* end H5VL__iod_server_adjust_buffer */ +} /* end H5VL__iod_server_type_is_vl */ /*------------------------------------------------------------------------- |