Merge of branch cdo-pio into trunk cdi

.gitattributes

@@ -95,6 +95,13 @@ examples/cdi_write_ens.c -text
 examples/cdi_write_f.f -text
 examples/cdi_write_f2003.f90 -text
 examples/compf -text
+examples/pio/Makefile.am -text
+examples/pio/Makefile.in -text
+examples/pio/cdi.inc -text
+examples/pio/collectData.c -text
+examples/pio/collectData2003.F90 -text
+examples/pio/collectDataNStreams.c -text
+examples/pio/compareResourcesArray.c -text
 interfaces/CdiInfo.cpp -text
 interfaces/Makefile.am -text
 interfaces/Makefile.in -text

examples/Makefile.in

@@ -220,6 +220,7 @@ MAINT = @MAINT@
 MAKEINFO = @MAKEINFO@
 MANIFEST_TOOL = @MANIFEST_TOOL@
 MKDIR_P = @MKDIR_P@
+MPI_LAUNCH = @MPI_LAUNCH@
 NC_CONFIG = @NC_CONFIG@
 NETCDF_INCLUDE = @NETCDF_INCLUDE@
 NETCDF_LIBS = @NETCDF_LIBS@
@@ -256,6 +257,7 @@ SZLIB_LIBS = @SZLIB_LIBS@
 THREADS_INCLUDE = @THREADS_INCLUDE@
 THREADS_LIBS = @THREADS_LIBS@
 USER_NAME = @USER_NAME@
+USE_MPI = @USE_MPI@
 VERSION = @VERSION@
 ZLIB_INCLUDE = @ZLIB_INCLUDE@
 ZLIB_LIBS = @ZLIB_LIBS@

examples/pio/Makefile.am

+noinst_PROGRAMS=collectData collectDataNStreams
+
+if USE_MPI
+noinst_PROGRAMS+=compareResourcesArray
+endif
+
+if USE_FC
+ noinst_PROGRAMS+=collectData2003
+endif
+
+AM_CFLAGS=-I$(top_srcdir)/src
+
+collectData_SOURCES=collectData.c
+
+collectData_LDADD=-L../src -lcdi
+
+collectData2003_SOURCES=collectData2003.F90
+
+collectData2003_LDADD=-L../src -lcdi
+
+collectDataNStreams_SOURCES=collectDataNStreams.c
+
+collectDataNStreams_LDADD=-L../src -lcdi
+
+compareResourcesArray_SOURCES=compareResourcesArray.c
+
+compareResourcesArray_LDADD=-L../src -lcdi
\ No newline at end of file

examples/pio/cdi.inc

+../../src/cdi.inc
\ No newline at end of file

examples/pio/collectData.c

