[Benchmark] Implement timeout for hyper

- Revert timout changes as it overrides multi core

- Implement timout tracking for hyper

Author: Tobias Kopp

benchmark/database_connectors/hyper.py

@@ -8,6 +8,7 @@
 import sys
 import subprocess
 from tqdm import tqdm
+import multiprocessing
 
 
 # Converting table names to lower case is needed because then
@@ -17,6 +18,17 @@ def table_name_to_lower_case(name: str):
     return name.lower()
 
 
+# Used as function for new process to track time / timeout
+def run_multiple_queries(connection, queries, data, queue):
+    times = hyperconf.benchmark_execution_times(connection, queries, data)
+    queue.put(times)
+
+def run_single_query(connection, query):
+    with connection.execute_query(query) as result:
+        for row in result:
+            pass
+
+
 class HyPer(Connector):
 
     def __init__(self, args = dict()):
@@ -33,6 +45,12 @@ def execute(self, n_runs, params: dict()):
         result = None
         if self.multithreaded:
             result = HyPer._execute(n_runs, params)
+            try:
+                result = HyPer._execute(n_runs, params)
+            except Exception as ex:
+                tqdm.write(str(ex))
+                return dict()
+
         else:
             path = os.getcwd()
             script = f'''
@@ -44,18 +62,26 @@ def execute(self, n_runs, params: dict()):
             args = ['taskset', '-c', '2', 'python3', '-c', script]
             if self.verbose:
                 tqdm.write(f"    $ {' '.join(args)}")
+            # try:
+            #     P = subprocess.run(
+            #         args=args,
+            #         capture_output=True,
+            #         text=True,
+            #         cwd=path
+            #     )
+            #     result = eval(P.stdout)
+            # except Exception as ex:
+            #     tqdm.write(str(ex))
+            #     return dict()
             P = subprocess.run(
-                args=args,
-                capture_output=True,
-                text=True,
-                cwd=path
-            )
-            # TODO error handling
-
+                    args=args,
+                    capture_output=True,
+                    text=True,
+                    cwd=path
+                )
             result = eval(P.stdout)
 
         patched_result = dict()
-
         for key, val in result.items():
             patched_result[f'{key}{suffix}'] = val
         return patched_result
@@ -73,6 +99,8 @@ def _execute(n_runs, params: dict()):
 
         hyperconf.init() # prepare for measurements
 
+        num_timeout_cases = 0
+
         for run_id in range(n_runs):
             # If tables contain scale factors, they have to be loaded separately for every case
             if (with_scale_factors or not bool(params.get('readonly'))):
@@ -95,8 +123,11 @@ def _execute(n_runs, params: dict()):
 
                         # Execute cases
                         for case, query in params['cases'].items():
+                            # Set up tables
                             for table_name, table in params['data'].items():
                                 table_name = table_name_to_lower_case(table_name)
+                                connection.execute_command(f'DELETE FROM {table_name};')  # Empty table first
+
                                 if table.get('scale_factors'):
                                     sf = table['scale_factors'][case]
                                 else:
@@ -105,44 +136,70 @@ def _execute(n_runs, params: dict()):
                                 num_rows = round((table['lines_in_file'] - header) * sf)
                                 connection.execute_command(f'INSERT INTO {table_name} SELECT * FROM {table_name}_tmp LIMIT {num_rows};')
 
-                            with connection.execute_query(query) as result:
-                                for row in result:
-                                    pass
-                            for table_name, table in params['data'].items():
-                                connection.execute_command(f'DELETE FROM {table_name_to_lower_case(table_name)};')
-
-                        # extract results
-                        matches = hyperconf.filter_results(
-                            hyperconf.extract_results(),
-                            { 'k': 'query-end'},
-                            [ hyperconf.MATCH_SELECT ]
-                        )
-                        times = map(lambda m: m['v']['execution-time'] * 1000, matches)
-                        times = list(map(lambda t: f'{t:.3f}', times))
-                        times = times[run_id * len(list(params['cases'].keys())) : ]    # get only times of this run, ignore previous runs
-                        times = list(zip(params['cases'].keys(), times))
-                        for case, time in times:
-                            if case not in measurement_times.keys():
-                                measurement_times[case] = list()
-                            measurement_times[case].append(time)
+                            timeout = TIMEOUT_PER_CASE
+                            times = None
+                            p = multiprocessing.Process(target=run_single_query, args=(connection, query))
+                            try:
+                                p.start()
+                                p.join(timeout=timeout)
+                                if p.is_alive():
+                                    # timeout happened
+                                    num_timeout_cases += 1
+                                    time = timeout * 1000 # in ms
+                                    p.terminate()
+                                    p.join()
+                                else:
+                                    # no timeout, extract result
+                                    matches = hyperconf.filter_results(
+                                        hyperconf.extract_results(),
+                                        { 'k': 'query-end'},
+                                        [ hyperconf.MATCH_SELECT ]
+                                    )
+                                    times = map(lambda m: m['v']['execution-time'] * 1000, matches)
+                                    times = list(map(lambda t: f'{t:.3f}', times))
+                                    case_idx = list(params['cases'].keys()).index(case)
+                                    time = times[run_id * len(list(params['cases'].keys())) + case_idx - num_timeout_cases]
+
+                                if case not in measurement_times.keys():
+                                    measurement_times[case] = list()
+                                measurement_times[case].append(time)
+
+                            except Exception as ex:
+                                raise(ConnectorException(ex))
+
 
 
             else:
+                open(hyperconf.HYPER_LOG_FILE, 'w').close() # to clear the log file and ignore previous runs
                 # Otherwise, tables have to be created just once before the measurements
                 with HyperProcess(telemetry=Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as hyper:
                     with Connection(endpoint=hyper.endpoint, database='benchmark.hyper', create_mode=CreateMode.CREATE_AND_REPLACE) as connection:
                         table_defs = HyPer.get_all_table_defs(params)
                         queries = HyPer.get_cases_queries(params, table_defs)
                         data = HyPer.get_data(params, table_defs)
 
-                        times = hyperconf.benchmark_execution_times(connection, queries.values(), data)
-                        times = times[run_id * len(queries.keys()):]    # get only times of this run, ignore previous runs
-                        times = list(zip(queries.keys(), list(map(lambda t: float(f'{t:.3f}'), times))))
+                        timeout = DEFAULT_TIMEOUT + TIMEOUT_PER_CASE * len(params['cases'])
+                        times = None
+                        q = multiprocessing.Queue()
+                        p = multiprocessing.Process(target=run_multiple_queries, args=(connection, queries.values(), data, q))
+                        try:
+                            p.start()
+                            p.join(timeout=timeout)
+                            if p.is_alive():
+                                # timeout happened
+                                times = times = list(zip(queries.keys(), [TIMEOUT_PER_CASE for _ in range(len(params['cases']))]))
+                            else:
+                                times = q.get()
+                                times = list(zip(queries.keys(), list(map(lambda t: float(f'{t:.3f}'), times))))
+
+                            for case, time in times:
+                                if case not in measurement_times.keys():
+                                    measurement_times[case] = list()
+                                measurement_times[case].append(time)
+
+                        except Exception as ex:
+                                raise(ConnectorException(ex))
 
-                        for case, time in times:
-                            if case not in measurement_times.keys():
-                                measurement_times[case] = list()
-                            measurement_times[case].append(time)
 
 
         return {'HyPer': measurement_times}