Skip to content

k4bench.runner.process

k4bench.runner.process

Run one command under /usr/bin/time -v and harvest what it measured.

This is the tool-independent half of a benchmark run: building the timed shell command, streaming the process output into its log, stopping the whole process group on Ctrl-C, and parsing the time -v report. It knows nothing about which program it times; :mod:k4bench.runner.ddsim and :mod:k4bench.runner.k4run supply the command line, and :mod:k4bench.runner.result turns the outcome into a :class:~k4bench.results.model.RunResult.

TimedProcess dataclass 

TimedProcess(returncode: int, metrics: dict)

Outcome of one timed process: its exit code and the parsed time -v metrics (see :func:~k4bench.runner.parser.parse_time_output for the keys; each is None when the report lacked it).

timed_shell_command 

timed_shell_command(argv: list[str], setup_script: Path | None = None, after_setup: str = '') -> str

Return the bash command that runs argv under GNU time -v.

setup_script, when given, is sourced first, and the bash after_setup runs next, in the environment the script left. Every argument is shell-quoted, and one argument goes on each line so the command reads cleanly when it is echoed into a log.

Source code in k4bench/runner/process.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def timed_shell_command(
    argv: list[str], setup_script: Path | None = None, after_setup: str = ""
) -> str:
    """Return the bash command that runs *argv* under GNU ``time -v``.

    *setup_script*, when given, is sourced first, and the bash *after_setup*
    runs next, in the environment the script left. Every argument is
    shell-quoted, and one argument goes on each line so the command reads
    cleanly when it is echoed into a log.
    """
    gnu_time = shutil.which("time")
    if gnu_time is None:
        raise RuntimeError(
            "GNU time not found in PATH. Install it (e.g. 'dnf install time' or 'apt install time')."
        )

    source_line = (
        f"source {shlex.quote(str(setup_script))}\n" if setup_script is not None else ""
    )
    program, *args = argv
    joined = " \\\n    ".join(shlex.quote(a) for a in args)
    return f"{source_line}{after_setup}{gnu_time} -v {shlex.quote(program)} \\\n    {joined}"

run_timed 

run_timed(command: str, *, env: dict[str, str], log_path: Path, verbose: bool = False, cwd: Path | None = None) -> TimedProcess

Run command (from :func:timed_shell_command) in bash and wait for it.

All output goes to log_path, and to stdout as well when verbose. The process runs in cwd (default: the current directory) and in its own session, so a Ctrl-C stops it and every child it started before the KeyboardInterrupt is re-raised.

Source code in k4bench/runner/process.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def run_timed(
    command: str,
    *,
    env: dict[str, str],
    log_path: Path,
    verbose: bool = False,
    cwd: Path | None = None,
) -> TimedProcess:
    """Run *command* (from :func:`timed_shell_command`) in bash and wait for it.

    All output goes to *log_path*, and to stdout as well when *verbose*. The
    process runs in *cwd* (default: the current directory) and in its own
    session, so a Ctrl-C stops it and every child it started before the
    ``KeyboardInterrupt`` is re-raised.
    """
    log_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        proc = subprocess.Popen(
            command,
            shell=True,
            executable="/bin/bash",
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            env=env,
            cwd=cwd,
            start_new_session=True,
        )

        tail: deque[str] = deque(maxlen=_TAIL_LINES)

        with log_path.open("w") as log_file:

            if proc.stdout is None:
                raise RuntimeError(f"Failed to capture the output of {log_path.stem}.")

            for line in proc.stdout:
                if verbose:
                    print(line, end="", flush=True)
                log_file.write(line)
                tail.append(line)

            proc.wait()  # ensure returncode is populated

    except KeyboardInterrupt:
        print(f"\nStopping {log_path.stem}...", flush=True)

        try:
            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            proc.wait(timeout=5)

        except (subprocess.TimeoutExpired, ProcessLookupError):

            try:
                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)

            except ProcessLookupError:
                pass

        raise

    metrics = parse_time_output("".join(tail))
    if metrics["wall_time_raw"] is None or metrics["peak_rss_mb"] is None:
        _warn_unparsed(log_path)

    return TimedProcess(returncode=proc.returncode, metrics=metrics)