Use stripes for intersection computation.

* This is configurable.

src/cdipio.h

@@ -143,5 +143,12 @@ void cdiPioConfSetXmapNew(int confResH,
 Xt_xmap (*cdiPioConfGetXmapNew(int confResH))(Xt_idxlist src_idxlist,
                                               Xt_idxlist dst_idxlist,
                                               MPI_Comm comm);
+/* convert index lists to stripes prior to intersection computation,
+ * defaults to true */
+void cdiPioConfSetStripeConversion(int confResH, int doStripify);
+
+/* are index lists of parts converted stripes before being passed
+ * to the xmap constructor? */
+int cdiPioConfGetStripeConversion(int confResH);
 
 #endif

src/cdipio.inc

@@ -174,3 +174,20 @@
 !                                    (INTEGER         confResH)
       EXTERNAL        cdiPioConfGetRedistCache
 
+!
+!  convert index lists to stripes prior to intersection computation,
+!  defaults to true
+!
+!                     cdiPioConfSetStripeConversion
+!                                    (INTEGER         confResH,
+!                                     INTEGER         doStripify)
+      EXTERNAL        cdiPioConfSetStripeConversion
+
+!
+!  are index lists of parts converted stripes before being passed
+!  to the xmap constructor?
+!
+      INTEGER         cdiPioConfGetStripeConversion
+!                                    (INTEGER         confResH)
+      EXTERNAL        cdiPioConfGetStripeConversion
+

src/cdipioFortran.c

@@ -96,4 +96,14 @@ FCALLSCFUN1 (INT, cdiPioConfGetLargePageAlign, CDIPIOCONFGETLARGEPAGEALIGN, cdip
 FCALLSCSUB2 (cdiPioConfSetRedistCache, CDIPIOCONFSETREDISTCACHE, cdipioconfsetredistcache, INT, INT)
 FCALLSCFUN1 (INT, cdiPioConfGetRedistCache, CDIPIOCONFGETREDISTCACHE, cdipioconfgetredistcache, INT)
 
+/*  convert index lists to stripes prior to intersection computation,
+ * defaults to true  */
+
+FCALLSCSUB2 (cdiPioConfSetStripeConversion, CDIPIOCONFSETSTRIPECONVERSION, cdipioconfsetstripeconversion, INT, INT)
+
+/*  are index lists of parts converted stripes before being passed
+ * to the xmap constructor?  */
+
+FCALLSCFUN1 (INT, cdiPioConfGetStripeConversion, CDIPIOCONFGETSTRIPECONVERSION, cdipioconfgetstripeconversion, INT)
+
 #endif

src/pio_conf.c

@@ -165,6 +165,7 @@ int cdiPioConfCreate(void)
   conf->cacheRedists = true;
   conf->recordAggBufLimMB = 128;
   conf->xmap_new = xt_xmap_dist_dir_new;
+  conf->stripify = true;
   int resH = reshPut(conf, &cdiPioConfOps);
   /* configuration objects are never forwarded */
   reshSetStatus(resH, &cdiPioConfOps,
@@ -281,3 +282,15 @@ xmap_new_func_ptr cdiPioConfGetXmapNew(int confResH)
   return conf->xmap_new;
 }
 
+void cdiPioConfSetStripeConversion(int confResH, int doConversion)
+{
+  struct cdiPioConf *conf = reshGetVal(confResH, &cdiPioConfOps);
+  conf->stripify = doConversion;
+}
+
+int cdiPioConfGetStripeConversion(int confResH)
+{
+  struct cdiPioConf *conf = reshGetVal(confResH, &cdiPioConfOps);
+  return conf->stripify;
+}
+

src/pio_conf.h

@@ -31,6 +31,7 @@ struct cdiPioConf {
   xmap_new_func_ptr xmap_new;
   bool largePageAlign;
   bool cacheRedists;
+  bool stripify;
 };
 
 extern const resOps cdiPioConfOps;
@@ -79,4 +80,8 @@ void cdiPioConfSetXmapNew(int confResH, xmap_new_func_ptr xmap_new);
 
 xmap_new_func_ptr cdiPioConfGetXmapNew(int confResH);
 
+void cdiPioConfSetStripeConversion(int confResH, int doConversion);
+
+int cdiPioConfGetStripeConversion(int confResH);
+
 #endif

src/pio_server.c

@@ -257,6 +257,12 @@ buildVarRedist(int headerIdx, size_t streamIdx,
   for (size_t clientIdx = 0; clientIdx < (size_t)numClients_; ++clientIdx)
     xt_idxlist_delete(part[clientIdx]);
   Free(part);
+  if (conf->stripify)
+    {
+      Xt_idxlist srcListStriped = xt_idxstripes_from_idxlist_new(srcList);
+      xt_idxlist_delete(srcList);
+      srcList = srcListStriped;
+    }
   Xt_xmap gatherXmap = conf->xmap_new(srcList, dstList, collComm);
   xt_idxlist_delete(srcList);
   struct Xt_offset_ext gatherExt