Add interface for user-controlled roles.

.gitattributes

@@ -294,6 +294,7 @@ src/pio_posixasynch.c -text
 src/pio_posixfpguardsendrecv.c -text
 src/pio_posixnonb.c -text
 src/pio_record_send.c -text
+src/pio_roles.c -text
 src/pio_rpc.h -text
 src/pio_serialize.c -text
 src/pio_serialize.h -text

doc/pio/tex/c_quick_ref.tex

@@ -4,6 +4,17 @@
 This appendix provide a brief listing of the C language bindings of the
 CDI library routines:
 
+\section*{\tt \htmlref{cdiPioInit}{cdiPioInit}}
+
+\begin{verbatim}
+    MPI_Comm cdiPioInit (MPI_Comm commSuper, int IOMode, int clientServerRole,
+                         int *pioNamespace, float partInflate,
+                         void (*postCommSetupActions)(void));
+\end{verbatim}
+
+initialize I/O server processes and communication.
+
+
 \section*{\tt \htmlref{cdiPioNoPostCommSetup}{cdiPioNoPostCommSetup}}
 
 \begin{verbatim}

doc/pio/tex/f_quick_ref.tex

@@ -4,6 +4,17 @@
 This appendix provide a brief listing of the Fortran language bindings of the
 CDI library routines:
 
+\section*{\tt \htmlref{cdiPioInit}{cdiPioInit}}
+
+\begin{verbatim}
+    INTEGER FUNCTION cdiPioInit (INTEGER commSuper, INTEGER IOMode,
+                                 INTEGER clientServerRole, INTEGER pioNamespace,
+                                 REAL partInflate, PROCEDURE postCommSetupActions)
+\end{verbatim}
+
+initialize I/O server processes and communication.
+
+
 \section*{\tt \htmlref{cdiPioNoPostCommSetup}{cdiPioNoPostCommSetup}}
 
 \begin{verbatim}

src/Makefile.am

@@ -182,9 +182,10 @@ libcdipio_la_SOURCES =							\
 	pio_list_set.c							\
 	resource_unpack.h						\
 	resource_unpack.c						\
-	pio_rpc.h							\
 	pio_client.c							\
 	pio_client.h							\
+	pio_roles.c							\
+	pio_rpc.h							\
 	pio_server.c							\
 	pio_server.h							\
 	pio_serialize.h							\

src/Makefile.in

@@ -222,9 +222,10 @@ am__libcdipio_la_SOURCES_DIST = cdipio.h cdipioFortran.c cfortran.h \
 	pio_interface.c pio_interface.h pio_mpinonb.c \
 	pio_record_send.c pio_posixasynch.c pio_posixfpguardsendrecv.c \
 	pio_posixnonb.c pio_list_set.c resource_unpack.h \
-	resource_unpack.c pio_rpc.h pio_client.c pio_client.h \
-	pio_server.c pio_server.h pio_serialize.h pio_serialize.c \
-	pio_util.c pio_util.h pio_cdf_int.h pio_cdf_int.c
+	resource_unpack.c pio_client.c pio_client.h pio_roles.c \
+	pio_rpc.h pio_server.c pio_server.h pio_serialize.h \
+	pio_serialize.c pio_util.c pio_util.h pio_cdf_int.h \
+	pio_cdf_int.c
 am__objects_5 = pio_cdf_int.lo
 @HAVE_PARALLEL_NC4_TRUE@@USE_MPI_TRUE@am__objects_6 =  \
 @HAVE_PARALLEL_NC4_TRUE@@USE_MPI_TRUE@	$(am__objects_5)
@@ -232,7 +233,7 @@ am_libcdipio_la_OBJECTS = cdipioFortran.lo pio.lo pio_comm.lo \
 	pio_dbuffer.lo pio_interface.lo pio_mpinonb.lo \
 	pio_record_send.lo pio_posixasynch.lo \
 	pio_posixfpguardsendrecv.lo pio_posixnonb.lo pio_list_set.lo \
-	resource_unpack.lo pio_client.lo pio_server.lo \
+	resource_unpack.lo pio_client.lo pio_roles.lo pio_server.lo \
 	pio_serialize.lo pio_util.lo $(am__objects_6)
 libcdipio_la_OBJECTS = $(am_libcdipio_la_OBJECTS)
 @ENABLE_CDI_LIB_FALSE@@USE_MPI_TRUE@am_libcdipio_la_rpath =
@@ -550,10 +551,10 @@ libcdipio_la_SOURCES = cdipio.h cdipioFortran.c cfortran.h pio.c pio.h \
 	pio_comm.c pio_comm.h pio_dbuffer.c pio_impl.h pio_interface.c \
 	pio_interface.h pio_mpinonb.c pio_record_send.c \
 	pio_posixasynch.c pio_posixfpguardsendrecv.c pio_posixnonb.c \
