From 9b94617d3f9934a1297e756980fbdecff81894aa Mon Sep 17 00:00:00 2001 From: Christian Kuss Date: Wed, 7 Dec 2022 14:51:37 -0800 Subject: [PATCH] Issue #200 - Move SLE class vars to instance vars When running more than one class that inherits from SLE, _state, _handlers, _data_queue, and _invoke_id become shared. The most common issue this creates is if data is being inserted into the data queue that is being shared. Data will be given to any of the children even if the PDU does not match what the child knows about. By making these instance level attributes data is kept confined to the individual member and is not shared. Additionally, if a socket is closed abruptly, the _state should be reset to "unbound" because the only way to reset it would be to destroy the object. The socket should be reset to None if it is ever closed. patch_all is not required as the only thing being monkey patched for gevents is time. By running patch_all, issues arise when running non-gevent queues in calling applications. _telem_sock should only be created when connecting, not on class instantiation Fixes #200 --- ait/dsn/sle/common.py | 17 +++++++++++------ setup.py | 2 +- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/ait/dsn/sle/common.py b/ait/dsn/sle/common.py index d9e7f6d..9b5d391 100644 --- a/ait/dsn/sle/common.py +++ b/ait/dsn/sle/common.py @@ -58,7 +58,7 @@ import gevent.queue import gevent.socket import gevent.select -import gevent.monkey; gevent.monkey.patch_all() +import gevent.monkey; gevent.monkey.patch_time() import pyasn1.error from pyasn1.codec.ber.encoder import encode @@ -92,13 +92,13 @@ class SLE(object): for interfacing with SLE. ''' - _state = 'unbound' - _handlers = defaultdict(list) - _data_queue = gevent.queue.Queue() - _invoke_id = 0 def __init__(self, *args, **kwargs): '''''' + self._state = 'unbound' + self._handlers = defaultdict(list) + self._data_queue = gevent.queue.Queue() + self._invoke_id = 0 self._downlink_frame_type = ait.config.get('dsn.sle.downlink_frame_type', kwargs.get('downlink_frame_type', 'TMTransFrame')) self._heartbeat = ait.config.get('dsn.sle.heartbeat', @@ -117,7 +117,6 @@ def __init__(self, *args, **kwargs): kwargs.get('peer_password', None)) self._responder_port = ait.config.get('dsn.sle.responder_port', kwargs.get('responder_port', 'default')) - self._telem_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self._auth_level = ait.config.get('dsn.sle.auth_level', kwargs.get('auth_level', 'none')) @@ -166,9 +165,11 @@ def send(self, data): for i in writeable: i.sendall(data) except socket.error as e: + self._state = 'unbound' if e.errno == errno.ECONNRESET: ait.core.log.error('Socket connection lost to DSN') self._socket.close() + self._socket = None else: ait.core.log.error('Unexpected error encountered when sending data. Aborting ...') raise e @@ -306,6 +307,8 @@ def connect(self): ait.core.log.error('Connection failure with DSN. Aborting ...') raise Exception('Unable to connect to DSN through any provided hostnames.') + self._telem_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + ait.core.log.info("Starting conn monitor for %s", type(self)) self._conn_monitor = gevent.spawn(conn_handler, self) self._data_processor = gevent.spawn(data_processor, self) @@ -335,7 +338,9 @@ def disconnect(self): greenlets for monitoring and processing data. ''' self._socket.close() + self._socket = None self._telem_sock.close() + self._telem_sock = None self._conn_monitor.kill() self._data_processor.kill() diff --git a/setup.py b/setup.py index e6a2672..971fe49 100644 --- a/setup.py +++ b/setup.py @@ -26,7 +26,7 @@ setup( name = 'ait-dsn', - version = '2.0.1-rc1', + version = '2.0.2', description = description, long_description = long_description, long_description_content_type = 'text/x-rst',