Merge branch 'main' into NGC-1264-f-dither-tidy

Author: bmerry

qualification/Jenkinsfile

@@ -101,19 +101,39 @@ pipeline {
          junit testResults: 'result.xml', skipPublishingChecks: true
 
          // Compress and publish json file so that at least we have that.
+         // 'includes' specifies which files from the current directory to archive;
+         // we only want the linked file itself
          sh 'xz --keep --force -T 0 report.json'
-         publishHTML(target: [keepAll: true, reportName: 'Qualification Test Intermediate JSON', reportDir: '', reportFiles: 'report.json.xz'])
+         publishHTML(target: [
+           keepAll: true,
+           reportName: 'Qualification Test Intermediate JSON',
+           reportDir: '',
+           reportFiles: 'report.json.xz',
+           includes: 'report.json.xz'
+         ])
 
          script {
            // Generate and publish test report
            COMMIT_ID = sh(script: 'qualification/report/generate_pdf.py report.json report.pdf -c', returnStdout: true)
            currentBuild.displayName = "#${BUILD_NUMBER} (${COMMIT_ID})"
          }
-         publishHTML(target: [keepAll: true, reportName: 'Qualification Test Report', reportDir: '', reportFiles: 'report.pdf'])
+         publishHTML(target: [
+           keepAll: true,
+           reportName: 'Qualification Test Report',
+           reportDir: '',
+           reportFiles: 'report.pdf',
+           includes: 'report.pdf'
+         ])
 
          // If that worked, we can probably generate and publish a procedure as well
          sh 'qualification/report/generate_pdf.py report.json procedure.pdf --generate-procedure-doc'
-         publishHTML(target: [keepAll: true, reportName: 'Qualification Test Procedure', reportDir: '', reportFiles: 'procedure.pdf'])
+         publishHTML(target: [
+           keepAll: true,
+           reportName: 'Qualification Test Procedure',
+           reportDir: '',
+           reportFiles: 'procedure.pdf',
+           includes: 'procedure.pdf'
+         ])
        }
      }
    }

qualification/cbf.py

@@ -48,8 +48,8 @@ class CBFBase:
     mode_config: dict  # Configuration values used for MeerKAT mode string
     uuid: UUID
 
-    async def close(self) -> None:
-        """Shut down all the connections."""
+    async def close(self, master_controller_client: aiokatcp.Client | None) -> None:
+        """Shut down all the connections and deconfigure the subarray product."""
         pass  # Derived classes can override
 
 
@@ -153,13 +153,15 @@ async def steady_state_timestamp(self, *, max_delay: int = DEFAULT_MAX_DELAY) ->
         logger.debug("steady_state_timestamp: %d", timestamp)
         return timestamp
 
-    async def close(self) -> None:
-        """Shut down all the connections."""
+    async def close(self, master_controller_client: aiokatcp.Client | None) -> None:
+        """Shut down all the connections and deconfigure the subarray product."""
         clients = self.dsim_clients + [self.product_controller_client]
         async with asyncio.TaskGroup() as tg:
             for client in clients:
                 client.close()
                 tg.create_task(client.wait_closed())
