Repository: PandABlocks/PandABlocks-client

Issue:
How to get data frames immediately, instead of one frame late?
Hi Team PandABlocks

At MAX IV, we are trying to implement a near "real-time" indicator of the number of samples received from the PandABox via the data capture port.  We are currently underestimating the expected value while capturing.

### Symptoms

Simplified code is something like this:

```python
client = BlockingClient(host)
client.connect()

num_samples_acquired = 0
for data in client.data(scaled=True):
    print(f"** data: {data}")
    if isinstance(data, ReadyData):
        client.send(Arm())
    elif isinstance(data, FrameData):
        num_samples_acquired += len(data.data)
        print(f"** samples acquired: {num_samples_acquired}")
    elif isinstance(data, EndData):
        break
print("done")
```

If we trigger the PandABox PCAP block slowly (on the order of seconds), we see the counter is always 1 behind the number of triggers.  This continues until we disarm, when we get the remaining data frames and the correct count.  We have `flush_every_frame=True`. 

As an example, consider 2 triggers, each a few seconds apart.  (The `>>` lines are actions we performed via the web interface).

Output we get:
```
** data: ReadyData()
** data: StartData(...)
>> Trigger PCAP
>> Trigger PCAP
** data: FrameData(...)
** samples acquired: 1
>> Disarm PCAP
** data: FrameData(...)
** samples acquired: 2
** data: EndData(samples=2, reason=<EndReason.DISARMED: 'Disarmed'>)
done
```

Output we **would like**:
```
** data: ReadyData()
** data: StartData(...)
>> Trigger PCAP
** data: FrameData(...)
** samples acquired: 1
>> Trigger PCAP
** data: FrameData(...)
** samples acquired: 2
>> Disarm PCAP
** data: EndData(samples=2, reason=<EndReason.DISARMED: 'Disarmed'>)
done
```

### Analysis

