Fix reconnect logic - #57
Conversation
The reconnect logic was removed years ago: #36 The motivation for this was to reduce the amount of time the system was left in an unconnected state if a _publisher_ went down. This is because a severed connection to the publisher caused subscribers to attempt a reconnect within rospy but if that failed it would go into a retry loop and backoff the retries up to 32 seconds thus causing delays. The assumption was that when the publisher came back it would cause the ROS master to send a publisherUpdate, and subscribers would then reconnect immediately. This logic is sound but it inadvertantly introduced a different problem: What if the subscribers connection died but the publishers did not? What happens here is the subscriber never attempts a reconnect (waiting for publisherUpdate) but the publisherUpdate never comes because the publisher is still up and working. This leads to subscribers being left unconnected indefinitely. In the mean time after pull 36 was merged we since pulled in additional upstream changes which added another exception catch that effectively did the same thing: break out of the loop and never reconnect. This patch fixes both those cases and handles the exception by starting a reconnect loop.
The reconnect logic was hard coded to wait a maximum of 32 seconds which isn't aggressive enough in a lot of use cases. In addition the time it took to call self.connect() was not factored into the overall sleep time which could result in delays much longer than 32 seconds in the worst case. This adds parameters to control the connect timeout as well as the maximum timeout we wait before retrying. In addition the time spent in self.connect() is now tracked and factored into the sleep so we wait at most the configured amount of time as expected.
There was a problem hiding this comment.
Pull request overview
Reworks rospy's TCPROS subscriber reconnect behavior so that TransportTerminated and TransportException no longer terminate receive_loop, but instead tear down the socket and retry via _reconnect(). Adds configurable backoff/timeout via two new ROS parameters and accounts for connection-attempt time when sleeping between retries.
Changes:
- Add
_get_reconnect_config()with lazy cached lookup of/tcp_reconnect_max_backoff_secand/tcp_reconnect_connect_timeout_sec(defaults 30s). - Make
_reconnect()use the configurable connect timeout and cap exponential backoff at the configured max, subtracting the elapsed connection time from the sleep budget. - Add
_reset_socket_for_reconnect()and changereceive_loop()soTransportTerminated/TransportExceptionreset the socket and continue (also call_reconnect()whenself.socket is None).
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| else: | ||
| self._reconnect() | ||
|
|
||
| except TransportTerminated as e: | ||
| logdebug("[%s] failed to receive incoming message : %s" % (self.name, str(e))) | ||
| rospydebug("[%s] failed to receive incoming message: %s" % (self.name, traceback.format_exc())) | ||
| break | ||
| # Treat hard transport termination the same as other transient | ||
| # transport failures so subscribers can auto-reconnect. | ||
| self._reset_socket_for_reconnect() | ||
| continue | ||
|
|
||
| except TransportException as e: | ||
| # A transport exception means the connection has been closed from the other side. | ||
| # Clean up our side. | ||
| self.close() | ||
| # Set socket to None so we reconnect. | ||
| self._reset_socket_for_reconnect() | ||
| continue |
There was a problem hiding this comment.
I'm not clear on what happens if, e.g., we restart a robot. Now the port numbers it's using for a given topic are different, and so we effectively have a totally separate connection for that topic, right? Does this mean we'll forever retain the previous connection?
| def _reset_socket_for_reconnect(self): | ||
| """Best-effort socket teardown while keeping transport alive for reconnect.""" | ||
| try: | ||
| if self.socket is not None: | ||
| try: | ||
| self.socket.shutdown(socket.SHUT_RDWR) | ||
| except: | ||
| pass | ||
| finally: | ||
| self.socket.close() | ||
| except: | ||
| pass | ||
| self.socket = None |
| except TransportInitError: | ||
| self.socket = None |
Copilot brought up a good point that if the publisher goes down the reconnect logic prevents reconnecting to the new publisher once its back up. To handle this check if the publisher is still active before reconnecting. If the publisher goes down we break out and avoid the reconnect logic and handle the reconnect via publisherUpdate, basically using the logic of the old commit removing reconnections.
| if self.dest_address is None: | ||
| raise ROSInitException("internal error with reconnection state: address not stored") | ||
|
|
||
| reconnect_start = time.time() |
There was a problem hiding this comment.
Any reason why we aren't using monotonic time here and below?
|
|
||
| return False | ||
|
|
||
| def _reconnect(self): |
There was a problem hiding this comment.
This function (_reconnect) returns a value now and before it didn't. Can we add the docstring type hints?
| try: | ||
| s = xmlrpcclient.ServerProxy(self.endpoint_id) | ||
| code, msg, val = s.getPublications(rospy.names.get_name()) | ||
| if code == 1: | ||
| return len([t for t in val if t[0] == self.name]) > 0 | ||
| except Exception: | ||
| return False |
There was a problem hiding this comment.
Does this connect to the remote endpoint and asks it if it's still publishing? If the subscribers connection has just failed because of bad network conditions then this query may fail right after (the network may still be down/bad). If that is the case, we will fail here and return False. That will cause us to bail out right away in the _reconnect function. The publisher may still be advertising but we fail to ask it.
There was a problem hiding this comment.
Won't we still just try again?
There was a problem hiding this comment.
We would break out immediately from the _reconnect
if not self._publisher_is_still_advertising():
rospywarn("publisher for [%s] is no longer advertised; giving up reconnect", self.name)
self.close()
return False
That would then break out of receive_loop:
else:
if not self._reconnect():
break
Then we would have to wait for another publisher to unwedge it, no?
We are treating failure to verify the same as "it isn't publishing anymore."
automatom-locus
left a comment
automatom-locus
left a comment
There was a problem hiding this comment.
I'm not sure I'm very qualified for reviewing this, but we do need to make sure a change of this magnitude doesn't go into 26.0.1, IMO. This could have fairly major ramifications, and I'd like more time to vet them.
No description provided.