+#if defined (HAVE_CONFIG_H)
+#  include "config.h"
+#endif
+
+#ifdef USE_MPI
+#include <unistd.h>
+#include <mpi.h>
+#include "pio_c_temp.h"
+#endif
+
+#include "cdi.h"
+#include "pio_util.h"
+
+
+void modelRun ()
+{
+  enum {
+    filetype    = FILETYPE_GRB,
+    ntfiles     = 2,
+    ntsteps     = 3,
+    nVars       = 5,
+    nlon        = 12,
+    nlat        = 6,
+    maxlev      = 5 };
+
+  static int nlev[nVars]    = {1,1,5,5,2};
+  static char * name        = "example";
+
+  int gridID, zaxisID[nVars], taxisID;
+  int vlistID, varID[nVars], streamID, tsID, tfID = 0;
+  int i, j, nmiss = 0, rank;
+  double lons[nlon] = {0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330};
+  double lats[nlat] = {-75, -45, -15, 15, 45, 75};
+  double levs[maxlev] = {101300, 92500, 85000, 50000, 20000};
+  double var[nlon*nlat*maxlev];
+  int vdate = 19850101, vtime = 120000;
+  int start = CDI_UNDEFID, chunk = CDI_UNDEFID, stop = CDI_UNDEFID;
+  char filename[1024];
+
+  gridID = gridCreate ( GRID_LONLAT, nlon*nlat );
+  gridDefXsize ( gridID, nlon );
+  gridDefYsize ( gridID, nlat );
+  gridDefXvals ( gridID, lons );
+  gridDefYvals ( gridID, lats );
+
+  for ( i = 0; i < nVars; i++ )
+    {
+      zaxisID[i] = zaxisCreate ( ZAXIS_PRESSURE, nlev[i] );
+      zaxisDefLevels ( zaxisID[i], levs );
+    }
+  vlistID = vlistCreate ();
+
+  for ( i = 0; i < nVars; i++ )
+    varID[i] = vlistDefVar ( vlistID, gridID, zaxisID[i], TIME_VARIABLE );
+
+  taxisID = taxisCreate ( TAXIS_ABSOLUTE );
+  vlistDefTaxis ( vlistID, taxisID );
+
+  sprintf ( &filename[0], "%s_%d.grb", name, tfID );
+  streamID = streamOpenWrite ( filename, filetype );
+  xassert ( streamID >= 0 );
+  streamDefVlist ( streamID, vlistID);
+
+  pioEndDef ();
+
+  for ( tfID = 0; tfID < ntfiles; tfID++ )
+    {
+      if ( tfID > 0 )
+	{
+	  streamClose ( streamID );
+	  sprintf ( &filename[0], "%s_%d.grb", name, tfID );
+	  streamID = streamOpenWrite ( filename, filetype );
+	  xassert ( streamID >= 0 );
+	  streamDefVlist ( streamID, vlistID );
+	}
+      for ( tsID = 0; tsID < ntsteps; tsID++ )
+	{
+	  taxisDefVdate ( taxisID, vdate );
+	  taxisDefVtime ( taxisID, vtime );
+	  streamDefTimestep ( streamID, tsID );
+	  for ( i = 0; i < nVars; i++ )
+	    {
+	      start = pioInqVarDecoOff   ( vlistID, varID[i] );
+	      chunk = pioInqVarDecoChunk ( vlistID, varID[i] );
+	      stop = start + chunk;
+	      for ( j = start; j < stop; j++ ) var[j] = 2.2;
+	      streamWriteVar ( streamID, varID[i], &var[start], nmiss );
+	      start = CDI_UNDEFID;
+	      chunk = CDI_UNDEFID;
+	    }
+	  pioWriteTimestep ( tsID, vdate, vtime );
+	}
+    }
+  pioEndTimestepping ();
+  streamClose ( streamID );
+  vlistDestroy ( vlistID );
+  taxisDestroy ( taxisID );
+  for ( i = 0; i < nVars; i++ )
+    zaxisDestroy ( zaxisID[i] );
+  gridDestroy ( gridID );
+}
+
+struct {
+  char *text;
+  int mode;
+} mode_map[] = {
+  { "PIO_MPI", PIO_MPI },
+  { "PIO_FPGUARD", PIO_FPGUARD },
+  { "PIO_ASYNCH", PIO_ASYNCH },
+  { "PIO_WRITER", PIO_WRITER }
+};
+
+
+int main (int argc, char *argv[])
+{
+  enum {
+    nNamespaces = 1 };
+
+  static int hasLocalFile[nNamespaces] = { 0 };
+
+#ifdef USE_MPI
+  MPI_Comm commGlob, commModel;
+  int sizeGlob;
+  int rankGlob;
+  int IOMode = PIO_MPI;
+  int nProcsIO = 2;
+
+  xmpi ( MPI_Init ( &argc, &argv));
+  xmpi ( MPI_Comm_dup ( MPI_COMM_WORLD, &commGlob ));
+  xmpi ( MPI_Comm_set_errhandler ( commGlob, MPI_ERRORS_RETURN ));
+  xmpi ( MPI_Comm_size ( commGlob, &sizeGlob ));
+  xmpi ( MPI_Comm_rank ( commGlob, &rankGlob ));
+
+  {
+    int opt;
+    while ((opt = getopt(argc, argv, "p:w:")) != -1)
+      switch (opt) {
+      case 'p':
+        {
+          int i, found=0;
+          for (i = 0;
+               i < sizeof (mode_map) / sizeof (mode_map[0]);
+               ++i)
+            if (!strcmp(optarg, mode_map[i].text))
+              {
+                found = 1;
+                IOMode = mode_map[i].mode;
+              }
+          if (!found)
+            {
+              fprintf(stderr, "Unsupported PIO mode requested: %s\n", optarg);
+              exit(EXIT_FAILURE);
+            }
+        }
+        break;
+      case 'w':
+        {
+          nProcsIO = strtol(optarg, NULL, 0);
+          break;
+        }
+      }
+  }
+
+  commModel = pioInit_c ( commGlob, nProcsIO, IOMode, nNamespaces, hasLocalFile );
+#endif
+
+  modelRun ();
+
+#ifdef USE_MPI
+  pioFinalize ();
+  MPI_Finalize ();
+#endif
+  return 0;
+}
+
+/*
+ * Local Variables:
+ * c-file-style: "Java"
+ * c-basic-offset: 2
+ * indent-tabs-mode: nil
+ * show-trailing-whitespace: t
+ * require-trailing-newline: t
+ * End:
+ */