+        if master_controller_client is not None:
+            await master_controller_client.request("product-deconfigure", self.name)
 
     async def dsim_time(self, dsim_idx: int = 0) -> float:
         """Get the current UNIX time, as reported by a dsim.
@@ -271,6 +273,8 @@ async def get_task_details(suffix: str, type: type[_T]) -> dict[str, _T]:
     interfaces = await get_task_details_multi(r"interfaces\.([^.]+)\.name", str)
     tasks: dict[str, TaskDict] = {}
     for task_name, hostname in hosts.items():
+        if task_name not in git_version_futures:
+            raise RuntimeError(f"could not get katcp port for {task_name}: possible DNS failure")
         task_interfaces = interfaces.get(task_name, {})
         tasks[task_name] = {
             "host": hostname,
@@ -326,10 +330,8 @@ async def _close_cbf(self) -> None:
         if self._cbf is not None:
             name = self._cbf.name
             logger.info("Tearing down CBF %s.", name)
-            await self._cbf.close()
+            await self._cbf.close(self._master_controller_client)
             self._cbf = None
-            if self._master_controller_client is not None:
-                await self._master_controller_client.request("product-deconfigure", name)
 
     async def _get_master_controller_client(self) -> aiokatcp.Client:
         if self._master_controller_client is not None:
@@ -351,8 +353,8 @@ async def get_cbf(self, cbf_config: dict, cbf_mode_config: dict) -> CBFBase:
         if self._cbf is not None and self._cbf.config == cbf_config:
             return self._cbf
 
-        await self._close_cbf()
         try:
+            await self._close_cbf()
             master_controller_client = await self._get_master_controller_client()
             product_name = self._pytestconfig.getini("product_name")
             try:

qualification/requirements.txt

@@ -398,7 +398,7 @@ yarl==1.9.4
     #   -c qualification/../requirements-dev.txt
     #   -c qualification/../requirements.txt
     #   aiohttp
-zipp==3.17.0
+zipp==3.19.2
     # via
     #   -c qualification/../requirements-dev.txt
     #   -c qualification/../requirements.txt

requirements-dev.txt

@@ -425,7 +425,7 @@ yarl==1.9.4
     # via
     #   -c requirements.txt
     #   aiohttp
-zipp==3.17.0
+zipp==3.19.2
     # via
     #   -c requirements.txt
     #   importlib-metadata

requirements.txt

@@ -174,5 +174,5 @@ xarray==2023.12.0
     # via katgpucbf (setup.cfg)
 yarl==1.9.4
     # via aiohttp
-zipp==3.17.0
+zipp==3.19.2
     # via importlib-metadata

scratch/fgpu/benchmarks/compute_bench.py

@@ -21,7 +21,7 @@
 import numpy as np
 from katsdpsigproc import accel
 
-from katgpucbf import DIG_SAMPLE_BITS
+from katgpucbf import DEFAULT_JONES_PER_BATCH, DIG_SAMPLE_BITS
 from katgpucbf.fgpu.compute import ComputeTemplate, NarrowbandConfig
 from katgpucbf.fgpu.engine import generate_ddc_weights, generate_pfb_weights
 from katgpucbf.fgpu.main import DEFAULT_DDC_TAPS_RATIO
@@ -33,7 +33,7 @@ def main():  # noqa: C901
     parser.add_argument("--recv-chunk-samples", type=int, default=32 * 1024 * 1024)
     parser.add_argument("--send-chunk-jones", type=int, default=8 * 1024 * 1024)
     parser.add_argument("--channels", type=int, default=32768)
-    parser.add_argument("--spectra-per-heap", type=int, default=256)
+    parser.add_argument("--jones-per-batch", type=int, default=DEFAULT_JONES_PER_BATCH)
     parser.add_argument("--dig-sample-bits", type=int, default=DIG_SAMPLE_BITS)
     parser.add_argument("--send-sample-bits", type=int, default=8, choices=[4, 8])
     parser.add_argument("--passes", type=int, default=1000)
@@ -57,6 +57,9 @@ def main():  # noqa: C901
             parser.error("--sem must divide into --passes")
     if args.kernel == "ddc" and not args.narrowband:
         parser.error("--kernel=ddc requires --narrowband")
+    if args.jones_per_batch % args.channels != 0:
+        parser.error("--jones-per-batch must be a multiple of --channels")
+    spectra_per_heap = args.jones_per_batch // args.channels
 
     rng = np.random.default_rng(seed=1)
     context = accel.create_some_context(device_filter=lambda device: device.is_cuda)
@@ -77,14 +80,14 @@ def main():  # noqa: C901
             context, args.taps, args.channels, args.dig_sample_bits, args.send_sample_bits, narrowband=narrowband_config
         )
         command_queue = context.create_tuning_command_queue()
-        out_spectra = accel.roundup(args.send_chunk_jones // args.channels, args.spectra_per_heap)
+        out_spectra = accel.roundup(args.send_chunk_jones // args.channels, spectra_per_heap)
         frontend_spectra = min(args.recv_chunk_samples // spectra_samples, out_spectra)
         extra_samples = window - spectra_samples
         fn = template.instantiate(
             command_queue,
             samples=args.recv_chunk_samples + extra_samples,
             spectra=out_spectra,
-            spectra_per_heap=args.spectra_per_heap,
+            spectra_per_heap=spectra_per_heap,
             seed=123,
             sequence_first=456,
         )

scratch/fgpu/compare_fengs.py

@@ -1,5 +1,21 @@
 #!/usr/bin/env python3
 
+################################################################################
+# Copyright (c) 2022, 2024, National Research Foundation (SARAO)
+#
+# Licensed under the BSD 3-Clause License (the "License"); you may not use
+# this file except in compliance with the License. You may obtain a copy
+# of the License at
+#
+#   https://opensource.org/licenses/BSD-3-Clause
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
 """Compare output of two F-engines.
 
 This can be used to ensure that a change to F-engine code produces
