Cleanups for FlexRay tests

src/gallia/db/handler.py

@@ -385,15 +385,19 @@ async def execute() -> None:
                     logger.warning(
                         f"Could not log message for {query_parameter[5]} to database. Retrying ..."
                     )
+                except asyncio.CancelledError:
+                    logger.warning("Database query was cancelled.")
+                    done = True
 
             if commit:
                 await self.connection.commit()
 
-        self.tasks.append(asyncio.create_task(execute()))
-        self.tasks[-1].add_done_callback(
+        task = asyncio.create_task(execute())
+        task.add_done_callback(
             handle_task_error,
             context=set_task_handler_ctx_variable(__name__, "DbHandler"),
         )
+        self.tasks.append(task)
 
     async def insert_session_transition(self, destination: int, steps: list[int]) -> None:
         assert self.connection is not None, "Not connected to the database"

src/gallia/transports/flexray_vector.py

@@ -421,6 +421,19 @@ async def read_tp_frame(self) -> FlexRayTPFrame:
         logger.trace("read FlexRayTPFrame %s", repr(frame))
         return frame
 
+    def _require_fc_frame(self, block_size: int, read_bytes: int) -> bool:
+        # 6 bytes already read in first frame.
+        return ((read_bytes - 6) & block_size) == 0
+
+    async def _send_flow_control_frame(self) -> None:
+        block_size = self.config.fc_block_size
+        fc_frame = FlexRayTPFlowControlFrame(
+            flag=FlexRayTPFlowControlFlag.CONTINUE_TO_SEND,
+            separation_time=self.config.fc_separation_time,
+            block_size=block_size,
+        )
+        await self.write_tp_frame(fc_frame)
+
     async def _handle_fragmented(self, expected_len: int) -> bytes:
         # 6 bytes already read in first frame.
         # Headersize is 2 byte.
@@ -431,7 +444,14 @@ async def _handle_fragmented(self, expected_len: int) -> bytes:
         while read_bytes < expected_len:
             # Reordering is not implemented.
             logger.debug(f"expected_len: {expected_len}; read_bytes: {read_bytes}")
-            frame = await self.read_tp_frame()
+
+            if self._require_fc_frame(self.config.fc_block_size, read_bytes):
+                await self._send_flow_control_frame()
+
+            # TODO: Make this configurable. Maybe align with separation_time.
+            async with asyncio.timeout(10):
+                frame = await self.read_tp_frame()
+
             if not isinstance(frame, FlexRayTPConsecutiveFrame):
                 raise RuntimeError(f"expected consecutive frame, got: {frame}")
             if frame.counter != (counter & 0x0F):
@@ -455,15 +475,6 @@ async def read_unsafe(
                 case FlexRayTPSingleFrame():
                     return frame.data
                 case FlexRayTPFirstFrame():
-                    fc_frame = FlexRayTPFlowControlFrame(
-                        flag=FlexRayTPFlowControlFlag.CONTINUE_TO_SEND,
-                        separation_time=self.config.fc_separation_time,
-                        # TODO: send again after block_size number of frames is read.
-                        # Maybe move sending the flow control frame into the
-                        # _handle_fragmented() function and create a loop.
-                        block_size=self.config.fc_block_size,
-                    )
-                    await self.write_tp_frame(fc_frame)
                     data = frame.data + await self._handle_fragmented(frame.size)
                     data = data[: frame.size]
                     logger.debug("read data: %s", data.hex())