def perfstat()

in perfrunbook/utilities/measure_aggregated_pmu_stats.py [0:0]
54 lines of code
16 McCabe index (conditional complexity)

def perfstat(counter_groups, timeout=None, cpus=None):
    """
    Measure performance counters using perf-stat in a subprocess.
    Stores results into a CSV file.  Uses our own multiplexing loop
    which is cheaper than letting perf in the kernel do multiplexing.
    """
    try:
        if not cpus:
            res = subprocess.run(["lscpu", "-p=CPU"], check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
            cpus = []
            for line in res.stdout.decode("utf-8").splitlines():
                match = re.search(r"""^(\d+)$""", line)
                if match is not None:
                    cpus.append(match.group(1))

        sig = SignalWatcher()
        if timeout:
            signal.alarm(timeout)
        out = open(RESULTS_CSV, "a")

        while not sig.kill_now:  # waits until a full measurement cycle is done.
            i = 0
            for pmu in counter_groups.keys():
                for ctrset in counter_groups[pmu]:
                    group = f"group{i}"
                    # Forms the perf counter format with a unique group and hash to form the name
                    #XXX: For arm_cmn, getting a None type for a counter to program, so that's odd....
                    counters = [f"{pmu}/{ctr.get_event_to_program()},name={group}-{ctr.get_canonical_name()}/" for ctr in ctrset]
                    # Collect everything un-aggregated.  So we can see lightly loaded CPUs

                    perf_cmd = [
                        "perf",
                        "stat",
                        f"-I{SAMPLE_INTERVAL * 1000}",
                        "-A",
                        "-x|",
                        "-a",
                        "-e",
                        f"{','.join(counters)}",
                    ]

                    # Assumes we don't mix counter types (we shouldn't as its per pmu)
                    for ctr in ctrset:
                        if ctr.is_per_cpu():
                            perf_cmd.append(f"-C{','.join(cpus)}")
                            break
                        else:
                            perf_cmd.append(f"-C0")
                            break

                    perf_cmd.extend([
                        "--",
                        "sleep",
                        f"{SAMPLE_INTERVAL}",
                    ])

                    # TODO: How to work with CMN and tell perf how to program the PMU
                    proc = subprocess.Popen(
                        perf_cmd,
                        preexec_fn=mask_signals,
                        stdout=subprocess.PIPE,
                        stderr=out,
                    )
                    proc.wait()
                    i += 1
        out.close()
        if timeout:
            # Cancel the timeout if any before leaving the loop
            signal.alarm(0)
    except subprocess.CalledProcessError:
        print("Failed to measure performance counters.")