From 14b078521c0e7f0346890cae62ad33c5134b9f2e Mon Sep 17 00:00:00 2001
From: Koltan Hauersperger <khauersperger@jlg.com>
Date: Mon, 20 Jul 2026 17:06:37 +0000
Subject: [PATCH 1/6] feat: implement a queue for the notifier

---
 j1939/electronic_control_unit.py | 63 +++++++++++++++++++++++++++++++-
 1 file changed, 62 insertions(+), 1 deletion(-)

diff --git a/j1939/electronic_control_unit.py b/j1939/electronic_control_unit.py
index a0fe4cb..13de2f4 100644
--- a/j1939/electronic_control_unit.py
+++ b/j1939/electronic_control_unit.py
@@ -121,8 +121,29 @@ def __init__(
         )
         self._timer_thread.daemon = True
 
+        # Dispatch thread: drains incoming frames from the Notifier thread and
+        # calls j1939_dll.notify() (and therefore _notify_subscribers) serially.
+        #
+        # This is Option A from threadingCheck.md.  The python-can Notifier
+        # thread previously called ecu.notify() → j1939_dll.notify() directly,
+        # blocking the OS socket reader for the entire duration of all CA
+        # callbacks.  With a dispatch queue the Notifier thread enqueues a
+        # (can_id, data, timestamp) tuple and returns to bus.recv() in ~1 µs,
+        # eliminating frame bursting and reducing the GIL hold time seen by the
+        # timer thread.
+        #
+        # Ordering is preserved: SimpleQueue is FIFO and the single dispatch
+        # thread processes frames serially — identical semantics to before.
+        logger.info("Starting ECU dispatch thread")
+        self._dispatch_queue: queue.SimpleQueue = queue.SimpleQueue()
+        self._dispatch_thread = threading.Thread(
+            target=self._dispatch_job_thread, name="j1939.ecu dispatch_thread"
+        )
+        self._dispatch_thread.daemon = True
+
         self._protocol_thread.start()
         self._timer_thread.start()
+        self._dispatch_thread.start()
 
     def stop(self):
         """Stops the ECU background handling
@@ -154,6 +175,7 @@ def stop(self):
         self._timer_wakeup_queue.put(1)
         self._protocol_thread.join()
         self._timer_thread.join()
+        self._dispatch_thread.join()
 
     def register_dependent(self, dependent):
         """Register a helper whose ``stop()`` should be called by :meth:`stop`.
@@ -445,6 +467,11 @@ def notify(self, can_id, data, timestamp):
         If a custom interface is used, this function must be called for each
         29-bit standard message read from the CAN bus.
 
+        The frame is enqueued onto the dispatch queue and processed by the
+        dedicated dispatch thread.  This returns to the caller (typically the
+        python-can Notifier thread) in ~1 µs regardless of how long subscriber
+        callbacks take, preventing frame bursting in the OS socket buffer.
+
         :param int can_id:
             CAN-ID of the message (always 29-bit)
         :param bytearray data:
@@ -455,7 +482,7 @@ def notify(self, can_id, data, timestamp):
             seconds.
             Where possible this will be timestamped in hardware.
         """
-        self.j1939_dll.notify(can_id, data, timestamp)
+        self._dispatch_queue.put((can_id, data, timestamp))
 
     def add_bus_filters(self, filters: can.typechecking.CanFilters | None):
         """Add bus filters to the underlying CAN bus.
@@ -468,6 +495,40 @@ def add_bus_filters(self, filters: can.typechecking.CanFilters | None):
             raise RuntimeError("Not connected to CAN bus")
         self._bus.set_filters(filters)
 
+    def _dispatch_job_thread(self):
+        """Dispatch thread: drains the incoming frame queue and calls the DLL.
+
+        Loops while the ECU is running, blocking on the dispatch queue with a
+        short timeout so it can observe ``_job_thread_end`` being set by
+        :meth:`stop`.  Uses the same ``while not self._job_thread_end.is_set()``
+        exit condition as the protocol and timer threads — no sentinel value
+        or second exit mechanism needed.
+
+        After the loop exits any frames that arrived concurrently with the stop
+        signal are drained so that in-flight TP reassembly is not truncated.
+        """
+        while not self._job_thread_end.is_set():
+            try:
+                can_id, data, timestamp = self._dispatch_queue.get(timeout=0.1)
+            except queue.Empty:
+                continue
+            try:
+                self.j1939_dll.notify(can_id, data, timestamp)
+            except Exception:
+                logger.exception("Exception in dispatch thread")
+
+        # Drain any frames that arrived between the last get() and stop() so
+        # that in-flight TP sessions are not truncated mid-reassembly.
+        while True:
+            try:
+                can_id, data, timestamp = self._dispatch_queue.get_nowait()
+            except queue.Empty:
+                break
+            try:
+                self.j1939_dll.notify(can_id, data, timestamp)
+            except Exception:
+                logger.exception("Exception in dispatch thread (drain)")
+
     def _protocol_job_thread(self):
         """Protocol thread: handles TP/BAM timeout management only.
 

From b349398dd0119c0b1821813d44bf4e82ed07580d Mon Sep 17 00:00:00 2001
From: Koltan Hauersperger <khauersperger@jlg.com>
Date: Tue, 21 Jul 2026 18:19:16 +0000
Subject: [PATCH 2/6] test: add coverage for queue

---
 test/test_threading.py | 290 +++++++++++++++++++++++++++++++----------
 1 file changed, 218 insertions(+), 72 deletions(-)

diff --git a/test/test_threading.py b/test/test_threading.py
index 7cd4b5d..5307955 100644
--- a/test/test_threading.py
+++ b/test/test_threading.py
@@ -4,10 +4,12 @@
 This module tests:
 - Timer accuracy and drift prevention (heapq-based scheduling)
 - Protocol/timer thread separation (slow callbacks don't block protocol)
+- Dispatch queue: Notifier thread unblocked, ordering, drain on stop
 - Thread-safe subscriber list operations
 - MemoryAccess servicer thread lifecycle
 - Dependent registry and cascaded shutdown from ECU
 """
