Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use run_testpmd_concurrent method in lisa

Best Python code snippet using lisa_python

dpdksuite.py

Source:dpdksuite.py

...272        # Want to only switch receiver sriov to avoid timing weirdness273        receiver.switch_sriov = True274        sender.switch_sriov = False275        kit_cmd_pairs = generate_send_receive_run_info("failsafe", sender, receiver)276        run_testpmd_concurrent(277            kit_cmd_pairs, DPDK_VF_REMOVAL_MAX_TEST_TIME, log, rescind_sriov=True278        )279        rescind_tx_pps_set = receiver.testpmd.get_mean_rx_pps_sriov_rescind()280        self._check_rx_or_tx_pps_sriov_rescind("RX", rescind_tx_pps_set)281    @TestCaseMetadata(282        description="""283            test sriov failsafe during vf revoke (send only version)284        """,285        priority=2,286        requirement=simple_requirement(287            min_core_count=8,288            min_nic_count=2,289            network_interface=Sriov(),290            unsupported_features=[Gpu, Infiniband],291            supported_features=[IsolatedResource],292        ),293    )294    def verify_dpdk_sriov_rescind_failover_send_only(295        self, node: Node, log: Logger, variables: Dict[str, Any]296    ) -> None:297        test_kit = initialize_node_resources(node, log, variables, "failsafe")298        testpmd = test_kit.testpmd299        test_nic = node.nics.get_nic_by_index()300        testpmd_cmd = testpmd.generate_testpmd_command(301            test_nic, 0, "txonly", "failsafe"302        )303        kit_cmd_pairs = {304            test_kit: testpmd_cmd,305        }306        run_testpmd_concurrent(307            kit_cmd_pairs, DPDK_VF_REMOVAL_MAX_TEST_TIME, log, rescind_sriov=True308        )309        rescind_tx_pps_set = testpmd.get_mean_tx_pps_sriov_rescind()310        self._check_rx_or_tx_pps_sriov_rescind("TX", rescind_tx_pps_set)311    def _check_rx_or_tx_pps_sriov_rescind(312        self, tx_or_rx: str, pps: Tuple[int, int, int]313    ) -> None:314        before_rescind, during_rescind, after_reenable = pps315        self._check_rx_or_tx_pps(tx_or_rx, before_rescind, sriov_enabled=True)316        self._check_rx_or_tx_pps(tx_or_rx, during_rescind, sriov_enabled=False)317        self._check_rx_or_tx_pps(tx_or_rx, after_reenable, sriov_enabled=True)318    def _check_rx_or_tx_pps(319        self, tx_or_rx: str, pps: int, sriov_enabled: bool = True320    ) -> None:...

dpdkutil.py

Source:dpdkutil.py