@@ -22,7 +38,7 @@
 
 import katgpucbf.recv
 import katgpucbf.xbgpu.recv
-from katgpucbf import COMPLEX, N_POLS
+from katgpucbf import COMPLEX, DEFAULT_JONES_PER_BATCH, N_POLS
 
 
 def parse_args() -> argparse.Namespace:
@@ -31,21 +47,25 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--interface", type=get_interface_address, help="Interface on which to listen")
     parser.add_argument("--ibv", action="store_true", help="Use ibverbs")
     parser.add_argument("--array-size", type=int, required=True)
-    parser.add_argument("--channels-per-stream", type=int, required=True)
-    parser.add_argument("--spectra-per-heap", type=int, default=256)
+    parser.add_argument("--channels", type=int, required=True)
+    parser.add_argument("--channels-per-substream", type=int, required=True)
+    parser.add_argument("--jones-per-batch", type=int, default=DEFAULT_JONES_PER_BATCH)
     parser.add_argument("--samples-between-spectra", type=int, required=True)
-    parser.add_argument("--heaps-per-fengine-per-chunk", type=int, default=16)
+    parser.add_argument("--heaps-per-fengine-per-chunk", type=int, default=32)
     args = parser.parse_args()
+    if args.jones_per_batch % args.channels != 0:
+        parser.error("--jones-per-batch must be a multiple of --channels")
     return args
 
 
 async def main() -> None:
     args = parse_args()
+    spectra_per_heap = args.jones_per_batch // args.channels
     layout = katgpucbf.xbgpu.recv.Layout(
         n_ants=args.array_size,
-        n_channels_per_stream=args.channels_per_stream,
-        n_spectra_per_heap=args.spectra_per_heap,
-        timestamp_step=args.samples_between_spectra * args.spectra_per_heap,
+        n_channels_per_substream=args.channels_per_substream,
+        n_spectra_per_heap=spectra_per_heap,
+        timestamp_step=args.samples_between_spectra * spectra_per_heap,
         sample_bits=8,
         heaps_per_fengine_per_chunk=args.heaps_per_fengine_per_chunk,
     )
@@ -56,14 +76,14 @@ async def main() -> None:
         shape = (
             layout.heaps_per_fengine_per_chunk,
             layout.n_ants,
-            layout.n_channels_per_stream,
+            layout.n_channels_per_substream,
             layout.n_spectra_per_heap,
             N_POLS,
             COMPLEX,
         )
         data = np.ones(shape, np.int8)
         present = np.zeros(shape[:2], np.uint8)
-        chunk = katgpucbf.recv.Chunk(data=data, present=present, stream=stream)
+        chunk = katgpucbf.recv.Chunk(data=data, present=present, sink=stream)
         chunk.recycle()
 
     srcs = [(ep.host, ep.port) for ep in args.src]

scratch/xbgpu/benchmarks/beamform_bench.py

@@ -20,24 +20,36 @@
 
 import katsdpsigproc.accel
 
+from katgpucbf import DEFAULT_JONES_PER_BATCH
 from katgpucbf.xbgpu.beamform import BeamformTemplate
 
 
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("--array-size", type=int, default=80, help="Antennas in the array [%(default)s]")
+    parser.add_argument(
+        "--channels", type=int, default=1024, help="Total number of channels in the stream [%(default)s]"
+    )
     parser.add_argument(
         "--channels-per-substream", type=int, default=16, help="Channels processed by one engine [%(default)s]"
     )