+
 import threading
 import time
 
@@ -25,7 +27,7 @@ def _make_ecu():
 
 def _wait_thread_exit(thread, timeout=0.5):
     """Wait for a thread to exit, polling every 10ms.
-    
+
     :param thread: The thread to wait for.
     :param timeout: Maximum time to wait in seconds.
     :return: True if thread exited, False if timeout reached.
@@ -38,7 +40,7 @@ def _wait_thread_exit(thread, timeout=0.5):
 
 def _wait_no_threads_named(name, timeout=0.5):
     """Wait until no alive threads have the given name.
-    
+
     :param name: Thread name to check for.
     :param timeout: Maximum time to wait in seconds.
     :return: True if no matching threads remain, False if timeout reached.
@@ -50,14 +52,15 @@ def _wait_no_threads_named(name, timeout=0.5):
         time.sleep(0.01)
     return False
 
+
 def test_timer_no_drift():
     """Verify heapq-based timer fires reliably and doesn't deadlock.
-    
+
     This test validates that the timer thread:
     - Fires reliably (gets all expected callbacks)
     - Doesn't deadlock or hang indefinitely
     - Doesn't accumulate extreme outlier delays (> 500ms)
-    
+
     Note: Strict interval timing is not validated on slow CI machines.
     The focus is on correctness (fire count) and absence of hangs.
     """
@@ -69,8 +72,8 @@ def callback(cookie):
         timestamps.append(time.monotonic())
         if len(timestamps) >= 10:
             done.set()
-            return False   # stop rescheduling
-        return True        # reschedule
+            return False  # stop rescheduling
+        return True  # reschedule
 
     ecu.add_timer(0.050, callback)
     fired = done.wait(timeout=10.0)  # generous timeout for very slow CI
@@ -79,20 +82,20 @@ def callback(cookie):
     assert fired, "Timer did not fire 10 times within 10 seconds - possible deadlock"
     assert len(timestamps) == 10, f"Expected 10 callbacks, got {len(timestamps)}"
 
-    intervals = [timestamps[i+1] - timestamps[i] for i in range(9)]
-    
+    intervals = [timestamps[i + 1] - timestamps[i] for i in range(9)]
+
     # Only check for extreme outliers that would indicate a broken timer
     # (e.g., long GC pause, system under extreme load, or actual deadlock)
     max_interval = max(intervals)
     assert max_interval < 1.0, (
-        f"Max interval was {max_interval*1000:.1f}ms, which is extreme "
+        f"Max interval was {max_interval * 1000:.1f}ms, which is extreme "
         "(expected < 1000ms even on slow CI). Timer may be deadlocked or broken."
     )
-    
+
     # Log intervals for debugging CI issues
     avg_interval = sum(intervals) / len(intervals)
     assert avg_interval > 0.025, (
-        f"Average interval was {avg_interval*1000:.1f}ms, "
+        f"Average interval was {avg_interval * 1000:.1f}ms, "
         "which is too fast (timer may be firing twice per cycle)"
     )
 
@@ -104,7 +107,7 @@ def test_slow_callback_no_protocol_impact(feeder):
 
     def slow_callback(cookie):
         slow_fired.set()
-        time.sleep(0.150)   # simulate heavy work
+        time.sleep(0.150)  # simulate heavy work
         return True
 
     feeder.ecu.add_timer(0.020, slow_callback)
@@ -115,18 +118,19 @@ def slow_callback(cookie):
     # 20-byte BAM: BAM announce + 3 DT frames
     pgn_value = 0xFEC8  # arbitrary broadcast PGN
     # Build raw CAN message sequence (same pattern as test_ecu.py)
-    can_id_bam = 0x1CECFF01   # TP.CM BAM from 0x01 to global
-    can_id_dt  = 0x1CEBFF01   # TP.DT  from 0x01 to global
+    can_id_bam = 0x1CECFF01  # TP.CM BAM from 0x01 to global
+    can_id_dt = 0x1CEBFF01  # TP.DT  from 0x01 to global
 
     feeder.can_messages = [
-        (Feeder.MsgType.CANRX, can_id_bam,
-         [32, 20, 0, 3, 255, pgn_value & 0xFF, (pgn_value >> 8) & 0xFF, 0], 0.0),
-        (Feeder.MsgType.CANRX, can_id_dt,
-         [1, 1, 2, 3, 4, 5, 6, 7], 0.0),
-        (Feeder.MsgType.CANRX, can_id_dt,
-         [2, 8, 9, 10, 11, 12, 13, 14], 0.0),
-        (Feeder.MsgType.CANRX, can_id_dt,
-         [3, 15, 16, 17, 18, 19, 20, 255], 0.0),
+        (
+            Feeder.MsgType.CANRX,
+            can_id_bam,
+            [32, 20, 0, 3, 255, pgn_value & 0xFF, (pgn_value >> 8) & 0xFF, 0],
+            0.0,
+        ),
+        (Feeder.MsgType.CANRX, can_id_dt, [1, 1, 2, 3, 4, 5, 6, 7], 0.0),
+        (Feeder.MsgType.CANRX, can_id_dt, [2, 8, 9, 10, 11, 12, 13, 14], 0.0),
+        (Feeder.MsgType.CANRX, can_id_dt, [3, 15, 16, 17, 18, 19, 20, 255], 0.0),
     ]
 
     received = threading.Event()
