Track and join socket threads on shutdown

Improve BaseProcessorSocket shutdown and restart behavior by tracking and joining threads.

- Import multiprocessing.connection.Client to wake a blocking accept() during stop.
- Add _accept_thread, _rx_thread and _rx_threads to track the accept thread and per-connection receiver threads.
- Start and store the accept thread instead of launching it anonymously; store and start rx threads when clients connect.
- On stop(), set _stop, attempt a Client connect to unblock accept(), close connections/listener, then join the accept thread and best-effort join tracked rx threads before clearing them.
- Clear _stop in start_server if previously set to allow restart, and minor whitespace/newline cleanup.

These changes reduce race conditions on restart, make shutdowns cleaner (especially on Windows), and help avoid lingering daemon threads.

Author: C-Achard

dlclivegui/processors/dlc_processor_socket.py

@@ -8,7 +8,7 @@
 import time
 from collections import deque
 from math import acos, atan2, copysign, degrees, pi, sqrt
-from multiprocessing.connection import Listener
+from multiprocessing.connection import Client, Listener
 from pathlib import Path
 from threading import Event, Thread
 
@@ -140,6 +140,9 @@ def __init__(
         self.address = bind
         self.authkey = authkey if authkey is not None else (b"secret password" if bind is not None else None)
         self.listener = None
+        self._accept_thread = None
+        self._rx_thread = None
+        self._rx_threads = set()
         self._socket_timeout = float(socket_timeout)
 
         self._stop = Event()
@@ -180,6 +183,9 @@ def start_server(self, bind, authkey=b"secret password", *, timeout: float = 1.0
         if self.listener is not None:
             return
 
+        if self._stop.is_set():
+            self._stop.clear()
+
         self.address = bind
         self.authkey = authkey
 
@@ -190,7 +196,8 @@ def start_server(self, bind, authkey=b"secret password", *, timeout: float = 1.0
         except Exception:
             pass
 
-        Thread(target=self._accept_loop, daemon=True).start()
+        self._accept_thread = Thread(target=self._accept_loop, daemon=True)
+        self._accept_thread.start()
         logger.info(f"Processor server started on {bind[0]}:{bind[1]}")
 
     # --------------------------------------------------------------------------------------
@@ -215,7 +222,9 @@ def _accept_loop(self):
                 logger.debug(f"Client connected from {self.listener.last_accepted}")
                 self.conns.add(conn)
 
-                Thread(target=self._rx_loop, args=(conn,), daemon=True).start()
+                self._rx_thread = Thread(target=self._rx_loop, args=(conn,), daemon=True)
+                self._rx_threads.add(self._rx_thread)
+                self._rx_thread.start()
 
             except (TimeoutError, OSError, EOFError):
                 if self._stop.is_set():
@@ -323,11 +332,36 @@ def stop(self):
         logger.info("Stopping processor...")
         self._stop.set()
 
+        # Wake accept() so the accept loop exits quickly (especially helpful on Windows)
+        # This is safe even if no clients are connected.
+        try:
+            if self.address is not None and self.authkey is not None:
+                c = Client(self.address, authkey=self.authkey)
+                c.close()
+        except Exception:
+            pass
+
         for conn in list(self.conns):
             self._close_conn(conn)
 
         self._close_listener()
 
+        # Join accept thread to avoid race conditions on restart
+        if self._accept_thread is not None:
+            self._accept_thread.join(timeout=2.0)
+            if self._accept_thread.is_alive():
+                logger.warning("Accept thread did not terminate cleanly")
+            self._accept_thread = None
+
+        # Join rx threads briefly (best effort)
+        for t in list(self._rx_threads):
+            try:
+                t.join(timeout=1.0)
+            except Exception:
+                pass
+        self._rx_threads.clear()
+        self._rx_thread = None
+
         # Small Windows delay to help TIME_WAIT cleanup
         if sys.platform.startswith("win"):
             if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):