Testbench stop when TP is done, allow perf control

MooreOnlineConf/options/perf_profile.py

+###############################################################################
+# (c) Copyright 2024 CERN for the benefit of the LHCb Collaboration           #
+#                                                                             #
+# This software is distributed under the terms of the GNU General Public      #
+# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".   #
+#                                                                             #
+# In applying this licence, CERN does not waive the privileges and immunities #
+# granted to it by virtue of its status as an Intergovernmental Organization  #
+# or submit itself to any jurisdiction.                                       #
+###############################################################################
+from Configurables import PerfProfile, HLTControlFlowMgr
+HLTControlFlowMgr(
+    'HLTControlFlowMgr'
+).PreambleAlgs = HLTControlFlowMgr('HLTControlFlowMgr').PreambleAlgs + [
+    PerfProfile(
+        FIFOPath="perf_control.fifo", StartFromEventN=100, StopAtEventN=10000)
+]

MooreScripts/job/runHLT2Perf.sh

+#!/bin/bash
+###############################################################################
+# (c) Copyright 2024 CERN for the benefit of the LHCb Collaboration      #
+#                                                                             #
+# This software is distributed under the terms of the GNU General Public      #
+# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".   #
+#                                                                             #
+# In applying this licence, CERN does not waive the privileges and immunities #
+# granted to it by virtue of its status as an Intergovernmental Organization  #
+# or submit itself to any jurisdiction.                                       #
+###############################################################################
+set -euo pipefail
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+source "$DIR/setupTask.sh"
+
+setup_options_path
+settings=$(python -c 'import OnlineEnvBase; print(OnlineEnvBase.HLTType)')
+if [ -z "$settings" ]; then
+    echo "ERROR settings (HLTType) is empty!"
+    exit 3
+fi
+exec_gaudirun \
+    $MOOREONLINECONFROOT/options/verbosity.py \
+    $MOOREONLINECONFROOT/options/tags-OnlineEnv.py \
+    $MOOREONLINECONFROOT/options/hlt2.py \
+    $HLT2CONFROOT/options/${settings}.py \
+    $MOOREONLINECONFROOT/options/online.py \
+    $MOOREONLINECONFROOT/options/perf_profile.py \
+    --post-option="from Configurables import HiveDataBrokerSvc, PerfProfile" \
+    --post-option="p = PerfProfile()" \
+    --post-option="if p not in HiveDataBrokerSvc().DataProducers: HiveDataBrokerSvc().DataProducers.append(p)"
\ No newline at end of file

MooreScripts/python/MooreScripts/testbench/architecture.py

@@ -114,6 +114,6 @@ def overwrite_dict_value(data: Any, mapping: dict) -> Any:
     elif isinstance(data, list):
         return [overwrite_dict_value(item, mapping) for item in data]
     elif isinstance(data, tuple):
-        return (overwrite_dict_value(item, mapping) for item in data)
+        return tuple(overwrite_dict_value(item, mapping) for item in data)
     else:
         return data

MooreScripts/python/MooreScripts/testbench/emulator.py

@@ -216,15 +216,22 @@ class Task:
         self._status_task = None
         self.process = None
         self._perf_data = None
+        self._perf_control = None
         self.utgid = args["args"][0]
         for a in args["args"]:
             if a.startswith("-type="):
                 self.type = a.removeprefix("-type=")
 
-    def use_perf(self):
+    def use_perf(self, control=False):
         if self.process is not None:
             raise RuntimeError("use_perf() must be called before load()")
         self._perf_data = Path(f"{self.utgid}.perf.data").absolute()
+        if control:
+            self._perf_control = Path("perf_control.fifo").absolute()
+            try:
+                os.mkfifo(self._perf_control)
+            except FileExistsError:
+                pass
 
     async def __aenter__(self):
         pass
@@ -242,7 +249,7 @@ class Task:
 
         if self._perf_data:
             args["args"][0] = args["executable"]
-            args["args"] = [
+            perf_cmd = [
                 "perf",
                 "record",
                 f"--output={self._perf_data}",
@@ -250,7 +257,10 @@ class Task:
                 "--aio",
                 "--call-graph=dwarf,65528",
                 "--count=100000000",
-            ] + args["args"]
+            ]
+            if self._perf_control:
+                perf_cmd.append(f"--control=fifo:{self._perf_control}")
+            args["args"] = perf_cmd + args["args"]
             args["executable"] = "perf"
 
         def exit_callback(p):

MooreScripts/python/MooreScripts/testbench/scenarios/default.py

@@ -34,6 +34,10 @@ async def run(tasks: List[emulator.Task], args, extra_argv):
         "--use-perf",
         action="store_true",
         help="perf record the main tasks and create a flamegraph")
+    parser.add_argument(
+        "--use-perf-control",
+        action="store_true",
+        help="perf record the event loop and create a flamegraph")
     parser.add_argument(
         "--wait-after-load",
         action="store_true",
@@ -49,9 +53,9 @@ async def run(tasks: List[emulator.Task], args, extra_argv):
         raise ValueError("There must be exactly one *Prod task")
     prod_task = prod_tasks[0]
 
-    if extra_args.use_perf:
+    if extra_args.use_perf or extra_args.use_perf_control:
         for t in main_tasks:
-            t.use_perf()
+            t.use_perf(control=extra_args.use_perf_control)
 
     await tasks_load(tasks)
     # TODO for some reason HLT2 publishes OFFLINE before NOT_READY, but only sometimes
@@ -78,7 +82,7 @@ async def run(tasks: List[emulator.Task], args, extra_argv):
 
     if args.measure_throughput > 0:
         # wait a bit for things to settle and measure throughput
-        await asyncio.sleep(args.measure_throughput / 8)
+        await asyncio.sleep(args.delay_tp_measurement)
         await tasks_measure_throughput(
             tasks, max_duration=args.measure_throughput)
     else:
@@ -101,7 +105,7 @@ async def run(tasks: List[emulator.Task], args, extra_argv):
         # if there is a writer, wait for the output rate to be 0
         if any("Writer" in task.utgid for task in tasks):
             await tasks_wait_for_output(tasks)
-    elif "HLT2" in main_tasks[0].utgid:
+    elif "HLT2" in main_tasks[0].utgid and not args.measure_throughput > 0:
         log.info(f"Waiting to process all {n_events_produced} events")
         n_events_processed = sum(await tasks_wait_for_value(
             main_tasks,

MooreScripts/scripts/testbench.py

@@ -112,6 +112,12 @@ parser.add_argument(
     type=float,
     help="How long to measure throughput for.",
 )
+parser.add_argument(
+    "--delay-tp-measurement",
+    default=60,
+    type=float,
+    help="How long to measure throughput for.",
+)
 parser.add_argument(
     "--tfdb-nfiles",
     type=int,
@@ -158,8 +164,10 @@ replacements = {
 
 arch = architecture.read_xml(args.architecture)
 if args.write_encoding_keys:
-    arch = architecture.overwrite_dict_value(arch,
-                                             {"WRITE_ENCODING_KEYS": "1"})
+    arch = architecture.overwrite_dict_value(
+        arch,
+        {"WRITE_ENCODING_KEYS": "1"},
+    )
 task_instance_args = architecture.instance_args(arch, replacements)
 
 emulator.check_for_orphans([a["args"][0] for a in task_instance_args])

MooreScripts/tests/options/HLT2Perf/Arch.xml

+<!--
+    (c) Copyright 2021-2022 CERN for the benefit of the LHCb Collaboration
+
+    This software is distributed under the terms of the GNU General Public
+    Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".
+
+    In applying this licence, CERN does not waive the privileges and immunities
+    granted to it by virtue of its status as an Intergovernmental Organization
+    or submit itself to any jurisdiction.
+-->
+<tasks_inventory>
+
+  <task name="MBM" user="${USER}" group="${GROUP}">
+    <command>${MOORESCRIPTSROOT}/scripts/runDFTask.sh</command>
+    <argument name="-type" value="${NAME}" />
+    <argument name="-runinfo" value="${RUNINFO}" />
+    <argument name="-options" value="${MOORESCRIPTSROOT}/options/HLT2MBM.opts" />
+    <argument name="-class" value="Class0" />
+    <fmcparam name="utgid" value="${PARTITION}_${NODE}_${NAME}_${INSTANCE}" />
+    <fmcparam name="define" value="BINARY_TAG=${BINARY_TAG}" />
+    <fmcparam name="define" value="WORKING_DIR=${WORKING_DIR}" />
+    <timeout action="Any" value="20" />
+  </task>
+
+  <task name="MDFProd" user="${USER}" group="${GROUP}">
+    <command>${MOORESCRIPTSROOT}/scripts/runDFTask.sh</command>
+    <argument name="-type" value="${NAME}" />
+    <argument name="-runinfo" value="${RUNINFO}" />
+    <argument name="-options" value="${MOORESCRIPTSROOT}/options/HLT2MDFProd.opts" />
+    <argument name="-class" value="Class2" />
+    <fmcparam name="utgid" value="${PARTITION}_${NODE}_${NAME}_${INSTANCE}" />
+    <fmcparam name="define" value="BINARY_TAG=${BINARY_TAG}" />
+    <fmcparam name="define" value="WORKING_DIR=${WORKING_DIR}" />
+    <fmcparam name="define" value="DATA_DIR=${DATA_DIR}" />
+    <timeout action="Any" value="30" />
+  </task>
+
+  <task name="HLT2" user="${USER}" group="${GROUP}" instances="1">
+    <command>${MOORESCRIPTSROOT}/job/runHLT2Perf.sh</command>
+    <argument name="-type" value="${NAME}" />
+    <argument name="-runinfo" value="${RUNINFO}" />
+    <argument name="-class" value="Class1" />
+    <argument name="-numthreads" value="20" />
+    <fmcparam name="utgid" value="${PARTITION}_${NODE}_${NAME}_${INSTANCE}" />
+    <fmcparam name="define" value="BINARY_TAG=${BINARY_TAG}" />
+    <fmcparam name="define" value="WORKING_DIR=${WORKING_DIR}" />
+    <fmcparam name="define" value="BIND_NUMA=1" />
+    <fmcparam name="define" value="WRITE_ENCODING_KEYS=1" />
+    <timeout action="Any" value="120" />
+    <timeout action="load" value="20" />
+  </task>
+
+  <task name="Writer" user="${USER}" group="${GROUP}">
+    <command>${MOORESCRIPTSROOT}/scripts/runDFTask.sh</command>
+    <argument name="-type" value="${NAME}" />
+    <argument name="-runinfo" value="${RUNINFO}" />
+    <argument name="-options" value="${MOORESCRIPTSROOT}/options/TestWriter.opts" />
+    <argument name="-class" value="Class1" />
+    <fmcparam name="utgid" value="${PARTITION}_${NODE}_${NAME}_${INSTANCE}" />
+    <fmcparam name="define" value="BINARY_TAG=${BINARY_TAG}" />
+    <fmcparam name="define" value="WORKING_DIR=${WORKING_DIR}" />
+    <timeout action="Any" value="20" />
+    <timeout action="load" value="20" />
+  </task>
+
+</tasks_inventory>

MooreScripts/tests/options/HLT2Perf/OnlineEnv.opts

+OnlineEnv.PartitionID         = 65535;
+OnlineEnv.PartitionName       = "TESTBEAMGUI";
+OnlineEnv.Activity            = "PHYSICS";
+OnlineEnv.OutputLevel         = 3;
+//
+OnlineEnv.Reader_Rescan       = 0;
+OnlineEnv.Reader_Directories  = {"/calib/online/tmpHlt1Dumps/LHCb/0000263844"};
+OnlineEnv.Reader_FilePrefix   = "Run_0000263844";

MooreScripts/tests/options/HLT2Perf/OnlineEnvBase.py

+###############################################################################
+# (c) Copyright 2022 CERN for the benefit of the LHCb Collaboration           #
+#                                                                             #
+# This software is distributed under the terms of the GNU General Public      #
+# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".   #
+#                                                                             #
+# In applying this licence, CERN does not waive the privileges and immunities #
+# granted to it by virtue of its status as an Intergovernmental Organization  #
+# or submit itself to any jurisdiction.                                       #
+###############################################################################
+PartitionID = 65535
+PartitionName = "TESTBEAMGUI"
+Activity = "PHYSICS"
+HltArchitecture = "dummy"
+OnlineVersion = "v0"
+MooreVersion = "v0"
+MooreOnlineVersion = "v0"
+OutputLevel = 3