@@ -152,7 +156,7 @@ def on_message(priority, pgn, sa, timestamp, data):
     feeder.ecu.remove_timer(slow_callback)
 
     assert delivered, (
-        f"BAM message was not reassembled within 400ms (elapsed {elapsed*1000:.0f}ms). "
+        f"BAM message was not reassembled within 400ms (elapsed {elapsed * 1000:.0f}ms). "
         "Slow callback may be blocking the protocol thread."
     )
 
@@ -185,9 +189,10 @@ def hammer():
 
     assert not errors, f"Exceptions during concurrent timer ops: {errors}"
 
+
 def test_memory_access_event_latency():
     """MemoryAccess servicer thread responds to events within reasonable latency.
-    
+
     This test validates that the servicer thread wakes up and responds to events
     without excessive delay. Rather than enforcing sub-10ms latency (which is
     unrealistic on slow CI machines with variable scheduler load), we check that:
@@ -197,17 +202,20 @@ def test_memory_access_event_latency():
     from j1939.memory_access import DMState, MemoryAccess
 
     ecu = _make_ecu()
-    ca = ecu.add_ca(name=j1939.Name(
-        arbitrary_address_capable=0,
-        industry_group=j1939.Name.IndustryGroup.Industrial,
-        vehicle_system_instance=1,
-        vehicle_system=1,
-        function=1,
-        function_instance=1,
-        ecu_instance=1,
-        manufacturer_code=1,
-        identity_number=1,
-    ), device_address=0x80)
+    ca = ecu.add_ca(
+        name=j1939.Name(
+            arbitrary_address_capable=0,
+            industry_group=j1939.Name.IndustryGroup.Industrial,
+            vehicle_system_instance=1,
+            vehicle_system=1,
+            function=1,
+            function_instance=1,
+            ecu_instance=1,
+            manufacturer_code=1,
+            identity_number=1,
+        ),
+        device_address=0x80,
+    )
 
     ma = MemoryAccess(ca)
 
@@ -233,7 +241,7 @@ def notify():
     assert callback_times, "notify callback was never called after _proceed_event.set()"
     latency = callback_times[0] - set_time[0]
     assert latency < 0.500, (
-        f"MemoryAccess notify latency was {latency*1000:.2f}ms, "
+        f"MemoryAccess notify latency was {latency * 1000:.2f}ms, "
         f"expected < 500ms (thread should not be blocked/deadlocked)"
     )
 
@@ -266,8 +274,9 @@ def subscribe_loop():
     can_id = 0x18FEC801  # broadcast from 0x01, PGN 0xFEC8
     inject_deadline = time.monotonic() + 0.5
     while time.monotonic() < inject_deadline:
-        feeder.message_queue.put((Feeder.MsgType.CANRX, can_id,
-                                  bytearray([1, 2, 3, 4, 5, 6, 7, 8]), 0.0))
+        feeder.message_queue.put(
+            (Feeder.MsgType.CANRX, can_id, bytearray([1, 2, 3, 4, 5, 6, 7, 8]), 0.0)
+        )
         time.sleep(0.01)
 
     sub_thread.join(timeout=2.0)
@@ -276,25 +285,30 @@ def subscribe_loop():
     assert not errors, f"Exceptions during subscribe/unsubscribe race: {errors}"
     assert received_count[0] > 0, "No messages were received during the race"
 
+
 def _make_ca(ecu, device_address=0x80):
     """Create a ControllerApplication with minimal valid Name."""
-    return ecu.add_ca(name=j1939.Name(
-        arbitrary_address_capable=0,
-        industry_group=j1939.Name.IndustryGroup.Industrial,
-        vehicle_system_instance=1,
-        vehicle_system=1,
-        function=1,
-        function_instance=1,
-        ecu_instance=1,
-        manufacturer_code=1,
-        identity_number=1,
-    ), device_address=device_address)
+    return ecu.add_ca(
+        name=j1939.Name(
+            arbitrary_address_capable=0,
+            industry_group=j1939.Name.IndustryGroup.Industrial,
+            vehicle_system_instance=1,
+            vehicle_system=1,
+            function=1,
+            function_instance=1,
+            ecu_instance=1,
+            manufacturer_code=1,
+            identity_number=1,
+        ),
+        device_address=device_address,
+    )
 
 
 def _j1939_threads():
     """Return list of alive threads with names starting with 'j1939.'."""
-    return [t for t in threading.enumerate()
-            if t.name.startswith('j1939.') and t.is_alive()]
+    return [
+        t for t in threading.enumerate() if t.name.startswith("j1939.") and t.is_alive()
+    ]
 
 
 class _FakeDependent:
@@ -322,37 +336,40 @@ def test_ecu_stop_cascades_to_memory_access():
     MemoryAccess(ca)
 
     # Sanity: servicer thread is running
-    assert any(t.name == 'j1939.memory_access servicer_thread' for t in _j1939_threads())
+    assert any(
+        t.name == "j1939.memory_access servicer_thread" for t in _j1939_threads()
+    )
 
     ecu.stop()
 
-    assert _wait_no_threads_named('j1939.memory_access servicer_thread', timeout=1.0), \
+    assert _wait_no_threads_named("j1939.memory_access servicer_thread", timeout=1.0), (
         "MemoryAccess servicer thread still running after ecu.stop()"
+    )
 
 
 def test_ecu_stop_cascades_lifo():
     """Dependents must be stopped in reverse registration order."""
     ecu = _make_ecu()
     log = []
-    a = _FakeDependent(log, 'A')
-    b = _FakeDependent(log, 'B')
-    c = _FakeDependent(log, 'C')
+    a = _FakeDependent(log, "A")
+    b = _FakeDependent(log, "B")
+    c = _FakeDependent(log, "C")
     ecu.register_dependent(a)
     ecu.register_dependent(b)
     ecu.register_dependent(c)
 
     ecu.stop()
 
-    assert log == ['C', 'B', 'A'], log
+    assert log == ["C", "B", "A"], log
 
 
 def test_ecu_stop_continues_on_dependent_failure():
     """A failing dependent.stop() must not prevent others from running."""
     ecu = _make_ecu()
     log = []
-    a = _FakeDependent(log, 'A')
-    b = _FakeDependent(log, 'B', raise_on_stop=True)
-    c = _FakeDependent(log, 'C')
+    a = _FakeDependent(log, "A")
+    b = _FakeDependent(log, "B", raise_on_stop=True)
+    c = _FakeDependent(log, "C")
     ecu.register_dependent(a)
     ecu.register_dependent(b)
     ecu.register_dependent(c)
@@ -360,7 +377,7 @@ def test_ecu_stop_continues_on_dependent_failure():
     ecu.stop()  # must not raise
 
     # All three should have had stop() called despite B raising.
-    assert log == ['C', 'B', 'A']
+    assert log == ["C", "B", "A"]
     # And ECU's own threads are stopped.
     assert not ecu._protocol_thread.is_alive()
     assert not ecu._timer_thread.is_alive()
@@ -378,9 +395,12 @@ def test_memory_access_explicit_stop_no_leak():
     ma.stop()
     elapsed = time.monotonic() - t0
 
-    assert elapsed < 0.050, f"MemoryAccess.stop() took {elapsed*1000:.1f}ms, expected < 50ms"
-    assert _wait_no_threads_named('j1939.memory_access servicer_thread'), \
+    assert elapsed < 0.050, (
+        f"MemoryAccess.stop() took {elapsed * 1000:.1f}ms, expected < 50ms"
+    )
+    assert _wait_no_threads_named("j1939.memory_access servicer_thread"), (
         "Servicer thread still running after ma.stop()"
+    )
 
     ecu.stop()
 
@@ -394,7 +414,9 @@ def test_memory_access_context_manager():
     with MemoryAccess(ca) as ma:
         assert ma._job_thread.is_alive()
 
-    assert _wait_thread_exit(ma._job_thread), "Servicer thread did not stop after context exit"
+    assert _wait_thread_exit(ma._job_thread), (
+        "Servicer thread did not stop after context exit"
+    )
     ecu.stop()
 
 
@@ -415,7 +437,7 @@ def test_register_unregister_dependent_idempotent():
     """Duplicate register/unregister calls are silently handled."""
     ecu = _make_ecu()
     log = []
-    a = _FakeDependent(log, 'A')
+    a = _FakeDependent(log, "A")
 
     ecu.register_dependent(a)
     ecu.register_dependent(a)  # duplicate - silently deduped
@@ -438,12 +460,12 @@ def test_register_dependent_rejected_during_shutdown():
     """Registering new dependent during shutdown raises RuntimeError."""
     ecu = _make_ecu()
     log = []
-    blocker = _FakeDependent(log, 'blocker')
-    late = _FakeDependent(log, 'late')
+    blocker = _FakeDependent(log, "blocker")
+    late = _FakeDependent(log, "late")
     captured = []
 
     def blocker_stop():
-        log.append('blocker')
+        log.append("blocker")
         try:
             ecu.register_dependent(late)
         except RuntimeError as e:
@@ -455,7 +477,7 @@ def blocker_stop():
     ecu.stop()
 
     assert captured, "Expected RuntimeError when registering during shutdown"
-    assert log == ['blocker']
+    assert log == ["blocker"]
 
 
 def test_send_pgn_concurrent_no_crash():
@@ -512,8 +534,9 @@ def capture_send(can_id, extended, data, fd_format=False):
     def send_once():
         try:
             barrier.wait()  # start both threads simultaneously
-            result = ecu.send_pgn(0, 0xFE, ParameterGroupNumber.Address.GLOBAL,
-                                  6, 0x01, list(range(20)))
+            result = ecu.send_pgn(
+                0, 0xFE, ParameterGroupNumber.Address.GLOBAL, 6, 0x01, list(range(20))
+            )
             results.append(result)
         except Exception as exc:
             errors.append(exc)
@@ -545,7 +568,130 @@ def test_dependent_registration_stress_no_leak():
         ma = MemoryAccess(ca)
         ma.stop()
 
-    assert _wait_no_threads_named('j1939.memory_access servicer_thread', timeout=1.0), \
+    assert _wait_no_threads_named("j1939.memory_access servicer_thread", timeout=1.0), (
         "Leaked servicer thread(s) after stress test"
+    )
+
+    ecu.stop()
+
+
+def test_dispatch_thread_exists_and_named():
+    """ECU must have a live, correctly named dispatch thread after construction."""
+    ecu = _make_ecu()
+    try:
+        assert hasattr(ecu, "_dispatch_thread"), "ECU has no _dispatch_thread attribute"
+        assert ecu._dispatch_thread.is_alive(), "dispatch thread is not alive"
+        assert ecu._dispatch_thread.name == "j1939.ecu dispatch_thread"
+    finally:
+        ecu.stop()
+
+
+def test_notify_returns_immediately_while_dispatch_is_busy():
+    """notify() must return in well under 1 ms even when a slow subscriber
+    callback is running in the dispatch thread.
+
+    This is the core guarantee of Option A: the Notifier thread (which calls
+    ecu.notify()) is never blocked by subscriber callback work.
+    """
+    ecu = _make_ecu()
+    try:
+        callback_entered = threading.Event()
+        callback_release = threading.Event()
+
+        def slow_subscriber(priority, pgn, sa, timestamp, data):
+            callback_entered.set()
+            callback_release.wait()  # block until the test releases it
+
+        ecu.subscribe(slow_subscriber)
+        # PGN 0xFF00 — PDU2 broadcast, SA 0x01
+        can_id = 0x18FF0001
+
+        # Fire the first notify so the slow callback is now running
+        ecu.notify(can_id, bytearray(8), 0.0)
+        assert callback_entered.wait(timeout=2.0), "Slow callback never started"
+
+        # The dispatch thread is now stuck in slow_subscriber.
+        # notify() must still return instantly from the Notifier thread's POV.
+        t0 = time.perf_counter()
+        ecu.notify(can_id, bytearray(8), 0.0)
+        elapsed = time.perf_counter() - t0
+
+        assert elapsed < 0.001, (
+            f"notify() took {elapsed * 1000:.2f} ms while dispatch was busy — "
+            "Notifier thread is being blocked by subscriber callbacks."
+        )
+    finally:
+        callback_release.set()  # unblock callback so threads can exit cleanly
+        ecu.stop()
+
+
+def test_dispatch_preserves_frame_order():
+    """Frames must be delivered to subscribers in the order notify() was called."""
+    ecu = _make_ecu()
+    try:
+        received_sas = []
+        done = threading.Event()
+        expected_count = 10
+
+        def record(priority, pgn, sa, timestamp, data):
+            received_sas.append(sa)
+            if len(received_sas) >= expected_count:
+                done.set()
+
+        ecu.subscribe(record)
+
+        # Inject frames with SA = 0..9 in order; PDU2 broadcast PGN
+        for sa in range(expected_count):
+            can_id = 0x18FF0000 | sa
+            ecu.notify(can_id, bytearray(8), 0.0)
+
+        assert done.wait(timeout=2.0), "Not all frames were delivered"
+        assert received_sas == list(range(expected_count)), (
+            f"Frame delivery order incorrect: {received_sas}"
+        )
+    finally:
+        ecu.stop()
+
+
+def test_dispatch_thread_stops_cleanly_on_ecu_stop():
+    """The dispatch thread must exit within 500 ms of ecu.stop()."""
+    ecu = _make_ecu()
+    dispatch_thread = ecu._dispatch_thread
+    assert dispatch_thread.is_alive()
 
     ecu.stop()
+
+    assert _wait_thread_exit(dispatch_thread, timeout=0.5), (
+        "dispatch_thread did not exit within 500 ms of ecu.stop()"
+    )
+
+
+def test_dispatch_drains_queued_frames_before_stop():
+    """Frames enqueued just before stop() must still be delivered.
+
+    The dispatch thread drains the queue after _job_thread_end is set, so
+    in-flight frames are not silently dropped on shutdown.
+    """
+    ecu = _make_ecu()
+    try:
+        received = []
+        done = threading.Event()
+        n = 5
+
+        def record(priority, pgn, sa, timestamp, data):
+            received.append(sa)
+            if len(received) >= n:
+                done.set()
+
+        ecu.subscribe(record)
+
+        # Enqueue several frames then stop immediately
+        for sa in range(n):
+            ecu.notify(0x18FF0000 | sa, bytearray(8), 0.0)
+    finally:
+        ecu.stop()
+
+    # After stop(), drain should have processed everything already enqueued
+    assert len(received) == n, (
+        f"Expected {n} frames after stop-drain, got {len(received)}: {received}"
+    )

From fbdf6d9032660378f7dda8b2929e9e2ad862654d Mon Sep 17 00:00:00 2001
From: Koltan Hauersperger <khauersperger@jlg.com>
Date: Tue, 21 Jul 2026 18:28:00 +0000
Subject: [PATCH 3/6] docs: remove unneeded documentation

---
 j1939/electronic_control_unit.py | 9 ---------
 1 file changed, 9 deletions(-)

diff --git a/j1939/electronic_control_unit.py b/j1939/electronic_control_unit.py
index 13de2f4..12e9de5 100644
--- a/j1939/electronic_control_unit.py
+++ b/j1939/electronic_control_unit.py
@@ -123,15 +123,6 @@ def __init__(
 
         # Dispatch thread: drains incoming frames from the Notifier thread and
         # calls j1939_dll.notify() (and therefore _notify_subscribers) serially.
-        #
-        # This is Option A from threadingCheck.md.  The python-can Notifier
-        # thread previously called ecu.notify() → j1939_dll.notify() directly,
-        # blocking the OS socket reader for the entire duration of all CA
-        # callbacks.  With a dispatch queue the Notifier thread enqueues a
-        # (can_id, data, timestamp) tuple and returns to bus.recv() in ~1 µs,
-        # eliminating frame bursting and reducing the GIL hold time seen by the
-        # timer thread.
-        #
         # Ordering is preserved: SimpleQueue is FIFO and the single dispatch
         # thread processes frames serially — identical semantics to before.
         logger.info("Starting ECU dispatch thread")

From 707753451f8f23bc22634a83a1f536799e194c64 Mon Sep 17 00:00:00 2001
From: Koltan Hauersperger <khauersperger@jlg.com>
Date: Tue, 21 Jul 2026 18:33:33 +0000
Subject: [PATCH 4/6] docs: remove some more unneeded documentation

---
 test/test_threading.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/test/test_threading.py b/test/test_threading.py
index 5307955..aba6351 100644
--- a/test/test_threading.py
+++ b/test/test_threading.py
@@ -589,9 +589,6 @@ def test_dispatch_thread_exists_and_named():
 def test_notify_returns_immediately_while_dispatch_is_busy():
     """notify() must return in well under 1 ms even when a slow subscriber
     callback is running in the dispatch thread.
