[Benchmark] Explicitly flush after tqdm.write().

Apparently, calls to `tqdm.write()` do *not* flush the output to the
output stream (`sys.stdout`).  We now do this manually after calls to
`tqdm.write()`.

Author: ImmanuelHaffner

benchmark/Benchmark.py

@@ -255,6 +255,7 @@ def perform_experiment(yml, conn, info, results) -> list():
         experiment_times = conn.execute(N_RUNS, params)
     except connector.ConnectorException as ex:
         tqdm.write(f"\nAn error occurred for {system} while executing {path_to_file}: {str(ex)}\n")
+        sys.stdout.flush()
         raise BenchmarkError()
 
     # Add measurements to result
@@ -350,10 +351,12 @@ def perform_experiment(yml, conn, info, results) -> list():
     if output_csv_file:
         if not os.path.isfile(output_csv_file): # file does not yet exist
             tqdm.write(f'Writing measurements to \'{output_csv_file}\'.')
+            sys.stdout.flush()
             with open(output_csv_file, 'w') as csv:
                 csv.write('commit,date,version,suite,benchmark,experiment,name,config,case,time\n')
         else:
             tqdm.write(f'Adding measurements to \'{output_csv_file}\'.')
+            sys.stdout.flush()
 
     # A central object to collect all measurements of all experiments.  Has the following structure:
     #
@@ -384,6 +387,7 @@ def perform_experiment(yml, conn, info, results) -> list():
         # Validate schema
         if not validate_schema(path_to_file, YML_SCHEMA):
             tqdm.write(f'Benchmark file "{path_to_file}" violates schema.')
+            sys.stdout.flush()
             continue
 
         with open(path_to_file, 'r') as yml_file:
@@ -409,6 +413,7 @@ def perform_experiment(yml, conn, info, results) -> list():
             tqdm.write('\n\n==========================================================')
             tqdm.write(f'Perform benchmarks in \'{path_to_file}\'.')
             tqdm.write('==========================================================')
+            sys.stdout.flush()
 
             # Perform experiment for each system
             for system in yml.get('systems').keys():

benchmark/__init__.py

@@ -1,9 +1,10 @@
 from .connector import *
 
-import os
+from tqdm import tqdm
 import json
+import os
 import subprocess
-from tqdm import tqdm
+import sys
 
 
 TMP_DB = 'tmp.duckdb'
@@ -22,6 +23,7 @@ def execute(self, n_runs, params: dict):
         experiment = params['name']
         configname = f'DuckDB ({get_num_cores()} cores)' if self.multithreaded else 'DuckDB (single core)'
         tqdm.write(f'` Perform experiment {suite}/{benchmark}/{experiment} with configuration {configname}.')
+        sys.stdout.flush()
 
         self.clean_up()
 
@@ -74,6 +76,7 @@ def execute(self, n_runs, params: dict):
                         if self.verbose and not verbose_printed:
                             verbose_printed = True
                             tqdm.write(combined_query)
+                            sys.stdout.flush()
 
                         benchmark_info = f"{suite}/{benchmark}/{experiment} [{configname}]"
                         try:
@@ -102,6 +105,7 @@ def execute(self, n_runs, params: dict):
                     if self.verbose and not verbose_printed:
                         verbose_printed = True
                         tqdm.write(combined_query)
+                        sys.stdout.flush()
 
                     benchmark_info = f"{suite}/{benchmark}/{experiment} [{configname}]"
                     try:
@@ -209,6 +213,7 @@ def run_query(self, query, timeout, benchmark_info):
         except subprocess.TimeoutExpired:
             process.kill()
             tqdm.write(f"    ! Query \n'{query}'\n' timed out after {timeout} seconds")
+            sys.stdout.flush()
             raise ExperimentTimeoutExpired(f'Query timed out after {timeout} seconds')
         finally:
             if process.poll() is None: # if process is still alive
