"""CPU utilities library."""
+from random import choice
+
from robot.libraries.BuiltIn import BuiltIn
from resources.libraries.python.Constants import Constants
@staticmethod
def get_affinity_trex(
- node, if_key, tg_mtc=1, tg_dtc=1, tg_ltc=1):
+ node, if_key, tg_mtc=1, tg_dtc=1, tg_ltc=1, tg_dtc_offset=0):
"""Get affinity for T-Rex. Result will be used to pin T-Rex threads.
:param node: TG node.
:param tg_mtc: TG main thread count.
:param tg_dtc: TG dataplane thread count.
:param tg_ltc: TG latency thread count.
+ :param tg_dtc_offset: TG dataplane thread offset.
:type node: dict
:type if_key: str
:type tg_mtc: int
:type tg_dtc: int
:type tg_ltc: int
+ :type tg_dtc_offset: int
:returns: List of CPUs allocated to T-Rex including numa node.
:rtype: int, int, int, list
"""
smt_used=False)
threads = CpuUtils.cpu_slice_of_list_per_node(
- node, cpu_node, skip_cnt=tg_mtc, cpu_cnt=tg_dtc,
- smt_used=False)
+ node, cpu_node, skip_cnt=tg_mtc + tg_ltc + tg_dtc_offset,
+ cpu_cnt=tg_dtc, smt_used=False)
latency_thread_id = CpuUtils.cpu_slice_of_list_per_node(
- node, cpu_node, skip_cnt=tg_mtc + tg_dtc, cpu_cnt=tg_ltc,
- smt_used=False)
+ node, cpu_node, skip_cnt=tg_mtc, cpu_cnt=tg_ltc, smt_used=False)
return master_thread_id[0], latency_thread_id[0], cpu_node, threads
continue
# Number of Data Plane physical cores.
dp_cores_count = BuiltIn().get_variable_value(
- f"${{dp_cores_count}}", phy_cores
+ "${dp_cores_count}", phy_cores
)
# Number of Feature Plane physical cores.
fp_cores_count = BuiltIn().get_variable_value(
- f"${{fp_cores_count}}", phy_cores - dp_cores_count
+ "${fp_cores_count}", phy_cores - dp_cores_count
)
# Ratio between RX queues and data plane threads.
rxq_ratio = BuiltIn().get_variable_value(
- f"${{rxq_ratio}}", 1
+ "${rxq_ratio}", 1
)
dut_pf_keys = BuiltIn().get_variable_value(
)
# SMT override in case of non standard test cases.
smt_used = BuiltIn().get_variable_value(
- f"${{smt_used}}", CpuUtils.is_smt_enabled(node["cpuinfo"])
+ "${smt_used}", CpuUtils.is_smt_enabled(node["cpuinfo"])
)
cpu_node = Topology.get_interfaces_numa_node(node, *dut_pf_keys)
cpu_main = CpuUtils.cpu_list_per_node_str(
node, cpu_node,
skip_cnt=skip_cnt,
- cpu_cnt=Constants.CPU_CNT_MAIN,
+ cpu_cnt=Constants.CPU_CNT_MAIN if phy_cores else 0,
smt_used=False
)
+ cpu_main = cpu_main if phy_cores else choice(cpu_main.split(","))
skip_cnt += Constants.CPU_CNT_MAIN
cpu_dp = CpuUtils.cpu_list_per_node_str(
node, cpu_node,