good locking

Author: JinZhou5042

taskvine_report/cli/report.py

@@ -50,24 +50,24 @@ def create_app(logs_dir):
     package_dir = os.path.dirname(os.path.dirname(__file__))
     template_dir = os.path.join(package_dir, 'templates')
     static_dir = os.path.join(package_dir, 'static')
-    
+
     app = Flask(__name__, 
                 template_folder=template_dir,
                 static_folder=static_dir)
 
     def setup_request_logging(app):
         @app.before_request
         def log_request_info():
-            app.config["RUNTIME_STATE"].template_lock.renew()
             request_folder = request.args.get('folder')
+            app.config["PROCESSING_REQUESTS_COUNT"] += 1
             app.config["RUNTIME_STATE"].log_request(request)
             request._start_time = time.time()
-            
             app.config["RUNTIME_STATE"].ensure_runtime_template(request_folder)
 
         @app.after_request
         def log_response_info(response):
             app.config["RUNTIME_STATE"].template_lock.renew()
+            app.config["PROCESSING_REQUESTS_COUNT"] -= 1
             if hasattr(request, '_start_time'):
                 duration = time.time() - request._start_time
                 app.config["RUNTIME_STATE"].log_response(response, request, duration)
@@ -123,6 +123,7 @@ def log_response_info(response):
     app.config["RUNTIME_STATE"].set_logger()
     app.config["RUNTIME_STATE"].set_logs_dir(logs_dir)
     app.config["RUNTIME_STATE"].log_info(f"Using logs directory: {logs_dir}")
+    app.config["PROCESSING_REQUESTS_COUNT"] = 0
 
     @app.route('/')
     def index():

taskvine_report/routes/runtime_state.py

@@ -6,27 +6,34 @@
 import time
 import traceback
 import threading
-import cloudpickle
+from flask import current_app
 
 
 class LeaseLock:
-    def __init__(self, lease_duration_sec=60):
+    def __init__(self, lease_duration_sec=20):
         self._lock = threading.Lock()
         self._expiry_time = 0
         self._lease_duration = lease_duration_sec
 
     def acquire(self):
         now = time.time()
+        # release lock if it's expired
         if self._lock.locked() and now > self._expiry_time:
             try:
                 self._lock.release()
             except RuntimeError:
                 pass
+        # release lock if there is no request being processed (the lock request itself is a request)
+        if current_app.config["PROCESSING_REQUESTS_COUNT"] <= 1:
+            try:
+                self._lock.release()
+            except RuntimeError:
+                pass
 
+        # acquire lock as non-blocking
         if not self._lock.acquire(blocking=False):
             return False
 
-        self._expiry_time = time.time() + self._lease_duration
         return True
 
     def release(self):
@@ -67,11 +74,11 @@ def __init__(self):
         self.tick_size = 12
 
         # for preventing multiple instances of the same runtime template
-        self.template_lock = LeaseLock(lease_duration_sec=60)
+        self.template_lock = LeaseLock()
 
         # for preventing multiple reloads of the data
         self._pkl_files_fingerprint = None
-        self.reload_lock = LeaseLock(lease_duration_sec=180)
+        self.reload_lock = LeaseLock()
 
     def set_logger(self):
         self.logger = Logger()

taskvine_report/src/data_parser.py

@@ -845,7 +845,6 @@ def parse_logs(self):
         self.parse_debug()
 
     def generate_subgraphs(self):
-        time_start = time.time()
         # exclude library tasks from subgraph generation to match RUNTIME_STATE filtering
         tasks_keys = set(key for key, task in self.tasks.items() if not task.is_library_task)
         parent = {key: key for key in tasks_keys}
@@ -878,7 +877,7 @@ def union(x, y):
                 rank[root_x] += 1
 
         with self._create_progress_bar() as progress:
-            task_id = progress.add_task("Parsing subgraphs", total=len(self.files))
+            task_id = progress.add_task("Parsing file dependencies", total=len(self.files))
 
             for file in self.files.values():
                 progress.update(task_id, advance=1)
@@ -904,9 +903,6 @@ def union(x, y):
         self.subgraphs = {i: subgraph for i,
                           subgraph in enumerate(sorted_subgraphs, 1)}
 
-        time_end = time.time()
-        print(f"Parsing subgraphs took {round(time_end - time_start, 4)} seconds")
-
         self.checkpoint_subgraphs()
 
     def checkpoint_debug(self):

taskvine_report/utils.py

@@ -524,3 +524,9 @@ def ensure_dir(path):
     if os.path.exists(path):
         return
     os.makedirs(path)
+
+def get_current_runtime_template():
+    return current_app.config["RUNTIME_STATE"].runtime_template
+
+def request_template_matches_current_runtime_template(request):
+    return request.args.get('folder') == get_current_runtime_template()