-	pio_list_set.c resource_unpack.h resource_unpack.c pio_rpc.h \
-	pio_client.c pio_client.h pio_server.c pio_server.h \
-	pio_serialize.h pio_serialize.c pio_util.c pio_util.h \
-	pio_cdf_int.h $(am__append_7)
+	pio_list_set.c resource_unpack.h resource_unpack.c \
+	pio_client.c pio_client.h pio_roles.c pio_rpc.h pio_server.c \
+	pio_server.h pio_serialize.h pio_serialize.c pio_util.c \
+	pio_util.h pio_cdf_int.h $(am__append_7)
 libcdipio_la_HAVE_PARALLEL_NC4_extra_sources = \
 	pio_cdf_int.c
 
@@ -724,6 +725,7 @@ distclean-compile:
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_posixfpguardsendrecv.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_posixnonb.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_record_send.Plo@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_roles.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_serialize.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_server.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pio_util.Plo@am__quote@

src/cdipio.h

@@ -22,6 +22,12 @@
 #define  PIO_MAXIOMODE                  PIO_FPGUARD
 #define  PIO_MINIOMODEWITHSPECIALPROCS  PIO_WRITER
 
+#define PIO_ROLE_CLIENT 0
+#define PIO_ROLE_COLLECTOR 1
+#define PIO_ROLE_WRITER 2
+#define PIO_ROLE_WRITER_COLLECTOR 3
+#define PIO_ROLE_FPGUARD 4
+
 /* parallel IO routines */
 #include <yaxt.h>
 
@@ -35,6 +41,10 @@ void cdiPioNoPostCommSetup(void);
 MPI_Comm pioInit(MPI_Comm commSuper, int nProcsIO, int IOMode,
                  int *pioNamespace, float partInflate,
                  void (*postCommSetupActions)(void));
+/*      cdiPioInit: initialize I/O server processes and communication */
+MPI_Comm cdiPioInit(MPI_Comm commSuper, int IOMode, int clientServerRole,
+                    int *pioNamespace, float partInflate,
+                    void (*postCommSetupActions)(void));
 void     pioWriteTimestep(void);
 void     cdiPioRDMAProgress(void);
 
@@ -46,4 +56,6 @@ void     streamWriteScatteredVarPart(int streamID, int varID, const void *data,
                                      const int displacements[],
                                      int nmiss, Xt_idxlist partDesc);
 
+int cdiPioCSRLastN(MPI_Comm commSuper, int IOMode, int nProcsIO);
+
 #endif

src/cdipio.inc

@@ -20,6 +20,16 @@
       PARAMETER (PIO_ASYNCH             =  3)
       INTEGER    PIO_FPGUARD           
       PARAMETER (PIO_FPGUARD            =  4)
+      INTEGER    PIO_ROLE_CLIENT       
+      PARAMETER (PIO_ROLE_CLIENT        =  0)
+      INTEGER    PIO_ROLE_COLLECTOR    
+      PARAMETER (PIO_ROLE_COLLECTOR     =  1)
+      INTEGER    PIO_ROLE_WRITER       
+      PARAMETER (PIO_ROLE_WRITER        =  2)
+      INTEGER    PIO_ROLE_WRITER_COLLECTOR
+      PARAMETER (PIO_ROLE_WRITER_COLLECTOR =  3)
+      INTEGER    PIO_ROLE_FPGUARD      
+      PARAMETER (PIO_ROLE_FPGUARD       =  4)
 !
 !  parallel IO routines
 !
@@ -44,6 +54,15 @@
 !                                     PROCEDURE       postCommSetupActions)
       EXTERNAL        pioInit
 
+      INTEGER         cdiPioInit
+!                                    (INTEGER         commSuper,
+!                                     INTEGER         IOMode,
+!                                     INTEGER         clientServerRole,
+!                                     INTEGER         pioNamespace,
+!                                     REAL            partInflate,
+!                                     PROCEDURE       postCommSetupActions)
+      EXTERNAL        cdiPioInit
+
 !                     pioWriteTimestep
       EXTERNAL        pioWriteTimestep
 
@@ -69,3 +88,9 @@
 !                                     TYPE(XT_IDXLIST)partDesc)
       EXTERNAL        streamWriteScatteredVarPart
 
+      INTEGER         cdiPioCSRLastN
+!                                    (INTEGER         commSuper,
+!                                     INTEGER         IOMode,
+!                                     INTEGER         nProcsIO)
+      EXTERNAL        cdiPioCSRLastN
+

src/cdipioFortran.c

@@ -34,6 +34,16 @@ static int pioInit_fwrap(int commSuper, int nProcsIO, int IOMode, int *pioNamesp
   return MPI_Comm_c2f(v);
 }
 FCALLSCFUN6 (INT, pioInit_fwrap, PIOINIT, pioinit, INT, INT, INT, PINT, FLOAT, ROUTINE)
