Made updates to support automatic reload of the TestGroups after a test run

src/architectures.py

@@ -17,7 +17,7 @@ from huggingface_hub import Repository
 from queue import Queue
 from threading import Thread, Timer
 from time import time
-from typing import List, Optional, Dict
+from typing import List, Optional, Dict, Callable
 from better_profanity import profanity
 
 from src.common import config_dir, data_dir, hf_api_token, escape_dollars
@@ -197,16 +197,22 @@ class LogWorker(Thread):
     trace_file_name = "trace.json"
     trace_file = os.path.join(trace_dir, trace_file_name)
     queue = Queue()
-    commit_time = 10  # Number of seconds after which to commit with no activity
-    commit_after = 10  # Number of records after which to commit irrespective of time
+    commit_time = 5  # Number of seconds after which to commit with no activity
+    commit_after = 20  # Number of records after which to commit irrespective of time
     commit_count = 0  # Current uncommitted records
     commit_timer = None  # The actual commit timer - we will schedule the commit on this
+    timeout_functions: List[Callable[[], None]] = []  # Callbacks which will be fired on timeout
 
     def run(self):
         while True:
             arch_name, request, trace, trace_tags, trace_comment = LogWorker.queue.get()
             if request is None:
-                LogWorker.commit_repo()
+                for func in LogWorker.timeout_functions:
+                    print(f"LogWorker commit running {func.__name__}")
+                    try:
+                        func()
+                    except Exception as e:
+                        print(f"Timeout func {func.__name__} had error {e}")
             else:
                 if LogWorker.commit_timer is not None and LogWorker.commit_timer.is_alive():
                     LogWorker.commit_timer.cancel()
@@ -249,12 +255,16 @@ class LogWorker(Thread):
 
     @classmethod
     def commit_repo(cls):
-        print(f"LogWorker committing {LogWorker.commit_count} open records")
-        cls.save_repo.push_to_hub()
-        LogWorker.commit_count = 0
+        if cls.commit_count > 0:
+            print(f"LogWorker committing {LogWorker.commit_count} open records")
+            cls.save_repo.push_to_hub()
+            LogWorker.commit_count = 0
 
     @classmethod
     def signal_commit(cls):
+        # Signalling this back via the queue and not doing the work here as it would
+        # be executed on the Timer thread and may conflict with resources if the main
+        # LogWorker starts doing work concurrently.
         print("LogWorker signalling commit based on time elapsed")
         cls.queue.put((None, None, None, None, None))
 
@@ -271,6 +281,7 @@ if LogWorker.instance is None:
     LogWorker.instance = LogWorker()
     LogWorker.daemon = True
     LogWorker.instance.start()
+    LogWorker.timeout_functions.append(LogWorker.commit_repo)
 
 
 class Architecture:

src/testing.py

@@ -9,11 +9,11 @@ import sys
 from huggingface_hub import Repository
 from queue import Queue
 from random import sample
-from threading import Thread, Timer
+from threading import Thread
 from typing import Dict, List, Optional, Tuple
 
-from src.architectures import Architecture, ArchitectureRequest, ArchitectureTrace
-from src.common import data_dir, hf_api_token
+from src.architectures import Architecture, ArchitectureRequest, LogWorker
+from src.common import data_dir
 
 
 class ArchitectureTestWorker(Thread):
@@ -323,12 +323,24 @@ class TestGroup:
             test_groups.append(tg)
         return test_groups
 
+    @classmethod
+    def force_load_all(cls):
+        """
+        Convenience wrapper to allow a no parameter call to force the reload of the
+        TestGroups without any parameters, for the LogWorker callback
+        """
+        cls.load_all(True)
+
     @classmethod
     def load_all(cls, reload: bool = False):
         """
         Load all the available TestGroups, from both the json file and the DB
         into the class variable - for efficiency do not reload unless requested
         """
+        if cls.force_load_all not in LogWorker.timeout_functions:
+            print("TestGroup adding forced refresh to LogWorker timeout")
+            LogWorker.timeout_functions.append(TestGroup.force_load_all)
+
         if cls.all is None or reload:
             working_test_groups = {}