Hi �ngel,
On Fri, Jan 17, 2025 at 01:14:04AM +0100, �ngel wrote:
The script looks good, and easy to read. It wouldn't be hard to
translate it to another language if needed to drop the python
dependency (but that would increase the nloc)
thank you for the detailed review. I also picked up Julien's request for detecting cores rater than threads and attach an updated version. I
suppose moving to some git would be good.
https://salsa.debian.org/helmutg/guess_concurrency
To get things going, I created it in my personal namespace, but I'm
happy to move it to debian or elsewhere.
I find this behavior a bit surprising:
$ python3 guess_concurrency.py --min 10 --max 2
10
If there is a minimum limit, it is returned, even if that violates the
max. It makes some sense to pick something but I as actually expecting
an error to the above.
How could one disagree with that? Fixed.
The order of processing the cpus is a bit awkward as well.
The order it uses is CMAKE_BUILD_PARALLEL_LEVEL, DEB_BUILD_OPTIONS, RPM_BUILD_NCPUS, --detect <n>, nproc/os.cpu_count()
But the order in the code is 4, 5, 3, 2, 1
Not straightforward.
Again, how would I disagree? I have changed the order of invocation to
match the order of preference and in that process also changed
--detect=N to take precendence over all environment variables while
still preferring environment variables over other detectors. Hope that
makes sense.
Also, it is doing actions such as running external program nproc even
it if's going to be discarded later. (nproc is in an essential package,
I know, but still)
This is fixed as a result of your earlier remark.
Also, why would the user want to manually specify between nptoc and os.cpu_count()?
While nproc is universally available on Debian, I anticipate that the
tool could be useful on non-Linux platforms and there os.cpu_count() may
be more portable. I mainly added it due to the low effort and to
demonstrate the ability to use different detection methods. The values I
see users pass to --detect are actual numbers or "cores".
I would unconditionally call nproc, with a fallback to os.cpu_count()
if that fails (I'm assuming nproc may be smarter than os.cpu_count(), otherwise one could use cpu_count() always)
Indeed nproc kinda is smarter as it honours some environment variables
such as OMP_NUM_THREADS and OMP_THREAD_LIMIT. Not sure we need that
fallback.
I suggest doing:
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--count",
action="store",
default=None,
metavar="NUMBER",
help="supply a processor count",
)
Is there a reason to rename it from --detect to --count? The advantage
of reusing --detect I see is that you cannot reasonably combine those
two options and --detect=9 intuitively makes sufficient sense to me.
args = parser.parse_args()
guess = None
try:
if args.count:
guess = positive_integer(args.count)
except ValueError:
parser.error("invalid argument to --count")
guess = guess or guess_from_environment("CMAKE_BUILD_PARALLEL_LEVEL")
guess = guess or guess_deb_build_parallel()
guess = guess or guess_from_environment("RPM_BUILD_NCPUS")
if not guess:
try:
guess = guess_nproc()
finally:
guess = guess or guess_python()
I see three main proposal in this code:
* Order of preference reflects order of code. (implemented)
* A given count overrides all other detection mechanisms. (implemented)
* Fallback from nproc to os.cpucount(). (not convinced)
Did I miss anything?
I'm more inclined to drop --detect=python entirely as you say its use is
fairly limited.
Additionally, the --ignore argument of nproc(1) might be of use for
this script as well.
Can you give a practical use case for this?
Helmut
#!/usr/bin/python3
# Copyright 2024 Helmut Grohne <
[email protected]>
# SPDX-License-Identifier: GPL-3
import argparse
import itertools
import os
import pathlib
import re
import subprocess
import sys
import typing
def positive_integer(decstr: str) -> int:
"""Parse a positive, integral number from a string and raise a ValueError
otherwise.
>>> positive_integer(-1)
Traceback (most recent call last):
...
ValueError: integer must be positive
>>> positive_integer(0)
Traceback (most recent call last):
...
ValueError: integer must be positive
>>> positive_integer(1)
1
"""
value = int(decstr) # raises ValueError
if value < 1:
raise ValueError("integer must be positive")
return value
def parse_size(expression: str) -> int:
"""Parse an expression representing a data size with an optional unit
suffix into an integer. Raises a ValueError on failure.
>>> parse_size("5")
5
>>> parse_size("4KB")
4096
>>> parse_size("0.9g")
966367641
>>> parse_size("-1")
Traceback (most recent call last):
...
ValueError: number must be positive
"""
expression = expression.lower()
if expression.endswith("b"):
expression = expression[:-1]
suffixes = {
"k": 2**10,
"m": 2**20,
"g": 2**30,
"t": 2**40,
"e": 2**50,
}
factor = 1
if expression[-1:] in suffixes:
factor = suffixes[expression[-1]]
expression = expression[:-1]
fval = float(expression) # propagate ValueError
value = int(fval * factor) # propagate ValueError
if value < 1:
raise ValueError("number must be positive")
return value
def guess_python() -> int | None:
"""Estimate the number of processors using Python's os.cpu_count().
>>> guess_python() > 0
True
"""
return os.cpu_count()
def guess_nproc() -> int:
"""Estimate number of processors using coreutils' nproc.
>>> guess_nproc() > 0
True
"""
return positive_integer(
subprocess.check_output(["nproc"], encoding="ascii")
)
def guess_cores() -> int | None:
"""Estimate the number of cores (not SMT threads) using /proc/cpuinfo.
This is done by counting the number of distinct "core id" values.
"""
cpus = set()
try:
with open("/proc/cpuinfo", encoding="utf8") as cf:
for line in cf:
if m := re.match(r"^core id\s+:\s+(\d+)$", line, re.ASCII):
cpus.add(int(m.group(1)))
except FileNotFoundError:
return None
if not cpus:
return None
return len(cpus)
def guess_deb_build_parallel(
environ: typing.Mapping[str, str] = os.environ
) -> int | None:
"""Parse a possible parallel= assignment in a DEB_BUILD_OPTIONS environment
variable.
>>> guess_deb_build_parallel({})
>>> guess_deb_build_parallel({"DEB_BUILD_OPTIONS": "nocheck parallel=3"})
3
"""
try:
options = environ["DEB_BUILD_OPTIONS"]
except KeyError:
return None
for option in options.split():
if option.startswith("parallel="):
option = option.removeprefix("parallel=")
try:
return positive_integer(option)
except ValueError:
pass
return None
def guess_from_environment(
variable: str, environ: typing.Mapping[str, str] = os.environ
) -> int | None:
"""Read a number from an environment variable.
>>> guess_from_environment("CPUS", {"CPUS": 4})
4
>>> guess_from_environment("CPUS", {"other": 3})
"""
try:
return positive_integer(environ[variable])
except (KeyError, ValueError):
return None
def guess_memavailable() -> int:
"""Estimate the available memory from /proc/meminfo in bytes."""
with open("/proc/meminfo", encoding="ascii") as fh:
for line in fh:
if line.startswith("MemAvailable:"):
line = line.removeprefix("MemAvailable:").strip()
return 1024 * positive_integer(line.removesuffix("kB"))
raise RuntimeError("no MemAvailable line found in /proc/meminfo")
def guess_cgroup_memory() -> int | None:
"""Return the smalles "memory.high" or "memory.max" limit of the current
cgroup or any parent of it if any.
"""
guess: int | None = None
mygroup = pathlib.PurePath(
pathlib.Path("/proc/self/cgroup")
.read_text(encoding=sys.getfilesystemencoding())
.strip()
.split(":", 2)[2]
).relative_to("/")
sfc = pathlib.Path("/sys/fs/cgroup")
for group in itertools.chain((mygroup,), mygroup.parents):
for entry in ("memory.max", "memory.high"):
try:
value = positive_integer(
(sfc / group / entry).read_text(encoding="ascii")
)
except (FileNotFoundError, ValueError):
pass
else:
if guess is None:
guess = value
else:
guess = min(guess, value)
return guess
def parse_required_memory(expression: str) -> list[int]:
"""Parse comma-separated list of memory expressions. Empty expressions copy
the previous entry.
>>> parse_required_memory("1k,9,,1")
[1024, 9, 9, 1]
"""
values: list[int] = []
for memexpr in expression.split(","):
if not memexpr:
if values:
values.append(values[-1])
else:
raise ValueError("initial memory expression cannot be empty")
else:
values.append(parse_size(memexpr))
return values
def guess_memory_concurrency(memory: int, usage: list[int]) -> int:
"""Estimate the maximum number of cores that can be used given the
available memory and a sequence of per-core memory consumption.
>>> guess_memory_concurrency(4, [1])
4
>>> guess_memory_concurrency(10, [5, 4, 3])
2
>>> guess_memory_concurrency(2, [3])
1
"""
concurrency = 0
for use in usage[:-1]:
if use > memory:
break
memory -= use
concurrency += 1
else:
concurrency += memory // usage[-1]
return max(1, concurrency)
def clamp(
value: int, lower: int | None = None, upper: int | None = None
) -> int:
"""Return an adjusted value that does not exceed the lower or upper limits
if any.
>>> clamp(5, upper=4)
4
>>> clamp(9, 2)
9
"""
if upper is not None and upper < value:
value = upper
if lower is not None and lower > value:
value = lower
return value
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--detect",
action="store",
default="nproc",
metavar="METHOD",
help="supply a processor count or select a detection method"
"(nproc, python or cores)",
)
parser.add_argument(
"--max",
action="store",
type=positive_integer,
default=None,
metavar="N",
help="limit the number of detected cores to a given maximum",
)
parser.add_argument(
"--min",
action="store",
type=positive_integer,
default=None,
metavar="N",
help="limit the number of detected cores to a given minimum",
)
parser.add_argument(
"--require-mem",
action="store",
type=parse_required_memory,
default=[],
metavar="MEMLIST",
help="specify per-core required memory as a comma separated list",
)
args = parser.parse_args()
if args.min is not None and args.max is not None and args.min > args.max:
parser.error("--min value larger than --max value")
guess = None
detectfunc = None
for detector in args.detect.split(","):
try:
guess = positive_integer(detector)
except ValueError:
try:
detectfunc = {
"nproc": guess_nproc,
"python": guess_python,
"cores": guess_cores,
}[detector]
except KeyError:
parser.error("invalid argument to --detect")
if guess is None:
assert detectfunc is not None
guess = (
guess_from_environment("CMAKE_BUILD_PARALLEL_LEVEL")
or guess_deb_build_parallel()
or guess_from_environment("RPM_BUILD_NCPUS")
or detectfunc()
)
if guess is None:
print("failed to guess processor count", file=sys.stderr)
sys.exit(1)
if args.require_mem and guess > 1:
memory = clamp(guess_memavailable(), upper=guess_cgroup_memory())
guess = clamp(
guess, upper=guess_memory_concurrency(memory, args.require_mem)
)
guess = clamp(guess, args.min, args.max)
print(guess)
if __name__ == "__main__":
main()
--- SoupGate-Win32 v1.05
* Origin: fsxNet Usenet Gateway (21:1/5)