-
-    This is the core guarantee of Option A: the Notifier thread (which calls
-    ecu.notify()) is never blocked by subscriber callback work.
     """
     ecu = _make_ecu()
     try:

From 5fabfd734160a3da724265b0d4ca2f6f58806ac1 Mon Sep 17 00:00:00 2001
From: Koltan Hauersperger <khauersperger@jlg.com>
Date: Tue, 21 Jul 2026 19:03:20 +0000
Subject: [PATCH 5/6] feat: add more docs and clean up some potential

bugs
---
 j1939/electronic_control_unit.py | 34 ++++++++++++++++++++++++++++----
 test/test_threading.py           |  2 +-
 2 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/j1939/electronic_control_unit.py b/j1939/electronic_control_unit.py
index 12e9de5..6960fc2 100644
--- a/j1939/electronic_control_unit.py
+++ b/j1939/electronic_control_unit.py
@@ -125,6 +125,14 @@ def __init__(
         # calls j1939_dll.notify() (and therefore _notify_subscribers) serially.
         # Ordering is preserved: SimpleQueue is FIFO and the single dispatch
         # thread processes frames serially — identical semantics to before.
+        #
+        # The queue is unbounded: it trades memory for backpressure — if
+        # subscriber callbacks are slower than the incoming frame rate the queue
+        # grows without bound.  In practice the OS socket buffer provides a
+        # natural upstream limit, and the J1939 bus rates in this codebase are
+        # well within subscriber throughput.  If bounded buffering is needed,
+        # replace SimpleQueue with queue.Queue(maxsize=N) and handle the Full
+        # exception in notify() (drop, log, or block to taste).
         logger.info("Starting ECU dispatch thread")
         self._dispatch_queue: queue.SimpleQueue = queue.SimpleQueue()
         self._dispatch_thread = threading.Thread(
@@ -136,7 +144,7 @@ def __init__(
         self._timer_thread.start()
         self._dispatch_thread.start()
 
-    def stop(self):
+    def stop(self, dispatch_join_timeout: float = 3.0):
         """Stops the ECU background handling
 
         This Function explicitly stops the background handling of the ECU.
