[svn-r25306] view and analysis shipping fixes and optimizations.

6 files changed, 199 insertions, 43 deletions

diff --git a/examples/h5ff_client_analysis.c b/examples/h5ff_client_analysis.c
index 3b033c4..daea3c6 100644
--- a/examples/h5ff_client_analysis.c
+++ b/examples/h5ff_client_analysis.c
@@ -12,7 +12,7 @@
 
 /* #define USE_NATIVE */
 
-#define NTUPLES 64
+#define NTUPLES 512
 
 static int my_rank = 0, my_size = 1;
 
diff --git a/src/H5FF.c b/src/H5FF.c
index c597d40..00c16ba 100644
--- a/src/H5FF.c
+++ b/src/H5FF.c
@@ -49,6 +49,7 @@
 #include "H5Iprivate.h"		/* IDs			  		*/
 #include "H5MMprivate.h"	/* Memory management			*/
 #include "H5Pprivate.h"		/* Property lists			*/
+#include "H5Qprivate.h"		/* Query        			*/
 #include "H5Tpkg.h"             /* Datatype access			*/
 #include "H5VLprivate.h"	/* VOL plugins				*/
 #include "H5VLiod.h"		/* IOD plugin - tmp      		*/
@@ -70,7 +71,8 @@
 /********************/
 /* Local Prototypes */
 /********************/
-
+static herr_t H5D__apply_index_query(void *idx_handle, H5X_class_t *idx_class, 
+                                     hid_t query_id, hid_t xxpl_id, hid_t *space_id);
 
 /*********************/
 /* Package Variables */
@@ -5021,7 +5023,6 @@ hid_t
 H5Dquery_ff(hid_t dset_id, hid_t query_id, hid_t scope_id, hid_t rcxt_id)
 {
     void *dset = NULL;
-    H5P_genplist_t *xxpl_plist = NULL, *plist = NULL; /* Property list pointer */
     hid_t xxpl_id = FAIL;
     unsigned plugin_id;
     void *buf = NULL;
@@ -5040,6 +5041,7 @@ H5Dquery_ff(hid_t dset_id, hid_t query_id, hid_t scope_id, hid_t rcxt_id)
     if (H5X_PLUGIN_NONE != (plugin_id = H5VL_iod_dataset_get_index_plugin_id(dset))) {
         void *idx_handle = NULL; /* index */
         H5X_class_t *idx_class = NULL;
+        H5P_genplist_t *xxpl_plist = NULL, *plist = NULL; /* Property list pointer */
 
         if (NULL == (idx_handle = H5VL_iod_dataset_get_index(dset)))
             HGOTO_ERROR(H5E_INDEX, H5E_CANTGET, FAIL, "can't get index handle from dataset");
@@ -5059,7 +5061,8 @@ H5Dquery_ff(hid_t dset_id, hid_t query_id, hid_t scope_id, hid_t rcxt_id)
 
         if (NULL == idx_class->query)
             HGOTO_ERROR(H5E_INDEX, H5E_BADVALUE, FAIL, "plugin query callback is not defined");
-        if (FAIL == idx_class->query(idx_handle, query_id, xxpl_id, &ret_value))
+
+        if(FAIL == H5D__apply_index_query(idx_handle, idx_class, query_id, xxpl_id, &ret_value))
             HGOTO_ERROR(H5E_INDEX, H5E_CANTCLOSEOBJ, FAIL, "cannot query index");
     }
     else 
@@ -5097,7 +5100,7 @@ H5Dquery_ff(hid_t dset_id, hid_t query_id, hid_t scope_id, hid_t rcxt_id)
             HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "can't read data from dataset");
 
         if(FAIL == (udata.space_query = H5Scopy(space_id)))
-            HGOTO_ERROR(H5E_DATASPACE, H5E_CANTCOPY, NULL, "can't copy dataspace")
+            HGOTO_ERROR(H5E_DATASPACE, H5E_CANTCOPY, FAIL, "can't copy dataspace")
         if(H5Sselect_none(udata.space_query) < 0)
             HGOTO_ERROR(H5E_DATASPACE, H5E_CANTDELETE, FAIL, "can't change selection")
 
