fix more threading issues; add users metric

owrx/metrics.py

@@ -1,3 +1,5 @@
+import threading
+
 class Metric(object):
     def getValue(self):
         return 0
@@ -24,11 +26,13 @@ class DirectMetric(Metric):
 
 class Metrics(object):
     sharedInstance = None
+    creationLock = threading.Lock()
 
     @staticmethod
     def getSharedInstance():
-        if Metrics.sharedInstance is None:
-            Metrics.sharedInstance = Metrics()
+        with Metrics.creationLock:
+            if Metrics.sharedInstance is None:
+                Metrics.sharedInstance = Metrics()
         return Metrics.sharedInstance
 
     def __init__(self):

owrx/source.py

@@ -4,6 +4,7 @@ from owrx.feature import FeatureDetector, UnknownFeatureException
 from owrx.meta import MetaParser
 from owrx.wsjt import WsjtParser
 from owrx.aprs import AprsParser
+from owrx.metrics import Metrics, DirectMetric
 import threading
 import csdr
 import time
@@ -698,15 +699,18 @@ class TooManyClientsException(Exception):
 
 class ClientRegistry(object):
     sharedInstance = None
+    creationLock = threading.Lock()
 
     @staticmethod
     def getSharedInstance():
-        if ClientRegistry.sharedInstance is None:
-            ClientRegistry.sharedInstance = ClientRegistry()
+        with ClientRegistry.creationLock:
+            if ClientRegistry.sharedInstance is None:
+                ClientRegistry.sharedInstance = ClientRegistry()
         return ClientRegistry.sharedInstance
 
     def __init__(self):
         self.clients = []
+        Metrics.getSharedInstance().addMetric("openwebrx.users", DirectMetric(self.clientCount))
         super().__init__()
 
     def broadcast(self):

owrx/wsjt.py

@@ -38,12 +38,14 @@ class WsjtQueueWorker(threading.Thread):
 
 class WsjtQueue(Queue):
     sharedInstance = None
+    creationLock = threading.Lock()
 
     @staticmethod
     def getSharedInstance():
-        if WsjtQueue.sharedInstance is None:
-            pm = PropertyManager.getSharedInstance()
-            WsjtQueue.sharedInstance = WsjtQueue(maxsize=pm["wsjt_queue_length"], workers=pm["wsjt_queue_workers"])
+        with WsjtQueue.creationLock:
+            if WsjtQueue.sharedInstance is None:
+                pm = PropertyManager.getSharedInstance()
+                WsjtQueue.sharedInstance = WsjtQueue(maxsize=pm["wsjt_queue_length"], workers=pm["wsjt_queue_workers"])
         return WsjtQueue.sharedInstance
 
     def __init__(self, maxsize, workers):
@@ -125,7 +127,8 @@ class WsjtChopper(threading.Thread):
 
     def _scheduleNextSwitch(self):
         with self.schedulerLock:
-            self.scheduler.enterabs(self.getNextDecodingTime(), 1, self.switchFiles)
+            if self.doRun:
+                self.scheduler.enterabs(self.getNextDecodingTime(), 1, self.switchFiles)
 
     def switchFiles(self):
         self.switchingLock.acquire()