After some digging, this seems to be due to the feature of handling data overruns, added in the commit:  067c4f431241a3aea00317b96f97c41bfda10a25.  If I understand it correctly, in [`_handle_data_frame()`](https://github.com/PandABlocks/PandABlocks-client/blob/8729a25c195f1592a54e9aff6802c46388d154aa/pandablocks/connections.py#L361-L363) the new data frame is added to `_pending_data`, and the previous data frame is added to `_partial_data`.  The `flush()` methods looks a `_partial_data`, so a flush from `_handle_data_frame()` will always be one frame behind.  Finally, in `_handle_data_end()`, we combine the pending and partial data which will gives us the remaining frames (unless there was a data overrun).

This implementation makes sense for writing data to a file, where the file will only be accessed after the PCAP block was disarmed, and all data is finalised.

In `_handle_data_frame()`, the easy way to get the data frames immediately is to skip the pending data step.  Obviously, that has the downside of invalid data on the last frame if there's a data overrun.
```diff
--- a/pandablocks/connections.py
+++ b/pandablocks/connections.py
@@ -359,8 +359,7 @@ class DataConnection:
         # we already read "BIN ", so read the rest
         data = self._buf.read_bytes(length - 4)[4:]
         # The pending data is now valid, and what we got is pending
-        self._partial_data += self._pending_data
-        self._pending_data = data
+        self._partial_data += data
         # if told to flush now, then yield what we have
         if self._flush_every_frame:
             yield from self.flush()
```

In the case of an overrun, I was wondering how we manage to get all the data frame bytes, `len("BIN " + 4_bytes_encoded_length + data))`, from the data socket, and the "END " data, but the last frame is corrupted.  That led me to PandaBlock-server code.  If I understand it correctly,  [`passthrough_capture_block()`](https://github.com/PandABlocks/PandABlocks-server/blob/750f8618bfd2b7f0a8a46fc572190c93d5fb7a33/server/data_server.c#L578-L581) can send a corrupt data frame, while [`process_capture_block`](https://github.com/PandABlocks/PandABlocks-server/blob/750f8618bfd2b7f0a8a46fc572190c93d5fb7a33/server/data_server.c#L514-L516) avoids sending the corrupt data frame.  If that is so, then we only need the pending data + discard on the client side if we are using `DataConnection.connect(scaled=False)` (the passthrough mode).

### Suggestion

Could we skip pending data + discard when using `DataConnection.connect(scaled=True)`?  That would suit our use case.  Our data rate is comparatively low and we're using the scaled processing.

Does this seems like a workable solution to you?  If so, we can provide a pull request.

### Versions

PandABlocks-client version:  0.5.0
PandABlocks-server version:  ~2.1.0 (+ PCAP ARM timestamp)

P.s.  Your codebase is really easy to read!  Well done, and thanks for making such nice code 👍 

Discussion:

Patch:
diff --git a/docs/explanations/performance.rst b/docs/explanations/performance.rst
index 18f384c..f2952d9 100644
--- a/docs/explanations/performance.rst
+++ b/docs/explanations/performance.rst
@@ -51,7 +51,7 @@ Scale the data on the client
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 `AsyncioClient.data` and `BlockingClient.data` accept a ``scaled`` argument.
-Setting this to True will transfer the raw unscaled data, allowing for up to
+Setting this to False will transfer the raw unscaled data, allowing for up to
 50% more data to be sent depending on the datatype of the field. You can
 use the `StartData.fields` information to scale the data on the client.
 The `write_hdf_files` function uses this approach.
@@ -98,4 +98,17 @@ When panda-webcontrol was not installed, the following results were achieved:
 - PandA CPU usage about 65% (of both cores)
 - local client CPU usage about 60% (of a single core)
 
-Increasing above these throughputs failed most scans with DATA_OVERRUN.
+Increasing above these throughputs failed most scans with `DATA_OVERRUN`.
+
+Data overruns
+-------------
+
+If there is a `DATA_OVERRUN`, the server will stop sending data. The most recently
+received `FrameData` from either `AsyncioClient.data` or `BlockingClient.data` may
+be corrupt. This is the case if the ``scaled`` argument is set to False. The mechanism
+the server uses to send raw unscaled data is only able to detect the corrupt frame after
+it has already been sent. Conversely, the mechanism used to send scaled data aborts prior
+to sending a corrupt frame.
+
+The `write_hdf_files` function uses ``scaled=False``, so your HDF file may include some
+corrupt data in the event of an overrun.
diff --git a/pandablocks/connections.py b/pandablocks/connections.py
index 4eeecda..50e210d 100644
--- a/pandablocks/connections.py
+++ b/pandablocks/connections.py
@@ -274,8 +274,6 @@ class DataConnection:
         self._next_handler: Optional[Callable[[], Optional[Iterator[Data]]]] = None
         # numpy dtype of produced FrameData
         self._frame_dtype = None
-        # frame data that has been received but should be discarded on Data Overrun
-        self._pending_data = bytearray()
         # frame data that has been received but not flushed yet
         self._partial_data = bytearray()
         # whether to flush after every frame
@@ -358,9 +356,7 @@ class DataConnection:
         length = struct.unpack("<I", self._buf.peek_bytes(4))[0]
         # we already read "BIN ", so read the rest
         data = self._buf.read_bytes(length - 4)[4:]
-        # The pending data is now valid, and what we got is pending
-        self._partial_data += self._pending_data
-        self._pending_data = data
+        self._partial_data += data
         # if told to flush now, then yield what we have
         if self._flush_every_frame:
             yield from self.flush()
@@ -370,9 +366,6 @@ class DataConnection:
         # Handle the end reason
         samples, reason = self._buf.read_line().split(maxsplit=1)
         reason_enum = EndReason(reason.decode())
-        if reason_enum != EndReason.DATA_OVERRUN:
-            # The last bit of pending data is now valid
-            self._partial_data += self._pending_data
         # Flush whatever is not already flushed
         yield from self.flush()
         yield EndData(samples=int(samples), reason=reason_enum)
diff --git a/pandablocks/hdf.py b/pandablocks/hdf.py
index 56ca537..d220909 100644
--- a/pandablocks/hdf.py
+++ b/pandablocks/hdf.py
@@ -10,13 +10,14 @@ from pandablocks.commands import Arm
 
 from .asyncio import AsyncioClient
 from .connections import SAMPLES_FIELD
-from .responses import EndData, FieldCapture, FrameData, ReadyData, StartData
+from .responses import EndData, EndReason, FieldCapture, FrameData, ReadyData, StartData
 
 # Define the public API of this module
 __all__ = [
     "Pipeline",
     "HDFWriter",
     "FrameProcessor",
+    "HDFDataOverrunException",
     "create_pipeline",
     "create_default_pipeline",
     "stop_pipeline",
@@ -24,6 +25,10 @@ __all__ = [
 ]
 
 
+class HDFDataOverrunException(Exception):
+    """Raised if `DATA_OVERRUN` occurs while receiving data for HDF file"""
+
+
 class Pipeline(threading.Thread):
     """Helper class that runs a pipeline consumer process in its own thread"""
 
@@ -211,13 +216,17 @@ async def write_hdf_files(
         num: The number of acquisitions to store in separate files. 0 = Infinite capture
         arm: Whether to arm PCAP at the start, and after each successful acquisition
 
+    Raises:
+        HDFDataOverrunException: if there is a data overrun.
     """
     counter = 0
+    end_data = None
     pipeline = create_default_pipeline(file_names)
     try:
         async for data in client.data(scaled=False, flush_period=flush_period):
             pipeline[0].queue.put_nowait(data)
             if type(data) == EndData:
+                end_data = data
                 counter += 1
                 if counter == num:
                     # We produced the right number of frames
@@ -225,5 +234,9 @@ async def write_hdf_files(
             if type(data) in (ReadyData, EndData) and arm:
                 # Told to arm at the beginning, and after each acquisition ends
                 await client.send(Arm())
+        if end_data and end_data.reason == EndReason.DATA_OVERRUN:
+            raise HDFDataOverrunException(
+                "Data overrun - streaming aborted! Last frame may be corrupt."
+            )
     finally:
         stop_pipeline(pipeline)
diff --git a/pandablocks/responses.py b/pandablocks/responses.py
index 021e1db..2df55be 100644
--- a/pandablocks/responses.py
+++ b/pandablocks/responses.py
@@ -200,7 +200,9 @@ class EndReason(Enum):
     DISARMED = "Disarmed"
     #: Client disconnect detected
     EARLY_DISCONNECT = "Early disconnect"
-    #: Client not taking data quickly or network congestion, internal buffer overflow
+    #: Client not taking data quickly or network congestion, internal buffer overflow.
+    #: In raw unscaled mode (i.e., no server-side scaling), the most recent
+    #: `FrameData` is likely corrupted.
     DATA_OVERRUN = "Data overrun"
     #: Triggers too fast for configured data capture
     FRAMING_ERROR = "Framing error"


Test patch:
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 0e3174d..38f3337 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -8,6 +8,7 @@ import pytest
 from click.testing import CliRunner
 
 from pandablocks import cli
+from pandablocks.hdf import HDFDataOverrunException
 from tests.conftest import STATE_RESPONSES, STATE_SAVEFILE, DummyServer
 
 
@@ -31,18 +32,7 @@ def test_writing_fast_hdf(dummy_server_in_thread: DummyServer, raw_dump, tmp_pat
         "PCAP.TS_START.Value",
     ]
     assert dummy_server_in_thread.received == ["*PCAP.ARM="]
-
-    def multiples(num, offset=0):
-        return pytest.approx(np.arange(1, 10001) * num + offset)
-
-    assert hdf_file["/COUNTER1.OUT.Max"][:] == multiples(1)
-    assert hdf_file["/COUNTER1.OUT.Mean"][:] == multiples(1)
-    assert hdf_file["/COUNTER1.OUT.Min"][:] == multiples(1)
-    assert hdf_file["/COUNTER2.OUT.Mean"][:] == multiples(2)
-    assert hdf_file["/COUNTER3.OUT.Value"][:] == multiples(3)
-    assert hdf_file["/PCAP.BITS2.Value"][:] == multiples(0)
-    assert hdf_file["/PCAP.SAMPLES.Value"][:] == multiples(0, offset=125)
-    assert hdf_file["/PCAP.TS_START.Value"][:] == multiples(2e-6, offset=7.2e-8 - 2e-6)
+    assert_all_data_in_hdf_file(hdf_file)
 
 
 def test_writing_overrun_hdf(
@@ -54,13 +44,16 @@ def test_writing_overrun_hdf(
     result = runner.invoke(
         cli.cli, ["hdf", "localhost", str(tmp_path / "%d.h5"), "--arm"]
     )
-    assert result.exit_code == 0
+    assert result.exit_code == 1
+    assert isinstance(result.exception, HDFDataOverrunException)
     hdf_file = h5py.File(tmp_path / "1.h5", "r")
+    assert_all_data_in_hdf_file(hdf_file)
+
 
+def assert_all_data_in_hdf_file(hdf_file):
     def multiples(num, offset=0):
-        return pytest.approx(np.arange(1, 8936) * num + offset)
+        return pytest.approx(np.arange(1, 10001) * num + offset)
 
-    # Check we didn't write the last chunk
     assert hdf_file["/COUNTER1.OUT.Max"][:] == multiples(1)
     assert hdf_file["/COUNTER1.OUT.Mean"][:] == multiples(1)
     assert hdf_file["/COUNTER1.OUT.Min"][:] == multiples(1)