@@ -231,6 +236,7 @@ def run_query(self, query, timeout, benchmark_info):
     {err}
     ==================
     ''')
+            sys.stdout.flush()
             if process.returncode:
                 raise ConnectorException(f'Benchmark failed with return code {process.returncode}.')
 
@@ -240,4 +246,4 @@ def run_query(self, query, timeout, benchmark_info):
         durations.remove('')
         durations = [float(i.replace("\n", "").replace(",", ".")) for i in durations]
 
-        return durations
+        return durations

benchmark/database_connectors/__init__.py

@@ -3,12 +3,13 @@
 
 from tableauhyperapi import HyperProcess, Telemetry, Connection, CreateMode, NOT_NULLABLE, NULLABLE, SqlType, \
         TableDefinition, Inserter, escape_name, escape_string_literal, HyperException, TableName
-import time
-import os
-import sys
-import subprocess
 from tqdm import tqdm
 import multiprocessing
+import os
+import subprocess
+import sys
+import sys
+import time
 
 
 # Converting table names to lower case is needed because then
@@ -41,6 +42,7 @@ def execute(self, n_runs, params: dict()):
         experiment = params['name']
         suffix = f' ({get_num_cores()} cores)' if self.multithreaded else ' (single core)'
         tqdm.write(f'` Perform experiment {suite}/{benchmark}/{experiment} with configuration HyPer{suffix}.')
+        sys.stdout.flush()
 
         result = None
         if self.multithreaded:
@@ -49,6 +51,7 @@ def execute(self, n_runs, params: dict()):
                 result = HyPer._execute(n_runs, params)
             except Exception as ex:
                 tqdm.write(str(ex))
+                sys.stdout.flush()
                 return dict()
 
         else:
@@ -62,6 +65,7 @@ def execute(self, n_runs, params: dict()):
             args = ['taskset', '-c', '2', 'python3', '-c', script]
             if self.verbose:
                 tqdm.write(f"    $ {' '.join(args)}")
+                sys.stdout.flush()
             # try:
             #     P = subprocess.run(
             #         args=args,

benchmark/database_connectors/duckdb.py

@@ -3,8 +3,9 @@
 from tqdm import tqdm
 import os
 import pandas
-import subprocess
 import re
+import subprocess
+import sys
 
 class BenchmarkError(Exception):
     pass
@@ -84,6 +85,7 @@ def run_configuration(self, experiment, config_name, config, yml):
             tqdm.write(f'` Perform experiment {suite}/{benchmark}/{experiment} with configuration {config_name}.')
         else:
             tqdm.write(f'` Perform experiment {suite}/{benchmark}/{experiment}.')
+        sys.stdout.flush()
 
         # Get database schema
         schema = os.path.join(os.path.dirname(path_to_file), 'data', 'schema.sql')
@@ -125,6 +127,7 @@ def run_configuration(self, experiment, config_name, config, yml):
                     durations = self.benchmark_query(command, query, yml['pattern'], timeout, path_to_file)
                 except BenchmarkTimeoutException as ex:
                     tqdm.write(str(ex))
+                    sys.stdout.flush()
                     # Add timeout durations
                     for case in cases.keys():
                         execution_times[case] = TIMEOUT_PER_CASE * 1000
@@ -147,11 +150,13 @@ def run_configuration(self, experiment, config_name, config, yml):
                         durations = self.benchmark_query(command, query_str, yml['pattern'], timeout, path_to_file)
                     except BenchmarkTimeoutException as ex:
                         tqdm.write(str(ex))
+                        sys.stdout.flush()
                         execution_times[case] = timeout * 1000
                     else:
                         execution_times[case] = durations[0]
         except BenchmarkError as ex:
             tqdm.write(str(ex))
+            sys.stdout.flush()
 
         return execution_times
 
@@ -195,6 +200,7 @@ def benchmark_query(self, command, query, pattern, timeout, path_to_file):
     {err}
     ==================
     ''')
+            sys.stdout.flush()
             if process.returncode:
                 raise ConnectorException(f'Benchmark failed with return code {process.returncode}.')
 
@@ -253,3 +259,4 @@ def print_command(command :list, query :str, indent = ''):
     query_str = query.strip().replace('\n', ' ').replace('"', '\\"')
     command_str = ' '.join(command)
     tqdm.write(f'{indent}$ echo "{query_str}" | {command_str}')
+    sys.stdout.flush()

benchmark/database_connectors/hyper.py

@@ -1,14 +1,15 @@
 from .connector import *
 
-import time
+from psycopg2.extras import LoggingConnection, LoggingCursor
+from tqdm import tqdm
+import logging
 import os
 import psycopg2
 import psycopg2.extensions
-from psycopg2.extras import LoggingConnection, LoggingCursor
-import logging
-import subprocess
 import shlex
-from tqdm import tqdm
+import subprocess
+import sys
+import time
 
 db_options = {
     'dbname': 'benchmark_tmp',
@@ -29,6 +30,7 @@ def execute(self, n_runs, params: dict):
         benchmark = params['benchmark']
         experiment = params['name']
         tqdm.write(f'` Perform experiment {suite}/{benchmark}/{experiment} with configuration PostgreSQL.')
+        sys.stdout.flush()
 
         # map that is returned with the measured times
         measurement_times = dict()
@@ -91,6 +93,7 @@ def execute(self, n_runs, params: dict):
                             verbose_printed = True
                             with open(TMP_SQL_FILE) as tmp:
                                 tqdm.write("    " + "    ".join(tmp.readlines()))
+                        sys.stdout.flush()
 
                     timeout = TIMEOUT_PER_CASE
                     benchmark_info = f"{suite}/{benchmark}/{experiment} [PostgreSQL]"
@@ -126,6 +129,7 @@ def execute(self, n_runs, params: dict):
                         verbose_printed = True
                         with open(TMP_SQL_FILE) as tmp:
                             tqdm.write("    " + "    ".join(tmp.readlines()))
+                    sys.stdout.flush()
 
                 timeout = DEFAULT_TIMEOUT + TIMEOUT_PER_CASE * len(params['cases'])
                 benchmark_info = f"{suite}/{benchmark}/{experiment} [PostgreSQL]"
@@ -269,6 +273,7 @@ def run_command(self, command, timeout, benchmark_info):
     {err}
     ==================
     ''')
+            sys.stdout.flush()
             if process.returncode:
                 raise ConnectorException(f'Benchmark failed with return code {process.returncode}.')