@@ -5130,4 +5133,62 @@ done:
 
     FUNC_LEAVE_API(ret_value)
 }
+
+static herr_t
+H5D__apply_index_query(void *idx_handle, H5X_class_t *idx_class, hid_t query_id, 
+                       hid_t xxpl_id, hid_t *space_id)
+{
+    H5Q_combine_op_t comb_type;
+    H5Q_type_t q_type;
+    herr_t ret;
+    herr_t ret_value = SUCCEED;
+
+    FUNC_ENTER_STATIC
+
+    if(H5Qget_combine_op(query_id, &comb_type) < 0)
+        HGOTO_ERROR_FF(FAIL, "can't get query op type");
+
+    if(H5Qget_type(query_id, &q_type) < 0)
+        HGOTO_ERROR_FF(FAIL, "can't get query op type");
+
+    if(H5Q_SINGLETON == comb_type) {
+        if (FAIL == idx_class->query(idx_handle, query_id, xxpl_id, space_id))
+            HGOTO_ERROR(H5E_INDEX, H5E_CANTCLOSEOBJ, FAIL, "cannot query index");
+    }
+    else {
+        hid_t qid1, qid2;
+        hid_t sid1 = FAIL , sid2 = FAIL;
+
+        if(H5Qget_components(query_id, &qid1, &qid2) < 0)
+            HGOTO_ERROR_FF(FAIL, "can't get query components");
+
+        ret = H5D__apply_index_query(idx_handle, idx_class, qid1, xxpl_id, &sid1);
+        if(ret != 0)
+            HGOTO_ERROR_FF(ret, "Error applying index query");
+
+        ret = H5D__apply_index_query(idx_handle, idx_class, qid2, xxpl_id, &sid2);
+        if(ret != 0)
+            HGOTO_ERROR_FF(ret, "Error applying index query");
+
+        /* combine result1 and result2 */
+        if(H5Q_COMBINE_AND == comb_type) {
+            if(FAIL == (*space_id = H5Scombine_select(sid1, H5S_SELECT_AND, sid2)))
+                HGOTO_ERROR_FF(ret, "Unable to AND 2 dataspace selections");
+        }
+        else if(H5Q_COMBINE_OR == comb_type) {
+            if(FAIL == (*space_id = H5Scombine_select(sid1, H5S_SELECT_OR, sid2)))
+                HGOTO_ERROR_FF(ret, "Unable to AND 2 dataspace selections");
+        }
+        else
+            HGOTO_ERROR_FF(FAIL, "invalid query combine OP");
+
+        if(sid1!=FAIL && H5Sclose(sid1) < 0)
+            HGOTO_ERROR_FF(FAIL, "unable to release dataspace");
+        if(sid2!=FAIL && H5Sclose(sid2) < 0)
+            HGOTO_ERROR_FF(FAIL, "unable to release dataspace");
+    }
+
+done:
+    FUNC_LEAVE_NOAPI(ret_value)
+}
 #endif /* H5_HAVE_EFF */
diff --git a/src/H5Q.c b/src/H5Q.c
index 58b8718..e21e62b 100644
--- a/src/H5Q.c
+++ b/src/H5Q.c
@@ -619,6 +619,8 @@ H5Qget_type(hid_t query_id, H5Q_type_t *query_type)
     /* Check args and get the query objects */
     if (NULL == (query = (H5Q_t *) H5I_object_verify(query_id, H5I_QUERY)))
         HGOTO_ERROR(H5E_ARGS, H5E_BADTYPE, FAIL, "not a query ID");
+    if (!query_type)
+        HGOTO_ERROR(H5E_ARGS, H5E_BADVALUE, FAIL, "NULL pointer to query type");
 
     /* Get match info */
     if (FAIL == H5Q_get_type(query, query_type))
@@ -645,10 +647,10 @@ H5Q_get_type(H5Q_t *query, H5Q_type_t *query_type)
     FUNC_ENTER_NOAPI_NOINIT_NOERR
 
     HDassert(query);
+    HDassert(query_type);
 
-    if (query_type)
-        *query_type = (query->is_combined) ? query->query.combine.type :
-                        query->query.select.type;
+    *query_type = (query->is_combined) ? query->query.combine.type :
+            query->query.select.type;
 
     FUNC_LEAVE_NOAPI(ret_value)
 } /* end H5Q_get_type() */
@@ -680,6 +682,8 @@ H5Qget_match_op(hid_t query_id, H5Q_match_op_t *match_op)
     /* Check args and get the query objects */
     if (NULL == (query = (H5Q_t *) H5I_object_verify(query_id, H5I_QUERY)))
         HGOTO_ERROR(H5E_ARGS, H5E_BADTYPE, FAIL, "not a query ID");
+    if (!match_op)
+        HGOTO_ERROR(H5E_ARGS, H5E_BADVALUE, FAIL, "NULL pointer to match op");
 
     /* Get match info */
     if (FAIL == H5Q_get_match_op(query, match_op))
@@ -706,11 +710,12 @@ H5Q_get_match_op(H5Q_t *query, H5Q_match_op_t *match_op)
     FUNC_ENTER_NOAPI_NOINIT
 
     HDassert(query);
+    HDassert(match_op);
 
     if (query->is_combined)
         HGOTO_ERROR(H5E_QUERY, H5E_CANTGET, FAIL, "cannot retrieve op from combined query");
 
-    if (match_op) *match_op = query->query.select.match_op;
+    *match_op = query->query.select.match_op;
 
 done:
     FUNC_LEAVE_NOAPI(ret_value)
@@ -914,6 +919,7 @@ H5Q_encode(H5Q_t *query, unsigned char *buf, size_t *nalloc)
     if (query->is_combined) {
         size_t l_buf_size = 0, r_buf_size = 0;
 
+        H5Q_encode_memcpy(&buf_ptr, &buf_size, &query->query.combine.type, sizeof(H5Q_type_t));
         H5Q_encode_memcpy(&buf_ptr, &buf_size, &query->query.combine.op, sizeof(H5Q_combine_op_t));
         H5Q_encode(query->query.combine.l_query, buf_ptr, &l_buf_size);
         buf_size += l_buf_size;
@@ -1041,6 +1047,7 @@ H5Q_decode(const unsigned char **buf_ptr)
 
     H5Q_decode_memcpy(&query->is_combined, sizeof(hbool_t), buf_ptr);
     if (query->is_combined) {
+        H5Q_decode_memcpy(&query->query.combine.type, sizeof(H5Q_type_t), buf_ptr);
         H5Q_decode_memcpy(&query->query.combine.op, sizeof(H5Q_combine_op_t), buf_ptr);
         query->query.combine.l_query = H5Q_decode(buf_ptr);
         query->query.combine.r_query = H5Q_decode(buf_ptr);
diff --git a/src/H5VLiod_analysis.c b/src/H5VLiod_analysis.c
index a5aeb10..48eca85 100644
--- a/src/H5VLiod_analysis.c
+++ b/src/H5VLiod_analysis.c
@@ -796,8 +796,8 @@ H5VL__iod_farm_work(iod_obj_map_t *obj_map, iod_handle_t *cohs,
     size_t *split_num_elmts;
     hid_t split_type_id = FAIL;
     hg_request_t *hg_reqs = NULL;
-    uint32_t u;
-    unsigned int num_targets = obj_map->u_map.array_map.n_range;
+    uint32_t u, i;
+    unsigned int num_targets = num_ions_g;//obj_map->u_map.array_map.n_range;
     coords_t coords;
     analysis_farm_in_t farm_input;
     analysis_farm_out_t *farm_output = NULL;
@@ -823,10 +823,11 @@ H5VL__iod_farm_work(iod_obj_map_t *obj_map, iod_handle_t *cohs,
 
     coords.rank = H5Sget_simple_extent_ndims(region);
 
-#if 0
+#if 1
     /* Farm work to the IONs and combine the result */
     for(i=0 ; i<num_ions_g ; i++) {
         hid_t space_id;
+        hbool_t first_space = TRUE;
 
         /* Get all ranges that are on Node i into a dataspace */
         for(u = 0; u < obj_map->u_map.array_map.n_range ; u++) {
@@ -841,44 +842,106 @@ H5VL__iod_farm_work(iod_obj_map_t *obj_map, iod_handle_t *cohs,
                 if(FAIL == (space_layout = H5VL__iod_get_space_layout(coords, region)))
                     HGOTO_ERROR_FF(FAIL, "can't generate local dataspace selection");
 
-                if(0 == farm_input.num_cells) {
+                if(TRUE == first_space) {
                     space_id = H5Scopy(space_layout);
+                    if(space_id < 0)
+                        HGOTO_ERROR_FF(FAIL, "can't copy Dataspace");
+                    first_space = FALSE;
                 }
                 else {
-                    /* MSC - OR the space with the current one */
+                    if(H5Smodify_select(space_id, H5S_SELECT_OR, space_layout) < 0)
+                        HGOTO_ERROR_FF(FAIL, "Unable to OR 2 dataspace selections");
                 }
 
-                farm_input.num_cells += obj_map->u_map.array_map.array_range[u].n_cell;
-
                 if(H5Sclose(space_layout) < 0)
                     HGOTO_ERROR_FF(FAIL, "cannot close region");
             }
         }
 
-        if(farm_input.num_cells) {
-            farm_input.space_id = space_id;
-            farm_input.coh = cohs[i];
+        if(H5Smodify_select(space_id, H5S_SELECT_AND, region) < 0)
+            HGOTO_ERROR_FF(FAIL, "Unable to AND 2 dataspace selections");
+
+        farm_input.space_id = space_id;
+        farm_input.coh = cohs[i];
 
+        if(0 == H5Sget_select_npoints(farm_input.space_id)) {
+            hg_reqs[i] = HG_REQUEST_NULL;;
+            split_data[i] = NULL;
+            split_num_elmts[i] = 0;
+        }
+        else if (i == 0) {
+            /* Do a local split */
+            hg_reqs[i] = HG_REQUEST_NULL;
+            if(FAIL == H5VL__iod_farm_split(cohs[0], obj_map->oid, rtid, farm_input.space_id,
+                                            farm_input.type_id, split_script,
+                                            &split_data[i], &split_num_elmts[i], &split_type_id))
+                HGOTO_ERROR_FF(FAIL, "can't split in farmed job");
+        } else {
             /* forward the call to the target server */
-            if (i == 0) {
-                hg_reqs[i] = HG_REQUEST_NULL;
-                /* Do a local split */
-                if(FAIL == H5VL__iod_farm_split(cohs[0], obj_map->oid, rtid, space_layout,
-                                                farm_input.num_cells, farm_input.type_id, split_script,
-                                                &split_data[i], &split_num_elmts[i], &split_type_id))
-                    HGOTO_ERROR_FF(FAIL, "can't split in farmed job");
-            } else {
-                if(HG_Forward(server_addr_g[i], H5VL_EFF_ANALYSIS_FARM, &farm_input, 
-                              &farm_output[i], &hg_reqs[i]) < 0)
+            if(HG_Forward(server_addr_g[i], H5VL_EFF_ANALYSIS_FARM, &farm_input, 
+                          &farm_output[i], &hg_reqs[i]) < 0)
+                HGOTO_ERROR_FF(FAIL, "failed to ship operation");
+        }
+
+        if(H5Sclose(space_id) < 0)
+            HGOTO_ERROR_FF(FAIL, "cannot close region");
+    }
+
+    for(i=0 ; i<num_ions_g ; i++) {
+        if (hg_reqs[i] == HG_REQUEST_NULL) {
+            /* No request / was local or empty */
+        } 
+        else {
+            analysis_transfer_in_t transfer_input;
+            analysis_transfer_out_t transfer_output;
+            hg_bulk_t bulk_handle;
+            size_t split_data_size;
+
+            /* Wait for the farmed work to complete */
+            if(HG_Wait(hg_reqs[i], HG_MAX_IDLE_TIME, HG_STATUS_IGNORE) < 0)
+                HGOTO_ERROR_FF(FAIL, "HG_Wait Failed");
+
+            if(0 != farm_output[i].num_elmts) {
+                /* Get split type ID and num_elemts 
+                   (all the arrays should have the same native type id) */
+                split_type_id = farm_output[i].type_id;
+                split_num_elmts[i] = farm_output[i].num_elmts;
+                split_data_size = split_num_elmts[i] * H5Tget_size(split_type_id);
+
+                if(NULL == (split_data[i] = malloc(split_data_size)))
+                    HGOTO_ERROR_FF(FAIL, "can't allocate farm buffer");
+
+                HG_Bulk_handle_create(1, &split_data[i], &split_data_size,
+                                      HG_BULK_READWRITE, &bulk_handle);
+
+                transfer_input.axe_id = farm_output[i].axe_id;
+                transfer_input.bulk_handle = bulk_handle;
+
+                /* forward a free call to the target server */
+                if(HG_Forward(server_addr_g[i], H5VL_EFF_ANALYSIS_FARM_TRANSFER,
+                              &transfer_input, &transfer_output, &hg_reqs[i]) < 0)
                     HGOTO_ERROR_FF(FAIL, "failed to ship operation");
-            }
 
-            if(H5Sclose(space_id) < 0)
-                HGOTO_ERROR_FF(FAIL, "cannot close region");
+                /* Wait for the farmed work to complete */
+                if(HG_Wait(hg_reqs[i], HG_MAX_IDLE_TIME, HG_STATUS_IGNORE) < 0)
+                    HGOTO_ERROR_FF(FAIL, "HG_Wait Failed");
+
+                /* Free bulk handle */
+                HG_Bulk_handle_free(bulk_handle);
+            }
+            else {
+                split_num_elmts[i] = 0;
+                split_data[i] = NULL;
+            }
+            /* Free Mercury request */
+            if(HG_Request_free(hg_reqs[i]) != HG_SUCCESS)
+                HGOTO_ERROR_FF(FAIL, "Can't Free Mercury Request");
         }
     }
+
 #endif
 
+#if 0
     /* farm work for a specific range on a specific node */
     for(u = 0; u < obj_map->u_map.array_map.n_range ; u++) {
         hid_t space_layout;
@@ -899,15 +962,21 @@ H5VL__iod_farm_work(iod_obj_map_t *obj_map, iod_handle_t *cohs,
         farm_input.space_id = space_layout;
         farm_input.coh = cohs[worker];
 
-        /* forward the call to the target server */
-        if (worker == 0) {
+        if(0 == H5Sget_select_npoints(farm_input.space_id)) {
+            hg_reqs[u] = HG_REQUEST_NULL;;
+            split_data[u] = NULL;
+            split_num_elmts[u] = 0;
+        }
+        else if (worker == 0) {
             hg_reqs[u] = HG_REQUEST_NULL;
             /* Do a local split */
             if(FAIL == H5VL__iod_farm_split(cohs[0], obj_map->oid, rtid, farm_input.space_id,
                                             farm_input.type_id, split_script,
                                             &split_data[u], &split_num_elmts[u], &split_type_id))
                 HGOTO_ERROR_FF(FAIL, "can't split in farmed job");
-        } else {
+        }
+        else {
+            /* forward the call to the target server */ 
             if(HG_Forward(server_addr_g[worker], H5VL_EFF_ANALYSIS_FARM, &farm_input, 
                           &farm_output[u], &hg_reqs[u]) < 0)
                 HGOTO_ERROR_FF(FAIL, "failed to ship operation");
@@ -921,7 +990,7 @@ H5VL__iod_farm_work(iod_obj_map_t *obj_map, iod_handle_t *cohs,
         int worker = obj_map->u_map.array_map.array_range[u].nearest_rank;
 
         if (hg_reqs[u] == HG_REQUEST_NULL) {
-            /* No request / was local */
+            /* No request / was local or empty */
         } 
         else {
             analysis_transfer_in_t transfer_input;
@@ -970,6 +1039,7 @@ H5VL__iod_farm_work(iod_obj_map_t *obj_map, iod_handle_t *cohs,
                 HGOTO_ERROR_FF(FAIL, "Can't Free Mercury Request");
         }
     }
+#endif
 
 #if H5_EFF_DEBUG 
     fprintf(stderr, "(%d) Applying combine on data\n", my_rank_g);
diff --git a/src/H5VLiod_util.c b/src/H5VLiod_util.c
index b6bfb4c..806e877 100644
--- a/src/H5VLiod_util.c
+++ b/src/H5VLiod_util.c
@@ -402,7 +402,8 @@ H5VL_iod_get_file_desc(hid_t space_id, hssize_t *count, iod_hyperslab_t *hslabs)
                                                    hslabs[0].block) < 0)
                         HGOTO_ERROR_FF(FAIL, "Failed to retrieve hyperslab selection");
                     for(i=0 ; i<ndims ; i++) {
-                        hslabs[0].stride[i] = hslabs[0].block[i] * hslabs[0].stride[i];
+                        if(hslabs[0].stride[i] == 1)
+                            hslabs[0].stride[i] = hslabs[0].block[i];
                     }
                 }
             }
@@ -439,6 +440,17 @@ H5VL_iod_get_file_desc(hid_t space_id, hssize_t *count, iod_hyperslab_t *hslabs)
                         }
                     }
 
+#if 0
+                    fprintf(stderr, "Space contains %zu elements\n", 
+                            (size_t)H5Sget_select_npoints(space_id));
+                    for(n=0 ; n<num_descriptors ; n++) {
+                        for(i=0 ; i<ndims ; i++) {
+                            fprintf(stderr, "DIMS %d: %zu %zu %zu %zu\n", i,
+                                    (size_t)hslabs[n].start[i],(size_t)hslabs[n].count[i],
+                                    (size_t)hslabs[n].block[i],(size_t)hslabs[n].stride[i]);
+                        }
+                    }
+#endif
                     free(blocks);
                 }
             }
diff --git a/src/H5VLiod_view.c b/src/H5VLiod_view.c
index 294c8f5..ed3b4ea 100644
--- a/src/H5VLiod_view.c
+++ b/src/H5VLiod_view.c
@@ -433,9 +433,15 @@ H5VL__iod_apply_query(hid_t file_id, hid_t rcxt_id, hid_t qid, hid_t vcpl_id,
     if(H5Qget_combine_op(qid, &comb_type) < 0)
         HGOTO_ERROR_FF(FAIL, "can't get query op type");
 
-    if(H5Q_SINGLETON == comb_type) {
-        if(H5Qget_match_info(qid, &q_type, &match_op) < 0)
-            HGOTO_ERROR_FF(FAIL, "can't get query info");
+    if(H5Qget_type(qid, &q_type) < 0)
+        HGOTO_ERROR_FF(FAIL, "can't get query op type");
+
+    if(H5Q_SINGLETON == comb_type || 
+       q_type == H5Q_TYPE_DATA_ELEM ||
+       q_type == H5Q_TYPE_ATTR_VALUE) {
+
+        //if(H5Qget_match_info(qid, &q_type, &match_op) < 0)
+        //HGOTO_ERROR_FF(FAIL, "can't get query info");
 
         if(H5Q_TYPE_DATA_ELEM == q_type) {
             if(H5I_DATASET == obj_type) {
@@ -473,7 +479,6 @@ H5VL__iod_apply_query(hid_t file_id, hid_t rcxt_id, hid_t qid, hid_t vcpl_id,
                     *result = QTRUE;
                     *region = sid;
                 }
-
                 else {
                     if(H5Sclose(sid) < 0)
                         HGOTO_ERROR_FF(FAIL, "unable to release dataspace");
@@ -1020,15 +1025,16 @@ H5VL__iod_get_query_data_cb(void *elem, hid_t type_id, unsigned ndim,
 {
     H5VL__iod_get_query_data_t *udata = (H5VL__iod_get_query_data_t *)_udata;
     hbool_t result = FALSE;
+    unsigned i;
     herr_t ret_value = SUCCEED;
 
     /* Apply the query */
-    if(H5Qapply(udata->query_id, &result, type_id, elem) < 0)
+    if(H5Qapply_combine(udata->query_id, &result, type_id, elem) < 0)
         HGOTO_ERROR_FF(FAIL, "unable to apply query to data element");
 
     /* If element satisfies query, add it to the selection */
     if (result) {
-        hsize_t count[H5S_MAX_RANK], i;
+        hsize_t count[H5S_MAX_RANK];
 #if 0
         fprintf(stderr, "(%d) Element [", my_rank_g);
         for(i=0; i<ndim; i++)