Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
else:
isolated = False
cpus = parse_cpu_list(cpus)
if set_cpu_affinity(cpus):
if self.args.verbose:
if isolated:
text = ("Pin process to isolated CPUs: %s"
% format_cpu_list(cpus))
else:
text = ("Pin process to CPUs: %s"
% format_cpu_list(cpus))
print(text)
if isolated:
self.args.affinity = format_cpu_list(cpus)
else:
if not isolated:
print("ERROR: CPU affinity not available.", file=sys.stderr)
print("Use Python 3.3 or newer, or install psutil dependency")
sys.exit(1)
elif not self.args.quiet:
print("WARNING: unable to pin worker processes to "
"isolated CPUs, CPU affinity not available")
print("Use Python 3.3 or newer, or install psutil dependency")
cpus = get_isolated_cpus()
if not cpus:
# no isolated CPUs or unable to get the isolated CPUs
return
else:
isolated = False
cpus = parse_cpu_list(cpus)
if set_cpu_affinity(cpus):
if self.args.verbose:
if isolated:
text = ("Pin process to isolated CPUs: %s"
% format_cpu_list(cpus))
else:
text = ("Pin process to CPUs: %s"
% format_cpu_list(cpus))
print(text)
if isolated:
self.args.affinity = format_cpu_list(cpus)
else:
if not isolated:
print("ERROR: CPU affinity not available.", file=sys.stderr)
print("Use Python 3.3 or newer, or install psutil dependency")
sys.exit(1)
elif not self.args.quiet:
print("WARNING: unable to pin worker processes to "
"isolated CPUs, CPU affinity not available")
print("Use Python 3.3 or newer, or install psutil dependency")
def collect_cpu_affinity(metadata, cpu_affinity, cpu_count):
if not cpu_affinity:
return
if not cpu_count:
return
# CPU affinity
if set(cpu_affinity) == set(range(cpu_count)):
return
metadata['cpu_affinity'] = format_cpu_list(cpu_affinity)
affinity = self.read_irqs_affinity()
modified = []
for irq in self.get_irqs():
if self.permission_error:
break
cpus = affinity.get(irq)
if new_cpus == cpus:
continue
if self.write_irq(irq, new_cpus):
modified.append(irq)
if modified:
self.log_action("Set affinity of IRQ %s to CPU %s"
% (format_cpu_list(modified), format_cpu_list(new_cpus)))
def check_isolcpus(self):
isolated = get_isolated_cpus()
if isolated:
self.log_state('Isolated CPUs (%s/%s): %s'
% (len(isolated), self.ncpu,
format_cpu_list(isolated)))
elif self.ncpu > 1:
self.log_state('No CPU is isolated')
self.advice('Use isolcpus= kernel parameter '
'to isolate CPUs')
def write_default(self, new_affinity):
default_smp_affinity = self.read_default_affinity()
if new_affinity == default_smp_affinity:
return
mask = format_cpus_as_mask(new_affinity)
try:
write_text(self.default_affinity_path, mask)
except IOError as exc:
self.check_permission_error(exc)
self.error("Failed to write %r into %s: %s"
% (mask, self.default_affinity_path, exc))
else:
self.log_action("Set default affinity to CPU %s"
% format_cpu_list(new_affinity))