Flush results CSV file after each line

pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.ldbc.driver</groupId>
     <artifactId>jeeves</artifactId>
-    <version>0.3.3</version>
+    <version>0.3.4</version>
     <packaging>jar</packaging>
 
     <name>LDBC Driver</name>

src/main/java/com/ldbc/driver/client/CreateValidationParamsMode.java

@@ -133,7 +133,8 @@ public Object startExecutionAndAwaitCompletion() throws ClientException
             int rowsWrittenSoFar = 0;
             try ( SimpleCsvFileWriter simpleCsvFileWriter = new SimpleCsvFileWriter(
                     validationFileToGenerate,
-                    SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR ) )
+                    SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR,
+                    controlService.configuration().flushLog() ) )
             {
                 DecimalFormat decimalFormat = new DecimalFormat( "###,###,##0" );
                 while ( csvRows.hasNext() )

src/main/java/com/ldbc/driver/client/ExecuteWorkloadMode.java

@@ -160,7 +160,10 @@ private void doInit( boolean warmup ) throws ClientException
         {
             resultsLogWriter = (null == resultsLog)
                                ? new NullResultsLogWriter()
-                               : new SimpleResultsLogWriter( resultsLog, controlService.configuration().timeUnit() );
+                               : new SimpleResultsLogWriter(
+                                    resultsLog,
+                                    controlService.configuration().timeUnit(),
+                                    controlService.configuration().flushLog() );
         }
         catch ( IOException e )
         {

src/main/java/com/ldbc/driver/control/ConsoleAndFileDriverConfiguration.java

@@ -77,6 +77,11 @@ public class ConsoleAndFileDriverConfiguration implements DriverConfiguration
     public static final String HELP_DEFAULT_STRING = Boolean.toString( HELP_DEFAULT );
     private static final String HELP_DESCRIPTION = "print usage instruction";
 
+    public static final String FLUSH_LOG_ARG = "flush_log";
+    public static final boolean FLUSH_LOG_DEFAULT = false;
+    public static final String FLUSH_LOG_DEFAULT_STRING = Boolean.toString(FLUSH_LOG_DEFAULT);
+    private static final String FLUSH_LOG_DESCRIPTION = "flush log to disk after each operation";
+
     public static final String NAME_ARG = "nm";
     private static final String NAME_ARG_LONG = "name";
     public static final String NAME_DEFAULT = "LDBC";
@@ -185,6 +190,7 @@ public static Map<String,String> defaultsAsMap() throws DriverConfigurationExcep
         Map<String,String> defaultParamsMap = new HashMap<>();
         defaultParamsMap.put( IGNORE_SCHEDULED_START_TIMES_ARG, IGNORE_SCHEDULED_START_TIMES_DEFAULT_STRING );
         defaultParamsMap.put( HELP_ARG, HELP_DEFAULT_STRING );
+        defaultParamsMap.put( FLUSH_LOG_ARG, FLUSH_LOG_DEFAULT_STRING );
         defaultParamsMap.put( OPERATION_COUNT_ARG, OPERATION_COUNT_DEFAULT_STRING );
         defaultParamsMap.put( WORKLOAD_ARG, WORKLOAD_DEFAULT_STRING );
         defaultParamsMap.put( NAME_ARG, NAME_DEFAULT_STRING );
@@ -280,6 +286,7 @@ public static ConsoleAndFileDriverConfiguration fromParamsMap( Map<String,String
             boolean printHelp = Boolean.parseBoolean( paramsMap.get( HELP_ARG ) );
             boolean ignoreScheduledStartTimes =
                     Boolean.parseBoolean( paramsMap.get( IGNORE_SCHEDULED_START_TIMES_ARG ) );
+            boolean flushLog = Boolean.parseBoolean( paramsMap.get( FLUSH_LOG_ARG ) );
             return new ConsoleAndFileDriverConfiguration(
                     paramsMap,
                     name,
@@ -298,7 +305,8 @@ public static ConsoleAndFileDriverConfiguration fromParamsMap( Map<String,String
                     printHelp,
                     ignoreScheduledStartTimes,
                     warmupCount,
-                    skipCount
+                    skipCount,
+                    flushLog
             );
         }
         catch ( DriverConfigurationException e )
@@ -605,6 +613,10 @@ private static Options buildOptions()
         Option printHelpOption = OptionBuilder.withDescription( HELP_DESCRIPTION ).create( HELP_ARG );
         options.addOption( printHelpOption );
 
+
+        Option flushOption = OptionBuilder.withDescription(FLUSH_LOG_DESCRIPTION).create(FLUSH_LOG_ARG);
+        options.addOption( flushOption );
+
         Option ignoreScheduledStartTimesOption =
                 OptionBuilder.withDescription( IGNORE_SCHEDULED_START_TIMES_DESCRIPTION )
                         .create( IGNORE_SCHEDULED_START_TIMES_ARG );
@@ -684,6 +696,7 @@ public static String commandlineHelpString()
     private final boolean ignoreScheduledStartTimes;
     private final long warmupCount;
     private final long skipCount;
+    private final boolean flushLog;
 
     public ConsoleAndFileDriverConfiguration( Map<String,String> paramsMap,
             String name,
@@ -702,7 +715,8 @@ public ConsoleAndFileDriverConfiguration( Map<String,String> paramsMap,
             boolean printHelp,
             boolean ignoreScheduledStartTimes,
             long warmupCount,
-            long skipCount )
+            long skipCount,
+            boolean flushLog )
     {
         if ( null == paramsMap )
         {
@@ -726,6 +740,7 @@ public ConsoleAndFileDriverConfiguration( Map<String,String> paramsMap,
         this.ignoreScheduledStartTimes = ignoreScheduledStartTimes;
         this.warmupCount = warmupCount;
         this.skipCount = skipCount;
+        this.flushLog = flushLog;
 
         if ( null != name )
         {
@@ -762,6 +777,7 @@ public ConsoleAndFileDriverConfiguration( Map<String,String> paramsMap,
         paramsMap.put( IGNORE_SCHEDULED_START_TIMES_ARG, Boolean.toString( ignoreScheduledStartTimes ) );
         paramsMap.put( WARMUP_COUNT_ARG, Long.toString( warmupCount ) );
         paramsMap.put( SKIP_COUNT_ARG, Long.toString( skipCount ) );
+        paramsMap.put( FLUSH_LOG_ARG, Boolean.toString( flushLog ) );
     }
 
     @Override
@@ -879,6 +895,9 @@ public long skipCount()
         return skipCount;
     }
 
+    @Override
+    public boolean flushLog() { return flushLog; }
+
     @Override
     public Map<String,String> asMap()
     {
@@ -995,6 +1014,9 @@ public DriverConfiguration applyArgs( Map<String,String> newParamsMap ) throws D
         long newSkipCount = (newParamsMapWithShortKeys.containsKey( SKIP_COUNT_ARG )) ?
                             Long.parseLong( newParamsMapWithShortKeys.get( SKIP_COUNT_ARG ) ) :
                             skipCount;
+        boolean newFlushLog = (newParamsMapWithShortKeys.containsKey( FLUSH_LOG_ARG )) ?
+                        Boolean.parseBoolean( newParamsMapWithShortKeys.get( FLUSH_LOG_ARG ) ) :
+                        flushLog;
 
         return new ConsoleAndFileDriverConfiguration(
                 newOtherParams,
@@ -1014,7 +1036,8 @@ public DriverConfiguration applyArgs( Map<String,String> newParamsMap ) throws D
                 newPrintHelp,
                 newIgnoreScheduledStartTimes,
                 newWarmupCount,
-                newSkipCount
+                newSkipCount,
+                newFlushLog
         );
     }
 

src/main/java/com/ldbc/driver/control/DriverConfiguration.java

@@ -41,6 +41,8 @@ public interface DriverConfiguration
 
     long skipCount();
 
+    boolean flushLog();
+
     String toPropertiesString() throws DriverConfigurationException;
 
     Map<String,String> asMap();

src/main/java/com/ldbc/driver/csv/simple/SimpleCsvFileWriter.java

@@ -16,13 +16,15 @@ public class SimpleCsvFileWriter implements Closeable
 
     private final BufferedWriter bufferedWriter;
     private final String columnSeparator;
+    private final boolean flushLog;
 
-    public SimpleCsvFileWriter( File file, String columnSeparator ) throws IOException
+    public SimpleCsvFileWriter( File file, String columnSeparator, boolean flushLog ) throws IOException
     {
         this.bufferedWriter =
                 new BufferedWriter( new OutputStreamWriter( new FileOutputStream( file ), Charsets.UTF_8 ) );
 
         this.columnSeparator = columnSeparator;
+        this.flushLog = flushLog;
     }
 
     public void writeRows( Iterator<String[]> csvRows ) throws IOException
@@ -40,6 +42,9 @@ public void writeRow( String... columns ) throws IOException
         }
         bufferedWriter.write( columns[columns.length - 1] );
         bufferedWriter.newLine();
+        if (flushLog) {
+            bufferedWriter.flush();
+        }
     }
 
     @Override

src/main/java/com/ldbc/driver/runtime/metrics/SimpleResultsLogWriter.java

@@ -11,9 +11,9 @@ public class SimpleResultsLogWriter implements ResultsLogWriter
     private final SimpleCsvFileWriter writer;
     private final TimeUnit unit;
 
-    public SimpleResultsLogWriter( File resultsLog, TimeUnit unit ) throws IOException
+    public SimpleResultsLogWriter( File resultsLog, TimeUnit unit, boolean flushLog ) throws IOException
     {
-        this.writer = new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR );
+        this.writer = new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR, flushLog );
         this.unit = unit;
         resultsLog.createNewFile();
         writer.writeRow(

src/main/resources/configuration/ldbc_driver_default.properties

@@ -81,6 +81,12 @@ help=false
 # COMMAND: -ignore_scheduled_start_times
 ignore_scheduled_start_times=false
 
+# flush log to disk after each operation
+# useful for durability tests
+# BOOLEAN
+# COMMAND: -flush_log
+flush_log=false
+
 # ***************************************************************
 # *** the following should be set by workload implementations ***
 # ***************************************************************

src/test/java/com/ldbc/driver/control/ConsoleAndFileDriverConfigurationTest.java

@@ -107,6 +107,7 @@ public void toMapThenFromMapShouldReturnSameResultWhenAllParamsAreInitiallySetVi
         long warmupCount = 5;
         long skipCount = 6;
         Map<String,String> paramsMap = new HashMap<>();
+        boolean flushLog = false;
 
         ConsoleAndFileDriverConfiguration configurationBefore = new ConsoleAndFileDriverConfiguration(
                 paramsMap,
@@ -126,7 +127,8 @@ public void toMapThenFromMapShouldReturnSameResultWhenAllParamsAreInitiallySetVi
                 printHelp,
                 ignoreScheduledStartTimes,
                 warmupCount,
-                skipCount
+                skipCount,
+                flushLog
         );
 
         DriverConfiguration configurationAfter =
@@ -565,6 +567,7 @@ public void shouldReturnSameAsConstructedWith()
         boolean ignoreScheduledStartTimes = false;
         long warmupCount = 10;
         long skipCount = 100;
+        boolean flushLog = false;
 
         ConsoleAndFileDriverConfiguration params = new ConsoleAndFileDriverConfiguration(
                 paramsMap,
@@ -584,7 +587,8 @@ public void shouldReturnSameAsConstructedWith()
                 printHelp,
                 ignoreScheduledStartTimes,
                 warmupCount,
-                skipCount
+                skipCount,
+                flushLog
         );
 
         assertThat( params.asMap(), equalTo( paramsMap ) );

src/test/java/com/ldbc/driver/generator/TimeMappingOperationGeneratorTest.java

@@ -362,6 +362,7 @@ public void shouldNotBreakTheMonotonicallyIncreasingScheduledStartTimesOfOperati
         boolean ignoreScheduledStartTimes = false;
         long warmupCount = 0;
         long skipCount = 0;
+        boolean flushLog = false;
 
         ConsoleAndFileDriverConfiguration configuration = new ConsoleAndFileDriverConfiguration(
                 paramsMap,
@@ -381,7 +382,8 @@ public void shouldNotBreakTheMonotonicallyIncreasingScheduledStartTimesOfOperati
                 printHelp,
                 ignoreScheduledStartTimes,
                 warmupCount,
-                skipCount
+                skipCount,
+                flushLog
         );
 
         Map<String,String> updateStreamParams = MapUtils.loadPropertiesToMap(

src/test/java/com/ldbc/driver/runtime/QueuePerformanceTests.java

@@ -86,6 +86,7 @@ public void operationQueuePerformanceTest()
         boolean ignoreScheduledStartTimes = false;
         long warmupCount = 0;
         long skipCount = 0;
+        boolean flushLog = false;
 
         DriverConfiguration config = new ConsoleAndFileDriverConfiguration(
                 paramsMap,
@@ -105,7 +106,8 @@ public void operationQueuePerformanceTest()
                 printHelp,
                 ignoreScheduledStartTimes,
                 warmupCount,
-                skipCount
+                skipCount,
+                flushLog
         );
 
         GeneratorFactory gf = new GeneratorFactory( new RandomDataGeneratorFactory( 42L ) );

src/test/java/com/ldbc/driver/runtime/WorkloadRunnerTest.java

@@ -114,6 +114,7 @@ public void shouldRunReadOnlyLdbcWorkloadWithNothingDbAndCrashInSaneManner()
             boolean ignoreScheduledStartTimes = false;
             long warmupCount = 100;
             long skipCount = 10;
+            boolean flushLog = false;
 
             ConsoleAndFileDriverConfiguration configuration = new ConsoleAndFileDriverConfiguration(
                     paramsMap,
@@ -133,7 +134,8 @@ public void shouldRunReadOnlyLdbcWorkloadWithNothingDbAndCrashInSaneManner()
                     printHelp,
                     ignoreScheduledStartTimes,
                     warmupCount,
-                    skipCount
+                    skipCount,
+                    flushLog
             );
 
             configuration = (ConsoleAndFileDriverConfiguration) configuration
@@ -171,7 +173,7 @@ public void shouldRunReadOnlyLdbcWorkloadWithNothingDbAndCrashInSaneManner()
 
             File resultsLog = temporaryFolder.newFile();
             SimpleCsvFileWriter csvResultsLogWriter =
-                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR );
+                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR, flushLog );
             metricsService = ThreadedQueuedMetricsService.newInstanceUsingBlockingBoundedQueue(
                     timeSource,
                     errorReporter,
@@ -315,6 +317,7 @@ private void doShouldRunReadOnlyLdbcWorkloadWithNothingDbAndReturnExpectedMetric
             boolean ignoreScheduledStartTimes = false;
             long warmupCount = 100;
             long skipCount = 10;
+            boolean flushLog = false;
 
             ConsoleAndFileDriverConfiguration configuration = new ConsoleAndFileDriverConfiguration(
                     paramsMap,
@@ -334,7 +337,8 @@ private void doShouldRunReadOnlyLdbcWorkloadWithNothingDbAndReturnExpectedMetric
                     printHelp,
                     ignoreScheduledStartTimes,
                     warmupCount,
-                    skipCount
+                    skipCount,
+                    flushLog
             );
 
             configuration = (ConsoleAndFileDriverConfiguration) configuration
@@ -372,7 +376,7 @@ private void doShouldRunReadOnlyLdbcWorkloadWithNothingDbAndReturnExpectedMetric
 
             File resultsLog = temporaryFolder.newFile();
             SimpleCsvFileWriter csvResultsLogWriter =
-                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR );
+                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR, flushLog );
             metricsService = ThreadedQueuedMetricsService.newInstanceUsingBlockingBoundedQueue(
                     timeSource,
                     errorReporter,
@@ -534,6 +538,7 @@ public void doShouldRunReadWriteLdbcWorkloadWithNothingDbAndReturnExpectedMetric
             boolean ignoreScheduledStartTimes = false;
             long warmupCount = 100;
             long skipCount = 10;
+            boolean flushLog = false;
 
             ConsoleAndFileDriverConfiguration configuration = new ConsoleAndFileDriverConfiguration(
                     paramsMap,
@@ -553,7 +558,8 @@ public void doShouldRunReadWriteLdbcWorkloadWithNothingDbAndReturnExpectedMetric
                     printHelp,
                     ignoreScheduledStartTimes,
                     warmupCount,
-                    skipCount
+                    skipCount,
+                    flushLog
             );
 
             configuration = (ConsoleAndFileDriverConfiguration) configuration
@@ -591,7 +597,7 @@ public void doShouldRunReadWriteLdbcWorkloadWithNothingDbAndReturnExpectedMetric
 
             File resultsLog = temporaryFolder.newFile();
             SimpleCsvFileWriter csvResultsLogWriter =
-                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR );
+                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR, flushLog );
             metricsService = ThreadedQueuedMetricsService.newInstanceUsingBlockingBoundedQueue(
                     timeSource,
                     errorReporter,
@@ -799,6 +805,7 @@ private void doShouldRunReadOnlyLdbcWorkloadWithNothingDbWhileIgnoringScheduledS
             boolean ignoreScheduledStartTimes = true;
             long warmupCount = 100;
             long skipCount = 10;
+            boolean flushLog = false;
 
             ConsoleAndFileDriverConfiguration configuration = new ConsoleAndFileDriverConfiguration(
                     paramsMap,
@@ -818,7 +825,8 @@ private void doShouldRunReadOnlyLdbcWorkloadWithNothingDbWhileIgnoringScheduledS
                     printHelp,
                     ignoreScheduledStartTimes,
                     warmupCount,
-                    skipCount
+                    skipCount,
+                    flushLog
             );
 
             configuration = (ConsoleAndFileDriverConfiguration) configuration
@@ -859,7 +867,7 @@ private void doShouldRunReadOnlyLdbcWorkloadWithNothingDbWhileIgnoringScheduledS
 
             File resultsLog = temporaryFolder.newFile();
             SimpleCsvFileWriter csvResultsLogWriter =
-                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR );
+                    new SimpleCsvFileWriter( resultsLog, SimpleCsvFileWriter.DEFAULT_COLUMN_SEPARATOR, flushLog );
             metricsService = ThreadedQueuedMetricsService.newInstanceUsingBlockingBoundedQueue(
                     timeSource,
                     errorReporter,