w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-09-20 14:27:42 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Use _stop_event consistently across all threads as the way to signal that it should stop. Add thread joins with timeouts to allow the program to exit cleanly

Browse Source
This commit is contained in:
Ian Renton
2026-09-18 09:21:38 +01:00
parent d79c8f72c8
commit 29d8654234
28 changed files with 239 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
providers/spot/aprsis.py
+9 -8
View File
@@ -17,17 +17,16 @@ class APRSIS(SpotProvider):
def __init__(self, provider_config):
super().__init__("APRS-IS", provider_config)
self._thread = Thread(target=self._run, name="APRSISSpotProvider")
self._thread.daemon = True
self._thread = None
self._aprsis = None
self._running = True
self._stop_event = Event()
def start(self):
self._thread = Thread(target=self._run, name="APRSISSpotProvider", daemon=True)
self._thread.start()
def _run(self):
while self._running:
while not self._stop_event.is_set():
try:
self._aprsis = aprslib.IS(SERVER_OWNER_CALLSIGN)
self.status = "Connecting"
@@ -37,20 +36,22 @@ class APRSIS(SpotProvider):
self._aprsis.consumer(self._handle, immortal=True)
except Exception:
if self._running:
if not self._stop_event.is_set():
self.status = "Error"
logger.exception("Exception in APRS-IS provider")
if self._running:
if not self._stop_event.is_set():
self._stop_event.wait(timeout=5)
def stop(self):
self._running = False
self.status = "Shutting down"
self._stop_event.set()
if self._aprsis:
self._aprsis.close()
self._thread.join()
if self._thread:
self._thread.join(timeout=15)
if self._thread.is_alive():
logger.warning("APRS-IS worker thread did not exit on time and will be killed.")
def _handle(self, data):
try:
providers/spot/dxcluster.py
+18 -16
View File
@@ -1,8 +1,7 @@
import logging
import re
from datetime import datetime
from threading import Thread
from time import sleep
from threading import Event, Thread
import pytz
import telnetlib3
@@ -41,23 +40,26 @@ class DXCluster(SpotProvider):
self._LINE_PATTERN_ALLOW_RBN if self._allow_rbn_spots else self._LINE_PATTERN_EXCLUDE_RBN
)
self._telnet = None
self._thread = Thread(target=self._handle, name=f"DXClusterSpotProvider-{self.name}")
self._thread.daemon = True
self._running = True
self._thread = None
self._stop_event = Event()
def start(self):
self._thread = Thread(target=self._handle, name=f"DXClusterSpotProvider-{self.name}", daemon=True)
self._thread.start()
def stop(self):
self._running = False
self._stop_event.set()
if self._telnet:
self._telnet.close()
self._thread.join()
if self._thread:
self._thread.join(timeout=15)
if self._thread.is_alive():
logger.warning(f"DX Cluster {self._hostname} worker thread did not exit on time and will be killed.")
def _handle(self):
while self._running:
while not self._stop_event.is_set():
connected = False
while not connected and self._running:
while not connected and not self._stop_event.is_set():
try:
self.status = "Connecting"
logger.info(f"DX Cluster {self._hostname} connecting...")
@@ -69,14 +71,14 @@ class DXCluster(SpotProvider):
except ConnectionRefusedError:
self.status = "Error"
logger.warning(f"Connection refused to DX cluster {self._hostname}")
sleep(300)
self._stop_event.wait(timeout=300)
except Exception:
self.status = "Error"
logger.exception(f"Exception while connecting to DX Cluster Provider ({self._hostname}).")
sleep(5)
self._stop_event.wait(timeout=5)
self.status = "Waiting for Data"
while connected and self._running:
while connected and not self._stop_event.is_set():
try:
# Check new telnet info against regular expression
telnet_output = self._telnet.read_until("\n".encode("latin-1"))
@@ -106,19 +108,19 @@ class DXCluster(SpotProvider):
except EOFError:
connected = False
if self._running:
if not self._stop_event.is_set():
self.status = "Restarting"
logger.warning(f"Disconnected from DX Cluster {self._hostname}. Reconnecting...")
sleep(5)
self._stop_event.wait(timeout=5)
else:
logger.info(f"DX Cluster {self._hostname} shutting down...")
self.status = "Shutting down"
except Exception:
connected = False
if self._running:
if not self._stop_event.is_set():
self.status = "Error"
logger.exception(f"Exception in DX Cluster Provider ({self._hostname})")
sleep(5)
self._stop_event.wait(timeout=5)
else:
logger.info(f"DX Cluster {self._hostname} shutting down...")
self.status = "Shutting down"
providers/spot/http_spot_provider.py
+5 -1
View File
@@ -28,12 +28,16 @@ class HTTPSpotProvider(SpotProvider):
# Fire off the polling thread. It will poll immediately on startup, then sleep for poll_interval between
# subsequent polls, so start() returns immediately and the application can continue starting.
logger.info(f"Set up query of {self.name} spot API every {self._poll_interval!s} seconds.")
self._thread = Thread(target=self._run, name=f"HTTPSpotProvider-{self.name}")
self._thread = Thread(target=self._run, name=f"HTTPSpotProvider-{self.name}", daemon=True)
self._thread.start()
def stop(self):
self._stop_event.set()
self._wakeup_event.set()
if self._thread:
self._thread.join(timeout=35)
if self._thread.is_alive():
logger.warning(f"{self.name} spot worker thread did not exit on time and will be killed.")
def force_poll(self):
"""Trigger an immediate poll without waiting for the normal interval."""
providers/spot/rbn.py
+17 -15
View File
@@ -1,8 +1,7 @@
import logging
import re
from datetime import datetime
from threading import Thread
from time import sleep
from threading import Event, Thread
import pytz
import telnetlib3
@@ -30,23 +29,26 @@ class RBN(SpotProvider):
super().__init__(name, provider_config)
self._port = provider_config["port"]
self._telnet = None
self._thread = Thread(target=self._handle, name=f"RBNSpotProvider-{self.name}")
self._thread.daemon = True
self._running = True
self._thread = None
self._stop_event = Event()
def start(self):
self._thread = Thread(target=self._handle, name=f"RBNSpotProvider-{self.name}", daemon=True)
self._thread.start()
def stop(self):
self._running = False
self._stop_event.set()
if self._telnet:
self._telnet.close()
self._thread.join()
if self._thread:
self._thread.join(timeout=15)
if self._thread.is_alive():
logger.warning(f"RBN (port {self._port!s}) worker thread did not exit on time and will be killed.")
def _handle(self):
while self._running:
while not self._stop_event.is_set():
connected = False
while not connected and self._running:
while not connected and not self._stop_event.is_set():
try:
self.status = "Connecting"
logger.info(f"RBN port {self._port!s} connecting...")
@@ -58,10 +60,10 @@ class RBN(SpotProvider):
except Exception:
self.status = "Error"
logger.exception(f"Exception while connecting to RBN (port {self._port!s}).")
sleep(5)
self._stop_event.wait(timeout=5)
self.status = "Waiting for Data"
while connected and self._running:
while connected and not self._stop_event.is_set():
try:
# Check new telnet info against regular expression
telnet_output = self._telnet.read_until("\n".encode("latin-1"))
@@ -91,19 +93,19 @@ class RBN(SpotProvider):
except EOFError:
connected = False
if self._running:
if not self._stop_event.is_set():
self.status = "Restarting"
logger.warning(f"Disconnected from RBN provider (port {self._port!s}). Reconnecting...")
sleep(5)
self._stop_event.wait(timeout=5)
else:
logger.info(f"RBN provider (port {self._port!s}) shutting down...")
self.status = "Shutting down"
except Exception:
connected = False
if self._running:
if not self._stop_event.is_set():
self.status = "Error"
logger.exception(f"Exception in RBN provider (port {self._port!s})")
sleep(5)
self._stop_event.wait(timeout=5)
else:
logger.info(f"RBN provider (port {self._port!s}) shutting down...")
self.status = "Shutting down"
providers/spot/websocket_spot_provider.py
+11 -10
View File
@@ -1,7 +1,6 @@
import logging
from datetime import datetime
from threading import Thread
from time import sleep
from threading import Event, Thread
import pytz
from websocket import create_connection
@@ -20,22 +19,24 @@ class WebsocketSpotProvider(SpotProvider):
self._url = url
self._ws = None
self._thread = None
self._stopped = False
self._stop_event = Event()
self._last_event_id = None
def start(self):
logger.info(f"Set up websocket connection to {self.name} spot API.")
self._stopped = False
self._stop_event.clear()
self._thread = Thread(target=self._run, name=f"WebsocketSpotProvider-{self.name}")
self._thread.daemon = True
self._thread.start()
def stop(self):
self._stopped = True
self._stop_event.set()
if self._ws:
self._ws.close()
if self._thread:
self._thread.join()
self._thread.join(timeout=15)
if self._thread.is_alive():
logger.warning(f"{self.name} websocket worker thread did not exit on time and will be killed.")
def _on_open(self):
self.status = "Waiting for Data"
@@ -44,7 +45,7 @@ class WebsocketSpotProvider(SpotProvider):
self.status = "Connecting"
def _run(self):
while not self._stopped:
while not self._stop_event.is_set():
try:
logger.debug(f"Connecting to {self.name} spot API...")
self.status = "Connecting"
@@ -53,7 +54,7 @@ class WebsocketSpotProvider(SpotProvider):
# Keep reading from this same connection until it drops or we're asked to stop, rather than
# reconnecting for every message.
while not self._stopped:
while not self._stop_event.is_set():
data = self._ws.recv()
if not data:
break
@@ -82,8 +83,8 @@ class WebsocketSpotProvider(SpotProvider):
# No problem, we were getting rid of this object anyway.
pass
self._ws = None
if not self._stopped:
sleep(5) # Wait before trying to reconnect
if not self._stop_event.is_set():
self._stop_event.wait(timeout=5) # Wait before trying to reconnect
def _ws_message_to_spot(self, b):
"""Convert a WS message received from the API into a spot. The exact message data (in bytes) is provided here so the