fix: clean up CA subscription management and ensure proper lifecycle management

j1939/controller_application.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import threading
 
 import j1939
 
@@ -60,6 +61,7 @@ def __init__(self, name, device_address_preferred=None, bypass_address_claim=Fal
         self._subscribers_request = []
         self._subscribers_acknowledge = []
         self._started = False
+        self._lifecycle_lock = threading.RLock()
 
     @property
     def _ecu_ref(self) -> j1939.ElectronicControlUnit:
@@ -73,21 +75,28 @@ def associate_ecu(self, ecu):
             The ECU this CA should be bound to.
             A j1939 :class:`j1939.ElectronicControlUnit` instance
         """
-        self._ecu = ecu
+        with self._lifecycle_lock:
+            self._ecu = ecu
 
     def remove_ecu(self):
-
-        self._ecu = None
+        with self._lifecycle_lock:
+            self._ecu = None
 
     def subscribe(self, callback):
         """Add the given callback to the message notification stream.
+
+        The subscription belongs to this ControllerApplication. If the CA is
+        removed from its ECU, the subscription is removed with it. When a CA
+        is replaced, callbacks must be subscribed again through the new CA.
+
         :param callback:
             Function to call when message is received.
         """
-        self._ecu_ref.subscribe(callback, self.message_acceptable)
+        self._ecu_ref.subscribe(callback, self.message_acceptable, owner=self)
 
     def unsubscribe(self, callback):
         """Stop listening for message.
+
         :param callback:
             Function to call when message is received.
         """
@@ -159,45 +168,64 @@ def start(self, claim_delay=0.5):
         :param claim_delay:
             The time in seconds to wait before starting the address claim procedure.
         """
-        # TODO raise RuntimeError("Can't start CA. Seems to be already running.")? or just ignore?
-        # check if we are not already started and there is an ecu connected
-        if self._ecu and not self.started:
-            self._started = True
-            self._ecu_ref.add_timer(claim_delay, self._process_claim_async)
+        with self._lifecycle_lock:
+            # TODO raise RuntimeError("Can't start CA. Seems to be already running.")? or just ignore?
+            # check if we are not already started and there is an ecu connected
+            if self._ecu and not self.started:
+                self._started = True
+                ecu = self._ecu
+            else:
+                ecu = None
+        if ecu is not None:
+            ecu.add_timer(claim_delay, self._process_claim_async)
 
     def stop(self):
         """Stops the CA
         """
-        # check if we are already started and there is an ecu connected
-        if self._ecu and self.started:
-            self._started = False
-            self._ecu_ref.remove_timer(self._process_claim_async)
+        with self._lifecycle_lock:
+            # check if we are already started and there is an ecu connected
+            if self._ecu and self.started:
+                self._started = False
+                ecu = self._ecu
+            else:
+                ecu = None
+        if ecu is not None:
+            ecu.remove_timer(self._process_claim_async)
 
     def _process_claim_async(self, cookie):
-        time_to_sleep = 0.500
-        if self._device_address_state == ControllerApplication.State.NONE:
-            if self._device_address_preferred is not None:
-                self._device_address_announced = self._device_address_preferred
-                self._send_address_claimed(self._device_address_announced)
-                if self._device_address_announced > 127 and self._device_address_announced < 248:
-                    self._device_address_state = ControllerApplication.State.WAIT_VETO
-                    time_to_sleep = ControllerApplication.ClaimTimeout.VETO
-                else:
-                    # addresses from 0..127 and 248..253 should start immediately
-                    self._device_address = self._device_address_announced
-                    self._device_address_state = ControllerApplication.State.NORMAL
-        elif self._device_address_state == ControllerApplication.State.WAIT_VETO:
-            # if we reach this phase, there was no VETO to our address claimed message so far
-            self._device_address = self._device_address_announced
-            self._device_address_state = ControllerApplication.State.NORMAL
-        elif self._device_address_state == ControllerApplication.State.NORMAL:
-            # do nothing
-            pass
-        elif self._device_address_state == ControllerApplication.State.CANNOT_CLAIM:
-            # do nothing
-            pass
+        with self._lifecycle_lock:
+            if not self._ecu or not self.started:
+                return False
+
+            time_to_sleep = 0.500
+            if self._device_address_state == ControllerApplication.State.NONE:
+                if self._device_address_preferred is not None:
+                    self._device_address_announced = self._device_address_preferred
+                    self._send_address_claimed(self._device_address_announced)
+                    if self._device_address_announced > 127 and self._device_address_announced < 248:
+                        self._device_address_state = ControllerApplication.State.WAIT_VETO
+                        time_to_sleep = ControllerApplication.ClaimTimeout.VETO
+                    else:
+                        # addresses from 0..127 and 248..253 should start immediately
+                        self._device_address = self._device_address_announced
+                        self._device_address_state = ControllerApplication.State.NORMAL
+            elif self._device_address_state == ControllerApplication.State.WAIT_VETO:
+                # if we reach this phase, there was no VETO to our address claimed message so far
+                self._device_address = self._device_address_announced
+                self._device_address_state = ControllerApplication.State.NORMAL
+            elif self._device_address_state == ControllerApplication.State.NORMAL:
+                # do nothing
+                pass
+            elif self._device_address_state == ControllerApplication.State.CANNOT_CLAIM:
+                # do nothing
+                pass
+            # Capture the ECU while protected by the lifecycle lock, then
+            # schedule outside the lock to avoid lock inversion with the ECU
+            # timer thread.
+            ecu = self._ecu
+
         # add new event with (possibly) new timeout value
-        self._ecu_ref.add_timer(time_to_sleep, self._process_claim_async)
+        ecu.add_timer(time_to_sleep, self._process_claim_async)
         # returning false deletes the event from the list
         return False
 

j1939/electronic_control_unit.py

@@ -99,6 +99,7 @@ def __init__(
 
         self._subscribers = []
         self._subscribers_lock = threading.RLock()
+        self._ca_lock = threading.RLock()
 
         # Heap-based timer event list: (deadline, seq, callback, cookie, delta_time)
         self._timer_events = []
@@ -323,7 +324,7 @@ def disconnect(self):
             self._bus_created = False
         self._bus = None
 
-    def subscribe(self, callback, device_address=None):
+    def subscribe(self, callback, device_address=None, owner=None):
         """Add the given callback to the message notification stream.
 
         :param callback:
@@ -333,18 +334,45 @@ def subscribe(self, callback, device_address=None):
             This is a simple way for peer-to-peer reception without adding a controller-application.
             Only one device address can be entered. Multiple device addresses are only possible with controller applications.
             Note: TP.CMDT will only be received if the destination address is bound to a controller application.
+
+        Subscriptions made directly on the ECU are independent of any
+        ControllerApplication and remain active until explicitly removed.
+
+        :param owner:
+            Optional lifecycle-managed owner, such as a
+            ControllerApplication. Owned subscriptions are removed when the
+            owner is removed from the ECU.
         """
         with self._subscribers_lock:
-            self._subscribers.append({"cb": callback, "dev_adr": device_address})
+            self._subscribers.append(
+                {"cb": callback, "dev_adr": device_address, "owner": owner}
+            )
 
-    def unsubscribe(self, callback):
+    def unsubscribe(self, callback, owner=None):
         """Stop listening for message.
 
         :param callback:
             Function to call when message is received.
+        :param owner:
+            Optional owner used to limit removal to that owner's
+            registration. If omitted, all registrations for the callback
+            are removed, preserving the original behavior.
         """
         with self._subscribers_lock:
-            self._subscribers = [d for d in self._subscribers if d["cb"] != callback]
+            self._subscribers = [
+                d
+                for d in self._subscribers
+                if d["cb"] != callback or (owner is not None and d["owner"] is not owner)
+            ]
+
+    def _unsubscribe_owner(self, owner):
+        """Remove all message subscriptions registered by an owner.
+
+        :param owner: 
+            The owner whose message subscriptions should be removed.
+        """
+        with self._subscribers_lock:
+            self._subscribers = [d for d in self._subscribers if d["owner"] is not owner]
 
     def add_ca(self, **kwargs):
         """Add a ControllerApplication to the ECU.
@@ -374,8 +402,9 @@ def add_ca(self, **kwargs):
             da = kwargs.get("device_address", None)
             ca = ControllerApplication(name, da)
 
-        self.j1939_dll.add_ca(ca)
-        ca.associate_ecu(self)
+        with self._ca_lock:
+            self.j1939_dll.add_ca(ca)
+            ca.associate_ecu(self)
         return ca
 
     def remove_ca(self, device_address):
@@ -386,8 +415,20 @@ def remove_ca(self, device_address):
 
         :return:
             True if the ControllerApplication was successfully removed, otherwise False is returned.
+
+        Any message subscriptions registered through the removed CA are also
+        removed. Callbacks that were registered directly through the ECU are
+        unaffected. When replacing a CA, subscribe its callbacks again through
+        the new CA.
         """
-        return self.j1939_dll.remove_ca(device_address)
+        with self._ca_lock:
+            ca = self.j1939_dll.remove_ca(device_address)
+            if ca is None:
+                return False
+            ca.stop()
+            self._unsubscribe_owner(ca)
+            ca.remove_ecu()
+            return True
 
     def add_bus(self, bus):
         """Add a bus to the ECU.

j1939/j1939_21.py

@@ -48,6 +48,7 @@ def __init__(self, send_message, job_thread_wakeup, notify_subscribers, max_cmdt
 
         # List of ControllerApplication
         self._cas = []
+        self._cas_lock = threading.RLock()
 
         # set minimum time between two tp-rts/cts messages
         self._minimum_tp_rts_cts_dt_interval = minimum_tp_rts_cts_dt_interval
@@ -71,14 +72,20 @@ def __init__(self, send_message, job_thread_wakeup, notify_subscribers, max_cmdt
         self.__ecu_is_message_acceptable = ecu_is_message_acceptable
 
     def add_ca(self, ca):
-        self._cas.append(ca)
+        with self._cas_lock:
+            self._cas.append(ca)
+
+    def _cas_snapshot(self):
+        with self._cas_lock:
+            return list(self._cas)
 
     def remove_ca(self, device_address):
-        for ca in self._cas:
-            if device_address == ca._device_address_preferred:
-                self._cas.remove(ca)
-                return True
-        return False
+        with self._cas_lock:
+            for ca in self._cas:
+                if device_address == ca._device_address_preferred:
+                    self._cas.remove(ca)
+                    return ca
+        return None
 
     def _buffer_hash(self, src_address, dest_address):
         """Calcluates a hash value for the given address pair
@@ -425,7 +432,7 @@ def _process_tp_dt(self, mid, dest_address, data, timestamp):
                     # route Commanded Address (J1939-81) to the registered CAs and
                     # consume it (do not forward to generic subscribers, consistent
                     # with ADDRESSCLAIM/REQUEST handling in notify())
-                    for ca in self._cas:
+                    for ca in self._cas_snapshot():
                         ca._process_commanded_address(src_address, self._rcv_buffer[buffer_hash]['data'], timestamp)
                 else:
                     self.__notify_subscribers(mid.priority, self._rcv_buffer[buffer_hash]['pgn'], src_address, dest_address, timestamp, self._rcv_buffer[buffer_hash]['data'])
@@ -534,20 +541,20 @@ def notify(self, can_id, data, timestamp):
             if self.__ecu_is_message_acceptable(dest_address): # simple peer-to-peer reception without adding a controller-application
                 owns_dest = True
             else:
-                for ca in self._cas:
+                for ca in self._cas_snapshot():
                     if ca.message_acceptable(dest_address):
                         owns_dest = True
                         break
 
         if pgn_value == ParameterGroupNumber.PGN.ADDRESSCLAIM:
-            for ca in self._cas:
+            for ca in self._cas_snapshot():
                 ca._process_addressclaim(mid, data, timestamp)
             # Address claims are broadcast and observable by any node on the bus;
             # forward them to subscribers as well so passive monitors can see the
             # NAME/source-address of other nodes.
             self.__notify_subscribers(mid.priority, pgn_value, mid.source_address, dest_address, timestamp, data)
         elif pgn_value == ParameterGroupNumber.PGN.REQUEST:
-            for ca in self._cas:
+            for ca in self._cas_snapshot():
                 if ca.message_acceptable(dest_address):
                     ca._process_request(mid, dest_address, data, timestamp)
         elif pgn_value == ParameterGroupNumber.PGN.TP_CM:
@@ -564,4 +571,3 @@ def notify(self, can_id, data, timestamp):
             # stack having to own the destination address.
             self.__notify_subscribers(mid.priority, pgn_value, mid.source_address, dest_address, timestamp, data)
             return
-