@@ -146,6 +154,13 @@ def stop(self):
         invoked in LIFO order. Exceptions raised by a dependent's ``stop()``
         are logged and swallowed so a single misbehaving dependent cannot
         prevent the rest of the shutdown from completing.
+
+        :param float dispatch_join_timeout:
+            Maximum seconds to wait for the dispatch thread to finish its
+            current subscriber callback before giving up and continuing
+            shutdown.  The dispatch thread is daemonic so it will not prevent
+            interpreter exit, but a warning is logged if it is still alive
+            after this timeout.  Defaults to 3s
         """
         # Snapshot dependents under lock, then mark the ECU as stopping so any
         # late registrations are rejected.
@@ -166,7 +181,13 @@ def stop(self):
         self._timer_wakeup_queue.put(1)
         self._protocol_thread.join()
         self._timer_thread.join()
-        self._dispatch_thread.join()
+        self._dispatch_thread.join(timeout=dispatch_join_timeout)
+        if self._dispatch_thread.is_alive():
+            logger.warning(
+                "dispatch_thread did not exit within %.1f s — a subscriber "
+                "callback may be blocking. Continuing shutdown.",
+                dispatch_join_timeout,
+            )
 
     def register_dependent(self, dependent):
         """Register a helper whose ``stop()`` should be called by :meth:`stop`.
