Merge branch 'thread-cleanup'

Author: jsl12

appdaemon/appdaemon.py

@@ -426,6 +426,7 @@ async def stop(self) -> None:
                 self.logger.warning("Timed out stopping plugins, continuing shutdown")
         self.sched.stop()
         self.state.stop()
+        self.threading.stop()
 
         self.executor.shutdown(wait=True)
 

appdaemon/threads.py

@@ -117,6 +117,14 @@ def total_threads(self) -> int | None:
     def total_threads(self, new: int):
         self.AD.config.total_threads = new
 
+    def stop(self):
+        """Stop all threads."""
+        for thread in self.threads.values():
+            match thread:
+                case {"queue": Queue() as q, "thread": Thread() as t}:
+                    q.put(None)
+                    t.join(timeout=1)
+
     async def get_q_update(self):
         """Updates queue sizes"""
         for thread in self.threads:
@@ -233,7 +241,7 @@ async def create_initial_threads(self) -> None:
             },
         )
 
-    def get_q(self, thread_id: str) -> Queue:
+    def get_q(self, thread_id: str) -> Queue[dict[str, Any] | None]:
         return self.threads[thread_id]["queue"]
 
     @staticmethod
@@ -1016,25 +1024,31 @@ def worker(self):  # noqa: C901
         thread_id = threading.current_thread().name
         q = self.get_q(thread_id)
         while True:
-            args = q.get()
-            _type = args["type"]
-            funcref = args["function"]
-            _id = args["id"]
-            objectid = args["objectid"]
-            name = args["name"]
-            error_logger = logging.getLogger(f"Error.{name}")
-            args["kwargs"]["__thread_id"] = thread_id
-
-            silent = False
-            if "__silent" in args["kwargs"]:
-                silent = args["kwargs"]["__silent"]
+            match args := q.get():
+                case {
+                    "type": _type,
+                    "function": funcref,
+                    "id": _id,
+                    "objectid": objectid,
+                    "name": name,
+                    "kwargs": kwargs
+                }:
+                    args["kwargs"]["__thread_id"] = thread_id
+                    error_logger = logging.getLogger(f"Error.{name}")
+                    silent = kwargs.get("__silent", False)
+                case None:
+                    self.logger.debug("Stopping worker thread %s", thread_id)
+                    break
+                case _:
+                    self.logger.warning("Unknown callback type for %s - discarding", name)
+                    return
 
             app = self.AD.app_management.get_app_instance(name, objectid)
             if app is not None:
                 try:
                     pos_args = tuple()
                     kwargs = dict()
-                    match args["type"]:
+                    match _type:
                         case "scheduler":
                             kwargs = self.AD.sched.sanitize_timer_kwargs(app, args["kwargs"])
 
@@ -1108,6 +1122,8 @@ def safe_callback():
                     self.logger.warning(f"Found stale callback for {name} - discarding")
                 q.task_done()
 
+        self.logger.debug("Shutdown worker thread queue %s", thread_id)
+
     def report_callback_sig(self, name, type, funcref, args):
         error_logger = logging.getLogger("Error.{}".format(name))
 

tests/unit/datetime/utils.py

@@ -64,6 +64,9 @@ def sun_params() -> tuple[tuple[str, ...], list[tuple[Any, ...]]]:
         sun_params = {
             "now_str": ["early", "midday", "late"],
             "when": ["today", "next"],
-            "input_": [f'{type_} {'+' if str_offset > 0 else '-'} {utils.parse_timedelta(abs(str_offset))}' for type_, str_offset in sun_params],
+            "input_": [
+                f'{type_} {"+" if str_offset > 0 else "-"} {utils.parse_timedelta(abs(str_offset))}'
+                for type_, str_offset in sun_params
+            ],
         }
         return _process_params(sun_params)