Repository: dls-controls/pymalcolm

Issue:
HDFWriterPart does not track frames written to disk
Currently scans with AreaDetector HDF5 plugins that fail (e.g. due to permission errors or invalid configuration parameters) will succeed as long as all of the other parts run as expected. I believe instead the scan should fail with an appropriate error.

In my case the value of frames cached per chunk was too large for the frame chunking dimensions, so the writer stopped before the first frame was written.

I think there should be two checks:

- Monitoring for any early errors. The capture_RBV PV will go low if an error occurs, and with defer being True will happen during the first frame, and I think just checking for the first frame to be written should capture most runtime errors.
- Checking at the end that the correct number of frames (NumCaptured_RBV) matches what we expect. 

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) Report
> Merging [#361](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) (d0cec74) into [master](https://codecov.io/gh/dls-controls/pymalcolm/commit/166dc80326f9c691046037196e08c7d77f175861?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) (166dc80) will **decrease** coverage by `0.01%`.
> The diff coverage is `84.78%`.

[![Impacted file tree graph](https://codecov.io/gh/dls-controls/pymalcolm/pull/361/graphs/tree.svg?width=650&height=150&src=pr&token=qykILXuaMk&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls)](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls)

```diff
@@            Coverage Diff             @@
##           master     #361      +/-   ##
==========================================
- Coverage   93.21%   93.19%   -0.02%     
==========================================
  Files         217      217              
  Lines       11763    11777      +14     
==========================================
+ Hits        10965    10976      +11     
- Misses        798      801       +3     
```


| [Impacted Files](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) | Coverage Δ | |
|---|---|---|
| [malcolm/modules/ADCore/parts/hdfwriterpart.py](https://codecov.io/gh/dls-controls/pymalcolm/pull/361/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls#diff-bWFsY29sbS9tb2R1bGVzL0FEQ29yZS9wYXJ0cy9oZGZ3cml0ZXJwYXJ0LnB5) | `95.04% <80.64%> (-1.17%)` | :arrow_down: |
| [malcolm/core/context.py](https://codecov.io/gh/dls-controls/pymalcolm/pull/361/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls#diff-bWFsY29sbS9jb3JlL2NvbnRleHQucHk=) | `96.01% <88.88%> (+0.04%)` | :arrow_up: |
| [malcolm/core/views.py](https://codecov.io/gh/dls-controls/pymalcolm/pull/361/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls#diff-bWFsY29sbS9jb3JlL3ZpZXdzLnB5) | `96.24% <100.00%> (ø)` | |
| [malcolm/testutil.py](https://codecov.io/gh/dls-controls/pymalcolm/pull/361/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls#diff-bWFsY29sbS90ZXN0dXRpbC5weQ==) | `90.54% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls). Last update [166dc80...d0cec74](https://codecov.io/gh/dls-controls/pymalcolm/pull/361?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls).

BenBradnick: > This looks reasonable. So we no longer compare uniqueID number, but look at the arrayCounter to see if we're finished?

Kind of - the arrayCounter is used just to wait for the first array to arrive in the plugin.

Then instead of using uniqueID it looks at the numCaptured_RBV which is the actual number of frames written to the HDF5 file. This is because if the file writing falls over for any reason, it will now be captured (as the file will be missing frames) rather than still succeeding due to having receiving the last frame with the correct uniqueID.
thomascobb: In which case this looks fine. I marginally prefer my `greater_than` implementation to the `**kwargs`, but there's not much in it

Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index c8995405..5fe906b5 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -14,6 +14,10 @@ Changed:
   generation dominates. A Yield has also been added so the thread suspends
   after each batch so it doesn't block other threads for the entire profile
   calculation, which previously caused timeout problems with other components.
+- HdfWriterPart now looks at the number of frames written to the HDF5 file to
+  track progress and check for stalling issues. This means a scan will now fail
+  instead of succeed if the file plugin receives all frames but doesn't write
+  them.
 
 Fixed:
 
diff --git a/malcolm/core/views.py b/malcolm/core/views.py
index d8c5e135..a6ae351f 100644
--- a/malcolm/core/views.py
+++ b/malcolm/core/views.py
@@ -186,7 +186,12 @@ class Block(View):
         )
 
     def when_value_matches(
-        self, attr, good_value, bad_values=None, timeout=None, event_timeout=None
+        self,
+        attr,
+        good_value,
+        bad_values=None,
+        timeout=None,
+        event_timeout=None,
     ):
         future = self.when_value_matches_async(attr, good_value, bad_values)
         self._context.wait_all_futures(
diff --git a/malcolm/modules/ADCore/blocks/hdf_writer_block.yaml b/malcolm/modules/ADCore/blocks/hdf_writer_block.yaml
index b046cd75..58011572 100644
--- a/malcolm/modules/ADCore/blocks/hdf_writer_block.yaml
+++ b/malcolm/modules/ADCore/blocks/hdf_writer_block.yaml
@@ -63,6 +63,11 @@
     pv: $(prefix):NumCapture
     rbv_suffix: _RBV
 
+- ca.parts.CALongPart:
+    name: numCapturedReadback
+    description: Number of frames captured
+    rbv: $(prefix):NumCaptured_RBV
+
 - ca.parts.CALongPart:
     name: flushDataPerNFrames
     description: Number of frames to capture between HDF dataset flushes
diff --git a/malcolm/modules/ADCore/parts/hdfwriterpart.py b/malcolm/modules/ADCore/parts/hdfwriterpart.py
index d78fb857..7067cce5 100644
--- a/malcolm/modules/ADCore/parts/hdfwriterpart.py
+++ b/malcolm/modules/ADCore/parts/hdfwriterpart.py
@@ -371,12 +371,12 @@ class HDFWriterPart(builtin.parts.ChildPart):
         super().__init__(name, mri)
         # Future for the start action
         self.start_future: Optional[Future] = None
-        self.array_future: Optional[Future] = None
-        self.done_when_reaches = 0
-        # This is when uniqueId last updated
-        self.last_id_update: Optional[float] = None
-        # CompletedSteps = arrayCounter + self.uniqueid_offset
-        self.uniqueid_offset = 0
+        self.first_array_future: Optional[Future] = None
+        self.done_when_captured = 0
+        # This is when the readback for number of frames captured last updated
+        self.last_capture_update: Optional[float] = None
+        # Store number of frames captured for progress reporting
+        self.num_captured_offset = 0
         # The HDF5 layout file we write to say where the datasets go
         self.layout_filename: Optional[str] = None
         self.runs_on_windows = runs_on_windows
@@ -436,9 +436,9 @@ class HDFWriterPart(builtin.parts.ChildPart):
         formatName: scanning.hooks.AFormatName = "det",
         fileTemplate: scanning.hooks.AFileTemplate = "%s.h5",
     ) -> scanning.hooks.UInfos:
-        # On initial configure, expect to get the demanded number of frames
-        self.done_when_reaches = completed_steps + steps_to_do
-        self.uniqueid_offset = 0
+        # What the capture readback value should say at the end of the run
+        self.done_when_captured = completed_steps + steps_to_do
+        self.num_captured_offset = 0
         # Calculate how long to wait before marking this scan as stalled
         self.frame_timeout = FRAME_TIMEOUT
         if generator.duration > 0:
@@ -494,12 +494,13 @@ class HDFWriterPart(builtin.parts.ChildPart):
         context.wait_all_futures(futures)
         # Reset numCapture back to 0
         child.numCapture.put_value(0)
-        # Start the plugin
+        # Create a future for the file capture
         self.start_future = child.start_async()
-        # Start a future waiting for the first array
-        self.array_future = child.when_value_matches_async(
+        # Create a future waiting for the first array
+        self.first_array_future = child.when_value_matches_async(
             "arrayCounterReadback", greater_than_zero
         )
+        # Check XML
         self._check_xml_is_valid(child)
         # Return the dataset information
         dataset_infos = list(
@@ -516,38 +517,54 @@ class HDFWriterPart(builtin.parts.ChildPart):
     @add_call_types
     def on_seek(
         self,
+        context: scanning.hooks.AContext,
         completed_steps: scanning.hooks.ACompletedSteps,
         steps_to_do: scanning.hooks.AStepsToDo,
     ) -> None:
-        # This is rewinding or setting up for another batch, so the detector
-        # will skip to a uniqueID that has not been produced yet
-        self.uniqueid_offset = completed_steps - self.done_when_reaches
-        self.done_when_reaches += steps_to_do
+        # Reset array counter to zero so we can check for the first frame
+        # to arrive in the plugin
+        child = context.block_view(self.mri)
+        child.arrayCounter.put_value(0)
+        # Calculate the expected number of captured frames. This is in case
+        # pausing has resulted in extra frames that will be overwritten
+        # using the position plugin when resuming.
+        num_captured = child.numCapturedReadback.value
+        # Offset for progress updates
+        self.num_captured_offset = num_captured - completed_steps
+        # Expected captured readback after this batch
+        self.done_when_captured = num_captured + steps_to_do
 
     @add_call_types
     def on_run(self, context: scanning.hooks.AContext) -> None:
-        context.wait_all_futures(self.array_future)
-        context.unsubscribe_all()
-        self.last_id_update = None
+        # Wait for the first array to arrive in the plugin without a timeout
+        context.wait_all_futures(self.first_array_future)
+
+        # Get the child block
         child = context.block_view(self.mri)
-        child.uniqueId.subscribe_value(self.update_completed_steps)
-        f_done = child.when_value_matches_async("uniqueId", self.done_when_reaches)
+
+        # Update progress based on number of frames written
+        child.numCapturedReadback.subscribe_value(self.update_completed_steps)
+
+        # Wait for the number of captured frames to reach the target
+        f_done = child.when_value_matches_async(
+            "numCapturedReadback", self.done_when_captured
+        )
+        self.last_capture_update = None
         while True:
             try:
+                # Use a regular timeout so we can request a manual flush every second
                 context.wait_all_futures(f_done, timeout=1)
             except TimeoutError:
                 # This is ok, means we aren't done yet, so flush
                 self._flush_if_still_writing(child)
-                # Check it hasn't been too long
-                if self.last_id_update:
-                    if time.time() > self.last_id_update + self.frame_timeout:
-                        raise TimeoutError(
-                            "HDF writer stalled, last updated at %s"
-                            % (self.last_id_update)
-                        )
-                # TODO: what happens if we miss the last frame?
+                # Check it hasn't been too long since the last frame was written
+                if self._has_file_writing_stalled():
+                    timeout_message = self._get_file_writing_stalled_error_message(
+                        child
+                    )
+                    raise TimeoutError(timeout_message)
             else:
-                return
+                break
 
     def _flush_if_still_writing(self, child):
         # Check that the start_future hasn't errored
@@ -558,6 +575,22 @@ class HDFWriterPart(builtin.parts.ChildPart):
             # Flush the hdf frames to disk
             child.flushNow()
 
+    def _has_file_writing_stalled(self) -> bool:
+        if self.last_capture_update:
+            return time.time() > self.last_capture_update + self.frame_timeout
+        return False
+
+    def _get_file_writing_stalled_error_message(self, child) -> str:
+        unique_id = child.uniqueId.value
+        num_captured = child.numCapturedReadback.value - self.num_captured_offset
+        frames_expected = self.done_when_captured - self.num_captured_offset
+        return (
+            f"{self.name}: writing stalled at {num_captured}/{frames_expected} "
+            f"captured (readback offset is {self.num_captured_offset}). "
+            f"Last update was {self.last_capture_update}, "
+            f"last uniqueId received in plugin was {unique_id}"
+        )
+
     @add_call_types
     def on_post_run_ready(self, context: scanning.hooks.AContext) -> None:
         # Do one last flush and then we're done
@@ -570,10 +603,10 @@ class HDFWriterPart(builtin.parts.ChildPart):
         child.stop()
 
     def update_completed_steps(self, value: int) -> None:
-        completed_steps = value + self.uniqueid_offset
-        self.last_id_update = time.time()
-        assert self.registrar, "No registrar assigned"
+        completed_steps = value - self.num_captured_offset
+        self.last_capture_update = time.time()
         # Stop negative values being reported for first call when subscribing
         # when we have a non-zero offset.
         if completed_steps >= 0:
+            assert self.registrar, "No registrar assigned"
             self.registrar.report(scanning.infos.RunProgressInfo(completed_steps))


Test patch:
diff --git a/malcolm/testutil.py b/malcolm/testutil.py
index 6e19bf60..4f0b99db 100644
--- a/malcolm/testutil.py
+++ b/malcolm/testutil.py
@@ -66,16 +66,20 @@ class ChildTestCase(unittest.TestCase):
         return child
 
     def mock_when_value_matches(self, child):
-        def handle_when_value_matches_async(attr, good_value, bad_values=None):
+        def handle_when_value_matches_async(
+            attr, good_value, bad_values=None, **kwargs
+        ):
             # tell the mock we were called
-            child.handled_requests.when_value_matches(attr, good_value, bad_values)
+            child.handled_requests.when_value_matches(
+                attr, good_value, bad_values, **kwargs
+            )
             # poke the value we are looking for into the attribute so
             # that old_when_matches will immediately succeed
             # If it's callable then rely on the test code to do this
             if not callable(good_value):
                 self.set_attributes(child, **{attr: good_value})
             # now run the original code
-            return self.old_when_matches_async(attr, good_value, bad_values)
+            return self.old_when_matches_async(attr, good_value, bad_values, **kwargs)
 
         def block_view(context=None, old=child.block_view):
             self._context = context
diff --git a/tests/test_modules/test_ADCore/test_hdfwriterpart.py b/tests/test_modules/test_ADCore/test_hdfwriterpart.py
index 2ea2ca3c..be00c97a 100644
--- a/tests/test_modules/test_ADCore/test_hdfwriterpart.py
+++ b/tests/test_modules/test_ADCore/test_hdfwriterpart.py
@@ -3,10 +3,10 @@ import shutil
 from xml.etree import ElementTree
 
 import cothread
-from mock import MagicMock, call
+from mock import MagicMock, call, patch
 from scanpointgenerator import CompoundGenerator, LineGenerator, SpiralGenerator
 
-from malcolm.core import Context, Future, Process
+from malcolm.core import Context, Future, Process, TimeoutError
 from malcolm.modules.ADCore.blocks import hdf_writer_block
 from malcolm.modules.ADCore.infos import (
     CalculatedNDAttributeDatasetInfo,
@@ -357,18 +357,19 @@ class TestHDFWriterPart(ChildTestCase):
     def test_run(self):
         self.o = HDFWriterPart(name="m", mri="BLOCK:HDF5")
         self.context.set_notify_dispatch_request(self.o.notify_dispatch_request)
-        self.o.done_when_reaches = 38
-        self.o.completed_offset = 0
-        # Say that we're getting the first frame
-        self.o.array_future = Future(None)
-        self.o.array_future.set_result(None)
+        self.o.done_when_captured = 38
+        # Need a registrar object or we get AssertionError
         self.o.registrar = MagicMock()
-        # run waits for this value, so say we have finished immediately
-        self.set_attributes(self.child, uniqueId=self.o.done_when_reaches)
+        # Run waits for this value, so say we have finished immediately
+        self.set_attributes(self.child, numCapturedReadback=self.o.done_when_captured)
         self.mock_when_value_matches(self.child)
+
+        # Run
         self.o.on_run(self.context)
+
+        # Check calls
         assert self.child.handled_requests.mock_calls == [
-            call.when_value_matches("uniqueId", 38, None)
+            call.when_value_matches("numCapturedReadback", 38, None)
         ]
         assert self.o.registrar.report.called_once
         assert self.o.registrar.report.call_args_list[0][0][0].steps == 38
@@ -376,40 +377,65 @@ class TestHDFWriterPart(ChildTestCase):
     def test_run_and_flush(self):
         self.o = HDFWriterPart(name="m", mri="BLOCK:HDF5")
 
-        def set_unique_id():
+        def set_num_captured():
             # Sleep for 2.5 seconds to ensure 2 flushes, and then set value to finish
             cothread.Sleep(2.5)
-            self.set_attributes(self.child, uniqueId=self.o.done_when_reaches)
+            self.set_attributes(
+                self.child, numCapturedReadback=self.o.done_when_captured
+            )
 
-        self.o.done_when_reaches = 38
-        self.o.completed_offset = 0
+        self.o.done_when_captured = 5
         # Say that we're getting the first frame
-        self.o.array_future = Future(None)
-        self.o.array_future.set_result(None)
+        self.o.first_array_future = Future(None)
+        self.o.first_array_future.set_result(None)
         self.o.start_future = Future(None)
+        # Need a registrar object or we get AssertionError
         self.o.registrar = MagicMock()
-        self.o.frame_timeout = 60
+        # Reduce frame timeout so we don't hang on this test for too long
+        self.o.frame_timeout = 5
         # Spawn process to finish it after a few seconds
-        self.process.spawn(set_unique_id)
+        self.process.spawn(set_num_captured)
+
         # Run
         self.o.on_run(self.context)
+
+        # Check calls
         assert self.child.handled_requests.mock_calls == [
             call.post("flushNow"),
             call.post("flushNow"),
         ]
         assert self.o.registrar.report.called_once
         assert self.o.registrar.report.call_args_list[0][0][0].steps == 0
-        assert self.o.registrar.report.call_args_list[1][0][0].steps == 38
+        assert self.o.registrar.report.call_args_list[1][0][0].steps == 5
+
+    def test_run_raises_TimeoutError_for_stalled_writer(self):
+        self.o = HDFWriterPart(name="m", mri="BLOCK:HDF5")
+        self.context.set_notify_dispatch_request(self.o.notify_dispatch_request)
+        self.o.done_when_captured = 10
+        # Need a registrar object or we get AssertionError
+        self.o.registrar = MagicMock()
+        self.o.start_future = MagicMock()
+        # Mock the last update
+        self.o.last_capture_update = MagicMock()
+        self.o.last_capture_update.return_value = 0.0
+        # Set a short timeout for testing
+        self.o.frame_timeout = 0.1
+
+        # Now check the error is raised
+        self.assertRaises(TimeoutError, self.o.on_run, self.context)
 
     def test_seek(self):
-        self.mock_when_value_matches(self.child)
         self.o = HDFWriterPart(name="m", mri="BLOCK:HDF5")
         self.context.set_notify_dispatch_request(self.o.notify_dispatch_request)
-        self.o.done_when_reaches = 10
+        # Num captured readback usually reads a bit higher than the completed steps
+        # after a pause is requested
         completed_steps = 4
+        self.set_attributes(self.child, numCapturedReadback=6)
+        # Call the seek
         steps_to_do = 3
-        self.o.on_seek(completed_steps, steps_to_do)
-        assert self.o.done_when_reaches == 13
+        self.o.on_seek(self.context, completed_steps, steps_to_do)
+        # We expect done when captured to be the current captured readback + steps to do
+        assert self.o.done_when_captured == 9
 
     def test_post_run_ready(self):
         self.o = HDFWriterPart(name="m", mri="BLOCK:HDF5")
@@ -460,3 +486,21 @@ class TestHDFWriterPart(ChildTestCase):
         child.xmlErrorMsg.value = "XML description file cannot be opened"
 
         self.assertRaises(AssertionError, self.o._check_xml_is_valid, child)
+
+    @patch("malcolm.modules.ADCore.parts.hdfwriterpart.time.time")
+    def test_has_file_writing_stalled(self, time_mock):
+        self.o = HDFWriterPart(name="m", mri="BLOCK:HDF5")
+
+        # First case - no last capture update so return False
+        assert self.o._has_file_writing_stalled() is False
+
+        # Set up the attributes and mock for the last two cases
+        self.o.last_capture_update = 10.0
+        self.o.frame_timeout = 60.0
+        time_mock.side_effect = [30.0, 71.0]
+
+        # Second case - last capture update is within frame timeout
+        assert self.o._has_file_writing_stalled() is False
+
+        # Final case - last capture update is outside frame timeout
+        assert self.o._has_file_writing_stalled() is True