examples/pio/collectData2003.F90

+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+
+PROGRAM collectdata2003
+
+  IMPLICIT NONE
+
+  INCLUDE 'cdi.inc'
+
+#ifdef USE_MPI
+  INCLUDE 'mpif.h'
+#endif
+
+  ! For parallel IO:
+  !  Parameter and variables needed.
+  INTEGER, PARAMETER :: nProcsIO     = 3
+  ! INTEGER, PARAMETER :: IOMode       = PIO_NONE                  
+  ! INTEGER, PARAMETER :: IOMode       = PIO_MPI              
+  ! INTEGER, PARAMETER :: IOMode       = PIO_WRITER            
+  ! INTEGER, PARAMETER :: IOMode       = PIO_ASYNCH          
+  INTEGER, PARAMETER :: IOMode       = PIO_FPGUARD 
+  INTEGER, PARAMETER :: nNamespaces  = 1 
+  INTEGER, PARAMETER :: hasLocalFile(nNamespaces) = (/ 0 /)  
+
+  INTEGER ::commGlob, commModel, error
+
+  ! Start parallel environment
+#ifdef USE_MPI
+  CALL MPI_INIT ( error )
+  CALL MPI_COMM_DUP ( MPI_COMM_WORLD, commGlob, error )
+#endif
+
+  ! For parallel IO:
+  ! Initialize environment.
+  commModel = pioInit ( commGlob, nProcsIO, IOMode, nNamespaces, hasLocalFile )
+
+  CALL modelrun ( commModel )
+
+  ! For parallel IO: 
+  ! Cleanup environment.
+  CALL pioFinalize ()
+
+#ifdef USE_MPI
+  CALL MPI_FINALIZE ( error )
+#endif
+
+CONTAINS
+
+  SUBROUTINE modelrun ( commModel )
+
+    INTEGER, INTENT(in) :: commModel
+
+    INTEGER, PARAMETER :: nlon     = 12   ! Number of longitudes
+    INTEGER, PARAMETER :: nlat     =  6   ! Number of latitudes
+    INTEGER, PARAMETER :: nlev     =  5   ! Number of levels
+    INTEGER, PARAMETER :: nts      =  3   ! Number of time steps
+    INTEGER, PARAMETER :: vdate    = 19850101
+    INTEGER, PARAMETER :: vtime    = 120000
+    INTEGER, PARAMETER :: filetype = FILETYPE_GRB
+
+    INTEGER :: gridID, zaxisID1, zaxisID2, taxisID
+    INTEGER :: vlistID, varID1, varID2, streamID, tsID
+    INTEGER :: i, nmiss, status
+    DOUBLE PRECISION :: lons ( nlon ), lats ( nlat ), levs ( nlev )
+    DOUBLE PRECISION :: var1 ( nlon * nlat ), var2 ( nlon * nlat * nlev )
+    CHARACTER(len=256) :: varname
+    INTEGER :: startID, stopID, chunk
+
+    lons = (/0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330/)
+    lats = (/-75, -45, -15, 15, 45, 75/)
+    levs = (/101300, 92500, 85000, 50000, 20000/)
+
+    nmiss = 0
+
+    !     Create a regular lon/lat grid
+    gridID = gridCreate(GRID_LONLAT, nlon*nlat)
+    CALL gridDefXsize(gridID, nlon)
+    CALL gridDefYsize(gridID, nlat)
+    CALL gridDefXvals(gridID, lons)
+    CALL gridDefYvals(gridID, lats)
+
+    !     Create a surface level Z-axis
+    zaxisID1 = zaxisCreate(ZAXIS_SURFACE, 1)
+
+    !     Create a pressure level Z-axis
+    zaxisID2 = zaxisCreate(ZAXIS_PRESSURE, nlev)
+    CALL zaxisDefLevels(zaxisID2, levs)
+
+    !     Create a variable list
+    vlistID = vlistCreate()
+
+    !     Define the variables
+    varID1 = vlistDefVar(vlistID, gridID, zaxisID1, TIME_VARIABLE)
+    varID2 = vlistDefVar(vlistID, gridID, zaxisID2, TIME_VARIABLE)
+
+    !     Define the variable names
+    varname = 'varname1'
+    CALL vlistDefVarName(vlistID, varID1, varname)
+    varname = 'varname2'
+    CALL vlistDefVarName(vlistID, varID2, varname)
+
+    !     Create a Time axis
+    taxisID = taxisCreate(TAXIS_ABSOLUTE)
+
+    !     Assign the Time axis to the variable list
+    CALL vlistDefTaxis(vlistID, taxisID)
+
+    streamID = streamOpenWrite('example.grb', filetype)
+    IF ( streamID < 0 ) THEN
+      WRITE(0,*) cdiStringError(streamID)
+      STOP
+    END IF
+
+    !     Assign the variable list to the dataset
+    CALL streamDefVlist(streamID, vlistID)
+
+    !     For parallel IO: 
+    !          End definition stage for CDI resources, 
+    !          balance load of variables among collecting IO server,
+    !          Decompose data on model processes for IO
+    !          Transfer resources to the collecting IO server.
+    CALL pioEndDef (); 
+
+    !     Loop over the number of time steps
+    DO tsID = 0, nts-1
+      CALL taxisDefVdate ( taxisID, vdate + tsID )
+      CALL taxisDefVtime ( taxisID, vtime )
+      status = streamDefTimestep ( streamID, tsID )
+
+      ! For parallel IO:    
+      ! Inquire start index and chunk for IO transposition of var1
+      startID = pioInqVarDecoOff   ( vlistID, varID1 ) + 1
+      chunk   = pioInqVarDecoChunk ( vlistID, varID1 ) - 1
+      stopID = startID + chunk 
+      ! Init decomposed data for var1
+      DO i = startID, stopID
+        var1(i) = 1.1
+      END DO
+      ! Write var1
+      CALL streamWriteVar(streamID, varID1, var1 ( startID ), nmiss)
+
+      ! For parallel IO: 
+      ! Inquire start index and chunk for IO transposition of var2
+      startID = pioInqVarDecoOff ( vlistID, varID2 ) + 1
+      chunk = pioInqVarDecoChunk ( vlistID, varID2 ) - 1
+      stopID = startID + chunk
+      ! Init decomposed data for var2
+      DO i = startID, stopID
+        var2(i) = 2.2
+      END DO
+      ! Write var2
+      CALL streamWriteVar(streamID, varID2, var2 ( startID ), nmiss)
+
+      ! For parallel IO: 
+      ! Start transmission of all data for output in this timestep to IO server.
+      CALL pioWriteTimestep ( tsID, vdate, vtime )
+    END DO
+
+    ! For parallel IO:
+    ! Preparation for local cleanup
+    CALL pioEndTimestepping ()
+
+    !     Close the output stream
+    CALL streamClose(streamID)
+
+    !     Destroy the objects
+    CALL vlistDestroy(vlistID)
+    CALL taxisDestroy(taxisID)
+    CALL zaxisDestroy(zaxisID1)
+    CALL zaxisDestroy(zaxisID2)
+    CALL gridDestroy(gridID)
+
+  END SUBROUTINE modelrun
+
+END PROGRAM collectdata2003

