Commit 25b64e79 authored by Valentin Reis's avatar Valentin Reis

testing dict_values

parent 50946595
Pipeline #7873 failed with stages
in 2 minutes
......@@ -23,19 +23,19 @@ from nrm.subprograms import (
SingularityClient,
)
from typing import Any, NamedTuple
logger = logging.getLogger("nrm")
Container = namedtuple(
"Container",
[
"uuid",
"manifest",
"resources",
"power",
"processes",
"clientids",
"hwbindings",
],
)
class Container(NamedTuple):
uuid: Any
manifest: Any
resources: Any
power: Any
processes: Any
clientids: Any
hwbindings: Any
class ContainerManager(object):
......
......@@ -15,7 +15,18 @@ import xml.etree.ElementTree
import tornado.process as process # type: ignore
import subprocess
from typing import NamedTuple, List
from typing import (
Any,
Callable,
Dict,
Iterable,
List,
Sized,
Tuple,
Type,
TypeVar,
NamedTuple,
)
logger = logging.getLogger("nrm")
......@@ -215,6 +226,8 @@ class SingularityClient(object):
class ChrtClient(object):
flags: Dict[str, str]
prefix: str
"""Client to chrt command line wrapper."""
......@@ -227,11 +240,11 @@ class ChrtClient(object):
"SCHED_HPC": "--hpc",
}
def __init__(self):
def __init__(self) -> None:
"""Load configuration."""
self.prefix = "chrt"
def getwrappedcmd(self, params):
def getwrappedcmd(self, params) -> List[str]:
"""Return a list of args to prepend to a popen call."""
args = [self.prefix]
args.append(self.flags[params.policy])
......@@ -240,10 +253,11 @@ class ChrtClient(object):
class HwlocClient(object):
prefix: str
"""Client to hwloc binaries."""
def __init__(self, hwloc="hwloc"):
def __init__(self, hwloc="hwloc") -> None:
"""Load configuration."""
self.prefix = hwloc
......@@ -268,7 +282,7 @@ class HwlocClient(object):
ret.mems.append(0)
return ret
def all2fake(self, resources):
def all2fake(self, resources) -> str:
"""Convert resource description of the system into fake topology.
We need that because hwloc barfs on fake numa nodes.
......@@ -280,7 +294,9 @@ class HwlocClient(object):
pu = cpus // mems
return "numa: %s pu:%s".format(mems, pu)
def distrib(self, numprocs, restrict=None, fake=None):
def distrib(
self, numprocs, restrict=None, fake=None
) -> dict_values[Resources]:
"""Distribute numprocs across the hierarchy."""
# The original command only reports back cpusets. We do better, by
# reporting the mems that go with it. This requires some magic, using
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment