Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _packable(cls, node):
"""
:returns: Whether it makes sense to pack a node into a common entry
"""
return (SchedProcFSNode._ends_with_digits(node) and
SchedProcFSNode._node_name(node) in cls.PACKABLE_ENTRIES)
class SchedDomain(SchedProcFSNode):
"""
Represents a sched domain as seen through procfs
"""
def __init__(self, nodes):
super(SchedDomain, self).__init__(nodes)
obj_flags = set()
for flag in list(SchedDomainFlag):
if self.flags & flag.value == flag.value:
obj_flags.add(flag)
self.flags = obj_flags
class SchedProcFSData(SchedProcFSNode):
"""
Root class for creating & storing SchedProcFSNode instances
"""
_read_depth = 6
sched_domain_root = '/proc/sys/kernel/sched_domain'
@staticmethod
def available(target):
path = SchedProcFSData.sched_domain_root
cpus = target.list_directory(path) if target.file_exists(path) else []
if not cpus:
return False
# Even if we have a CPU entry, it can be empty (e.g. hotplugged out)
# Make sure some data is there
def _build_node(node_name, node_data):
if isinstance(node_data, dict):
return SchedProcFSNode._build_directory(node_name, node_data)
else:
return SchedProcFSNode._build_entry(node_data)
def _build_directory(node_name, node_data):
if node_name.startswith("domain"):
return SchedDomain(node_data)
else:
return SchedProcFSNode(node_data)
def __init__(self, nodes):
self.procfs = nodes
# First, reduce the procs fields by packing them if possible
# Find which entries can be packed into a common entry
packables = {
node : SchedProcFSNode._node_name(node) + "s"
for node in list(nodes.keys()) if SchedProcFSNode._packable(node)
}
self._dyn_attrs = {}
for dest in set(packables.values()):
self._dyn_attrs[dest] = {}
# Pack common entries
for key, dest in packables.items():
i = SchedProcFSNode._node_digits(key)
self._dyn_attrs[dest][i] = self._build_node(key, nodes[key])
# Build the other nodes
for key in nodes.keys():
if key in packables:
continue
def _node_name(node):
"""
:returns: The name of the procfs node
"""
match = re.search(SchedProcFSNode._re_procfs_node, node)
if match:
return match.group("name")
return node
def _packable(cls, node):
"""
:returns: Whether it makes sense to pack a node into a common entry
"""
return (SchedProcFSNode._ends_with_digits(node) and
SchedProcFSNode._node_name(node) in cls.PACKABLE_ENTRIES)
parts = target.kernel_version.parts
# Checked to be valid from v4.4
# Not saved as a class attribute else it'll be converted to an enum
ref_parts = (4, 4, 0)
if parts < ref_parts:
logger.warn(
"Sched domain flags are defined for kernels v{} and up, "
"but target is running v{}".format(ref_parts, parts)
)
def __str__(self):
return self.name
class SchedDomain(SchedProcFSNode):
"""
Represents a sched domain as seen through procfs
"""
def __init__(self, nodes):
super(SchedDomain, self).__init__(nodes)
obj_flags = set()
for flag in list(SchedDomainFlag):
if self.flags & flag.value == flag.value:
obj_flags.add(flag)
self.flags = obj_flags
class SchedProcFSData(SchedProcFSNode):
"""