-    parser.add_argument("--spectra-per-heap", type=int, default=256, help="Spectra in each batch [%(default)s]")
+    parser.add_argument(
+        "--jones-per-batch",
+        type=int,
+        default=DEFAULT_JONES_PER_BATCH,
+        help="Number of antenna-channelised-voltage Jones vectors in each F-engine batch [%(default)s]",
+    )
     parser.add_argument("--heaps-per-fengine-per-chunk", type=int, default=32, help="Batches per chunk [%(default)s]")
     parser.add_argument("--beams", type=int, default=4, help="Number of dual-pol beams [%(default)s]")
     parser.add_argument("--passes", type=int, default=10000, help="Number of times to repeat the test [%(default)s]")
     args = parser.parse_args()
+    if args.jones_per_batch % args.channels != 0:
+        parser.error("--jones-per-batch must be a multiple of --channels")
+    spectra_per_heap = args.jones_per_batch // args.channels
 
     ctx = katsdpsigproc.accel.create_some_context()
     command_queue = ctx.create_command_queue()
-    template = BeamformTemplate(ctx, [0, 1] * args.beams, n_spectra_per_batch=args.spectra_per_heap)
+    template = BeamformTemplate(ctx, [0, 1] * args.beams, n_spectra_per_batch=spectra_per_heap)
     fn = template.instantiate(
         command_queue,
         n_batches=args.heaps_per_fengine_per_chunk,
@@ -66,7 +78,7 @@ def main():
     voltages = (
         args.array_size
         * args.channels_per_substream
-        * args.spectra_per_heap
+        * spectra_per_heap
         * args.heaps_per_fengine_per_chunk
         * args.passes
     )

scratch/xbgpu/benchmarks/correlate_bench.py

@@ -20,23 +20,35 @@
 
 import katsdpsigproc.accel
 
+from katgpucbf import DEFAULT_JONES_PER_BATCH
 from katgpucbf.xbgpu.correlation import CorrelationTemplate
 
 
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("--array-size", type=int, default=80, help="Antennas in the array [%(default)s]")
+    parser.add_argument(
+        "--channels", type=int, default=1024, help="Total number of channels in the stream [%(default)s]"
+    )
     parser.add_argument(
         "--channels-per-substream", type=int, default=16, help="Channels processed by one engine [%(default)s]"
     )
-    parser.add_argument("--spectra-per-heap", type=int, default=256, help="Spectra in each batch [%(default)s]")
+    parser.add_argument(
+        "--jones-per-batch",
+        type=int,
+        default=DEFAULT_JONES_PER_BATCH,
+        help="Number of antenna-channelised-voltage Jones vectors in each F-engine batch [%(default)s]",
+    )
     parser.add_argument("--heaps-per-fengine-per-chunk", type=int, default=32, help="Frames per chunk [%(default)s]")
     parser.add_argument("--passes", type=int, default=10000, help="Number of times to repeat the test [%(default)s]")
     args = parser.parse_args()
+    if args.jones_per_batch % args.channels != 0:
+        parser.error("--jones-per-batch must be a multiple of --channels")
+    spectra_per_heap = args.jones_per_batch // args.channels
 
     ctx = katsdpsigproc.accel.create_some_context()
     command_queue = ctx.create_command_queue()
-    template = CorrelationTemplate(ctx, args.array_size, args.channels_per_substream, args.spectra_per_heap, 8)
+    template = CorrelationTemplate(ctx, args.array_size, args.channels_per_substream, spectra_per_heap, 8)
     fn = template.instantiate(command_queue, args.heaps_per_fengine_per_chunk)
 
     fn.ensure_all_bound()
@@ -54,7 +66,7 @@ def main():
     voltages = (
         args.array_size
         * args.channels_per_substream
-        * args.spectra_per_heap
+        * spectra_per_heap
         * args.heaps_per_fengine_per_chunk
         * args.passes
     )

src/katgpucbf/fsim/main.py

@@ -73,7 +73,7 @@ def parse_args(arglist: Sequence[str] | None = None) -> argparse.Namespace:
         "--array-size", type=int, default=80, help="Number of antennas in the simulated array [%(default)s]"
     )
     parser.add_argument(
-        "--channels", type=int, default=32768, help="Total number of channels in the simulated array [%(default)s]"
+        "--channels", type=int, default=32768, help="Total number of channels in the simulated stream [%(default)s]"
     )
     parser.add_argument(
         "--channels-per-substream", type=int, default=512, help="Number of channels sent by this fsim [%(default)s]"