update to protect against executor calls after unsubscribe/shutdown

Author: David Conner

flexbe_core/flexbe_core/proxy/proxy_action_client.py

@@ -72,7 +72,6 @@ def shutdown():
             ProxyActionClient._cancel_current_goal.clear()
             ProxyActionClient._has_active_goal.clear()
             ProxyActionClient._current_goal.clear()
-            ProxyActionClient._node = None
         except Exception as exc:  # pylint: disable=W0703
             print(f'Something went wrong during shutdown of proxy action clients!\n{ str(exc)}')
 

flexbe_core/flexbe_core/proxy/proxy_publisher.py

@@ -66,7 +66,6 @@ def shutdown():
 
             print("Shutdown proxy publisher  ...")
             ProxyPublisher._topics.clear()
-            ProxyPublisher._node = None
 
         except Exception as exc:  # pylint: disable=W0703
             Logger.error(f'Something went wrong during shutdown of proxy publisher !\n{str(exc)}')

flexbe_core/flexbe_core/proxy/proxy_service_caller.py

@@ -71,7 +71,6 @@ def shutdown():
 
             print("Shutdown proxy service caller  ...")
             ProxyServiceCaller._results.clear()
-            ProxyServiceCaller._node = None
 
         except Exception as exc:  # pylint: disable=W0703
             print(f'Something went wrong during shutdown of proxy service caller !\n{str(exc)}')

flexbe_core/flexbe_core/proxy/proxy_subscriber_cached.py

@@ -70,7 +70,6 @@ def shutdown():
             print("Shutdown proxy subscriber  ...")
             ProxySubscriberCached._topics.clear()
             ProxySubscriberCached._persistant_topics.clear()
-            ProxySubscriberCached._node = None
 
         except Exception as exc:  # pylint: disable=W0703
             print(f'Something went wrong during shutdown of proxy subscriber !\n{str(exc)}')
@@ -156,7 +155,8 @@ def subscribe(cls, topic, msg_type, callback=None, buffered=False, qos=None, ins
                                      f"({len(ProxySubscriberCached._topics[topic]['subscribers'])})")
 
         # Register the local callback for topic message
-        cls.set_callback(topic, callback, inst_id)
+        if callback is not None:
+            cls.set_callback(topic, callback, inst_id)
 
     @classmethod
     def _callback(cls, msg, topic):
@@ -173,16 +173,18 @@ def _callback(cls, msg, topic):
             Logger.localinfo(f"-- invalid topic={topic} for callback!")
             return
 
-        ProxySubscriberCached._topics[topic]['last_msg'] = msg
-        if ProxySubscriberCached._topics[topic]['buffered']:
-            ProxySubscriberCached._topics[topic]['msg_queue'].append(msg)
-
-        for inst_id, callback in ProxySubscriberCached._topics[topic]['callbacks'].items():
-            try:
-                callback(msg)
-            except Exception as exc:  # pylint: disable=W0703
-                Logger.error(f"Exception in callback for {topic}: "
-                             f"{callback.__module__}  {callback.__name__} @ {inst_id} \n {exc} ")
+        try:
+            ProxySubscriberCached._topics[topic]['last_msg'] = msg
+            if ProxySubscriberCached._topics[topic]['buffered']:
+                ProxySubscriberCached._topics[topic]['msg_queue'].append(msg)
+            for inst_id, callback in ProxySubscriberCached._topics[topic]['callbacks'].items():
+                try:
+                    callback(msg)
+                except Exception as exc:  # pylint: disable=W0703
+                    Logger.error(f"Exception in callback for {topic}: "
+                                f"{callback.__module__}  {callback.__name__} @ {inst_id} \n {exc} ")
+        except KeyError:
+                    Logger.error(f"Error: {topic} is no longer available for processing callback! ")
 
     @classmethod
     def set_callback(cls, topic, callback, inst_id=-1):
@@ -203,16 +205,24 @@ def set_callback(cls, topic, callback, inst_id=-1):
             return
 
         if callback is not None:
+            ProxySubscriberCached._node.executor.create_task(cls.__set_callback, topic, callback, inst_id)
+
+    @classmethod
+    def __set_callback(cls, topic, callback, inst_id):
+        """Set callback in executor thread."""
+        try:
             if inst_id not in ProxySubscriberCached._topics[topic]['callbacks']:
                 ProxySubscriberCached._topics[topic]['callbacks'][inst_id] = callback
                 Logger.localinfo(f"   Set local callback {callback.__name__} of "
-                                 f"{len(ProxySubscriberCached._topics[topic]['callbacks'])} for {topic}!")
+                                    f"{len(ProxySubscriberCached._topics[topic]['callbacks'])} for {topic}!")
             else:
                 Logger.localinfo("Update existing callback "
-                                 f"{ProxySubscriberCached._topics[topic]['callbacks'][inst_id].__name__} with "
-                                 f"{callback.__name__} of {len(ProxySubscriberCached._topics[topic]['callbacks'])}"
-                                 f" for {topic}!")
+                                    f"{ProxySubscriberCached._topics[topic]['callbacks'][inst_id].__name__} with "
+                                    f"{callback.__name__} of {len(ProxySubscriberCached._topics[topic]['callbacks'])}"
+                                    f" for {topic}!")
                 ProxySubscriberCached._topics[topic]['callbacks'][inst_id] = callback
+        except KeyError:
+            Logger.localwarn("Error: topic {topic} is not longer available - cannot set callback!")
 
     @classmethod
     def enable_buffer(cls, topic):
@@ -307,9 +317,13 @@ def remove_last_msg(cls, topic, clear_buffer=False):
         """
         if topic in ProxySubscriberCached._persistant_topics:
             return
-        ProxySubscriberCached._topics[topic]['last_msg'] = None
-        if clear_buffer:
-            ProxySubscriberCached._topics[topic]['msg_queue'] = []
+
+        try:
+            ProxySubscriberCached._topics[topic]['last_msg'] = None
+            if clear_buffer:
+                ProxySubscriberCached._topics[topic]['msg_queue'] = []
+        except KeyError:
+            Logger.localwarn(f"remove_last_msg: {topic} is not available!")
 
     @classmethod
     def make_persistant(cls, topic):
@@ -343,20 +357,18 @@ def unsubscribe_topic(cls, topic, inst_id=-1):
                     topic_dict['subscribers'].remove(inst_id)
                     Logger.localinfo(f"Unsubscribed {topic} from proxy! "
                                      f"({len(topic_dict['subscribers'])} remaining)")
-
-                if inst_id in topic_dict['callbacks']:
-                    topic_dict['callbacks'].pop(inst_id)
-                    Logger.localinfo(f"Removed callback from proxy subscription for {topic} "
-                                     f"from proxy! ({len(topic_dict['callbacks'])} remaining)")
-
-                remaining_subscribers = len(topic_dict['subscribers'])
-                remaining_callbacks = len(topic_dict['callbacks'])
-                if remaining_subscribers == 0:
-                    assert remaining_callbacks == 0, "Must have at least one subscriber tracked for every callback!"
-                    sub = topic_dict['subscription']
-                    ProxySubscriberCached._topics.pop(topic)
-                    ProxySubscriberCached._node.executor.create_task(ProxySubscriberCached.destroy_subscription,
-                                                                     sub, topic)
+                    
+                    remaining_subscribers = len(topic_dict['subscribers'])
+                    if remaining_subscribers == 0:
+                        ProxySubscriberCached._topics.pop(topic)  # remove from list of valid topics
+                        sub = topic_dict['subscription']
+                        ProxySubscriberCached._node.executor.create_task(ProxySubscriberCached.destroy_subscription,
+                                                                        sub, topic)
+                    elif inst_id in topic_dict['callbacks']:
+                            # Remove callback in executor thread to avoid changing size during callback
+                            ProxySubscriberCached._node.executor.create_task(topic_dict['callbacks'].pop, inst_id)
+                            Logger.localinfo(f"Removed callback from proxy subscription for {topic} "
+                                            f"from proxy! ({len(topic_dict['callbacks'])} remaining)")
 
             except Exception as exc:  # pylint: disable=W0703
                 Logger.error(f'Something went wrong unsubscribing {topic} of proxy subscriber!\n%s', str(exc))

flexbe_mirror/flexbe_mirror/flexbe_mirror.py

@@ -268,7 +268,8 @@ def _start_mirror(self, msg):
 
     def _stop_mirror(self, msg):
         self.get_logger().info('--> Mirror - request to stop mirror for '
-                               f'checksum id={msg.behavior_id} - waiting for sync lock ...')
+                               f'checksum id={msg.behavior_id if isinstance(msg, BEStatus) else None} '
+                               f'- waiting for sync lock ...')
         with self._sync_lock:
             # self.get_logger().info(f'--> Mirror - stopping mirror for checksum id={msg.behavior_id} with sync lock ...')
             self._stopping = True