examples/pio/collectDataNStreams.c

+#if defined (HAVE_CONFIG_H)
+#  include "config.h"
+#endif
+
+#ifdef USE_MPI
+#include <mpi.h>
+#include "pio_c_temp.h"
+#endif
+
+#include "cdi.h"
+#include "pio_util.h"
+
+void hoursPassingHack ( int * vdate, int * vtime, int hoursPassed )
+{
+  int sum, days, hours, oldDays;
+  
+  xassert ( hoursPassed % 10000 == 0 );
+  sum = * vtime + hoursPassed;
+  days = sum / 240000;
+  hours = sum % 240000;
+  oldDays = * vdate % 100;
+  if ( oldDays + days > 28 ) xabort ( "UNEXPECTED USE" );
+  * vtime = hours;
+  * vdate = * vdate + days;
+}
+ 
+void modelRun ()
+{
+  enum {
+    filetype    = FILETYPE_GRB,
+    nStreams    = 5,
+    MAXNSTREAMS = 25,
+    ntfiles     = 2,
+    ntsteps     = 3,
+    nVars       = 5,
+    nlon        = 12,
+    nlat        = 6,
+    MAXNLEV     = 5 };
+
+  static int nlev[nStreams][nVars] =
+    {{1,1,5,5,2},{3,5,2,2,1},{3,5,2,2,1},{5,2,2,2,1}, {3,3,3,3,3}};
+
+  static char nameExp[7]           = "example";
+  static int asciiA                = 65;
+  char nameStream[8];
+  char filename[1024];
+
+  int gridID, zaxisID[nStreams][nVars], taxisID;
+  int streamID[nStreams], vlistID[nStreams], varID[nStreams][nVars], tsID, tfID = 0;
+  int i, j, k, nmiss = 0, rank;
+  double lons[nlon] = {0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330};
+  double lats[nlat] = {-75, -45, -15, 15, 45, 75};
+  double levs[MAXNLEV] = {101300, 92500, 85000, 50000, 20000};
+  double var[nlon*nlat*MAXNLEV];
+  int vdate = 19850101, vtime = 120000, hourStep = 20000;
+  int start = CDI_UNDEFID, chunk = CDI_UNDEFID, stop = CDI_UNDEFID;
+
+  xassert ( nStreams < MAXNSTREAMS );
+
+  gridID = gridCreate ( GRID_LONLAT, nlon*nlat );
+  gridDefXsize ( gridID, nlon );
+  gridDefYsize ( gridID, nlat );
+  gridDefXvals ( gridID, lons );
+  gridDefYvals ( gridID, lats );
+
+  for ( i = 0; i < nStreams; i++ )
+    {
+      for ( j = 0; j < nVars; j++ )
+        {
+          xassert ( nlev[i][j] > 0 && nlev[i][j] <= MAXNLEV );
+          zaxisID[i][j] = zaxisCreate ( ZAXIS_PRESSURE, nlev[i][j] );
+          zaxisDefLevels ( zaxisID[i][j], levs );
+        }
+      vlistID[i] = vlistCreate ();
+    }
+
+  for ( i = 0; i < nStreams; i++ )
+    for ( j = 0; j < nVars; j++ )
+      varID[i][j] = vlistDefVar ( vlistID[i], gridID, zaxisID[i][j], TIME_VARIABLE );
+
+  taxisID = taxisCreate ( TAXIS_ABSOLUTE );
+  for ( i = 0; i < nStreams; i++ )
+    vlistDefTaxis ( vlistID[i], taxisID );
+
+  for ( i = 0; i < nStreams; i++ )
+    {
+      memset ( filename, 0, 1024 );
+      sprintf ( &filename[0], "%s%c_%d.grb", nameExp, asciiA + i, tfID );
+      streamID[i] = streamOpenWrite ( filename, filetype );
+      xassert ( streamID[i] >= 0 );
+      streamDefVlist ( streamID[i], vlistID[i]);
+    }
+
+  pioEndDef ();
+
+  for ( tfID = 0; tfID < ntfiles; tfID++ )
+    {
+      if ( tfID > 0 )
+	{
+          for ( i = 0; i < nStreams; i++ )
+            {
+              streamClose ( streamID[i] );
+              sprintf ( &filename[0], "%s%c_%d.grb", nameExp, asciiA + i, tfID );
+              streamID[i] = streamOpenWrite ( filename, filetype );
+              xassert ( streamID[i] >= 0 );
+              streamDefVlist ( streamID[i], vlistID[i] );
+            }
+	}
+
+      for ( tsID = 0; tsID < ntsteps; tsID++ )
+	{
+          hoursPassingHack ( &vdate, &vtime, hourStep );
+	  taxisDefVdate ( taxisID, vdate );
+	  taxisDefVtime ( taxisID, vtime );
+          for ( i = 0; i < nStreams; i++ )
+            {
+              streamDefTimestep ( streamID[i], tsID );
+              for ( j = 0; j < nVars; j++ )
+                {
+                  start = pioInqVarDecoOff   ( vlistID[i], varID[i][j] );
+                  chunk = pioInqVarDecoChunk ( vlistID[i], varID[i][j] );
+                  stop = start + chunk;
+                  for ( k = start; k < stop; k++ ) var[k] = 2.2;
+                  streamWriteVar ( streamID[i], varID[i][j], &var[start], nmiss );
+                  start = CDI_UNDEFID;
+                  chunk = CDI_UNDEFID;
+                }
+            }
+	  pioWriteTimestep ( tsID, vdate, vtime );
+	}
+    }
+
+  pioEndTimestepping ();
+
+  for ( i = 0; i < nStreams; i++ )
+    {
+      streamClose ( streamID[i] );
+      vlistDestroy ( vlistID[i] );
+    }
+  taxisDestroy ( taxisID );
+  for ( i = 0; i < nStreams; i++ )
+    for ( j = 0; j < nVars; j++ )
+      zaxisDestroy ( zaxisID[i][j] );
+  gridDestroy ( gridID );
+  xdebug("%s", "RETURN");
+}
+
+
+int main (int argc, char *argv[])
+{
+  enum {
+    nProcsIODef    = 3,
+    //IOModeDef       = PIO_NONE,
+    //IOModeDef       = PIO_MPI,
+    IOModeDef       = PIO_FPGUARD,
+    //IOModeDef       = PIO_ASYNCH,
+    //IOModeDef       = PIO_WRITER,
+    nNamespaces = 1 };
+
+  static int hasLocalFile[nNamespaces] = { 0 };
+
+#ifdef USE_MPI
+  MPI_Comm commGlob, commModel;
+  int sizeGlob;
+  int rankGlob;
+  int IOMode, nProcsIO, count;
+
+  xmpi ( MPI_Init ( &argc, &argv));
+  xmpi ( MPI_Comm_dup ( MPI_COMM_WORLD, &commGlob ));
+  xmpi ( MPI_Comm_set_errhandler ( commGlob, MPI_ERRORS_RETURN ));
+  xmpi ( MPI_Comm_size ( commGlob, &sizeGlob ));
+  xmpi ( MPI_Comm_rank ( commGlob, &rankGlob ));
+
+  if (argc > 1)
+    {
+      xassert ( argc >= 3 );