Enables threading in data server

1 files changed, 18 insertions, 15 deletions

diff --git a/salis.py b/salis.py
index afb8099..9a615f5 100755
--- a/salis.py
+++ b/salis.py
@@ -8,11 +8,12 @@ import shutil
 import sqlite3
 import subprocess
 import sys
+import threading
 import urllib.parse
 import urllib.request
 
 from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, ArgumentTypeError, RawTextHelpFormatter
-from http.server import HTTPServer, SimpleHTTPRequestHandler
+from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
 from tempfile import TemporaryDirectory
 
 # ------------------------------------------------------------------------------
@@ -117,17 +118,19 @@ args = main_parser.parse_args()
 # ------------------------------------------------------------------------------
 # Logging
 # ------------------------------------------------------------------------------
+log_lock = threading.Lock()
+
 def now():
     return f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S}"
 
 def info(msg, val=""):
-    print(f"\r{now()} ++ \033[1;34mINFO\033[0m {msg}", val)
+    with log_lock: print(f"\r{now()} ++ \033[1;34mINFO\033[0m {msg}", val, flush=True)
 
 def warn(msg, val=""):
-    print(f"\r{now()} ++ \033[1;33mWARN\033[0m {msg}", val)
+    with log_lock: print(f"\r{now()} ++ \033[1;33mWARN\033[0m {msg}", val, flush=True)
 
 def error(msg, val=""):
-    print(f"\r{now()} ++ \033[1;31mERROR\033[0m {msg}", val)
+    with log_lock: print(f"\r{now()} ++ \033[1;31mERROR\033[0m {msg}", val, flush=True)
     sys.exit(1)
 
 # ------------------------------------------------------------------------------
@@ -212,13 +215,6 @@ if args.command in ["serve"]:
                 with open(os.path.join("data/vendor", file), "wb") as f:
                     f.write(response.read())
 
-    sim_db = os.path.join(sim_dir, f"{args.name}.sqlite3")
-    info("Connecting to SQLite database:", sim_db)
-    db_con = sqlite3.connect(sim_db)
-    db_con.row_factory = sqlite3.Row
-    db_con.enable_load_extension(True)
-    db_cur = db_con.cursor()
-
     # Build SQLite event-array render extension
     sqlx_flags = set()
     sqlx_defines = set()
@@ -250,7 +246,6 @@ if args.command in ["serve"]:
 
     info("Using build command:", sqlx_build_cmd)
     subprocess.run(sqlx_build_cmd, check=True)
-    db_cur.execute(f"SELECT load_extension('{sqlx_so}')")
 
     # Generate configuration so front-end knows how to render the plots.
     # Each architecture may also provide its own set of plots, which will be merged with the
@@ -318,8 +313,15 @@ if args.command in ["serve"]:
     # NOTE: this server implementation is very minimal and has no built-in security.
     # Please do not put this on the internet! Only run the data server within secure
     # networks that you own.
+    sim_db = os.path.join(sim_dir, f"{args.name}.sqlite3")
+
     class Handler(SimpleHTTPRequestHandler):
         def __init__(self, *args, **kwargs):
+            self.db_con = sqlite3.connect(sim_db, timeout=600)
+            self.db_con.row_factory = sqlite3.Row
+            self.db_con.enable_load_extension(True)
+            self.db_con.execute("PRAGMA journal_mode=wal;")
+            self.db_con.execute(f"SELECT load_extension('{sqlx_so}');")
             super().__init__(*args, **kwargs, directory="data")
 
         def log_message(_, format, *args):
@@ -369,7 +371,7 @@ if args.command in ["serve"]:
                     selects = "*"
 
                 sql_query = f"SELECT * FROM (SELECT rowid, {selects} FROM {table} WHERE {x_axis} >= {x_low} AND {x_axis} <= {x_high} AND rowid % {nth} == 0 ORDER BY {x_axis} DESC LIMIT {entries}) ORDER BY {x_axis} ASC;"
-                sql_res = db_cur.execute(sql_query)
+                sql_res = self.db_con.execute(sql_query)
                 sql_list = [dict(row) for row in sql_res.fetchall()]
                 return self.send_as_json(sql_list)
 
@@ -377,15 +379,16 @@ if args.command in ["serve"]:
                 http_query = urllib.parse.parse_qs(bits.query)
                 x_axis = http_query["x_axis"][0]
                 sql_query = f"SELECT {x_axis} as x_high FROM general ORDER BY {x_axis} DESC LIMIT 1;"
-                sql_dict = dict(db_cur.execute(sql_query).fetchone())
+                sql_dict = dict(self.db_con.execute(sql_query).fetchone())
                 return self.send_as_json(sql_dict)
 
             self.log_error(f"Unsupported endpoint: {bits.path}")
             self.send_response(400)
             self.end_headers()
+            self.db_con.close()
 
     info("Launching data server")
-    server = HTTPServer(("", args.port), Handler)
+    server = ThreadingHTTPServer(("", args.port), Handler)
 
     try:
         server.serve_forever()