- def _get_app_cpu(self):
- if self.CORES:
- return self.CORES
-
- vnf_cfg = self.scenario_helper.vnf_cfg
- sys_obj = CpuSysCores(self.ssh_helper)
- self.sys_cpu = sys_obj.get_core_socket()
- num_core = int(vnf_cfg["worker_threads"])
- if vnf_cfg.get("lb_config", "SW") == 'HW':
- num_core += self.HW_DEFAULT_CORE
- else:
- num_core += self.SW_DEFAULT_CORE
- app_cpu = self.sys_cpu[str(self.socket)][:num_core]
- return app_cpu
-
- def _get_cpu_sibling_list(self, cores=None):
- if cores is None:
- cores = self._get_app_cpu()
- sys_cmd_template = "%s/cpu%s/topology/thread_siblings_list"
- awk_template = "awk -F: '{ print $1 }' < %s"
- sys_path = "/sys/devices/system/cpu/"
- cpu_topology = []
- try:
- for core in cores:
- sys_cmd = sys_cmd_template % (sys_path, core)
- cpu_id = self.ssh_helper.execute(awk_template % sys_cmd)[1]
- cpu_topology.extend(cpu.strip() for cpu in cpu_id.split(','))
-
- return cpu_topology
- except Exception:
- return []
-
- def _validate_cpu_cfg(self):
- return self._get_cpu_sibling_list()
-