Add logic to address threading issues with writing from memory access

j1939/Dm14Server.py

@@ -164,6 +164,7 @@ def parse_dm14(
                 self.reset_server()
 
             case _:
+                self.reset_server()
                 raise ValueError("Invalid state")
 
     def _send_dm15(
@@ -220,6 +221,7 @@ def _send_dm15(
                 data[length - 3] = edcp
 
             case _:
+                self.reset_server()
                 raise ValueError("Invalid state")
         self._ca.send_pgn(0, (pgn >> 8) & 0xFF, sa & 0xFF, 6, data)
 
@@ -253,6 +255,7 @@ def _parse_dm16(
 
         if pgn != j1939.ParameterGroupNumber.PGN.DM16 or sa != self.sa:
             return
+
         length = min(data[0], len(data) - 1)
         self.data_queue.put(data[1 : length + 1])
         self._ca.unsubscribe(self._parse_dm16)
@@ -388,5 +391,9 @@ def respond(
         self._wait_for_data()
         mem_data = None
         if self.state == ResponseState.WAIT_FOR_DM16:
-            mem_data = self.data_queue.get(block=True, timeout=max_timeout)
+            try:
+                mem_data = self.data_queue.get(block=True, timeout=max_timeout)
+            except queue.Empty:
+                self.reset_server()
+                raise RuntimeError("No data received from DM16 within timeout period")
         return mem_data

j1939/memory_access.py

@@ -1,7 +1,8 @@
 from enum import Enum
+import threading
+import time
 import j1939
 
-
 class DMState(Enum):
     IDLE = 1
     REQUEST_STARTED = 2
@@ -20,12 +21,38 @@ def __init__(self, ca: j1939.ControllerApplication) -> None:
         self._ca = ca
         self.query = j1939.Dm14Query(ca)
         self.server = j1939.DM14Server(ca)
+        self.proceed = False
         self._ca.subscribe(self._listen_for_dm14)
         self.state = DMState.IDLE
         self.seed_security = False
         self._notify_query_received = None
         self._proceed_function = None
 
+        self._job_thread_end = threading.Event()
+        self._job_thread = threading.Thread(target=self._servicer, name='j1939.memory_access servicer_thread')
+        # A thread can be flagged as a "daemon thread". The significance of
+        # this flag is that the entire Python program exits when only daemon
+        # threads are left.
+        self._job_thread.daemon = True
+        self._job_thread.start()
+
+    def __del__(self):
+        self._job_thread_end.set()
+        if self._job_thread.is_alive():
+            self._job_thread.join()
+
+    def _servicer(self):
+        """
+        Job thread to service memory access requests
+        """
+        while not self._job_thread_end.is_set():
+            if (self.state == DMState.WAIT_RESPONSE) and self.proceed:
+                self.proceed = False
+                if self._notify_query_received is not None:
+                    self._notify_query_received()  # notify incoming request
+            time.sleep(0.001)  # Add a small delay to yield control to other threads
+
+
     def _handle_error(self, priority: int, pgn: int, sa: int, timestamp: int, data: bytearray, error_code: int) -> None:
         """
         Handles errors by resetting the state and unsubscribing from DM14 messages
@@ -82,10 +109,10 @@ def _listen_for_dm14(
                                     self.server.access_level,
                                     0x0,  # placeholder for seed
                                 )  # call proceed function and pass in basic parameters
-                                if self.proceed:
-                                    self._notify_query_received()  # notify incoming request
-                                else:
+                                if not self.proceed:
                                     self._handle_error(priority, pgn, sa, timestamp, data, 0x100)
+                            else:
+                                self.proceed = True  # no security, so always proceed
 
                 case DMState.REQUEST_STARTED:
                     self.server.parse_dm14(priority, pgn, sa, timestamp, data)
@@ -111,17 +138,18 @@ def _listen_for_dm14(
                                         self.server.access_level,
                                         self.server.seed,
                                     )  # call proceed function and pass in basic parameters
-                                    if self.proceed:
-                                        self._notify_query_received()  # notify incoming request
-                                    else:
+                                    if not self.proceed:
                                         self._handle_error(priority, pgn, sa, timestamp, data, 0x100)
+                                else:
+                                    self.proceed = True  # no proceed function, so always proceed
                             else:
                                 self._handle_error(priority, pgn, sa, timestamp, data, 0x1003)
 
                 case DMState.WAIT_QUERY:
                     self.server.set_busy(True)
                     self.server.parse_dm14(priority, pgn, sa, timestamp, data)
                     self.server.set_busy(False)
+
                 case DMState.SERVER_CLEANUP:
                     self.state = DMState.IDLE
                 case _:
@@ -150,6 +178,7 @@ def respond(
         if self.state is not DMState.WAIT_RESPONSE:
             return data
 
+        self.proceed = False
         self._ca.unsubscribe(self._listen_for_dm14)
         return_data = self.server.respond(proceed, data, error, edcp, max_timeout)
         self.state = DMState.SERVER_CLEANUP if self.server.state.value != DMState.IDLE.value else DMState.IDLE
@@ -275,3 +304,4 @@ def reset(self) -> None:
         self._ca.subscribe(self._listen_for_dm14)
         self.server.reset_server()
         self.query.reset_query()
+        self.proceed = False

test/test_ca.py

@@ -68,7 +68,7 @@ def test_addr_claim_fixed_reduced_time(feeder):
         identity_number=1234567,
     )
     new_ca = feeder.ecu.add_ca(name=name, device_address=128)
-    new_ca.start(0.25)
+    new_ca.start(0.2)
 
     # wait until all messages are processed asynchronously 
     # rounded up to account for scheduling delays