+
+#undef ROUTINE_6
+#define ROUTINE_6 (void (*)(void))
+static int cdiPioInit_fwrap(int commSuper, int IOMode, int clientServerRole, int *pioNamespace, float partInflate, void (*postCommSetupActions)(void))
+{
+  MPI_Comm v;
+  v = cdiPioInit(MPI_Comm_f2c(commSuper), IOMode, clientServerRole, pioNamespace, partInflate, postCommSetupActions);
+  return MPI_Comm_c2f(v);
+}
+FCALLSCFUN6 (INT, cdiPioInit_fwrap, CDIPIOINIT, cdipioinit, INT, INT, INT, PINT, FLOAT, ROUTINE)
 FCALLSCSUB0 (pioWriteTimestep, PIOWRITETIMESTEP, piowritetimestep)
 FCALLSCSUB0 (cdiPioRDMAProgress, CDIPIORDMAPROGRESS, cdipiordmaprogress)
 static void streamWriteVarPart_fwrap(int streamID, int varID, const void *data, int nmiss, void *partDesc)
@@ -46,5 +56,12 @@ static void streamWriteScatteredVarPart_fwrap(int streamID, int varID, const voi
   streamWriteScatteredVarPart(streamID, varID, data, numBlocks, blocklengths, displacements, nmiss, (*(Xt_idxlist *)partDesc));
 }
 FCALLSCSUB8 (streamWriteScatteredVarPart_fwrap, STREAMWRITESCATTEREDVARPART, streamwritescatteredvarpart, INT, INT, PVOID, INT, INTV, INTV, INT, PVOID)
+static int cdiPioCSRLastN_fwrap(int commSuper, int IOMode, int nProcsIO)
+{
+  int v;
+  v = cdiPioCSRLastN(MPI_Comm_f2c(commSuper), IOMode, nProcsIO);
+  return v;
+}
+FCALLSCFUN3 (INT, cdiPioCSRLastN_fwrap, CDIPIOCSRLASTN, cdipiocsrlastn, INT, INT, INT)
 
 #endif

src/pio_interface.c

@@ -744,6 +744,46 @@ cdiPioNoPostCommSetup(void)
 static int pioNamespace_ = -1;
 static int xtInitByCDI = 0;
 
+MPI_Comm
+cdiPioInit(MPI_Comm commGlob, int IOMode, int clientServerRole,
+           int *pioNamespace, float partInflate,
+           void (*postCommSetupActions)(void))
+{
+  /* construct new communicator such that old interface can be used
+   * i.e. push all processes not being clients to ranks succeeding
+   * clients and non-collectors even further
+   */
+  int key;
+  switch (clientServerRole)
+    {
+    case PIO_ROLE_CLIENT:
+      key = 0;
+      break;
+    case PIO_ROLE_COLLECTOR:
+    case PIO_ROLE_WRITER_COLLECTOR:
+      key = 1;
+      break;
+    case PIO_ROLE_WRITER:
+    case PIO_ROLE_FPGUARD:
+      key = 2;
+      break;
+    default:
+      xabort("Invalid client/server-role specified: %d", clientServerRole);
+    }
+  MPI_Comm rearrangedCommGlob;
+  xmpi(MPI_Comm_split(commGlob, 0, key, &rearrangedCommGlob));
+  int serverCount = (key > 0);
+  xmpi(MPI_Allreduce(MPI_IN_PLACE, &serverCount,
+                     1, MPI_INT, MPI_SUM, commGlob));
+  MPI_Comm retComm
+    = pioInit(rearrangedCommGlob, IOMode, serverCount, pioNamespace,
+              partInflate, postCommSetupActions);
+  MPI_Comm_free(&rearrangedCommGlob);
+  return retComm;
+}
+
+
+
 /* pioInit definition must currently compile even in non-MPI configurations */
 /**
    @brief initializes the MPI_Communicators needed for the

src/pio_roles.c

+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+
+#include <stdlib.h>
+#include "cdipio.h"
+
+#include "pio_util.h"
+
+int
+cdiPioCSRLastN(MPI_Comm commSuper, int IOMode, int nProcsIO)
+{
+  int commRank, commSize;
+  xmpi(MPI_Comm_size(commSuper, &commSize));
+  xmpi(MPI_Comm_rank(commSuper, &commRank));
+  int role;
+  switch (IOMode)
+  {
+  case PIO_NONE:
+    xassert(commRank == 0 && commSize == 1);
+    role = PIO_NONE;
+    break;
+  case PIO_MPI:
+    role = (commRank < commSize - nProcsIO)
+      ? PIO_ROLE_CLIENT : PIO_ROLE_WRITER_COLLECTOR;
+    break;
+  case PIO_WRITER:
+  case PIO_ASYNCH:
+    role = (commRank < commSize - nProcsIO)
+      ? PIO_ROLE_CLIENT : ((commRank == commSize - 1)
+                           ? PIO_ROLE_WRITER
+                           : PIO_ROLE_COLLECTOR);
+    break;
+  case PIO_FPGUARD:
+    role = (commRank < commSize - nProcsIO)
+      ? PIO_ROLE_CLIENT : ((commRank == commSize - 1)
+                           ? PIO_ROLE_FPGUARD
+                           : PIO_ROLE_WRITER_COLLECTOR);
+    break;
+  default:
+    abort();
+  }
+  return role;
+}
+