linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adjustments on client.SecopClient

Browse Source 
- change self._state to self.state: may be read by calling code
- online and state attributes to be set after the nodStateChange
  callback in order to detect state changes
- remove ProxyClient.readParameter (no need to be implemented)
- reconnect thread must be killed on an external call to disconnect

Change-Id: I08d75dc8e29aa6e65a33ce36a911da4eaf0b62ef
Reviewed-on: https://forge.frm2.tum.de/review/c/sine2020/secop/playground/+/22551
Tested-by: JenkinsCodeReview <bjoern_pedersen@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker
2020-02-27 14:46:45 +01:00
parent a5ad527378
commit 199ff0419a
1 changed files with 37 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
secop/client/__init__.py
View File

@ -100,7 +100,7 @@ class ProxyClient:
CALLBACK_NAMES = ('updateEvent', 'descriptiveDataChange', 'nodeStateChange', 'unhandledMessage') CALLBACK_NAMES = ('updateEvent', 'descriptiveDataChange', 'nodeStateChange', 'unhandledMessage')
online = False # connected or reconnecting since a short time online = False # connected or reconnecting since a short time
validate_data = False validate_data = False
_state = 'disconnected' # further possible values: 'connecting', 'reconnecting', 'connected' state = 'disconnected' # further possible values: 'connecting', 'reconnecting', 'connected'
def __init__(self): def __init__(self):
self.callbacks = {cbname: defaultdict(list) for cbname in self.CALLBACK_NAMES} self.callbacks = {cbname: defaultdict(list) for cbname in self.CALLBACK_NAMES}
@ -141,7 +141,7 @@ class ProxyClient:
if mname == key: if mname == key:
cbfunc(mname, pname, *data) cbfunc(mname, pname, *data)
elif cbname == 'nodeStateChange': elif cbname == 'nodeStateChange':
cbfunc(self.online, self._state) cbfunc(self.online, self.state)
if kwds: if kwds:
raise TypeError('unknown callback: %s' % (', '.join(kwds))) raise TypeError('unknown callback: %s' % (', '.join(kwds)))
@ -180,17 +180,15 @@ class ProxyClient:
self.readParameter(module, parameter) self.readParameter(module, parameter)
return self.cache[module, parameter] return self.cache[module, parameter]
def readParameter(self, module, parameter):
"""forced read over connection"""
raise NotImplementedError()
class SecopClient(ProxyClient): class SecopClient(ProxyClient):
"""a general SECoP client""" """a general SECoP client"""
reconnect_timeout = 10 reconnect_timeout = 10
shutdown = False _running = False
_shutdown = False
_rxthread = None _rxthread = None
_txthread = None _txthread = None
_connthread = None
disconnect_time = 0 # time of last disconnect disconnect_time = 0 # time of last disconnect
secop_version = '' secop_version = ''
descriptive_data = {} descriptive_data = {}
@ -229,7 +227,7 @@ class SecopClient(ProxyClient):
else: else:
self._set_state(False, 'connecting') self._set_state(False, 'connecting')
deadline = time.time() + try_period deadline = time.time() + try_period
while True: while not self._shutdown:
try: try:
self.io = AsynConn(self.uri) # timeout 1 sec self.io = AsynConn(self.uri) # timeout 1 sec
self.io.writeline(IDENTREQUEST.encode('utf-8')) self.io.writeline(IDENTREQUEST.encode('utf-8'))
@ -242,6 +240,7 @@ class SecopClient(ProxyClient):
raise self.error_map('HardwareError')('bad answer to %s: %r' % raise self.error_map('HardwareError')('bad answer to %s: %r' %
(IDENTREQUEST, self.secop_version)) (IDENTREQUEST, self.secop_version))
# now its safe to do secop stuff # now its safe to do secop stuff
self._running = True
self._rxthread = mkthread(self.__rxthread) self._rxthread = mkthread(self.__rxthread)
self._txthread = mkthread(self.__txthread) self._txthread = mkthread(self.__txthread)
self.log.debug('connected to %s', self.uri) self.log.debug('connected to %s', self.uri)
@ -256,13 +255,14 @@ class SecopClient(ProxyClient):
# print(formatExtendedTraceback()) # print(formatExtendedTraceback())
if time.time() > deadline: if time.time() > deadline:
# stay online for now, if activated # stay online for now, if activated
self._set_state(self.online and self.activate, 'disconnected') self._set_state(self.online and self.activate)
raise raise
time.sleep(1) time.sleep(1)
self.log.info('%s ready', self.nodename) if not self._shutdown:
self.log.info('%s ready', self.nodename)
def __txthread(self): def __txthread(self):
while not self.shutdown: while self._running:
entry = self.txq.get() entry = self.txq.get()
if entry is None: if entry is None:
break break
@ -281,10 +281,10 @@ class SecopClient(ProxyClient):
self.log.debug('TX: %r', line) self.log.debug('TX: %r', line)
self.io.send(line) self.io.send(line)
self._txthread = None self._txthread = None
self.disconnect() self.disconnect(False)
def __rxthread(self): def __rxthread(self):
while not self.shutdown: while self._running:
try: try:
reply = self.io.readline() reply = self.io.readline()
if reply is None: if reply is None:
@ -340,10 +340,10 @@ class SecopClient(ProxyClient):
self.txq.put(self.pending.get()) self.txq.put(self.pending.get())
self._rxthread = None self._rxthread = None
self.disconnect() self.disconnect(False)
if self.activate: if self.activate:
self.log.info('reconnect to %s', self.uri) self.log.info('try to reconnect to %s', self.uri)
mkthread(self._reconnect) self._connthread = mkthread(self._reconnect)
else: else:
self.log.warning('%s disconnected', self.uri) self.log.warning('%s disconnected', self.uri)
self._set_state(False, 'disconnected') self._set_state(False, 'disconnected')
@ -354,10 +354,10 @@ class SecopClient(ProxyClient):
and trigger event when done and event is not None and trigger event when done and event is not None
""" """
self.disconnect_time = time.time() self.disconnect_time = time.time()
mkthread(self._reconnect, connected_callback) self._connthread = mkthread(self._reconnect, connected_callback)
def _reconnect(self, connected_callback=None): def _reconnect(self, connected_callback=None):
while True: while not self._shutdown:
try: try:
self.connect() self.connect()
if connected_callback: if connected_callback:
@ -372,17 +372,25 @@ class SecopClient(ProxyClient):
if self.online: # was recently connected if self.online: # was recently connected
self.disconnect_time = 0 self.disconnect_time = 0
self.log.warning('can not reconnect to %s (%r)' % (self.nodename, e)) self.log.warning('can not reconnect to %s (%r)' % (self.nodename, e))
self.log.info('continue trying to reconnect')
# self.log.warning(formatExtendedTraceback()) # self.log.warning(formatExtendedTraceback())
self._set_state(False, 'disconnected') self._set_state(False)
time.sleep(self.reconnect_timeout) time.sleep(self.reconnect_timeout)
else: else:
time.sleep(1) time.sleep(1)
self._connthread = None
def disconnect(self): def disconnect(self, shutdown=True):
self.shutdown = True self._running = False
if shutdown:
self._shutdown = True
self._set_state(False, 'shutdown')
if self._connthread: # wait for connection thread stopped
self._connthread.join()
self._connthread = None
self.disconnect_time = time.time() self.disconnect_time = time.time()
if self._txthread: if self._txthread:
self.txq.put(None) # shutdownmarker self.txq.put(None) # shutdown marker
self._txthread.join() self._txthread.join()
self._txthread = None self._txthread = None
if self._rxthread: if self._rxthread:
@ -404,7 +412,6 @@ class SecopClient(ProxyClient):
event.set() event.set()
except queue.Empty: except queue.Empty:
pass pass
self.shutdown = False
def _init_descriptive_data(self, data): def _init_descriptive_data(self, data):
"""rebuild descriptive data""" """rebuild descriptive data"""
@ -456,11 +463,13 @@ class SecopClient(ProxyClient):
def _set_state(self, online, state=None): def _set_state(self, online, state=None):
# treat reconnecting as online! # treat reconnecting as online!
self._state = state or self._state state = state or self.state
self.online = online self.callback(None, 'nodeStateChange', online, state)
self.callback(None, 'nodeStateChange', self.online, self._state)
for mname in self.modules: for mname in self.modules:
self.callback(mname, 'nodeStateChange', self.online, self._state) self.callback(mname, 'nodeStateChange', online, state)
# set online attribute after callbacks -> callback may check for old state
self.online = online
self.state = state
def queue_request(self, action, ident=None, data=None): def queue_request(self, action, ident=None, data=None):
"""make a request""" """make a request"""
@ -492,6 +501,7 @@ class SecopClient(ProxyClient):
return self.get_reply(entry) return self.get_reply(entry)
def readParameter(self, module, parameter): def readParameter(self, module, parameter):
"""forced read over connection"""
try: try:
self.request(READREQUEST, self.identifier[module, parameter]) self.request(READREQUEST, self.identifier[module, parameter])
except secop.errors.SECoPError: except secop.errors.SECoPError: