Getting my fork in sync with upstream (#10)

* updated documentation and help

* Update README.md

* fix typos in args help

* add travis ci

* fix ci build to return exist code 0

* add prepare step to ci build

* create user and db

* create user and db in before_script

* fix psql command arg

* use postgres version 9.6 for ci build

* fix warning for psycopg2

* set password in travis settings

* add query phase to ci build

* Use DSP travis ci build status

* change data size from 1GB to 10MB

* move env var with password from travis settings to yml

* move psycopg2-binary from ci yml to requirements

* minor changes in README.md

* minor changes in README.md

* Remove psycopg2 from requirements

remove sources, we have psycopg2-binary in there now

* added throughput tests

* avoid plus one

* add reboot functionality

* refactor code, run twice and add verbose and read-only options

* set number of stream based on scale factor per default

* update README.md

* skeleton for metrics

* reuse code

* load json from result files

* remove folder tree for results before running perf tests

* implemented metrics formulas

* added TODOs

* fix bug

* remove unnecessary print statements

* python 3.6 needed for dict syntax

* sudo is needed for restarting postgresql and clearing os caches

* fix bug in print result header/footer

* do not skip running perf tests

* remove results folder in the beginning

* add throughput total time

* do not version control results folder

* change from two runs to only one

* move Result outside of the loop

* merge metric and no reboot

* fix merge problem

Author: slavong

benchmark.py

@@ -69,7 +69,6 @@
         [13, 15, 17, 1, 22, 11, 3, 4, 7, 20, 14, 21, 9, 8, 2, 18, 16, 6, 10, 12, 5, 19]
         ]
 NUM_QUERIES = len(QUERY_ORDER[0]) # 22
-NUM_RUNS = 2 # as per TPC-H spec, the test should run twice, with a reboot between them
 
 ## End Constants
 
@@ -248,8 +247,7 @@ def generate_data(dbgen_dir, data_dir, scale, num_streams):
 
     ## Update/Delete phase data
     ## we generate num_streams + 1 number of updates because 1 is used by the power test
-    ## and multiplied by 2 because there are two runs
-    p = subprocess.Popen(["./dbgen", "-vf", "-s", str(scale), "-U", str(2 * (num_streams + 1))],
+    p = subprocess.Popen(["./dbgen", "-vf", "-s", str(scale), "-U", str(num_streams + 1)],
                          cwd = dbgen_dir)
     p.communicate()
     if (not p.returncode):
@@ -430,12 +428,11 @@ def insert_lineitem(cols, conn):
     conn.executeQuery(li_insert_stmt)
 
 
-def refresh_func1(conn, data_dir, run, stream, num_streams, verbose):
+def refresh_func1(conn, data_dir, stream, num_streams, verbose):
     try:
         if verbose:
-            print("Running refresh function #1 in run #%s stream #%s" % (run, stream))
+            print("Running refresh function #1 in stream #%s" % stream)
         file_nr = stream + 1 # generated files are named 1,2,3,... while streams are indexed 0,1,2,...
-        file_nr += run * (num_streams+1) # and we have two runs
         filepath_o = os.path.join(data_dir, UPDATE_DIR, "orders.tbl.u" + str(file_nr) + ".csv")
         filepath_l = os.path.join(data_dir, UPDATE_DIR, "lineitem.tbl.u" + str(file_nr) + ".csv")
         with open(filepath_o) as orders_file, open(filepath_l) as lineitem_file:
@@ -474,12 +471,11 @@ def refresh_func1(conn, data_dir, run, stream, num_streams, verbose):
         return 1
 
 
-def refresh_func2(conn, data_dir, run, stream, num_streams, verbose):
+def refresh_func2(conn, data_dir, stream, num_streams, verbose):
     try:
         if verbose:
-            print("Running refresh function #2 in run #%s stream #%s" % (run, stream))
+            print("Running refresh function #2 in stream #%s" % stream)
         file_nr = stream + 1
-        file_nr += run * (num_streams+1)
         filepath = os.path.join(data_dir, DELETE_DIR, "delete." + str(file_nr) + ".csv")
         with open(filepath, 'r') as in_file:
             for ids in grouper(in_file, 100, ''):
@@ -492,73 +488,73 @@ def refresh_func2(conn, data_dir, run, stream, num_streams, verbose):
         return 1
 
 
-def run_query_stream(conn, query_root, run, stream, num_streams, result, verbose):
-    index = (run * (num_streams+1) + stream) % len(QUERY_ORDER)
+def run_query_stream(conn, query_root, stream, num_streams, result, verbose):
+    index = stream % len(QUERY_ORDER)
     order = QUERY_ORDER[index]
     for i in range(0, 22):
         try:
             if verbose:
-                print("Running query #%s in run #%s stream #%s ..." % (order[i], run, stream))
+                print("Running query #%s in stream #%s ..." % (order[i], stream))
             filepath = os.path.join(query_root, GENERATED_QUERY_DIR, str(order[i]) + ".sql")
             result.startTimer()
             conn.executeQueryFromFile(filepath)
-            result.setMetric("run_%s_stream_%s_query_%s" % (run, stream, order[i]), result.stopTimer())
+            result.setMetric("query_stream_%s_query_%s" % (stream, order[i]), result.stopTimer())
         except Exception as e:
-            print("unable to execute query %s in run %s stream %s: %s" % (order[i], run, stream, e))
+            print("unable to execute query %s in stream %s: %s" % (order[i], stream, e))
             return 1
     return 0
 
 
 def run_power_test(query_root, data_dir, host, port, db_name, user, password,
-                   run, num_streams, verbose, read_only):
+                   num_streams, verbose, read_only):
     try:
-        print("Power test run #%s started ..." % run)
+        print("Power test started ...")
         conn = PGDB(host, port, db_name, user, password)
         result = Result("Power")
         result.startTimer()
         stream = 0 # constant for power test
         #
         if not read_only:
-            if refresh_func1(conn, data_dir, run, stream, num_streams, verbose):
+            if refresh_func1(conn, data_dir, stream, num_streams, verbose):
                 return 1
-        result.setMetric("refresh_run_%s_stream_%s_func1" % (run, stream), result.stopTimer())
+        result.setMetric("refresh_stream_%s_func_1" % stream, result.stopTimer())
         #
-        if run_query_stream(conn, query_root, run, stream, num_streams, result, verbose):
+        if run_query_stream(conn, query_root, stream, num_streams, result, verbose):
             return 1
         #
         result.startTimer()
         if not read_only:
-            if refresh_func2(conn, data_dir, run, stream, num_streams, verbose):
+            if refresh_func2(conn, data_dir, stream, num_streams, verbose):
                 return 1
-        result.setMetric("refresh_run_%s_stream_%s_func2" % (run, stream), result.stopTimer())
+        result.setMetric("refresh_stream_%s_func_2" % stream, result.stopTimer())
         #
-        print("Power test run #%s finished." % run)
+        print("Power test finished.")
         if verbose:
             result.printMetrics()
-        result.saveMetrics("power%s" % run)
+        result.saveMetrics("power")
     except Exception as e:
         print("unable to run power test. DB connection failed: %s" % e)
         return 1
 
 
 def run_throughput_inner(query_root, data_dir, host, port, db_name, user, password, 
-                         run, stream, num_streams, q, verbose):
+                         stream, num_streams, q, verbose):
     try:
         conn = PGDB(host, port, db_name, user, password)
         result = Result("ThroughputQueryStream%s" % stream)
-        if run_query_stream(conn, query_root, run, stream, num_streams, result, verbose):
-            print("unable to finish query run #%s stream #%s" % (run, stream))
+        if run_query_stream(conn, query_root, stream, num_streams, result, verbose):
+            print("unable to finish query in stream #%s" % stream)
             exit(1)
         q.put(result)
     except Exception as e:
