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

frappy.client: cleanup properly after a reply timeout

Browse Source 
when a reply to a request is not received within 10 s a timeout
error is raised, but any further requests with the same identifier
will be blocked - fix this

Change-Id: Id2fbc8bb6e6ecfd54bba1fa57b62e626e49b54c8
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/33385
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker
2024-03-26 15:13:49 +01:00
parent a73b7e7d88
commit 4f65ae7e46
1 changed files with 13 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
frappy/client/__init__.py
View File

@ -282,6 +282,7 @@ class SecopClient(ProxyClient):
self.nodename = uri self.nodename = uri
self._lock = RLock() self._lock = RLock()
self._shutdown = Event() self._shutdown = Event()
self.cleanup = []
def __del__(self): def __del__(self):
try: try:
@ -297,6 +298,10 @@ class SecopClient(ProxyClient):
with self._lock: with self._lock:
if self.io: if self.io:
return return
self.txq = queue.Queue(30)
self.pending = queue.Queue(30)
self.active_requests.clear()
self.cleanup.clear()
if self.online: if self.online:
self._set_state(True, 'reconnecting') self._set_state(True, 'reconnecting')
else: else:
@ -368,6 +373,12 @@ class SecopClient(ProxyClient):
noactivity = 0 noactivity = 0
try: try:
while self._running: while self._running:
while self.cleanup:
entry = self.cleanup.pop()
for key, prev in self.active_requests.items():
if prev is entry:
self.active_requests.pop(key)
break
# may raise ConnectionClosed # may raise ConnectionClosed
reply = self.io.readline() reply = self.io.readline()
if reply is None: if reply is None:
@ -591,8 +602,10 @@ class SecopClient(ProxyClient):
def get_reply(self, entry): def get_reply(self, entry):
"""wait for reply and return it""" """wait for reply and return it"""
if not entry[1].wait(10): # event if not entry[1].wait(10): # event
self.cleanup.append(entry)
raise TimeoutError('no response within 10s') raise TimeoutError('no response within 10s')
if not entry[2]: # reply if not entry[2]: # reply
# no cleanup needed as self.active_requests will be cleared on connect
raise ConnectionError('connection closed before reply') raise ConnectionError('connection closed before reply')
action, _, data = entry[2] # pylint: disable=unpacking-non-sequence action, _, data = entry[2] # pylint: disable=unpacking-non-sequence
if action.startswith(ERRORPREFIX): if action.startswith(ERRORPREFIX):