ExecutorBuilt incomes with cpbenchy
cpbenchy run instances/ -s ortools -t 60 --executor runlimit- Version
- cpbenchy 0.1.0.dev0
- Last updated
- 6 Oct 2026 · 1 commit
- Authors
- ThomSerg
- Requires
- Linux with cgroups v2 (see `cpbenchy doctor`)
- Tags
- executorbenchexecmeasurement
What it does
Runs every worker under BenchExec’s runexec: cgroups enforce the memory limit and measure the CPU
time and peak memory of the whole process tree, so measurements are reliable. Each parallel run gets its
own physical cores and the memory of their NUMA node.
It is the default wherever it works: cpbenchy doctor says whether it does.
Use it
import cpbenchy
cpbenchy.run("instances/", solvers=["ortools"], time_limit=60, jobs=8, executor="runlimit")
cpbenchy.run("instances/", solvers=["ortools"], time_limit=60, args=["--container"]) # no network, read-only file systemcpbenchy run instances/ -s ortools -t 60 -j 8 --executor runlimit
cpbenchy run instances/ -s ortools -t 60 --container # no network, read-only file systemOptions
| Option | |
|---|---|
--grace |
seconds after a time limit before a run is killed (default 10) |
--terminate |
stop runs at their limit, with SIGTERM at the CPU time limit; see --terminate |
--hyperthreading |
let runs use hyperthread siblings |
--container |
no network, read-only file system except the output directory |
Implementation
The RunlimitExecutor in src/cpbenchy/executors.py, lines 157–205 of 305, as of this version of the docs.
class RunlimitExecutor(Executor):
"""BenchExec's runexec, through runlimit: cgroups enforce the memory limit and measure CPU time and
memory of the whole process tree; each run is pinned to its own cores and NUMA memory."""
name = "runlimit"
reliable = True
pin = True
def __init__(self, *, container: bool = False, writable_dirs: tuple[Path, ...] = (), **kwargs):
super().__init__(**kwargs)
self.container = container
self.writable_dirs = writable_dirs
self._running: set = set()
def execute(self, job: WorkerJob) -> Measurement:
from cpbenchy import runlimit
handle = runlimit.start(
job.cmd,
output_file=job.log,
walltime=job.limits.time_s + self.grace_s,
cputime=job.limits.cputime_s + self.grace_s if job.limits.cputime_s is not None else None,
soft_cputime=job.limits.cputime_s if self.terminate else None,
memlimit_mib=job.limits.mem_mib,
cores=job.cpus,
memory_nodes=job.memory_nodes,
container=self.container,
writable_dirs=[job.log.parent, *self.writable_dirs],
env=job.env or None,
)
self._running.add(handle)
try:
result = handle.result()
finally:
self._running.discard(handle)
return Measurement(
walltime_s=result["walltime"],
cputime_s=result["cputime"],
memory_mib=result["memory"] / 2**20 if result["memory"] is not None else None,
termination=result["terminationreason"],
exitcode=result["exitcode"]["code"],
executor=self.name,
reliable=self.reliable,
)
def close(self) -> None:
for handle in list(self._running): # only left if interrupted
handle.terminate()
super().close()