Cleanup and start feature close files with defined resources in MPINONB

pioExamples/collectData.c

@@ -11,7 +11,7 @@
 #include "pio_util.h"
 #include <assert.h>
 
-#define nProcsIO 3
+#define nProcsIO 1
 
 static int hasLocalFile[] = { 0 };
 
@@ -24,8 +24,8 @@ static int nlev[nVars] = {1,1,5,5,2};
 #define ntsteps  3 
 
 //int IOMode = PIO_NONE;
-//int IOMode = PIO_MPI_NONB;
-int IOMode = PIO_POSIX_FPGUARD_SENDRECV;
+int IOMode = PIO_MPI_NONB;
+//int IOMode = PIO_POSIX_FPGUARD_SENDRECV;
 //int IOMode = PIO_POSIX_ASYNCH;
 //int IOMode = PIO_POSIX_NONB;
 
@@ -96,6 +96,8 @@ void modelRun ()
 	}
       if ( tsID == 1 )
 	streamID[1] = streamOpenWrite ( "example2.grb", FILETYPE_GRB );
+      if ( tsID == 2 )
+	streamClose ( streamID[1] );
       pioWriteTimestep ( tsID, vdate, vtime );
     }
 
@@ -104,7 +106,6 @@ void modelRun ()
    
   
   streamClose ( streamID[0] );
-  streamClose ( streamID[1] );
   vlistDestroy ( vlistID );
   taxisDestroy ( taxisID );
   for ( i = 0; i < nVars; i++ )

src/namespace.c

@@ -81,7 +81,7 @@ void pioNamespaceInit ( int nspn, int * argHasLocalFile )
       hasLocalFile = xmalloc ( nspn * sizeof ( hasLocalFile[0] ));
       for ( i = 0; i < nspn; i++ )  
 	hasLocalFile[i] = argHasLocalFile[i];
-      resASent = xmalloc ( nspn * sizeof ( resASent[0] ));
+      resASent = xmalloc ( nspn * sizeof ( resASent[0] )); 
     }
 #endif
 }
@@ -90,10 +90,11 @@ void pioNamespaceInit ( int nspn, int * argHasLocalFile )
 void pioNamespaceCleanup ( void )
 {
   if ( nNamespaces > 1 ) 
-      {
-          free ( hasLocalFile );
-          hasLocalFile = NULL;
-      }
+    {
+      free ( hasLocalFile );
+      hasLocalFile = NULL;
+      free ( resASent );
+    }
 }
 
 

src/pio.c

@@ -181,6 +181,21 @@ void backendInit ( void )
 
 /***************************************************************/
 
+void backendCleanup ( void )
+{
+  int IOMode = commInqIOMode ();
+  switch ( IOMode )
+    {
+    case PIO_MPI_NONB:
+      finalizeMPINONB ();
+      break;
+    default:
+      xdebug ( " backendCleanup function not implemented yet." );
+    }
+}
+
+/***************************************************************/
+
 void backendFinalize ( void )
 {
   commDestroy ();

src/pio.h

@@ -9,6 +9,7 @@
 #include <stdlib.h>
 #include <mpi.h>
 
+void   backendCleanup  ( void );
 void   backendInit     ( void );
 void   backendFinalize ( void );
 int    pioFileOpenW    ( const char* );

src/pio_impl.h

@@ -77,6 +77,7 @@ int       fowMPINONB ( const char * );
 int       fcMPINONB ( int );
 size_t    fwMPINONB( int, int, const void *, size_t );
 void      initMPINONB ( void );
+void      finalizeMPINONB ( void );
 
 /* pio_posixasynch.c */
 #ifndef _SX

src/pio_interface.c

@@ -254,16 +254,16 @@ void varsMapNDeco ( int nNodes, int * nodeSizes )
 
   nStreams = streamSize ();
 
-  resHs       = xmalloc ( nStreams * sizeof ( int ));
-  streamSizes = xmalloc ( nStreams * sizeof ( int ));
+  resHs       = xmalloc ( nStreams * sizeof ( resHs[0] ));
+  streamSizes = xmalloc ( nStreams * sizeof ( streamSizes[0] ));
   streamGetIndexArray ( nStreams, resHs );
 
   for ( i = 0; i < nStreams; i++ )
     streamSizes[i] = streamInqNvars ( * ( resHs + i )); 
 
   nVars = xsum ( nStreams, streamSizes );
-  varSizes   = xmalloc ( nVars * sizeof ( int ));
-  varMapping = xmalloc ( nVars * sizeof ( int ));
+  varSizes   = xmalloc ( nVars * sizeof ( varSizes[0] ));
+  varMapping = xmalloc ( nVars * sizeof ( varMapping[0] ));
 
   for ( i = 0; i < nStreams; i++ )
     for ( j = 0; j < * ( streamSizes + i ); j++ )
@@ -546,6 +546,9 @@ void pioBufferFuncCall ( const char * func, int argc, ... )
   
   switch ( funcID ) 
     {
+    case STREAMCLOSE:
+      xdebug ( "CASE STREAMCLOSE" );
+      break;
     case STREAMOPEN:
       {
         char * filename;

src/pio_mpinonb.c

@@ -361,6 +361,21 @@ int fowMPINONB ( const char *filename )
 
 /***************************************************************/
 
+void      finalizeMPINONB ( void )
+{
+  if ( bibAFiledataM->head != NULL )
+    {
+      xabort ( "queue bibAFiledataM not empty" );
+    }
+  else
+    {
+      xdebug ( "cleanup queue" );
+      free ( bibAFiledataM );
+    }
+}
+
+/***************************************************************/
+
 void initMPINONB ( void )
 {
   commDefCommColl ( 1 );

src/pio_server.c

@@ -398,7 +398,20 @@ void IOServer ()
                 if ( commInqRankGlob () == nProcsModel ) 
                   reshArrayPrint ( "reshArrayIOServer" );
 
+                {
+                  int nStreams = streamSize (), streamNo;
+                  int * resHs;
+
+                  resHs       = xmalloc ( nStreams * sizeof ( resHs[0] ));
+                  streamGetIndexArray ( nStreams, resHs );
+                  for ( streamNo = 0; streamNo < nStreams; streamNo++ )
+                    streamClose ( resHs[streamNo] );
+                  free ( resHs );
+                  backendCleanup ();
+                }
+
                 serverWinCleanup ();
+                xdebug ();
                 arrayDestroy();
                 return;
               }

src/stream.c

@@ -1125,9 +1125,13 @@ void streamClose(int streamID)
   int vlistID;
   stream_t *streamptr;
 
+  xdebug ( "streamID=%d", streamID );
+
   streamptr = stream_to_pointer(streamID);
+  xdebug ();
 
   stream_check_ptr(__func__, streamptr);
+  xdebug ();
 
   if ( CDI_Debug )
     Message("fileID = %d filename = %s", streamID, streamptr->filename);
@@ -1193,6 +1197,11 @@ void streamClose(int streamID)
 	    }
 	  }
     }
+#ifdef USE_MPI
+  else
+    if ( namespaceInqResASent () == 1 )
+      pioBufferFuncCall ( __func__, 1, streamID );
+#endif
       
   if ( streamptr->record )
       {