@@ -460,8 +481,8 @@ def notify(self, can_id, data, timestamp):
 
         The frame is enqueued onto the dispatch queue and processed by the
         dedicated dispatch thread.  This returns to the caller (typically the
-        python-can Notifier thread) in ~1 µs regardless of how long subscriber
-        callbacks take, preventing frame bursting in the OS socket buffer.
+        python-can Notifier thread) without running subscriber callbacks,
+        preventing frame bursting in the OS socket buffer.
 
         :param int can_id:
             CAN-ID of the message (always 29-bit)
@@ -473,6 +494,11 @@ def notify(self, can_id, data, timestamp):
             seconds.
             Where possible this will be timestamped in hardware.
         """
+        if self._job_thread_end.is_set():
+            # ECU is stopping or stopped; the dispatch thread has exited or is
+            # draining. Drop the frame rather than growing the queue
+            # with no consumer.
+            return
         self._dispatch_queue.put((can_id, data, timestamp))
 
     def add_bus_filters(self, filters: can.typechecking.CanFilters | None):
diff --git a/test/test_threading.py b/test/test_threading.py
index aba6351..957decf 100644
--- a/test/test_threading.py
+++ b/test/test_threading.py
@@ -613,7 +613,7 @@ def slow_subscriber(priority, pgn, sa, timestamp, data):
         ecu.notify(can_id, bytearray(8), 0.0)
         elapsed = time.perf_counter() - t0
 
-        assert elapsed < 0.001, (
+        assert elapsed < 0.1, (
             f"notify() took {elapsed * 1000:.2f} ms while dispatch was busy — "
             "Notifier thread is being blocked by subscriber callbacks."
         )

From 26c98209482b553eba556269c55be0e450040940 Mon Sep 17 00:00:00 2001
From: Koltan Hauersperger <khauersperger@jlg.com>
Date: Wed, 22 Jul 2026 18:19:07 +0000
Subject: [PATCH 6/6] feat: bound the queue to make it safer

---
 j1939/electronic_control_unit.py | 58 ++++++++++++++++++++------
 test/test_threading.py           | 70 +++++++++++++++++++++++++++++++-
 2 files changed, 114 insertions(+), 14 deletions(-)

diff --git a/j1939/electronic_control_unit.py b/j1939/electronic_control_unit.py
index 6960fc2..ef1d5d0 100644
--- a/j1939/electronic_control_unit.py
+++ b/j1939/electronic_control_unit.py
@@ -30,6 +30,7 @@ def __init__(
         minimum_tp_bam_dt_interval=None,
         send_message=None,
         bus: can.BusABC | None = None,
+        dispatch_queue_size: int = 1000,
     ):
         """
         :param data_link_layer:
@@ -44,6 +45,12 @@ def __init__(
             optional callback function to send a raw CAN message to the bus. If not provided, the default implementation will be used, which sends messages via the python-can bus.
         :param bus:
             optional python-can :class:`can.BusABC` instance. If not provided, the ECU will not be connected to a bus until :meth:`connect` is called.
+        :param int dispatch_queue_size:
+            Maximum number of CAN frames that may be buffered in the dispatch
+            queue between the python-can Notifier thread and the ECU dispatch
+            thread. If the queue is full when a new frame arrives the frame is
+            dropped and a warning is logged (suppressed until the queue drains).
+            Defaults to 1000.
         """
         if send_message:
             self.send_message = send_message
@@ -123,18 +130,21 @@ def __init__(
 
         # Dispatch thread: drains incoming frames from the Notifier thread and
         # calls j1939_dll.notify() (and therefore _notify_subscribers) serially.
-        # Ordering is preserved: SimpleQueue is FIFO and the single dispatch
+        # Ordering is preserved: the queue is FIFO and the single dispatch
         # thread processes frames serially — identical semantics to before.
         #
-        # The queue is unbounded: it trades memory for backpressure — if
-        # subscriber callbacks are slower than the incoming frame rate the queue
-        # grows without bound.  In practice the OS socket buffer provides a
-        # natural upstream limit, and the J1939 bus rates in this codebase are
-        # well within subscriber throughput.  If bounded buffering is needed,
-        # replace SimpleQueue with queue.Queue(maxsize=N) and handle the Full
-        # exception in notify() (drop, log, or block to taste).
+        # The queue is bounded (maxsize=dispatch_queue_size, default 1000).
+        # When full, notify() drops the incoming frame and logs a warning.
+        # The warning is suppressed after the first drop and re-emitted as a
+        # summary once the queue has room again, to avoid log flooding.
+        # Drop-tracking state is only accessed from the python-can Notifier
+        # thread (the sole caller of notify()), so no locking is required.
         logger.info("Starting ECU dispatch thread")
-        self._dispatch_queue: queue.SimpleQueue = queue.SimpleQueue()
+        self._dispatch_queue: queue.Queue = queue.Queue(maxsize=dispatch_queue_size)
+        # Number of frames dropped since the last time the queue drained.
+        self._dispatch_queue_drop_count: int = 0
+        # True while the queue is at capacity and frames are being dropped.
+        self._dispatch_queue_dropped: bool = False
         self._dispatch_thread = threading.Thread(
             target=self._dispatch_job_thread, name="j1939.ecu dispatch_thread"
         )
@@ -480,9 +490,14 @@ def notify(self, can_id, data, timestamp):
         29-bit standard message read from the CAN bus.
 
         The frame is enqueued onto the dispatch queue and processed by the
-        dedicated dispatch thread.  This returns to the caller (typically the
-        python-can Notifier thread) without running subscriber callbacks,
-        preventing frame bursting in the OS socket buffer.
+        dedicated dispatch thread.  This returns quickly to the caller
+        (typically the python-can Notifier thread) without waiting for
+        subscriber callbacks to complete.
+
+        If the dispatch queue is full the frame is dropped rather than
+        blocking.  A warning is logged on the first drop and suppressed until
+        the queue drains, at which point a summary of the total drop count is
+        logged.
 
         :param int can_id:
             CAN-ID of the message (always 29-bit)
@@ -499,7 +514,24 @@ def notify(self, can_id, data, timestamp):
             # draining. Drop the frame rather than growing the queue
             # with no consumer.
             return
-        self._dispatch_queue.put((can_id, data, timestamp))
+        try:
+            self._dispatch_queue.put_nowait((can_id, data, timestamp))
+            if self._dispatch_queue_dropped:
+                # Queue has room again — emit the suppressed summary and reset.
+                logger.warning(
+                    "dispatch_queue drained: %d frame(s) were dropped while the queue was full",
+                    self._dispatch_queue_drop_count,
+                )
+                self._dispatch_queue_dropped = False
+                self._dispatch_queue_drop_count = 0
+        except queue.Full:
+            self._dispatch_queue_drop_count += 1
+            if not self._dispatch_queue_dropped:
+                logger.warning(
+                    "dispatch_queue full (maxsize=%d): dropping incoming frames until queue drains",
+                    self._dispatch_queue.maxsize,
+                )
+                self._dispatch_queue_dropped = True
 
     def add_bus_filters(self, filters: can.typechecking.CanFilters | None):
         """Add bus filters to the underlying CAN bus.
diff --git a/test/test_threading.py b/test/test_threading.py
index 957decf..d8a292f 100644
--- a/test/test_threading.py
+++ b/test/test_threading.py
@@ -587,7 +587,7 @@ def test_dispatch_thread_exists_and_named():
 
 
 def test_notify_returns_immediately_while_dispatch_is_busy():
-    """notify() must return in well under 1 ms even when a slow subscriber
+    """notify() must return well under 100 ms even when a slow subscriber
     callback is running in the dispatch thread.
     """
     ecu = _make_ecu()
@@ -692,3 +692,71 @@ def record(priority, pgn, sa, timestamp, data):
     assert len(received) == n, (
         f"Expected {n} frames after stop-drain, got {len(received)}: {received}"
     )
+
+
+def test_dispatch_queue_drop_on_full():
+    """When the dispatch queue is full, notify() drops frames without blocking
+    or raising.  The drop counter increments and the dropped flag is set.
+    Once the queue drains the state resets.
+    """
+    # Use a tiny queue so it is easy to fill.
+    ecu = j1939.ElectronicControlUnit(
+        send_message=lambda *a, **kw: None,
+        dispatch_queue_size=5,
+    )
+    try:
+        callback_entered = threading.Event()
+        callback_release = threading.Event()
+
+        def blocking_subscriber(priority, pgn, sa, timestamp, data):
+            callback_entered.set()
+            callback_release.wait()
+
+        ecu.subscribe(blocking_subscriber)
+
+        # Trigger the first frame so the dispatch thread is blocked inside the
+        # slow callback, preventing the queue from draining.
+        can_id = 0x18FF0001
+        ecu.notify(can_id, bytearray(8), 0.0)
+        assert callback_entered.wait(timeout=2.0), "Blocking subscriber never entered"
+
+        # Queue can hold 5 items; one is already consumed (dispatch thread is
+        # inside the callback).  Fill and then overflow it.
+        for _ in range(10):
+            ecu.notify(can_id, bytearray(8), 0.0)
+
+        # Some frames must have been dropped (queue size 5, we sent 10 extra).
+        assert ecu._dispatch_queue_drop_count > 0, (
+            "Expected dropped frames but drop_count is 0"
+        )
+        assert ecu._dispatch_queue_dropped is True, (
+            "Expected _dispatch_queue_dropped to be True while queue is full"
+        )
+
+        # Release the blocking callback so the queue drains.
+        callback_release.set()
+
+        # Wait for the queue to drain fully, then send one more frame to
+        # trigger the "success path" of put_nowait which resets the drop state.
+        deadline = time.monotonic() + 2.0
+        while ecu._dispatch_queue.qsize() > 0 and time.monotonic() < deadline:
+            time.sleep(0.01)
+
+        # One more notify() to trigger the drain-reset path in notify().
+        ecu.notify(can_id, bytearray(8), 0.0)
+
+        # Give a moment for the reset to propagate (it happens synchronously in
+        # the put_nowait success branch, so it should be immediate).
+        deadline = time.monotonic() + 2.0
+        while ecu._dispatch_queue_dropped and time.monotonic() < deadline:
+            time.sleep(0.01)
+
+        assert not ecu._dispatch_queue_dropped, (
+            "dispatch_queue_dropped flag was not cleared after queue drained"
+        )
+        assert ecu._dispatch_queue_drop_count == 0, (
+            "dispatch_queue_drop_count was not reset after queue drained"
+        )
+    finally:
+        callback_release.set()  # ensure unblocked even if test fails early
+        ecu.stop()