Best Python code snippet using lisa_python
dpdktestpmd.py
Source:dpdktestpmd.py
...152 )153 else:154 major, minor = map(int, [match.group("major"), match.group("minor")])155 self._dpdk_version_info: VersionInfo = VersionInfo(major, minor)156 def generate_testpmd_include(157 self, node_nic: NicInfo, vdev_id: int, use_max_nics: bool = False158 ) -> str:159 # handle generating different flags for pmds/device combos for testpmd160 # identify which nics to inlude in test, exclude others161 # TODO: Test does not make full use of multiple nics yet.162 if use_max_nics:163 self.node.log.warn(164 "NOTE: Testpmd suite does not yet make full use of multiple nics"165 )166 # if use_max_nics:167 # ssh_nic = self.node.nics.get_nic(self.node.nics.default_nic)168 # include_nics = [169 # self.node.nics.get_nic(nic)170 # for nic in self.node.nics.get_upper_nics()171 # if nic != ssh_nic.upper172 # ]173 # exclude_nics = [ssh_nic]174 # else:175 include_nics = [node_nic]176 exclude_nics = [177 self.node.nics.get_nic(nic)178 for nic in self.node.nics.get_upper_nics()179 if nic != node_nic.upper180 ]181 # build list of vdev info flags for each nic182 vdev_info = ""183 self.node.log.info(f"Running test with {len(include_nics)} nics.")184 for nic in include_nics:185 if self._dpdk_version_info and self._dpdk_version_info >= "18.11.0":186 vdev_name = "net_vdev_netvsc"187 vdev_flags = f"iface={nic.upper},force=1"188 else:189 vdev_name = "net_failsafe"190 vdev_flags = (191 f"dev({nic.pci_slot}),dev(net_tap0,iface={nic.upper},force=1)"192 )193 if nic.bound_driver == "hv_netvsc":194 vdev_info += f'--vdev="{vdev_name}{vdev_id},{vdev_flags}" '195 elif nic.bound_driver == "uio_hv_generic":196 pass197 else:198 fail(199 (200 f"Unknown driver({nic.bound_driver}) bound to "201 f"{nic.upper}/{nic.lower}."202 "Cannot generate testpmd include arguments."203 )204 )205 # exclude pci slots not associated with the test nic206 exclude_flags = ""207 for nic in exclude_nics:208 exclude_flags += f' -b "{nic.pci_slot}"'209 return vdev_info + exclude_flags210 def _calculate_core_count(211 self,212 cores_available: int,213 txq: int,214 rxq: int,215 use_max_nics: bool,216 service_cores: int = 1,217 ) -> int:218 # Use either:219 # - as many cores as are available, minus a core for the system220 # - 1 per queue on each nic + one per NIC PMD221 # this is a no-op for now,222 # test does not correctly handle multiple nics yet223 if use_max_nics:224 pass225 nics_available = 1226 return min(227 cores_available - 1,228 (nics_available * (txq + rxq)) + (nics_available * service_cores),229 )230 def generate_testpmd_command(231 self,232 nic_to_include: NicInfo,233 vdev_id: int,234 mode: str,235 pmd: str,236 extra_args: str = "",237 txq: int = 0,238 rxq: int = 0,239 service_cores: int = 1,240 use_max_nics: bool = False,241 ) -> str:242 # testpmd \243 # -l <core-list> \244 # -n <num of mem channels> \245 # -w <pci address of the device you plan to use> \246 # --vdev="net_vdev_netvsc<id>,iface=<the iface to attach to>" \247 # -- --port-topology=chained \248 # --nb-cores <number of cores to use for test pmd> \249 # --forward-mode=txonly \250 # --eth-peer=<port id>,<receiver peer MAC address> \251 # --stats-period <display interval in seconds>252 # if test asks for multicore, it implies using more than one nic253 # otherwise default core count for single nic will be used254 # and then adjusted for queue count255 if not (rxq or txq):256 txq = 1257 rxq = 1258 # calculate the available cores per numa node, infer the offset259 # required for core selection argument260 cores_available = self.node.tools[Lscpu].get_core_count()261 numa_node_count = self.node.tools[Lscpu].get_numa_node_count()262 nic_numa_node = self.node.nics._get_nic_numa_node(nic_to_include.lower)263 cores_per_numa = cores_available // numa_node_count264 numa_core_offset = cores_per_numa * nic_numa_node265 # calculate how many cores to use based on txq/rxq per nic and how many nics266 use_core_count = self._calculate_core_count(267 cores_per_numa, txq, rxq, use_max_nics, service_cores=service_cores268 )269 nic_include_info = self.generate_testpmd_include(270 nic_to_include, vdev_id, use_max_nics271 )272 # set up queue arguments273 if txq or rxq:274 # set number of queues to use for txq and rxq (per nic, default is 1)275 assert_that(txq).described_as(276 "TX queue value must be greater than 0 if txq is used"277 ).is_greater_than(0)278 assert_that(rxq).described_as(279 "RX queue value must be greater than 0 if rxq is used"280 ).is_greater_than(0)281 extra_args += f" --txq={txq} --rxq={rxq} "282 assert_that(use_core_count).described_as(283 "Selection asked for more cores than were available for numa "...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!