Repository: dls-controls/pymalcolm

Issue:
Large number of points results in CA timeouts on child parts during configure
As the number of points in a scan increases, so does the configure time taken by a few of the parts. At some point a limit is reached where the pmacChildPart takes so long to configure that it blocks other threads for longer than the timeouts used on Channel Access requests using cothread.catools. Then when those threads start running again they throw timeout errors.

The limit on points varies on the type of generator (e.g. Static or Line with moving axes) and the performance of the host that Malcolm is running on. On P45-control you will get errors at ~2 million static points, but p45-ws001 doesn't error up to 4 million points.

On p45-ws001 the following generator:
```
{
  typeid: 'scanpointgenerator:generator/CompoundGenerator:1.0',
  generators: [
    {
      typeid: 'scanpointgenerator:generator/StaticPointGenerator:1.0',
      size: 8000000
    }
  ],
  excluders: [],
  mutators: [],
  duration: 0.2,
  continuous: true
}
```
Results in the following timeouts (usually multiple PVs across multiple parts):
```
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/modules/ADCore/parts/statspluginpart.py", line 98, in on_configure
    context.wait_all_futures(fs)
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/core/context.py", line 318, in wait_all_futures
    self._service_futures(filtered_futures, until)
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/core/context.py", line 426, in _service_futures
    raise response.message
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/core/controller.py", line 145, in _handle_request
    responses += handler(request)
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/core/controller.py", line 204, in _handle_put
    result = put_function(value)
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/modules/ca/parts/cabooleanpart.py", line 41, in caput
    self.caa.caput(int(value))
  File "/dls_sw/work/R3.14.12.7/ioc/BL45P/pymalcolm/malcolm/modules/ca/util.py", line 271, in caput
    throw=self.throw,
  File "/scratch/cvl62853/pipenv/pymalcolm-PZqGxTx3/lib/python3.7/site-packages/cothread/catools.py", line 761, in caget
    return caget_one(pvs, **kargs)
  File "/scratch/cvl62853/pipenv/pymalcolm-PZqGxTx3/lib/python3.7/site-packages/cothread/catools.py", line 130, in throw_wrapper
    return function(pv, *args, **kargs)
  File "/scratch/cvl62853/pipenv/pymalcolm-PZqGxTx3/lib/python3.7/site-packages/cothread/catools.py", line 648, in caget_one
    return ca_timeout(done, timeout, pv)
  File "/scratch/cvl62853/pipenv/pymalcolm-PZqGxTx3/lib/python3.7/site-packages/cothread/catools.py", line 157, in ca_timeout
    py23.raise_from(ca_nothing(name, cadef.ECA_TIMEOUT), timeout)
  File "<string>", line 4, in raise_from
cothread.catools.ca_nothing: BL45P-EA-DIFF-01:STAT:ComputeStatistics_RBV: User specified timeout on IO operation expired
```

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/dls-controls/pymalcolm/pull/360?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) Report
> Merging [#360](https://codecov.io/gh/dls-controls/pymalcolm/pull/360?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) (afcfe7a) into [master](https://codecov.io/gh/dls-controls/pymalcolm/commit/8b81adf1984b828db0955be7ca4b9fd63e9b50be?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) (8b81adf) will **decrease** coverage by `0.00%`.
> The diff coverage is `91.42%`.

[![Impacted file tree graph](https://codecov.io/gh/dls-controls/pymalcolm/pull/360/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/360?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     #360      +/-   ##
==========================================
- Coverage   93.13%   93.12%   -0.01%     
==========================================
  Files         217      217              
  Lines       11707    11756      +49     
==========================================
+ Hits        10903    10948      +45     
- Misses        804      808       +4     
```


| [Impacted Files](https://codecov.io/gh/dls-controls/pymalcolm/pull/360?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls) | Coverage Δ | |
|---|---|---|
| [malcolm/modules/pmac/parts/pmacchildpart.py](https://codecov.io/gh/dls-controls/pymalcolm/pull/360/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls#diff-bWFsY29sbS9tb2R1bGVzL3BtYWMvcGFydHMvcG1hY2NoaWxkcGFydC5weQ==) | `94.17% <91.42%> (-0.29%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/dls-controls/pymalcolm/pull/360?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/360?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dls-controls). Last update [8b81adf...afcfe7a](https://codecov.io/gh/dls-controls/pymalcolm/pull/360?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).


Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 856c2098..c8995405 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -6,6 +6,15 @@ This project adheres to `Semantic Versioning <http://semver.org/>`_ after 2-1.
 Unreleased
 ----------
 
+Changed:
+
+- PmacChildPart's profile generation is now split into two methods based on the
+  trigger type. The loop performance has been improved, reducing configure
+  times by ~20% for small scans and ~40% for large scans when profile point
+  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.
+
 Fixed:
 
 - PmacChildPart now only reports progress if the brick is providing a trigger
diff --git a/malcolm/modules/pmac/parts/pmacchildpart.py b/malcolm/modules/pmac/parts/pmacchildpart.py
index 3e8bd748..b2358f4c 100644
--- a/malcolm/modules/pmac/parts/pmacchildpart.py
+++ b/malcolm/modules/pmac/parts/pmacchildpart.py
@@ -2,6 +2,7 @@ import re
 from enum import Enum
 from typing import Dict, List, Optional, Union
 
+import cothread
 import numpy as np
 from annotypes import add_call_types
 from scanpointgenerator import CompoundGenerator, Point
@@ -432,9 +433,9 @@ class PmacChildPart(builtin.parts.ChildPart):
         # "how far through the pmac trajectory" rather than a generator
         # scan step
         if scanned > 0:
-            completed_steps = self.completed_steps_lookup[scanned - 1]
             # Only report progress if we are triggering for every point
             if self.output_triggers == scanning.infos.MotionTrigger.EVERY_POINT:
+                completed_steps = self.completed_steps_lookup[scanned - 1]
                 self.registrar.report(scanning.infos.RunProgressInfo(completed_steps))
             # Keep PROFILE_POINTS trajectory points in front
             if (
@@ -690,6 +691,7 @@ class PmacChildPart(builtin.parts.ChildPart):
         or not used at all. Indexing into points[] is expensive so this is
         intentional.
         """
+        profile_point_added = False
         if self.time_since_last_pvt > 0 and not points_are_joined:
             # assume we can skip if we are at the end of a row and we
             # just skipped the most recent point (i.e. time_since_last_pvt > 0)
@@ -712,6 +714,7 @@ class PmacChildPart(builtin.parts.ChildPart):
                 {name: point.positions[name] for name in self.axis_mapping},
             )
             self.time_since_last_pvt = point.duration / 2.0
+            profile_point_added = True
 
         # only add the lower bound if we did not skip this point OR if we are
         # at the end of a row where we always require a final point
@@ -741,6 +744,8 @@ class PmacChildPart(builtin.parts.ChildPart):
                 {name: point.upper[name] for name in self.axis_mapping},
             )
             self.time_since_last_pvt = 0
+            profile_point_added = True
+        return profile_point_added
 
     def get_some_points(self, start_index):
         # calculate the indices of the next batch of points to get for
@@ -759,6 +764,145 @@ class PmacChildPart(builtin.parts.ChildPart):
 
         return points, joined, velocities
 
+    def check_profile_length_exceeds_profile_points(self) -> bool:
+        # Check if we have exceeded the points number and need to write
+        # Strictly less than so we always add one more point to the time
+        # array so we can always stretch points in a subsequent add with
+        # the values already in the profiles
+        return len(self.profile["timeArray"]) > PROFILE_POINTS
+
+    def is_last_point_in_current_batch(
+        self, start_batch_index: int, num_points: int
+    ) -> bool:
+        return start_batch_index + num_points == self.steps_up_to
+
+    def create_generator_profile_every_point(self, start_index: int) -> bool:
+        # Ensure we have the correct trigger type
+        assert (
+            self.output_triggers == scanning.infos.MotionTrigger.EVERY_POINT
+        ), f"{self.name}: trigger should be every point, got {self.output_triggers}"
+
+        # Keep going until we finish or exceed the maximum length of a profile
+        start_batch_index = start_index
+        point_index = start_index
+        while True:
+            # Grab some points
+            points, joined, _ = self.get_some_points(start_batch_index)
+            if not points:
+                return True
+
+            # Number of points and check if the last point is in this batch
+            num_points = len(points)
+            last_point_in_batch = self.is_last_point_in_current_batch(
+                start_batch_index, num_points
+            )
+
+            # Don't do all points in the batch otherwise we can get an index error
+            # when adding the turnaround.
+            points_to_do = num_points - 1
+            for i in range(points_to_do):
+                self.add_generator_point_pair(points[i], point_index, joined[i])
+
+                # add in the turnaround between non-contiguous points
+                if not (joined[i]):
+                    self.insert_gap(points[i], points[i + 1], point_index + 1)
+
+                # Check if we have exceeded the profile points limit
+                if self.check_profile_length_exceeds_profile_points():
+                    self.end_index = point_index + 1
+                    return False
+
+                # Increment point index
+                point_index += 1
+
+            # Check for the last point
+            if last_point_in_batch:
+                # Add the final generator point
+                self.add_generator_point_pair(points[-1], point_index, False)
+
+                # Check if we have exceeded the profile points limit
+                if self.check_profile_length_exceeds_profile_points():
+                    self.end_index = point_index + 1
+                    return False
+
+                # Increment point index
+                point_index += 1
+
+            # Increment the index for the next batch
+            start_batch_index = point_index
+
+            # Check if we are done
+            if start_batch_index == self.steps_up_to:
+                # Return True for a tail-off
+                return True
+
+            # Yield so that we don't continuously block other threads
+            cothread.Yield()
+
+    def create_generator_profile_sparse(self, start_index: int) -> bool:
+        # Ensure we have the correct trigger type
+        assert (
+            self.output_triggers != scanning.infos.MotionTrigger.EVERY_POINT
+        ), f"{self.name}: trigger should not be every point"
+
+        # Keep going until we finish or exceed the maximum length of a profile
+        start_batch_index = start_index
+        point_index = start_index
+        while True:
+            # Grab some points
+            points, joined, velocities = self.get_some_points(start_batch_index)
+            if not points:
+                return True
+
+            # Number of points and check if the last point is in this batch
+            num_points = len(points)
+            last_point_in_batch = self.is_last_point_in_current_batch(
+                start_batch_index, num_points
+            )
+
+            # Don't do all points in the batch otherwise we can get an index error
+            # when adding the turnaround.
+            points_to_do = num_points - 1
+            for i in range(points_to_do):
+                point_added = self.add_sparse_point(points, i, joined[i], velocities[i])
+
+                # add in the turnaround between non-contiguous points
+                if not (joined[i]):
+                    self.insert_gap(points[i], points[i + 1], point_index + 1)
+
+                # Check if we have exceeded the profile points limit. Only check if we
+                # have actually added a point, otherwise we waste time.
+                if point_added and self.check_profile_length_exceeds_profile_points():
+                    self.end_index = point_index + 1
+                    return False
+
+                # Increment point index
+                point_index += 1
+
+            # Check for the last point
+            if last_point_in_batch:
+                point_added = self.add_sparse_point(points, points_to_do, False, False)
+
+                # Check if we have exceeded the profile points limit. Only check if we
+                # have actually added a point, otherwise we waste time.
+                if point_added and self.check_profile_length_exceeds_profile_points():
+                    self.end_index = point_index + 1
+                    return False
+
+                # Increment point index
+                point_index += 1
+
+            # Increment the index for the next batch
+            start_batch_index = point_index
+
+            # Check if we are done
+            if start_batch_index == self.steps_up_to:
+                # Return True for a tail-off
+                return True
+
+            # Yield so that we don't continuously block other threads
+            cothread.Yield()
+
     def calculate_generator_profile(self, start_index, do_run_up=False):
         # If we are doing the first build, do_run_up will be passed to flag
         # that we need a run up, else just continue from the previous point
@@ -789,45 +933,13 @@ class PmacChildPart(builtin.parts.ChildPart):
 
         self.time_since_last_pvt = 0
 
-        points, joined, velocities = self.get_some_points(start_index)
-        points_idx = start_index
-        for i in range(start_index, self.steps_up_to):
-            if i - points_idx >= BATCH_POINTS:
-                points, joined, velocities = self.get_some_points(i)
-                points_idx = i
-
-            last_point = i + 1 == self.steps_up_to
-            if not last_point:
-                # cope with the zero axes case (where joined == None)
-                points_are_joined = joined is None or joined[i - points_idx]
-                same_velocities = velocities is None or velocities[i - points_idx]
-            else:
-                same_velocities = points_are_joined = False
-
-            if self.output_triggers == scanning.infos.MotionTrigger.EVERY_POINT:
-                self.add_generator_point_pair(
-                    points[i - points_idx], i, points_are_joined
-                )
-            else:
-                self.add_sparse_point(
-                    points, i - points_idx, points_are_joined, same_velocities
-                )
-
-            # add in the turnaround between non-contiguous points
-            if not (points_are_joined or last_point):
-                self.insert_gap(
-                    points[i - points_idx], points[i - points_idx + 1], i + 1
-                )
-
-            # Check if we have exceeded the points number and need to write
-            # Strictly less than so we always add one more point to the time
-            # array so we can always stretch points in a subsequent add with
-            # the values already in the profiles
-            if len(self.profile["timeArray"]) > PROFILE_POINTS:
-                self.end_index = i + 1
-                return
-
-        self.add_tail_off()
+        # Create the generator profile
+        if self.output_triggers == scanning.infos.MotionTrigger.EVERY_POINT:
+            needs_tail_off = self.create_generator_profile_every_point(start_index)
+        else:
+            needs_tail_off = self.create_generator_profile_sparse(start_index)
+        if needs_tail_off:
+            self.add_tail_off()
 
     def add_tail_off(self):
         # Add the last tail off point
diff --git a/malcolm/modules/pmac/util.py b/malcolm/modules/pmac/util.py
index fdc233d4..e398607f 100644
--- a/malcolm/modules/pmac/util.py
+++ b/malcolm/modules/pmac/util.py
@@ -158,6 +158,9 @@ def all_points_same_velocities(points: Points) -> Array[bool]:
         v2 = velocities[1:]
         results[axis_name] = np.isclose(v1, v2)
     result = and_all_axes(results)
+    # If we have no axes then all velocities are the same
+    if result is None:
+        result = np.ones(len(points), dtype=bool)
     return result
 
 
@@ -174,6 +177,9 @@ def all_points_joined(points: Points) -> Array[bool]:
         lowers = points.lower[axis_name][1:]
         results[axis_name] = np.logical_and(uppers == lowers, no_delay)
     result = and_all_axes(results)
+    # If we have no axes then all points are joined
+    if result is None:
+        result = np.ones(len(points), dtype=bool)
     return result
 
 


Test patch:
diff --git a/tests/test_modules/test_pmac/test_pmacchildpart.py b/tests/test_modules/test_pmac/test_pmacchildpart.py
index 4d219dbc..2e586ad5 100644
--- a/tests/test_modules/test_pmac/test_pmacchildpart.py
+++ b/tests/test_modules/test_pmac/test_pmacchildpart.py
@@ -6,7 +6,9 @@ from typing import List
 import numpy as np
 import pytest
 from mock import ANY, Mock, call, patch
+from mock.mock import MagicMock
 from scanpointgenerator import CompoundGenerator, LineGenerator, StaticPointGenerator
+from scanpointgenerator.core.point import Point, Points
 
 from malcolm.core import Context, Process
 from malcolm.modules import scanning
@@ -1215,7 +1217,7 @@ class TestPMACChildPart(ChildTestCase):
         )
         elapsed = datetime.now() - start
         # todo goal was sub 1 second but we achieved sub 3 secs
-        assert elapsed.total_seconds() < 3.5
+        assert elapsed.total_seconds() < 3.0
 
     def test_configure_long_trajectory(self):
         # Skip on GitHub Actions and GitLab CI
@@ -1395,3 +1397,137 @@ class TestPMACChildPart(ChildTestCase):
         assert (
             len(failed_list) == 0
         ), f"Failed with tweaking a second time on following inputs: {failed_list}"
+
+    def test_add_sparse_point(self):
+        # Set up the part
+        self.o.output_triggers = MotionTrigger.ROW_GATE
+        self.o.profile = MagicMock()
+
+        # Sample point
+        point = Point()
+        point.duration = 1.0
+        points = Points()
+        points += point
+
+        # A point will not be added if they are joined and the velocity is the same
+        points_are_joined = True
+        same_velocities = True
+        point_added = self.o.add_sparse_point(
+            points, 0, points_are_joined, same_velocities
+        )
+        assert point_added is False
+
+        # A point will be added if joined but not the same velocity
+        points_are_joined = True
+        same_velocities = False
+        point_added = self.o.add_sparse_point(
+            points, 0, points_are_joined, same_velocities
+        )
+        assert point_added is True
+
+        # A lower bound is added if we are at the end of a row
+        points_are_joined = False
+        same_velocities = False
+        point_added = self.o.add_sparse_point(
+            points, 0, points_are_joined, same_velocities
+        )
+        assert point_added is True
+
+        # A point is added if we are not joined
+        points_are_joined = False
+        same_velocities = True
+        point_added = self.o.add_sparse_point(
+            points, 0, points_are_joined, same_velocities
+        )
+        assert point_added is True
+
+
+class TestSparseGeneratorProfileTailOffs(ChildTestCase):
+    def setUp(self) -> None:
+        self.o = PmacChildPart(name="pmac", mri="PMAC")
+        # Set up the triggering
+        self.o.output_triggers = MotionTrigger.ROW_GATE
+
+    @patch.object(PmacChildPart, "insert_gap")
+    @patch.object(PmacChildPart, "add_sparse_point")
+    @patch.object(PmacChildPart, "check_profile_length_exceeds_profile_points")
+    @patch.object(PmacChildPart, "get_some_points")
+    def test_adds_tail_off_for_zero_points(
+        self, get_some_points_mock, exceed_profile_points_mock, *_
+    ):
+        # Configure mocks
+        exceed_profile_points_mock.return_value = False
+        get_some_points_mock.return_value = None, None, None
+
+        add_tail_off = self.o.create_generator_profile_sparse(0)
+
+        assert add_tail_off is True
+
+    @patch.object(PmacChildPart, "insert_gap")
+    @patch.object(PmacChildPart, "add_sparse_point")
+    @patch.object(PmacChildPart, "check_profile_length_exceeds_profile_points")
+    @patch.object(PmacChildPart, "get_some_points")
+    def test_adds_tail_off_when_done(
+        self, get_some_points_mock, exceed_profile_points_mock, *_
+    ):
+        # Set the mock profile check to False
+        exceed_profile_points_mock.return_value = False
+
+        # Create a Points object with a single point
+        point = Point()
+        point.duration = 1.0
+        points = Points()
+        points += point
+
+        assert len(points) == 1
+
+        # Return those points
+        get_some_points_mock.return_value = points, False, False
+
+        # Last point is in batch
+        self.o.steps_up_to = 1
+
+        add_tail_off = self.o.create_generator_profile_sparse(0)
+
+        assert add_tail_off is True
+
+    @patch.object(PmacChildPart, "insert_gap")
+    @patch.object(PmacChildPart, "add_sparse_point")
+    @patch.object(PmacChildPart, "check_profile_length_exceeds_profile_points")
+    @patch.object(PmacChildPart, "get_some_points")
+    def test_no_tail_off_for_exceeding_profile_points(
+        self, get_some_points_mock, exceed_profile_points_mock, *_
+    ):
+        # Set the mock profile check to True
+        exceed_profile_points_mock.return_value = True
+
+        # In the last point in batch
+        # Create a Points object with a single point
+        point = Point()
+        point.duration = 1.0
+        points = Points()
+        points += point
+
+        # Return those points
+        self.o.get_some_points.return_value = points, False, False
+
+        # Set steps up to - important for last_point_in_batch flag
+        self.o.steps_up_to = 1
+
+        add_tail_off = self.o.create_generator_profile_sparse(0)
+
+        assert add_tail_off is False
+
+        # In the main loop
+        # Just add one more point
+        points += point
+
+        # Return those points
+        get_some_points_mock.return_value = points, [False, False], [False, False]
+
+        # Set steps up to
+        self.o.steps_up_to = 2
+
+        add_tail_off = self.o.create_generator_profile_sparse(0)
+
+        assert add_tail_off is False