...262                "dpdk",263                kit.testpmd.get_dpdk_version(),264                "-tx-ip flag for ip forwarding",265            )266def run_testpmd_concurrent(267    node_cmd_pairs: Dict[DpdkTestResources, str],268    seconds: int,269    log: Logger,270    rescind_sriov: bool = False,271) -> Dict[DpdkTestResources, str]:272    output: Dict[DpdkTestResources, str] = dict()273    task_manager = start_testpmd_concurrent(node_cmd_pairs, seconds, log, output)274    if rescind_sriov:275        time.sleep(10)  # run testpmd for a bit before disabling sriov276        test_kits = node_cmd_pairs.keys()277        # disable sriov (and wait for change to apply)278        for node_resources in [x for x in test_kits if x.switch_sriov]:279            node_resources.nic_controller.switch_sriov(280                enable=False, wait=True, reset_connections=False281            )282        # let run for a bit with SRIOV disabled283        time.sleep(10)284        # re-enable sriov285        for node_resources in [x for x in test_kits if x.switch_sriov]:286            node_resources.nic_controller.switch_sriov(287                enable=True, wait=True, reset_connections=False288            )289        # run for a bit with SRIOV re-enabled290        time.sleep(10)291        # kill the commands to collect the output early and terminate before timeout292        for node_resources in test_kits:293            node_resources.testpmd.kill_previous_testpmd_command()294    task_manager.wait_for_all_workers()295    return output296def start_testpmd_concurrent(297    node_cmd_pairs: Dict[DpdkTestResources, str],298    seconds: int,299    log: Logger,300    output: Dict[DpdkTestResources, str],301) -> TaskManager[Tuple[DpdkTestResources, str]]:302    cmd_pairs_as_tuples = deque(node_cmd_pairs.items())303    def _collect_dict_result(result: Tuple[DpdkTestResources, str]) -> None:304        output[result[0]] = result[1]305    def _run_command_with_testkit(306        run_kit: Tuple[DpdkTestResources, str]307    ) -> Tuple[DpdkTestResources, str]:308        testkit, cmd = run_kit309        return (testkit, testkit.testpmd.run_for_n_seconds(cmd, seconds))310    task_manager = run_in_parallel_async(311        [partial(_run_command_with_testkit, x) for x in cmd_pairs_as_tuples],312        _collect_dict_result,313    )314    return task_manager315def init_nodes_concurrent(316    environment: Environment, log: Logger, variables: Dict[str, Any], pmd: str317) -> List[DpdkTestResources]:318    # Use threading module to parallelize the IO-bound node init.319    test_kits = run_in_parallel(320        [321            partial(initialize_node_resources, node, log, variables, pmd)322            for node in environment.nodes.list()323        ],324        log,325    )326    return test_kits327def verify_dpdk_build(328    node: Node,329    log: Logger,330    variables: Dict[str, Any],331    pmd: str,332) -> None:333    # setup and unwrap the resources for this test334    test_kit = initialize_node_resources(node, log, variables, pmd)335    testpmd = test_kit.testpmd336    # grab a nic and run testpmd337    test_nic = node.nics.get_nic_by_index()338    testpmd_cmd = testpmd.generate_testpmd_command(339        test_nic,340        0,341        "txonly",342        pmd,343    )344    testpmd.run_for_n_seconds(testpmd_cmd, 10)345    tx_pps = testpmd.get_mean_tx_pps()346    log.info(347        f"TX-PPS:{tx_pps} from {test_nic.upper}/{test_nic.lower}:"348        + f"{test_nic.pci_slot}"349    )350    assert_that(tx_pps).described_as(351        f"TX-PPS ({tx_pps}) should have been greater than 2^20 (~1m) PPS."352    ).is_greater_than(2**20)353def verify_dpdk_send_receive(354    environment: Environment,355    log: Logger,356    variables: Dict[str, Any],357    pmd: str,358    use_max_nics: bool = False,359    use_service_cores: int = 1,360) -> Tuple[DpdkTestResources, DpdkTestResources]:361    # helpful to have the public ips labeled for debugging362    external_ips = []363    for node in environment.nodes.list():364        if isinstance(node, RemoteNode):365            external_ips += node.connection_info[366                constants.ENVIRONMENTS_NODES_REMOTE_ADDRESS367            ]368        else:369            raise SkippedException()370    log.debug((f"\nsender:{external_ips[0]}\nreceiver:{external_ips[1]}\n"))371    test_kits = init_nodes_concurrent(environment, log, variables, pmd)372    check_send_receive_compatibility(test_kits)373    sender, receiver = test_kits374    kit_cmd_pairs = generate_send_receive_run_info(375        pmd,376        sender,377        receiver,378        use_max_nics=use_max_nics,379        use_service_cores=use_service_cores,380    )381    results = run_testpmd_concurrent(kit_cmd_pairs, 15, log)382    # helpful to have the outputs labeled383    log.debug(f"\nSENDER:\n{results[sender]}")384    log.debug(f"\nRECEIVER:\n{results[receiver]}")385    rcv_rx_pps = receiver.testpmd.get_mean_rx_pps()386    snd_tx_pps = sender.testpmd.get_mean_tx_pps()387    log.info(f"receiver rx-pps: {rcv_rx_pps}")388    log.info(f"sender tx-pps: {snd_tx_pps}")389    # differences in NIC type throughput can lead to different snd/rcv counts390    assert_that(rcv_rx_pps).described_as(391        "Throughput for RECEIVE was below the correct order-of-magnitude"392    ).is_greater_than(2**20)393    assert_that(snd_tx_pps).described_as(394        "Throughput for SEND was below the correct order of magnitude"395    ).is_greater_than(2**20)396    return sender, receiver397def verify_dpdk_send_receive_multi_txrx_queue(398    environment: Environment,399    log: Logger,400    variables: Dict[str, Any],401    pmd: str,402    use_max_nics: bool = False,403    use_service_cores: int = 1,404) -> Tuple[DpdkTestResources, DpdkTestResources]:405    test_kits = init_nodes_concurrent(environment, log, variables, pmd)406    check_send_receive_compatibility(test_kits)407    sender, receiver = test_kits408    kit_cmd_pairs = generate_send_receive_run_info(409        pmd,410        sender,411        receiver,412        txq=4,413        rxq=4,414        use_max_nics=use_max_nics,415        use_service_cores=use_service_cores,416    )417    results = run_testpmd_concurrent(kit_cmd_pairs, 15, log)418    # helpful to have the outputs labeled419    log.debug(f"\nSENDER:\n{results[sender]}")420    log.debug(f"\nRECEIVER:\n{results[receiver]}")421    rcv_rx_pps = receiver.testpmd.get_mean_rx_pps()422    snd_tx_pps = sender.testpmd.get_mean_tx_pps()423    log.info(f"receiver rx-pps: {rcv_rx_pps}")424    log.info(f"sender tx-pps: {snd_tx_pps}")425    # differences in NIC type throughput can lead to different snd/rcv counts426    # check that throughput it greater than 1m pps as a baseline427    assert_that(rcv_rx_pps).described_as(428        "Throughput for RECEIVE was below the correct order-of-magnitude"429    ).is_greater_than(2**20)430    assert_that(snd_tx_pps).described_as(431        "Throughput for SEND was below the correct order of magnitude"...

Automation Testing Tutorials

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.