Fixes to issue #360. Regression caused by #352 . (#361)

Author: dilawar

basecode/header.h

@@ -99,7 +99,6 @@ class Neutral;
 #include "SharedFinfo.h"
 #include "FieldElementFinfo.h"
 #include "FieldElement.h"
-#include "../builtins/Streamer.h"
 #include "../shell/Neutral.h"
 
 

builtins/Streamer.cpp

@@ -1,17 +1,8 @@
 /***
- *       Filename:  Streamer.cpp
- *
  *    Description:  Stream table data.
  *
- *        Version:  0.0.1
- *        Created:  2016-04-26
-
- *       Revision:  none
- *
  *         Author:  Dilawar Singh <[email protected]>
  *   Organization:  NCBS Bangalore
- *
- *        License:  GNU GPL2
  */
 
 #include <algorithm>
@@ -45,12 +36,18 @@ const Cinfo* Streamer::initCinfo()
         , &Streamer::getFormat
     );
 
-    static ReadOnlyValueFinfo< Streamer, size_t > numTables (
+    static ReadOnlyValueFinfo<Streamer, size_t> numTables (
         "numTables"
         , "Number of Tables handled by Streamer "
         , &Streamer::getNumTables
     );
 
+    static ReadOnlyValueFinfo<Streamer, size_t> numWriteEvents(
+        "numWriteEvents"
+        , "Number of time streamer was called to write. (For debugging/performance reason only)"
+        , &Streamer::getNumWriteEvents
+    );
+
     /*-----------------------------------------------------------------------------
      *
      *-----------------------------------------------------------------------------*/
@@ -96,7 +93,7 @@ const Cinfo* Streamer::initCinfo()
      *-----------------------------------------------------------------------------*/
     static Finfo* procShared[] =
     {
-        &process , &reinit , &addTable, &addTables, &removeTable, &removeTables
+        &process, &reinit, &addTable, &addTables, &removeTable, &removeTables
     };
 
     static SharedFinfo proc(
@@ -107,7 +104,7 @@ const Cinfo* Streamer::initCinfo()
 
     static Finfo * tableStreamFinfos[] =
     {
-        &outfile, &format, &proc, &numTables
+        &outfile, &format, &proc, &numTables, &numWriteEvents
     };
 
     static string doc[] =
@@ -140,8 +137,9 @@ Streamer::Streamer()
 {
     // Not all compilers allow initialization during the declaration of class
     // methods.
-    format_ = "npy";
-    columns_.push_back( "time" );               /* First column is time. */
+    format_ = "csv";
+    numWriteEvents_ = 0;
+    columns_.push_back("time");               /* First column is time. */
     tables_.resize(0);
     tableIds_.resize(0);
     tableTick_.resize(0);
@@ -238,7 +236,7 @@ void Streamer::reinit(const Eref& e, ProcPtr p)
     // write now.
     currTime_ = 0.0;
     zipWithTime( );
-    StreamerBase::writeToOutFile( outfilePath_, format_, "w", data_, columns_);
+    StreamerBase::writeToOutFile(outfilePath_, format_, "w", data_, columns_);
     data_.clear( );
 }
 
@@ -261,15 +259,11 @@ void Streamer::cleanUp( )
  */
 void Streamer::process(const Eref& e, ProcPtr p)
 {
-    //LOG( moose::debug, "Writing to table" );
+    // LOG( moose::debug, "Writing Streamer data to file." );
     zipWithTime( );
-
-    // Write only if there are more than 100 entry in first table.
-    if( tables_[0]->getVecSize() > 100 )
-    {
-        StreamerBase::writeToOutFile( outfilePath_, format_, "a", data_, columns_ );
-        data_.clear( );
-    }
+    StreamerBase::writeToOutFile( outfilePath_, format_, "a", data_, columns_ );
+    data_.clear();
+    numWriteEvents_ += 1;
 }
 
 
@@ -356,6 +350,19 @@ size_t Streamer::getNumTables( void ) const
     return tables_.size();
 }
 
+/* --------------------------------------------------------------------------*/
+/**
+ * @Synopsis  Get number of write events in streamer. Useful for debugging and
+ * performance measuerments.
+ *
+ * @Returns   
+ */
+/* ----------------------------------------------------------------------------*/
+size_t Streamer::getNumWriteEvents( void ) const
+{
+    return numWriteEvents_;
+}
+
 
 string Streamer::getOutFilepath( void ) const
 {

builtins/Streamer.h

@@ -49,6 +49,7 @@ class Streamer : public StreamerBase
     void setFormat( string format );
 
     size_t getNumTables( void ) const;
+    size_t getNumWriteEvents( void ) const;
 
     void addTable( Id table );
     void addTables( vector<Id> tables);
@@ -74,6 +75,9 @@ class Streamer : public StreamerBase
 
     string outfilePath_;
     string format_;
+
+    size_t numWriteEvents_;
+
     bool isOutfilePathSet_;
 
     // dt_ and tick number of Table's clock

builtins/StreamerBase.cpp

@@ -91,14 +91,14 @@ void StreamerBase::writeToCSVFile( const string& filepath, const string& openmod
     }
 
     // If writing in "w" mode, write the header first.
-    if( openmode == "w" )
+    if(openmode == "w")
     {
         string headerText = "";
         for( vector<string>::const_iterator it = columns.begin();
             it != columns.end(); it++ )
             headerText += ( *it + delimiter_ );
         headerText += eol;
-        fprintf( fp, "%s", headerText.c_str() );
+        fprintf(fp, "%s", headerText.c_str());
     }
 
     string text = "";
@@ -111,8 +111,8 @@ void StreamerBase::writeToCSVFile( const string& filepath, const string& openmod
         // At the end of each row, we remove the delimiter_ and append newline_.
         *(text.end()-1) = eol;
     }
-    fprintf( fp, "%s", text.c_str() );
-    fclose( fp );
+    fprintf(fp, "%s", text.c_str() );
+    fclose(fp);
 }
 
 /*  write data to a numpy file */

shell/Shell.cpp

@@ -22,6 +22,7 @@
 #include "../msg/OneToAllMsg.h"
 #include "../msg/SparseMsg.h"
 #include "../builtins/SocketStreamer.h"
+#include "../builtins/Streamer.h"
 
 #include "Shell.h"
 #include "Wildcard.h"

tests/python/test_streamer.py

@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
-"""test_streamer.py:
+from __future__ import print_function
 
+"""test_streamer.py:
 Test script for Streamer class.
-
 """
 
 __author__           = "Dilawar Singh"
@@ -14,13 +14,17 @@
 __email__            = "[email protected]"
 __status__           = "Development"
 
-import os
-import sys
-import time
 import moose
+import threading
 import numpy as np
-print(( '[INFO] Using moose form %s' % moose.__file__ ))
+import time
+import os
+import sys
+print('[INFO] Using moose form %s' % moose.__file__)
+
+all_done_ = False
 
+# Poll the file to see that we are really writing to it.
 def sanity_test( ):
     a = moose.Table( '/t1' )
     b = moose.Table( '/t1/t1' )
@@ -64,6 +68,7 @@ def sanity_test( ):
 
 def test( ):
     compt = moose.CubeMesh( '/compt' )
+    assert compt
     r = moose.Reac( '/compt/r' )
     a = moose.Pool( '/compt/a' )
     a.concInit = 1
@@ -77,6 +82,10 @@ def test( ):
     r.Kf = 0.1
     r.Kb = 0.01
 
+    outfile = 'streamer_test.csv'
+    if os.path.exists(outfile):
+        os.remove(outfile)
+
     tabA = moose.Table2( '/compt/a/tab' )
     tabB = moose.Table2( '/compt/tabB' )
     tabC = moose.Table2( '/compt/tabB/tabC' )
@@ -88,33 +97,32 @@ def test( ):
 
     # Now create a streamer and use it to write to a stream
     st = moose.Streamer( '/compt/streamer' )
-    st.outfile = os.path.join( os.getcwd(), 'temp.npy' )
-    print(("outfile set to: %s " % st.outfile ))
-    assert st.outfile  == os.path.join( os.getcwd(), 'temp.npy' ), st.outfile
+    st.outfile = outfile
 
+    print("outfile set to: %s " % st.outfile )
     st.addTable( tabA )
     st.addTables( [ tabB, tabC ] )
-
     assert st.numTables == 3
 
     moose.reinit( )
-    print( '[INFO] Running for 57 seconds' )
-    moose.start( 57 )
+    t = 100
+    print( '[INFO] Running for %d seconds' % t )
+    moose.start(t)
     outfile = st.outfile
     moose.quit() # Otherwise Streamer won't flush the rest of entries.
 
+    print('Moose is done. Waiting for monitor to shut down...')
+
     # Now read the table and verify that we have written
     print( '[INFO] Reading file %s' % outfile )
     if 'csv' in outfile:
         data = np.loadtxt(outfile, skiprows=1 )
     else:
         data = np.load( outfile )
     # Total rows should be 58 (counting zero as well).
-    # print(data)
+    #  print(data)
     # print( data.dtype )
-    time = data['time']
-    print( time )
-    assert data.shape >= (58,), data.shape
+    assert data.shape >= (101,), data.shape
     print( '[INFO] Test 2 passed' )
     return 0
 

tests/python/test_table_streaming_support.py

@@ -59,7 +59,7 @@ def test( ):
     moose.reinit( )
     [ print_table( x) for x in [tabA, tabB, tabC] ]
     runtime = 1000
-    print( 'Starting moose for %s' % runtime )
+    print( 'Starting moose for %d secs' % runtime )
     moose.start( runtime, 1 )
     print( ' MOOSE is done' )