diff --git a/j1939/controller_application.py b/j1939/controller_application.py index 84d34cb..dc23038 100644 --- a/j1939/controller_application.py +++ b/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 diff --git a/j1939/electronic_control_unit.py b/j1939/electronic_control_unit.py index 6efa776..7637240 100644 --- a/j1939/electronic_control_unit.py +++ b/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. diff --git a/j1939/j1939_21.py b/j1939/j1939_21.py index e3e55ff..862ffb2 100644 --- a/j1939/j1939_21.py +++ b/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 - diff --git a/j1939/j1939_22.py b/j1939/j1939_22.py index 3722cbb..7f2eb1e 100644 --- a/j1939/j1939_22.py +++ b/j1939/j1939_22.py @@ -70,6 +70,7 @@ def __init__(self, send_message, job_thread_wakeup, notify_subscribers, max_cmdt # List of ControllerApplication self._cas = [] + self._cas_lock = threading.RLock() self._LUT_FD_DLC = ( list(range(9)) + @@ -107,14 +108,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, session_num, src_address, dest_address): """Calculates a hash value for the given address pair @@ -576,7 +583,7 @@ def _process_tp_cm(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, pgn, src_address, dest_address, timestamp, self._rcv_buffer[buffer_hash]['data']) @@ -717,7 +724,7 @@ def _process_multi_pg(self, mid : MessageId, 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, payload, timestamp) else: self.__notify_subscribers(mid.priority, cpgn, src_address, dest_address, timestamp, payload) @@ -834,7 +841,7 @@ 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 @@ -844,14 +851,14 @@ def notify(self, can_id, data, timestamp): # contained PGNs are delivered to subscribers regardless of ownership. self._process_multi_pg(mid, dest_address, data, timestamp) elif 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 (consistent with j1939-21). 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.FD_TP_CM: @@ -870,4 +877,3 @@ def notify(self, can_id, data, timestamp): self.__notify_subscribers(mid.priority, pgn.value, mid.source_address, ParameterGroupNumber.Address.GLOBAL, timestamp, data) else: self.__notify_subscribers(mid.priority, pgn_value, mid.source_address, dest_address, timestamp, data) - diff --git a/test/test_ecu.py b/test/test_ecu.py index 3d789b5..f720f53 100644 --- a/test/test_ecu.py +++ b/test/test_ecu.py @@ -1,6 +1,7 @@ import time import can +import pytest import j1939 from test.helpers.feeder import Feeder @@ -343,6 +344,139 @@ def callback(priority: int, pgn: int, sa: int, timestamp: int, data: bytearray): assert call_count == 1 +def test_remove_ca_cleans_ca_subscriptions_but_preserves_ecu_subscriptions(feeder): + """Removing a CA removes only subscriptions registered through that CA.""" + received = [] + + def callback(priority, pgn, sa, timestamp, data): + received.append(data) + + ca = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x80, bypass_address_claim=True + ) + ) + ca.subscribe(callback) + feeder.ecu.subscribe(callback) + + assert len(feeder.ecu._subscribers) == 2 + assert feeder.ecu.remove_ca(0x80) + assert len(feeder.ecu._subscribers) == 1 + assert feeder.ecu._subscribers[0]["owner"] is None + + feeder.ecu._notify_subscribers(6, 0xF000, 1, 0x80, 0.0, bytearray([1])) + assert received == [bytearray([1])] + + +def test_replacement_ca_must_be_resubscribed(feeder): + """A replacement CA receives callbacks only after explicit re-subscription.""" + received = [] + + def callback(priority, pgn, sa, timestamp, data): + received.append(data) + + first_ca = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x80, bypass_address_claim=True + ) + ) + first_ca.subscribe(callback) + assert feeder.ecu.remove_ca(0x80) + + replacement_ca = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x81, bypass_address_claim=True + ) + ) + feeder.ecu._notify_subscribers(6, 0xF000, 1, 0x81, 0.0, bytearray([1])) + assert received == [] + + replacement_ca.subscribe(callback) + feeder.ecu._notify_subscribers(6, 0xF000, 1, 0x81, 0.0, bytearray([2])) + assert received == [bytearray([2])] + + +def test_ca_unsubscribe_preserves_legacy_callback_behavior(feeder): + """CA unsubscribe removes all registrations for the callback as before.""" + received = [] + + def callback(priority, pgn, sa, timestamp, data): + received.append(data) + + first_ca = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x80, bypass_address_claim=True + ) + ) + second_ca = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x81, bypass_address_claim=True + ) + ) + first_ca.subscribe(callback) + second_ca.subscribe(callback) + first_ca.unsubscribe(callback) + + feeder.ecu._notify_subscribers(6, 0xF000, 1, 0x81, 0.0, bytearray([1])) + assert received == [] + + +def test_remove_started_ca_stops_timer_before_detaching(feeder): + """Removing a started CA stops its timer before detaching its ECU.""" + ca = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x80, bypass_address_claim=True + ) + ) + ca.start() + assert ca.started + + assert feeder.ecu.remove_ca(0x80) + assert not ca.started + assert ca._ecu is None + with feeder.ecu._timer_events_lock: + assert all(event[2] != ca._process_claim_async for event in feeder.ecu._timer_events) + + replacement = feeder.ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x80, bypass_address_claim=True + ) + ) + replacement.start() + assert replacement.started + + +@pytest.mark.parametrize("data_link_layer", ["j1939-21", "j1939-22"]) +def test_remove_ca_cleans_subscription_through_receive_path(data_link_layer): + """CA subscriptions are cleaned for both data-link receive paths.""" + ecu = j1939.ElectronicControlUnit( + send_message=lambda *args, **kwargs: None, + data_link_layer=data_link_layer, + ) + received = [] + + def callback(priority, pgn, sa, timestamp, data): + received.append(data) + + try: + ca = ecu.add_ca( + controller_application=j1939.ControllerApplication( + None, device_address_preferred=0x81, bypass_address_claim=True + ) + ) + ca.subscribe(callback) + ecu.notify(0x18DF8101, [1, 2, 3], 0.0) + time.sleep(0.05) + assert len(received) == 1 + + assert ecu.remove_ca(0x81) + ecu.notify(0x18DF8101, [4, 5, 6], 0.0) + time.sleep(0.05) + assert len(received) == 1 + finally: + ecu.stop() + + def test_constructor_accepts_bus_instance(): """Passing a bus instance to the constructor stores it without calling connect().""" bus = can.interface.Bus(interface="virtual", channel="test_ctor_bus")