improve shutdown handling

This commit is contained in:
Jakob Ketterl 2019-05-12 18:10:24 +02:00
parent 17a362fe7a
commit a85a6c694c
6 changed files with 58 additions and 31 deletions

View File

@ -2,7 +2,7 @@ from http.server import HTTPServer
from owrx.http import RequestHandler from owrx.http import RequestHandler
from owrx.config import PropertyManager from owrx.config import PropertyManager
from owrx.feature import FeatureDetector from owrx.feature import FeatureDetector
from owrx.source import SdrService from owrx.source import SdrService, ClientRegistry
from socketserver import ThreadingMixIn from socketserver import ThreadingMixIn
from owrx.sdrhu import SdrHuUpdater from owrx.sdrhu import SdrHuUpdater
@ -45,4 +45,8 @@ Author contact info: Andras Retzler, HA7ILM <randras@sdr.hu>
if __name__ == "__main__": if __name__ == "__main__":
try:
main() main()
except KeyboardInterrupt:
for c in ClientRegistry.getSharedInstance().clients:
c.close()

View File

@ -1,5 +1,5 @@
from owrx.config import PropertyManager from owrx.config import PropertyManager
from owrx.source import DspManager, CpuUsageThread, SdrService, ClientReporterThread from owrx.source import DspManager, CpuUsageThread, SdrService, ClientRegistry
import json import json
import logging import logging
@ -14,7 +14,7 @@ class OpenWebRxClient(object):
def __init__(self, conn): def __init__(self, conn):
self.conn = conn self.conn = conn
ClientReporterThread.getSharedInstance().addClient(self) ClientRegistry.getSharedInstance().addClient(self)
self.dsp = None self.dsp = None
self.sdr = None self.sdr = None
@ -68,10 +68,8 @@ class OpenWebRxClient(object):
def close(self): def close(self):
self.stopDsp() self.stopDsp()
CpuUsageThread.getSharedInstance().remove_client(self) CpuUsageThread.getSharedInstance().remove_client(self)
try: ClientRegistry.getSharedInstance().removeClient(self)
ClientReporterThread.getSharedInstance().removeClient(self) self.conn.close()
except ValueError:
pass
logger.debug("connection closed") logger.debug("connection closed")
def stopDsp(self): def stopDsp(self):

View File

@ -3,7 +3,7 @@ import mimetypes
from datetime import datetime from datetime import datetime
from owrx.websocket import WebSocketConnection from owrx.websocket import WebSocketConnection
from owrx.config import PropertyManager from owrx.config import PropertyManager
from owrx.source import ClientReporterThread from owrx.source import ClientRegistry
from owrx.connection import WebSocketMessageHandler from owrx.connection import WebSocketMessageHandler
from owrx.version import openwebrx_version from owrx.version import openwebrx_version
@ -41,7 +41,7 @@ class StatusController(Controller):
"status": "active", "status": "active",
"name": pm["receiver_name"], "name": pm["receiver_name"],
"op_email": pm["receiver_admin"], "op_email": pm["receiver_admin"],
"users": ClientReporterThread.getSharedInstance().clientCount(), "users": ClientRegistry.getSharedInstance().clientCount(),
"users_max": pm["max_clients"], "users_max": pm["max_clients"],
"gps": pm["receiver_gps"], "gps": pm["receiver_gps"],
"asl": pm["receiver_asl"], "asl": pm["receiver_asl"],

View File

@ -10,7 +10,7 @@ logger = logging.getLogger(__name__)
class SdrHuUpdater(threading.Thread): class SdrHuUpdater(threading.Thread):
def __init__(self): def __init__(self):
self.doRun = True self.doRun = True
super().__init__() super().__init__(daemon = True)
def update(self): def update(self):
pm = PropertyManager.getSharedInstance() pm = PropertyManager.getSharedInstance()

View File

@ -467,6 +467,8 @@ class CpuUsageThread(threading.Thread):
c.write_cpu_usage(cpu_usage) c.write_cpu_usage(cpu_usage)
time.sleep(3) time.sleep(3)
logger.debug("cpu usage thread shut down") logger.debug("cpu usage thread shut down")
if CpuUsageThread.sharedInstance == self:
CpuUsageThread.sharedInstance = None
def get_cpu_usage(self): def get_cpu_usage(self):
try: try:
@ -499,42 +501,49 @@ class CpuUsageThread(threading.Thread):
self.shutdown() self.shutdown()
def shutdown(self): def shutdown(self):
if self.doRun: self.doRun = False
if CpuUsageThread.sharedInstance == self:
CpuUsageThread.sharedInstance = None class ClientReportingThread(threading.Thread):
def __init__(self, registry):
self.doRun = True
self.registry = registry
super().__init__()
def run(self):
while self.doRun:
self.registry.broadcast()
time.sleep(3)
def stop(self):
self.doRun = False self.doRun = False
class TooManyClientsException(Exception): class TooManyClientsException(Exception):
pass pass
class ClientReporterThread(threading.Thread): class ClientRegistry(object):
sharedInstance = None sharedInstance = None
@staticmethod @staticmethod
def getSharedInstance(): def getSharedInstance():
if ClientReporterThread.sharedInstance is None: if ClientRegistry.sharedInstance is None:
ClientReporterThread.sharedInstance = ClientReporterThread() ClientRegistry.sharedInstance = ClientRegistry()
ClientReporterThread.sharedInstance.start() return ClientRegistry.sharedInstance
ClientReporterThread.sharedInstance.doRun = True
return ClientReporterThread.sharedInstance
def __init__(self): def __init__(self):
self.doRun = True
self.clients = [] self.clients = []
self.reporter = None
super().__init__() super().__init__()
def run(self): def broadcast(self):
while (self.doRun):
n = self.clientCount() n = self.clientCount()
for c in self.clients: for c in self.clients:
c.write_clients(n) c.write_clients(n)
time.sleep(3)
ClientReporterThread.sharedInstance = None
def addClient(self, client): def addClient(self, client):
pm = PropertyManager.getSharedInstance() pm = PropertyManager.getSharedInstance()
if len(self.clients) >= pm["max_clients"]: if len(self.clients) >= pm["max_clients"]:
raise TooManyClientsException() raise TooManyClientsException()
self.clients.append(client) self.clients.append(client)
if self.reporter is None:
self.reporter = ClientReportingThread(self)
self.reporter.start()
def clientCount(self): def clientCount(self):
return len(self.clients) return len(self.clients)
@ -544,5 +553,6 @@ class ClientReporterThread(threading.Thread):
self.clients.remove(client) self.clients.remove(client)
except ValueError: except ValueError:
pass pass
if not self.clients: if not self.clients and self.reporter is not None:
self.doRun = False self.reporter.stop()
self.reporter = None

View File

@ -72,5 +72,20 @@ class WebSocketConnection(object):
else: else:
logger.warning("unsupported opcode: {0}".format(opcode)) logger.warning("unsupported opcode: {0}".format(opcode))
def close(self):
try:
header = self.get_header(0, 8)
self.handler.wfile.write(header)
self.handler.wfile.flush()
except ValueError:
logger.exception("while writing close frame:")
try:
self.handler.finish()
self.handler.connection.close()
except Exception:
logger.exception("while closing connection:")
class WebSocketException(Exception): class WebSocketException(Exception):
pass pass