-        print("unable to connect to DB for query run #%s stream #%s: %s" % (run, stream, e))
+        print("unable to connect to DB for query in stream #%s: %s" % (stream, e))
         exit(1)
 
 
 def run_throughput_test(query_root, data_dir, host, port, db_name, user, password,
-                        run, num_streams, verbose, read_only):
+                        num_streams, verbose, read_only):
     try:
-        print("Throughput test run #%s started ..." % run)
+        print("Throughput test started ...")
         conn = PGDB(host, port, db_name, user, password)
         total = Result("ThroughputTotal")
         total.startTimer()
@@ -567,10 +563,10 @@ def run_throughput_test(query_root, data_dir, host, port, db_name, user, passwor
         for i in range(num_streams):
             stream = i + 1
             # queries
-            print("Throughput test run #%s stream #%s started ..." % (run, stream))
+            print("Throughput test in stream #%s started ..." % stream)
             p = Process(target=run_throughput_inner,
                         args=(query_root, data_dir, host, port, db_name, user, password,
-                              run, stream, num_streams, q, verbose))
+                              stream, num_streams, q, verbose))
             processes.append(p)
             p.start()
         result = Result("ThroughputRefreshStream")
@@ -579,33 +575,33 @@ def run_throughput_test(query_root, data_dir, host, port, db_name, user, passwor
             # refresh functions
             result.startTimer()
             if not read_only:
-                if refresh_func1(conn, data_dir, run, stream, num_streams,verbose):
+                if refresh_func1(conn, data_dir, stream, num_streams, verbose):
                     return 1
-            result.setMetric("refresh_run_%s_stream_%s_func1" % (run, stream), result.stopTimer())
+            result.setMetric("refresh_stream_%s_func_1" % stream, result.stopTimer())
             #
             result.startTimer()
             if not read_only:
-                if refresh_func2(conn, data_dir, run, stream, num_streams, verbose):
+                if refresh_func2(conn, data_dir, stream, num_streams, verbose):
                     return 1
-            result.setMetric("refresh_run_%s_stream_%s_func2" % (run, stream), result.stopTimer())
+            result.setMetric("refresh_stream_%s_func_2" % stream, result.stopTimer())
             #
         q.put(result)
         for p in processes:
             p.join()
-        print("Throughput test run #%s (all streams) finished." % run)
+        print("Throughput test finished.")
         for i in range(q.qsize()):
             res = q.get(False)
             if verbose:
                 res.printMetrics()
-            res.saveMetrics("throughput%s" % run)
+            res.saveMetrics("throughput")
         #
-        total.setMetric("throughput_test_total_run_%s" % run, total.stopTimer())
+        total.setMetric("throughput_test_total", total.stopTimer())
         if verbose:
             total.printMetrics()
-        total.saveMetrics("throughput%s" % run)
+        total.saveMetrics("throughput")
         #
     except Exception as e:
-        print("unable to execute throughput tests in run #%s. e" % (run, e))
+        print("unable to execute throughput test: %s" % e)
         return 1
 
 
@@ -615,21 +611,6 @@ def niceprint(txt, width):
     print("*"*w + " " + txt + " " + " "*x + "*"*w)
 
 
-def reboot():
-    # TODO: we need another solution, this is fine for a local DB and running with sudo rights
-    # but the DB can be remote, user running the test has no sudo, etc.
-    width = 60
-    print("*"*width)
-    niceprint("Restarting PostgreSQL ...", width)
-    command = ['sudo', 'service', 'postgresql', 'restart'];
-    subprocess.call(command, shell=False) # shell=FALSE for sudo to work.
-    print("*"*width)
-    niceprint("Clearing OS caches ...", width)
-    # https://linux-mm.org/Drop_Caches
-    os.system('sudo sh -c "sync; echo 3 > /proc/sys/vm/drop_caches"')
-    print("*"*width)
-
-
 def scale_to_num_streams(scale):
     num_streams = 2
     if scale <= 1:
@@ -664,8 +645,7 @@ def get_json_files_from(path):
 def get_json_files(path):
     json_files = []
     for mode in ['power', 'throughput']:
-        for run in range(2):
-            json_files += get_json_files_from(path + "/" + mode + str(run) + "/")
+        json_files += get_json_files_from(path + "/" + mode + "/")
     return json_files
 
 
@@ -695,29 +675,26 @@ def qi(jsons, i, s): # execution time for query Qi within the query stream s
     # s is 0 for the power function and the position of the query stream for the throughput test
     assert(1 <= i <= 22)
     assert(0 <= s)
-    metric_name = 'run_%s_stream_%s_query_%s'
-    s0 = get_timedelta_in_seconds(jsons, metric_name % (0, s, i))
-    s1 = get_timedelta_in_seconds(jsons, metric_name % (1, s, i))
-    return ( s0 + s1 ) / 2 # simple average of two values
+    metric_name = 'query_stream_%s_query_%s'
+    ret = get_timedelta_in_seconds(jsons, metric_name % (s, i))
+    return ret
 
 
 def ri(jsons, j, s): # execution time for the refresh function RFi within a refresh stream s
     # j is the ordering function of the refresh function ranging from 1 to 2
     # s is 0 for the power function and the position of the pair of refresh functions in the stream for the throughput test
     assert(j == 1 or j == 2)
     assert(0 <= s)
-    metric_name = 'refresh_run_%s_stream_%s_func%s'
-    s0 = get_timedelta_in_seconds(jsons, metric_name % (0, s, j))
-    s1 = get_timedelta_in_seconds(jsons, metric_name % (1, s, j))
-    return ( s0 + s1 ) / 2 # simple average of two values
+    metric_name = 'refresh_stream_%s_func_%s'
+    ret = get_timedelta_in_seconds(jsons, metric_name % (s, j))
+    return ret
 
 
 def ts(jsons): # total time needed to execute the throughput test
     # TODO: total time for throughput tests needs to be implemented
-    metric_name = 'throughput_test_total_run_%s'
-    s0 = get_timedelta_in_seconds(jsons, metric_name % 0)
-    s1 = get_timedelta_in_seconds(jsons, metric_name % 1)
-    return  ( s0 + s1 ) / 2
+    metric_name = 'throughput_test_total'
+    ret = get_timedelta_in_seconds(jsons, metric_name)
+    return ret
 
 def get_power_size(jsons, scale, num_streams):
     qi_product = 1
@@ -796,17 +773,16 @@ def main(phase, host, port, user, password, database, data_dir, query_root, dbge
         print("done creating indexes and foreign keys")
         result.printMetrics()
     elif phase == "query":
-        for run in [0,1]:
-            # Power test
-            if run_power_test(query_root, data_dir, host, port, database, user, password,
-                              run, num_streams, verbose, read_only):
-                print("running power test failed")
-                exit(1)
-            # Throughput test
-            if run_throughput_test(query_root, data_dir, host, port, database, user, password,
-                                   run, num_streams, verbose, read_only):
-                print("running throughput test failed")
-                exit(1)
+        if run_power_test(query_root, data_dir, host, port, database, user, password,
+                          num_streams, verbose, read_only):
+            print("running power test failed")
+            exit(1)
+        # Throughput test
+        if run_throughput_test(query_root, data_dir, host, port, database, user, password,
+                               num_streams, verbose, read_only):
+            print("running throughput test failed")
+            exit(1)
+        print("done performance test")
         metrics(scale, num_streams)
 
 
@@ -844,4 +820,3 @@ def main(phase, host, port, user, password, database, data_dir, query_root, dbge
 
     ## main
     main(phase, host, port, user, password, database, data_dir, query_root, dbgen_dir, scale, num_streams, verbose, read_only)
-