<div dir="ltr">This series has been applied to next-dts. <div> </div><div>I am noting here that patchwork CI results shows a checkpatches warning. However, I have run checkpatches on this patch and it's reporting no issues. My DPDK_CHECKPATCH_PATH and codespell dictionary seem fine. So, I am confident enough that there is no issue to apply to next-dts. Ali please let me know if you are concerned.</div></div> <div class="gmail_quote gmail_quote_container"><div dir="ltr" class="gmail_attr">On Fri, Aug 29, 2025 at 1:43 PM Paul Szczepanek <<a href="mailto:paul.szczepanek@arm.com">paul.szczepanek@arm.com</a>> wrote: </div><blockquote class="gmail_quote" style="margin:0px 0px 0px 0.8ex;border-left:1px solid rgb(204,204,204);padding-left:1ex">Testpmd moved into new API directory. Capabilities converted into vanilla enum and moved to API. Removed some ciruclar dependencies and incorrect imports. Signed-off-by: Paul Szczepanek <<a href="mailto:paul.szczepanek@arm.com" target="_blank">paul.szczepanek@arm.com</a>> --- doc/api/dts/api.capabilities.rst | 8 + doc/api/dts/api.rst | 20 + ...stpmd_shell.rst => api.testpmd.config.rst} | 4 +- doc/api/dts/api.testpmd.rst | 15 + doc/api/dts/api.testpmd.types.rst | 8 + doc/api/dts/framework.params.rst | 1 - doc/api/dts/framework.params.testpmd.rst | 8 - doc/api/dts/framework.remote_session.rst | 1 - doc/api/dts/index.rst | 1 + dts/api/__init__.py | 14 + dts/api/capabilities.py | 180 ++ dts/api/testpmd/__init__.py | 1294 ++++++++ .../testpmd.py => api/testpmd/config.py} | 9 +- dts/api/testpmd/types.py | 1406 ++++++++ dts/framework/config/__init__.py | 3 +- dts/framework/params/eal.py | 12 +- dts/framework/params/types.py | 4 +- dts/framework/remote_session/__init__.py | 44 - dts/framework/remote_session/testpmd_shell.py | 2844 ----------------- dts/framework/testbed_model/capability.py | 144 +- dts/framework/testbed_model/linux_session.py | 2 +- dts/framework/testbed_model/os_session.py | 14 +- dts/framework/testbed_model/topology.py | 30 +- 23 files changed, 3086 insertions(+), 2980 deletions(-) create mode 100644 doc/api/dts/api.capabilities.rst create mode 100644 doc/api/dts/api.rst rename doc/api/dts/{framework.remote_session.testpmd_shell.rst => api.testpmd.config.rst} (54%) create mode 100644 doc/api/dts/api.testpmd.rst create mode 100644 doc/api/dts/api.testpmd.types.rst delete mode 100644 doc/api/dts/framework.params.testpmd.rst create mode 100644 dts/api/__init__.py create mode 100644 dts/api/capabilities.py create mode 100644 dts/api/testpmd/__init__.py rename dts/{framework/params/testpmd.py => api/testpmd/config.py} (98%) create mode 100644 dts/api/testpmd/types.py delete mode 100644 dts/framework/remote_session/testpmd_shell.py diff --git a/doc/api/dts/api.capabilities.rst b/doc/api/dts/api.capabilities.rst new file mode 100644 index 0000000000..311872f61d --- /dev/null +++ b/doc/api/dts/api.capabilities.rst @@ -0,0 +1,8 @@ +.. SPDX-License-Identifier: BSD-3-Clause + +capabilities - SUT Capabilities +========================================== + +.. automodule:: api.capabilities + :members: + :show-inheritance: diff --git a/doc/api/dts/api.rst b/doc/api/dts/api.rst new file mode 100644 index 0000000000..d3cf1226eb --- /dev/null +++ b/doc/api/dts/api.rst @@ -0,0 +1,20 @@ +.. SPDX-License-Identifier: BSD-3-Clause + +api - DTS API +========================================== + +.. automodule:: api + :members: + :show-inheritance: + +.. toctree:: + :hidden: + :maxdepth: 2 + + api.testpmd + +.. toctree:: + :hidden: + :maxdepth: 1 + + api.capabilities \ No newline at end of file diff --git a/doc/api/dts/framework.remote_session.testpmd_shell.rst b/doc/api/dts/api.testpmd.config.rst similarity index 54% rename from doc/api/dts/framework.remote_session.testpmd_shell.rst rename to doc/api/dts/api.testpmd.config.rst index 81ca23337f..d338c07a36 100644 --- a/doc/api/dts/framework.remote_session.testpmd_shell.rst +++ b/doc/api/dts/api.testpmd.config.rst @@ -1,8 +1,8 @@ .. SPDX-License-Identifier: BSD-3-Clause -testpmd\_shell - Testpmd Interactive Remote Shell +config - Testpmd configuration ================================================= -.. automodule:: framework.remote_session.testpmd_shell +.. automodule:: api.testpmd.config :members: :show-inheritance: diff --git a/doc/api/dts/api.testpmd.rst b/doc/api/dts/api.testpmd.rst new file mode 100644 index 0000000000..75a92823f8 --- /dev/null +++ b/doc/api/dts/api.testpmd.rst @@ -0,0 +1,15 @@ +.. SPDX-License-Identifier: BSD-3-Clause + +testpmd - Testpmd Interactive Remote Shell +================================================= + +.. automodule:: api.testpmd + :members: + :show-inheritance: + +.. toctree:: + :hidden: + :maxdepth: 1 + + api.testpmd.types + api.testpmd.config \ No newline at end of file diff --git a/doc/api/dts/api.testpmd.types.rst b/doc/api/dts/api.testpmd.types.rst new file mode 100644 index 0000000000..75b197aa73 --- /dev/null +++ b/doc/api/dts/api.testpmd.types.rst @@ -0,0 +1,8 @@ +.. SPDX-License-Identifier: BSD-3-Clause + +types - Testpmd types +================================================= + +.. automodule:: api.testpmd.types + :members: + :show-inheritance: diff --git a/doc/api/dts/framework.params.rst b/doc/api/dts/framework.params.rst index 4e263e2e5c..d8c6af9667 100644 --- a/doc/api/dts/framework.params.rst +++ b/doc/api/dts/framework.params.rst @@ -12,5 +12,4 @@ params - Command Line Parameters Modelling :maxdepth: 1 framework.params.eal - framework.params.testpmd framework.params.types diff --git a/doc/api/dts/framework.params.testpmd.rst b/doc/api/dts/framework.params.testpmd.rst deleted file mode 100644 index 19583b01de..0000000000 --- a/doc/api/dts/framework.params.testpmd.rst +++ /dev/null @@ -1,8 +0,0 @@ -.. SPDX-License-Identifier: BSD-3-Clause - -testpmd - TestPMD Parameters Modelling -====================================== - -.. automodule:: framework.params.testpmd - :members: - :show-inheritance: diff --git a/doc/api/dts/framework.remote_session.rst b/doc/api/dts/framework.remote_session.rst index 27c9153e64..b7dbe71412 100644 --- a/doc/api/dts/framework.remote_session.rst +++ b/doc/api/dts/framework.remote_session.rst @@ -18,5 +18,4 @@ remote\_session - Node Connections Package framework.remote_session.shell_pool framework.remote_session.dpdk framework.remote_session.dpdk_shell - framework.remote_session.testpmd_shell framework.remote_session.python_shell diff --git a/doc/api/dts/index.rst b/doc/api/dts/index.rst index a11f395e11..c719297c11 100644 --- a/doc/api/dts/index.rst +++ b/doc/api/dts/index.rst @@ -15,6 +15,7 @@ Packages :maxdepth: 1 tests + api framework.testbed_model framework.remote_session framework.params diff --git a/dts/api/__init__.py b/dts/api/__init__.py new file mode 100644 index 0000000000..26b773bfbb --- /dev/null +++ b/dts/api/__init__.py @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2025 Arm Limited + +"""DTS API. + +This package exposes public API modules for test writers. + +All modules in this package are considered stable. Do not use framework +internal modules in your tests. Any missing functionality should be added +to the public API. + +Private methods and members are prefixed with an underscore and should not be +used outside of the framework. +""" diff --git a/dts/api/capabilities.py b/dts/api/capabilities.py new file mode 100644 index 0000000000..1a79413f6f --- /dev/null +++ b/dts/api/capabilities.py @@ -0,0 +1,180 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2024 PANTHEON.tech s.r.o. +# Copyright(c) 2025 Arm Limited + +"""Testbed capabilities. + +This module provides a protocol that defines the common attributes of test cases and suites +and support for test environment capabilities. + +Many test cases are testing features not available on all hardware. +On the other hand, some test cases or suites may not need the most complex topology available. + +The module allows developers to mark test cases or suites to require certain hardware capabilities +or a particular topology. + +There are differences between hardware and topology capabilities: + + * Hardware capabilities are assumed to not be required when not specified. + * However, some topology is always available, so each test case or suite is assigned + a default topology if no topology is specified in the decorator. + +Examples: + .. code:: python + + from framework.test_suite import TestSuite, func_test + from framework.testbed_model.capability import LinkTopology, requires_link_topology + # The whole test suite (each test case within) doesn't require any links. + @requires_link_topology(LinkTopology.NO_LINK) + @func_test + class TestHelloWorld(TestSuite): + def hello_world_single_core(self): + ... + + .. code:: python + + from framework.test_suite import TestSuite, func_test + from framework.testbed_model.capability import NicCapability, requires_nic_capability + class TestPmdBufferScatter(TestSuite): + # only the test case requires the SCATTERED_RX_ENABLED capability + # other test cases may not require it + @requires_nic_capability(NicCapability.SCATTERED_RX_ENABLED) + @func_test + def test_scatter_mbuf_2048(self): +""" + +from enum import IntEnum, auto +from typing import TYPE_CHECKING, Callable + +if TYPE_CHECKING: + from framework.test_suite import TestProtocol + + +class LinkTopology(IntEnum): + """Supported topology types.""" + + #: A topology with no Traffic Generator. + NO_LINK = 0 + #: A topology with one physical link between the SUT node and the TG node. + ONE_LINK = auto() + #: A topology with two physical links between the Sut node and the TG node. + TWO_LINKS = auto() + + @classmethod + def default(cls) -> "LinkTopology": + """The default topology required by test cases if not specified otherwise.""" + return cls.TWO_LINKS + + +class NicCapability(IntEnum): + """DPDK NIC capabilities. + + The capabilities are used to mark test cases or suites that require a specific + DPDK NIC capability to run. The capabilities are used by the test framework to + determine whether a test case or suite can be run on the current testbed. + """ + + #: Scattered packets Rx enabled. + SCATTERED_RX_ENABLED = 0 + #: Device supports VLAN stripping. + RX_OFFLOAD_VLAN_STRIP = auto() + #: Device supports L3 checksum offload. + RX_OFFLOAD_IPV4_CKSUM = auto() + #: Device supports L4 checksum offload. + RX_OFFLOAD_UDP_CKSUM = auto() + #: Device supports L4 checksum offload. + RX_OFFLOAD_TCP_CKSUM = auto() + #: Device supports Large Receive Offload. + RX_OFFLOAD_TCP_LRO = auto() + #: Device supports QinQ (queue in queue) offload. + RX_OFFLOAD_QINQ_STRIP = auto() + #: Device supports inner packet L3 checksum. + RX_OFFLOAD_OUTER_IPV4_CKSUM = auto() + #: Device supports MACsec. + RX_OFFLOAD_MACSEC_STRIP = auto() + #: Device supports filtering of a VLAN Tag identifier. + RX_OFFLOAD_VLAN_FILTER = auto() + #: Device supports VLAN offload. + RX_OFFLOAD_VLAN_EXTEND = auto() + #: Device supports receiving segmented mbufs. + RX_OFFLOAD_SCATTER = auto() + #: Device supports Timestamp. + RX_OFFLOAD_TIMESTAMP = auto() + #: Device supports crypto processing while packet is received in NIC. + RX_OFFLOAD_SECURITY = auto() + #: Device supports CRC stripping. + RX_OFFLOAD_KEEP_CRC = auto() + #: Device supports L4 checksum offload. + RX_OFFLOAD_SCTP_CKSUM = auto() + #: Device supports inner packet L4 checksum. + RX_OFFLOAD_OUTER_UDP_CKSUM = auto() + #: Device supports RSS hashing. + RX_OFFLOAD_RSS_HASH = auto() + #: Device supports scatter Rx packets to segmented mbufs. + RX_OFFLOAD_BUFFER_SPLIT = auto() + #: Device supports all checksum capabilities. + RX_OFFLOAD_CHECKSUM = auto() + #: Device supports all VLAN capabilities. + RX_OFFLOAD_VLAN = auto() + #: Device supports Rx queue setup after device started. + RUNTIME_RX_QUEUE_SETUP = auto() + #: Device supports Tx queue setup after device started. + RUNTIME_TX_QUEUE_SETUP = auto() + #: Device supports shared Rx queue among ports within Rx domain and switch domain. + RXQ_SHARE = auto() + #: Device supports keeping flow rules across restart. + FLOW_RULE_KEEP = auto() + #: Device supports keeping shared flow objects across restart. + FLOW_SHARED_OBJECT_KEEP = auto() + #: Device supports multicast address filtering. + MCAST_FILTERING = auto() + #: Device supports flow ctrl. + FLOW_CTRL = auto() + #: Device is running on a physical function. + PHYSICAL_FUNCTION = auto() + + +def requires_link_topology( + link_topology: LinkTopology, +) -> Callable[[type["TestProtocol"]], type["TestProtocol"]]: + """Decorator to set required topology type for a test case or test suite. + + Args: + link_topology: The topology type the test suite or case requires. + + Returns: + The decorated test case or test suite. + """ + from framework.testbed_model.capability import TopologyCapability + + def add_required_topology( + test_case_or_suite: type["TestProtocol"], + ) -> type["TestProtocol"]: + topology_capability = TopologyCapability.get_unique(link_topology) + topology_capability.set_required(test_case_or_suite) + return test_case_or_suite + + return add_required_topology + + +def requires_nic_capability( + nic_capability: NicCapability, +) -> Callable[[type["TestProtocol"]], type["TestProtocol"]]: + """Decorator to add a single required NIC capability to a test case or test suite. + + Args: + nic_capability: The NIC capability that is required by the test case or test suite. + + Returns: + The decorated test case or test suite. + """ + from framework.testbed_model.capability import DecoratedNicCapability + + def add_required_capability( + test_case_or_suite: type["TestProtocol"], + ) -> type["TestProtocol"]: + decorated = DecoratedNicCapability.get_unique(nic_capability) + decorated.add_to_required(test_case_or_suite) + return test_case_or_suite + + return add_required_capability diff --git a/dts/api/testpmd/__init__.py b/dts/api/testpmd/__init__.py new file mode 100644 index 0000000000..49cf3742dd --- /dev/null +++ b/dts/api/testpmd/__init__.py @@ -0,0 +1,1294 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2023 University of New Hampshire +# Copyright(c) 2023 PANTHEON.tech s.r.o. +# Copyright(c) 2024 Arm Limited + +"""Testpmd interactive shell. + +Typical usage example in a TestSuite:: + + testpmd = TestPmd(self.sut_node) + devices = testpmd.get_devices() + for device in devices: + print(device) + testpmd.close() +""" + +import functools +import re +import time +from collections.abc import MutableSet +from enum import Flag +from pathlib import PurePath +from typing import ( + Any, + Callable, + ClassVar, + Concatenate, + ParamSpec, + Tuple, +) + +from typing_extensions import Unpack + +from api.capabilities import LinkTopology, NicCapability +from api.testpmd.config import PortTopology, SimpleForwardingModes, TestPmdParams +from api.testpmd.types import ( + ChecksumOffloadOptions, + DeviceCapabilitiesFlag, + FlowRule, + RxOffloadCapabilities, + RxOffloadCapability, + TestPmdDevice, + TestPmdPort, + TestPmdPortFlowCtrl, + TestPmdPortStats, + TestPmdQueueInfo, + TestPmdRxqInfo, + TestPmdVerbosePacket, + VLANOffloadFlag, +) +from framework.context import get_ctx +from framework.exception import InteractiveCommandExecutionError, InternalError +from framework.params.types import TestPmdParamsDict +from framework.remote_session.dpdk_shell import DPDKShell +from framework.remote_session.interactive_shell import only_active +from framework.settings import SETTINGS + +P = ParamSpec("P") +TestPmdMethod = Callable[Concatenate["TestPmd", P], Any] + + +def _requires_stopped_ports(func: TestPmdMethod) -> TestPmdMethod: + """Decorator for :class:`TestPmd` commands methods that require stopped ports. + + If the decorated method is called while the ports are started, then these are stopped before + continuing. + + Args: + func: The :class:`TestPmd` method to decorate. + """ + + @functools.wraps(func) + def _wrapper(self: "TestPmd", *args: P.args, **kwargs: P.kwargs): + if self.ports_started: + self._logger.debug("Ports need to be stopped to continue.") + self.stop_all_ports() + + return func(self, *args, **kwargs) + + return _wrapper + + +def _requires_started_ports(func: TestPmdMethod) -> TestPmdMethod: + """Decorator for :class:`TestPmd` commands methods that require started ports. + + If the decorated method is called while the ports are stopped, then these are started before + continuing. + + Args: + func: The :class:`TestPmd` method to decorate. + """ + + @functools.wraps(func) + def _wrapper(self: "TestPmd", *args: P.args, **kwargs: P.kwargs): + if not self.ports_started: + self._logger.debug("Ports need to be started to continue.") + self.start_all_ports() + + return func(self, *args, **kwargs) + + return _wrapper + + +def _add_remove_mtu(mtu: int = 1500) -> Callable[[TestPmdMethod], TestPmdMethod]: + """Configure MTU to `mtu` on all ports, run the decorated function, then revert. + + Args: + mtu: The MTU to configure all ports on. + + Returns: + The method decorated with setting and reverting MTU. + """ + + def decorator(func: TestPmdMethod) -> TestPmdMethod: + @functools.wraps(func) + def wrapper(self: "TestPmd", *args: P.args, **kwargs: P.kwargs): + original_mtu = self.ports[0].mtu + self.set_port_mtu_all(mtu=mtu, verify=False) + retval = func(self, *args, **kwargs) + self.set_port_mtu_all(original_mtu if original_mtu else 1500, verify=False) + return retval + + return wrapper + + return decorator + + +class TestPmd(DPDKShell): + """Testpmd interactive shell. + + The testpmd shell users should never use + the :meth:`~.interactive_shell.InteractiveShell.send_command` method directly, but rather + call specialized methods. If there isn't one that satisfies a need, it should be added. + + Attributes: + ports_started: Indicates whether the ports are started. + """ + + _app_params: TestPmdParams + _ports: list[TestPmdPort] | None + + #: The testpmd's prompt. + _default_prompt: ClassVar[str] = "testpmd>" + + #: This forces the prompt to appear after sending a command. + _command_extra_chars: ClassVar[str] = "\n" + + ports_started: bool + + def __init__( + self, + name: str | None = None, + privileged: bool = True, + **app_params: Unpack[TestPmdParamsDict], + ) -> None: + """Overrides :meth:`~.dpdk_shell.DPDKShell.__init__`. Changes app_params to kwargs.""" + if "port_topology" not in app_params and get_ctx().topology.type is LinkTopology.ONE_LINK: + app_params["port_topology"] = PortTopology.loop + super().__init__(name, privileged, app_params=TestPmdParams(**app_params)) + self.ports_started = not self._app_params.disable_device_start + self._ports = None + + @property + def path(self) -> PurePath: + """The path to the testpmd executable.""" + return PurePath("app/dpdk-testpmd") + + @property + def ports(self) -> list[TestPmdPort]: + """The ports of the instance. + + This caches the ports returned by :meth:`show_port_info_all`. + To force an update of port information, execute :meth:`show_port_info_all` or + :meth:`show_port_info`. + + Returns: The list of known testpmd ports. + """ + if self._ports is None: + return self.show_port_info_all() + return self._ports + + @_requires_started_ports + def start(self, verify: bool = True) -> None: + """Start packet forwarding with the current configuration. + + Args: + verify: If :data:`True` , a second start command will be sent in an attempt to verify + packet forwarding started as expected. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and forwarding fails to + start or ports fail to come up. + """ + self.send_command("start") + if verify: + # If forwarding was already started, sending "start" again should tell us + start_cmd_output = self.send_command("start") + if "Packet forwarding already started" not in start_cmd_output: + self._logger.debug(f"Failed to start packet forwarding: \n{start_cmd_output}") + raise InteractiveCommandExecutionError("Testpmd failed to start packet forwarding.") + + def stop(self, verify: bool = True) -> str: + """Stop packet forwarding. + + Args: + verify: If :data:`True` , the output of the stop command is scanned to verify that + forwarding was stopped successfully or not started. If neither is found, it is + considered an error. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the command to stop + forwarding results in an error. + + Returns: + Output gathered from the stop command and all other preceding logs in the buffer. This + output is most often used to view forwarding statistics that are displayed when this + command is sent as well as any verbose packet information that hasn't been consumed + prior to calling this method. + """ + stop_cmd_output = self.send_command("stop") + if verify: + if ( + "Done." not in stop_cmd_output + and "Packet forwarding not started" not in stop_cmd_output + ): + self._logger.debug(f"Failed to stop packet forwarding: \n{stop_cmd_output}") + raise InteractiveCommandExecutionError("Testpmd failed to stop packet forwarding.") + return stop_cmd_output + + def get_devices(self) -> list[TestPmdDevice]: + """Get a list of device names that are known to testpmd. + + Uses the device info listed in testpmd and then parses the output. + + Returns: + A list of devices. + """ + dev_info: str = self.send_command("show device info all") + dev_list: list[TestPmdDevice] = [] + for line in dev_info.split("\n"): + if "device name:" in line.lower(): + dev_list.append(TestPmdDevice(line)) + return dev_list + + def wait_link_status_up(self, port_id: int, timeout=SETTINGS.timeout) -> bool: + """Wait until the link status on the given port is "up". + + Arguments: + port_id: Port to check the link status on. + timeout: Time to wait for the link to come up. The default value for this + argument may be modified using the :option:`--timeout` command-line argument + or the :envvar:`DTS_TIMEOUT` environment variable. + + Returns: + Whether the link came up in time or not. + """ + time_to_stop = time.time() + timeout + port_info: str = "" + while time.time() < time_to_stop: + port_info = self.send_command(f"show port info {port_id}") + if "Link status: up" in port_info: + break + time.sleep(0.5) + else: + self._logger.error(f"The link for port {port_id} did not come up in the given timeout.") + return "Link status: up" in port_info + + def set_forward_mode(self, mode: SimpleForwardingModes, verify: bool = True): + """Set packet forwarding mode. + + Args: + mode: The forwarding mode to use. + verify: If :data:`True` the output of the command will be scanned in an attempt to + verify that the forwarding mode was set to `mode` properly. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the forwarding mode + fails to update. + """ + set_fwd_output = self.send_command(f"set fwd {mode.value}") + if verify: + if f"Set {mode.value} packet forwarding mode" not in set_fwd_output: + self._logger.debug(f"Failed to set fwd mode to {mode.value}:\n{set_fwd_output}") + raise InteractiveCommandExecutionError( + f"Test pmd failed to set fwd mode to {mode.value}" + ) + + def stop_all_ports(self, verify: bool = True) -> None: + """Stops all the ports. + + Args: + verify: If :data:`True`, the output of the command will be checked for a successful + execution. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the ports were not + stopped successfully. + """ + self._logger.debug("Stopping all the ports...") + output = self.send_command("port stop all") + if verify and not output.strip().endswith("Done"): + raise InteractiveCommandExecutionError("Ports were not stopped successfully.") + + self.ports_started = False + + def start_all_ports(self, verify: bool = True) -> None: + """Starts all the ports. + + Args: + verify: If :data:`True`, the output of the command will be checked for a successful + execution. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the ports were not + started successfully. + """ + self._logger.debug("Starting all the ports...") + output = self.send_command("port start all") + if verify and not output.strip().endswith("Done"): + raise InteractiveCommandExecutionError("Ports were not started successfully.") + + self.ports_started = True + + @_requires_stopped_ports + def set_ports_queues(self, number_of: int) -> None: + """Sets the number of queues per port. + + Args: + number_of: The number of RX/TX queues to create per port. + + Raises: + InternalError: If `number_of` is invalid. + """ + if number_of < 1: + raise InternalError("The number of queues must be positive and non-zero.") + + self.send_command(f"port config all rxq {number_of}") + self.send_command(f"port config all txq {number_of}") + + @_requires_stopped_ports + def close_all_ports(self, verify: bool = True) -> None: + """Close all ports. + + Args: + verify: If :data:`True` the output of the close command will be scanned in an attempt + to verify that all ports were stopped successfully. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and at lease one port + failed to close. + """ + port_close_output = self.send_command("port close all") + if verify: + num_ports = len(self.ports) + if not all(f"Port {p_id} is closed" in port_close_output for p_id in range(num_ports)): + raise InteractiveCommandExecutionError("Ports were not closed successfully.") + + def show_port_info_all(self) -> list[TestPmdPort]: + """Returns the information of all the ports. + + Returns: + list[TestPmdPort]: A list containing all the ports information as `TestPmdPort`. + """ + output = self.send_command("show port info all") + + # Sample output of the "all" command looks like: + # + # <start> + # + # ********************* Infos for port 0 ********************* + # Key: value + # + # ********************* Infos for port 1 ********************* + # Key: value + # <end> + # + # Takes advantage of the double new line in between ports as end delimiter. But we need to + # artificially add a new line at the end to pick up the last port. Because commands are + # executed on a pseudo-terminal created by paramiko on the remote node, lines end with CRLF. + # Therefore we also need to take the carriage return into account. + iter = re.finditer(r"\*{21}.*?[\r\n]{4}", output + "\r\n", re.S) + self._ports = [TestPmdPort.parse(block.group(0)) for block in iter] + return self._ports + + def show_port_info(self, port_id: int) -> TestPmdPort: + """Returns the given port information. + + Args: + port_id: The port ID to gather information for. + + Raises: + InteractiveCommandExecutionError: If `port_id` is invalid. + + Returns: + TestPmdPort: An instance of `TestPmdPort` containing the given port's information. + """ + output = self.send_command(f"show port info {port_id}", skip_first_line=True) + if output.startswith("Invalid port"): + raise InteractiveCommandExecutionError("invalid port given") + + port = TestPmdPort.parse(output) + self._update_port(port) + return port + + def _update_port(self, port: TestPmdPort) -> None: + if self._ports: + self._ports = [ + existing_port if <a href="http://port.id" rel="noreferrer" target="_blank">port.id</a> != <a href="http://existing_port.id" rel="noreferrer" target="_blank">existing_port.id</a> else port + for existing_port in self._ports + ] + + def set_mac_addr(self, port_id: int, mac_address: str, add: bool, verify: bool = True) -> None: + """Add or remove a mac address on a given port's Allowlist. + + Args: + port_id: The port ID the mac address is set on. + mac_address: The mac address to be added to or removed from the specified port. + add: If :data:`True`, add the specified mac address. If :data:`False`, remove specified + mac address. + verify: If :data:'True', assert that the 'mac_addr' operation was successful. If + :data:'False', run the command and skip this assertion. + + Raises: + InteractiveCommandExecutionError: If the set mac address operation fails. + """ + mac_cmd = "add" if add else "remove" + output = self.send_command(f"mac_addr {mac_cmd} {port_id} {mac_address}") + if "Bad arguments" in output: + self._logger.debug("Invalid argument provided to mac_addr") + raise InteractiveCommandExecutionError("Invalid argument provided") + + if verify: + if "mac_addr_cmd error:" in output: + self._logger.debug(f"Failed to {mac_cmd} {mac_address} on port {port_id}") + raise InteractiveCommandExecutionError( + f"Failed to {mac_cmd} {mac_address} on port {port_id} \n{output}" + ) + + def set_multicast_mac_addr( + self, port_id: int, multi_addr: str, add: bool, verify: bool = True + ) -> None: + """Add or remove multicast mac address to a specified port's allow list. + + Args: + port_id: The port ID the multicast address is set on. + multi_addr: The multicast address to be added or removed from the filter. + add: If :data:'True', add the specified multicast address to the port filter. + If :data:'False', remove the specified multicast address from the port filter. + verify: If :data:'True', assert that the 'mcast_addr' operations was successful. + If :data:'False', execute the 'mcast_addr' operation and skip the assertion. + + Raises: + InteractiveCommandExecutionError: If either the 'add' or 'remove' operations fails. + """ + mcast_cmd = "add" if add else "remove" + output = self.send_command(f"mcast_addr {mcast_cmd} {port_id} {multi_addr}") + if "Bad arguments" in output: + self._logger.debug("Invalid arguments provided to mcast_addr") + raise InteractiveCommandExecutionError("Invalid argument provided") + + if verify: + if ( + "Invalid multicast_addr" in output + or f"multicast address {'already' if add else 'not'} filtered by port" in output + ): + self._logger.debug(f"Failed to {mcast_cmd} {multi_addr} on port {port_id}") + raise InteractiveCommandExecutionError( + f"Failed to {mcast_cmd} {multi_addr} on port {port_id} \n{output}" + ) + + def show_port_stats_all(self) -> Tuple[list[TestPmdPortStats], str]: + """Returns the statistics of all the ports. + + Returns: + Tuple[str, list[TestPmdPortStats]]: A tuple where the first element is the stats of all + ports as `TestPmdPortStats` and second is the raw testpmd output that was collected + from the sent command. + """ + output = self.send_command("show port stats all") + + # Sample output of the "all" command looks like: + # + # ########### NIC statistics for port 0 ########### + # values... + # ################################################# + # + # ########### NIC statistics for port 1 ########### + # values... + # ################################################# + # + iter = re.finditer(r"(^ #*.+#*$[^#]+)^ #*\r$", output, re.MULTILINE) + return ([TestPmdPortStats.parse(block.group(1)) for block in iter], output) + + def show_port_stats(self, port_id: int) -> TestPmdPortStats: + """Returns the given port statistics. + + Args: + port_id: The port ID to gather information for. + + Raises: + InteractiveCommandExecutionError: If `port_id` is invalid. + + Returns: + TestPmdPortStats: An instance of `TestPmdPortStats` containing the given port's stats. + """ + output = self.send_command(f"show port stats {port_id}", skip_first_line=True) + if output.startswith("Invalid port"): + raise InteractiveCommandExecutionError("invalid port given") + + return TestPmdPortStats.parse(output) + + def set_multicast_all(self, on: bool, verify: bool = True) -> None: + """Turns multicast mode on/off for the specified port. + + Args: + on: If :data:`True`, turns multicast mode on, otherwise turns off. + verify: If :data:`True` an additional command will be sent to verify + that multicast mode is properly set. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and multicast + mode is not properly set. + """ + multicast_cmd_output = self.send_command(f"set allmulti all {'on' if on else 'off'}") + if verify: + port_stats = self.show_port_info_all() + if on ^ all(stats.is_allmulticast_mode_enabled for stats in port_stats): + self._logger.debug( + f"Failed to set multicast mode on all ports.: \n{multicast_cmd_output}" + ) + raise InteractiveCommandExecutionError( + "Testpmd failed to set multicast mode on all ports." + ) + + @_requires_stopped_ports + def csum_set_hw( + self, layers: ChecksumOffloadOptions, port_id: int, verify: bool = True + ) -> None: + """Enables hardware checksum offloading on the specified layer. + + Args: + layers: The layer/layers that checksum offloading should be enabled on. + port_id: The port number to enable checksum offloading on, should be within 0-32. + verify: If :data:`True` the output of the command will be scanned in an attempt to + verify that checksum offloading was enabled on the port. + + Raises: + InteractiveCommandExecutionError: If checksum offload is not enabled successfully. + """ + for name, offload in ChecksumOffloadOptions.__members__.items(): + if offload in layers: + name = name.replace("_", "-") + csum_output = self.send_command(f"csum set {name} hw {port_id}") + if verify: + if ( + "Bad arguments" in csum_output + or f"Please stop port {port_id} first" in csum_output + or f"checksum offload is not supported by port {port_id}" in csum_output + ): + self._logger.debug(f"Csum set hw error:\n{csum_output}") + raise InteractiveCommandExecutionError( + f"Failed to set csum hw {name} mode on port {port_id}" + ) + success = False + if f"{name} checksum offload is hw" in csum_output.lower(): + success = True + if not success and verify: + self._logger.debug( + f"Failed to set csum hw mode on port {port_id}:\n{csum_output}" + ) + raise InteractiveCommandExecutionError( + f"""Failed to set csum hw mode on port + {port_id}:\n{csum_output}""" + ) + + def flow_create(self, flow_rule: FlowRule, port_id: int) -> int: + """Creates a flow rule in the testpmd session. + + This command is implicitly verified as needed to return the created flow rule id. + + Args: + flow_rule: :class:`FlowRule` object used for creating testpmd flow rule. + port_id: Integer representing the port to use. + + Raises: + InteractiveCommandExecutionError: If flow rule is invalid. + + Returns: + Id of created flow rule. + """ + flow_output = self.send_command(f"flow create {port_id} {flow_rule}") + match = re.search(r"#(\d+)", flow_output) + if match is not None: + match_str = match.group(1) + flow_id = int(match_str) + return flow_id + else: + self._logger.debug(f"Failed to create flow rule:\n{flow_output}") + raise InteractiveCommandExecutionError(f"Failed to create flow rule:\n{flow_output}") + + def flow_validate(self, flow_rule: FlowRule, port_id: int) -> bool: + """Validates a flow rule in the testpmd session. + + Args: + flow_rule: :class:`FlowRule` object used for validating testpmd flow rule. + port_id: Integer representing the port to use. + + Returns: + Boolean representing whether rule is valid or not. + """ + flow_output = self.send_command(f"flow validate {port_id} {flow_rule}") + if "Flow rule validated" in flow_output: + return True + return False + + def flow_delete(self, flow_id: int, port_id: int, verify: bool = True) -> None: + """Deletes the specified flow rule from the testpmd session. + + Args: + flow_id: ID of the flow to remove. + port_id: Integer representing the port to use. + verify: If :data:`True`, the output of the command is scanned + to ensure the flow rule was deleted successfully. + + Raises: + InteractiveCommandExecutionError: If flow rule is invalid. + """ + flow_output = self.send_command(f"flow destroy {port_id} rule {flow_id}") + if verify: + if "destroyed" not in flow_output: + self._logger.debug(f"Failed to delete flow rule:\n{flow_output}") + raise InteractiveCommandExecutionError( + f"Failed to delete flow rule:\n{flow_output}" + ) + + @_requires_started_ports + @_requires_stopped_ports + def set_port_mtu(self, port_id: int, mtu: int, verify: bool = True) -> None: + """Change the MTU of a port using testpmd. + + Some PMDs require that the port be stopped before changing the MTU, and it does no harm to + stop the port before configuring in cases where it isn't required, so ports are stopped + prior to changing their MTU. On the other hand, some PMDs require that the port had already + been started once since testpmd startup. Therefore, ports are also started before stopping + them to ensure this has happened. + + Args: + port_id: ID of the port to adjust the MTU on. + mtu: Desired value for the MTU to be set to. + verify: If `verify` is :data:`True` then the output will be scanned in an attempt to + verify that the mtu was properly set on the port. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the MTU was not + properly updated on the port matching `port_id`. + """ + set_mtu_output = self.send_command(f"port config mtu {port_id} {mtu}") + if verify and (f"MTU: {mtu}" not in self.send_command(f"show port info {port_id}")): + self._logger.debug( + f"Failed to set mtu to {mtu} on port {port_id}. Output was:\n{set_mtu_output}" + ) + raise InteractiveCommandExecutionError( + f"Test pmd failed to update mtu of port {port_id} to {mtu}" + ) + + def set_port_mtu_all(self, mtu: int, verify: bool = True) -> None: + """Change the MTU of all ports using testpmd. + + Runs :meth:`set_port_mtu` for every port that testpmd is aware of. + + Args: + mtu: Desired value for the MTU to be set to. + verify: Whether to verify that setting the MTU on each port was successful or not. + Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the MTU was not + properly updated on at least one port. + """ + for port in self.ports: + self.set_port_mtu(<a href="http://port.id" rel="noreferrer" target="_blank">port.id</a>, mtu, verify) + + @staticmethod + def extract_verbose_output(output: str) -> list[TestPmdVerbosePacket]: + """Extract the verbose information present in given testpmd output. + + This method extracts sections of verbose output that begin with the line + "port X/queue Y: sent/received Z packets" and end with the ol_flags of a packet. + + Args: + output: Testpmd output that contains verbose information + + Returns: + List of parsed packet information gathered from verbose information in `output`. + """ + out: list[TestPmdVerbosePacket] = [] + prev_header: str = "" + iter = re.finditer( + r"(?P<HEADER>(?:port \d+/queue \d+: (?:received|sent) \d+ packets)?)\s*" + r"(?P<PACKET>src=[\w\s=:-]+?ol_flags: [\w ]+)", + output, + ) + for match in iter: + if match.group("HEADER"): + prev_header = match.group("HEADER") + out.append(TestPmdVerbosePacket.parse(f"{prev_header}\n{match.group('PACKET')}")) + return out + + @_requires_stopped_ports + def set_vlan_filter(self, port: int, enable: bool, verify: bool = True) -> None: + """Set vlan filter on. + + Args: + port: The port number to enable VLAN filter on. + enable: Enable the filter on `port` if :data:`True`, otherwise disable it. + verify: If :data:`True`, the output of the command and show port info + is scanned to verify that vlan filtering was set successfully. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the filter + fails to update. + """ + filter_cmd_output = self.send_command(f"vlan set filter {'on' if enable else 'off'} {port}") + if verify: + vlan_settings = self.show_port_info(port_id=port).vlan_offload + if enable ^ (vlan_settings is not None and VLANOffloadFlag.FILTER in vlan_settings): + self._logger.debug( + f"""Failed to {"enable" if enable else "disable"} + filter on port {port}: \n{filter_cmd_output}""" + ) + raise InteractiveCommandExecutionError( + f"""Failed to {"enable" if enable else "disable"} + filter on port {port}""" + ) + + def set_mac_address(self, port: int, mac_address: str, verify: bool = True) -> None: + """Set port's MAC address. + + Args: + port: The number of the requested port. + mac_address: The MAC address to set. + verify: If :data:`True`, the output of the command is scanned to verify that + the mac address is set in the specified port. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the command + fails to execute. + """ + output = self.send_command(f"mac_addr set {port} {mac_address}", skip_first_line=True) + if verify: + if output.strip(): + self._logger.debug( + f"Testpmd failed to set MAC address {mac_address} on port {port}." + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to set MAC address {mac_address} on port {port}." + ) + + def set_flow_control( + self, port: int, flow_ctrl: TestPmdPortFlowCtrl, verify: bool = True + ) -> None: + """Set the given `port`'s flow control. + + Args: + port: The number of the requested port. + flow_ctrl: The requested flow control parameters. + verify: If :data:`True`, the output of the command is scanned to verify that + the flow control in the specified port is set. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the command + fails to execute. + """ + output = self.send_command(f"set flow_ctrl {flow_ctrl} {port}", skip_first_line=True) + if verify: + if output.strip(): + self._logger.debug(f"Testpmd failed to set the {flow_ctrl} in port {port}.") + raise InteractiveCommandExecutionError( + f"Testpmd failed to set the {flow_ctrl} in port {port}." + ) + + def show_port_flow_info(self, port: int) -> TestPmdPortFlowCtrl | None: + """Show port info flow. + + Args: + port: The number of the requested port. + + Returns: + The current port flow control parameters if supported, otherwise :data:`None`. + """ + output = self.send_command(f"show port {port} flow_ctrl") + if "Flow control infos" in output: + return TestPmdPortFlowCtrl.parse(output) + return None + + @_requires_stopped_ports + def rx_vlan(self, vlan: int, port: int, add: bool, verify: bool = True) -> None: + """Add specified vlan tag to the filter list on a port. Requires vlan filter to be on. + + Args: + vlan: The vlan tag to add, should be within 1-1005. + port: The port number to add the tag on. + add: Adds the tag if :data:`True`, otherwise removes the tag. + verify: If :data:`True`, the output of the command is scanned to verify that + the vlan tag was added to the filter list on the specified port. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the tag + is not added. + """ + rx_cmd_output = self.send_command(f"rx_vlan {'add' if add else 'rm'} {vlan} {port}") + if verify: + if ( + "VLAN-filtering disabled" in rx_cmd_output + or "Invalid vlan_id" in rx_cmd_output + or "Bad arguments" in rx_cmd_output + ): + self._logger.debug( + f"""Failed to {"add" if add else "remove"} tag {vlan} + port {port}: \n{rx_cmd_output}""" + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to {'add' if add else 'remove'} tag {vlan} on port {port}." + ) + + @_requires_stopped_ports + def set_vlan_strip(self, port: int, enable: bool, verify: bool = True) -> None: + """Enable or disable vlan stripping on the specified port. + + Args: + port: The port number to use. + enable: If :data:`True`, will turn vlan stripping on, otherwise will turn off. + verify: If :data:`True`, the output of the command and show port info + is scanned to verify that vlan stripping was enabled on the specified port. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and stripping + fails to update. + """ + strip_cmd_output = self.send_command(f"vlan set strip {'on' if enable else 'off'} {port}") + if verify: + vlan_settings = self.show_port_info(port_id=port).vlan_offload + if enable ^ (vlan_settings is not None and VLANOffloadFlag.STRIP in vlan_settings): + self._logger.debug( + f"""Failed to set strip {"on" if enable else "off"} + port {port}: \n{strip_cmd_output}""" + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to set strip {'on' if enable else 'off'} port {port}." + ) + + @_requires_stopped_ports + def tx_vlan_set( + self, port: int, enable: bool, vlan: int | None = None, verify: bool = True + ) -> None: + """Set hardware insertion of vlan tags in packets sent on a port. + + Args: + port: The port number to use. + enable: Sets vlan tag insertion if :data:`True`, and resets if :data:`False`. + vlan: The vlan tag to insert if enable is :data:`True`. + verify: If :data:`True`, the output of the command is scanned to verify that + vlan insertion was enabled on the specified port. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the insertion + tag is not set. + """ + if enable: + tx_vlan_cmd_output = self.send_command(f"tx_vlan set {port} {vlan}") + if verify: + if ( + "Please stop port" in tx_vlan_cmd_output + or "Invalid vlan_id" in tx_vlan_cmd_output + or "Invalid port" in tx_vlan_cmd_output + ): + self._logger.debug( + f"Failed to set vlan tag {vlan} on port {port}:\n{tx_vlan_cmd_output}" + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to set vlan insertion tag {vlan} on port {port}." + ) + else: + tx_vlan_cmd_output = self.send_command(f"tx_vlan reset {port}") + if verify: + if "Please stop port" in tx_vlan_cmd_output or "Invalid port" in tx_vlan_cmd_output: + self._logger.debug( + f"Failed to reset vlan insertion on port {port}: \n{tx_vlan_cmd_output}" + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to reset vlan insertion on port {port}." + ) + + def set_promisc(self, port: int, enable: bool, verify: bool = True) -> None: + """Enable or disable promiscuous mode for the specified port. + + Args: + port: Port number to use. + enable: If :data:`True`, turn promiscuous mode on, otherwise turn off. + verify: If :data:`True` an additional command will be sent to verify that + promiscuous mode is properly set. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and promiscuous mode + is not correctly set. + """ + promisc_cmd_output = self.send_command(f"set promisc {port} {'on' if enable else 'off'}") + if verify: + stats = self.show_port_info(port_id=port) + if enable ^ stats.is_promiscuous_mode_enabled: + self._logger.debug( + f"Failed to set promiscuous mode on port {port}: \n{promisc_cmd_output}" + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to set promiscuous mode on port {port}." + ) + + def set_verbose(self, level: int, verify: bool = True) -> None: + """Set debug verbosity level. + + Args: + level: 0 - silent except for error + 1 - fully verbose except for Tx packets + 2 - fully verbose except for Rx packets + >2 - fully verbose + verify: If :data:`True` the command output will be scanned to verify that verbose level + is properly set. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and verbose level + is not correctly set. + """ + verbose_cmd_output = self.send_command(f"set verbose {level}") + if verify: + if "Change verbose level" not in verbose_cmd_output: + self._logger.debug( + f"Failed to set verbose level to {level}: \n{verbose_cmd_output}" + ) + raise InteractiveCommandExecutionError( + f"Testpmd failed to set verbose level to {level}." + ) + + def rx_vxlan(self, vxlan_id: int, port_id: int, enable: bool, verify: bool = True) -> None: + """Add or remove vxlan id to/from filter list. + + Args: + vxlan_id: VXLAN ID to add to port filter list. + port_id: ID of the port to modify VXLAN filter of. + enable: If :data:`True`, adds specified VXLAN ID, otherwise removes it. + verify: If :data:`True`, the output of the command is checked to verify + the VXLAN ID was successfully added/removed from the port. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and VXLAN ID + is not successfully added or removed. + """ + action = "add" if enable else "rm" + vxlan_output = self.send_command(f"rx_vxlan_port {action} {vxlan_id} {port_id}") + if verify: + if "udp tunneling add error" in vxlan_output: + self._logger.debug(f"Failed to set VXLAN:\n{vxlan_output}") + raise InteractiveCommandExecutionError(f"Failed to set VXLAN:\n{vxlan_output}") + + def clear_port_stats(self, port_id: int, verify: bool = True) -> None: + """Clear statistics of a given port. + + Args: + port_id: ID of the port to clear the statistics on. + verify: If :data:`True` the output of the command will be scanned to verify that it was + successful, otherwise failures will be ignored. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and testpmd fails to + clear the statistics of the given port. + """ + clear_output = self.send_command(f"clear port stats {port_id}") + if verify and f"NIC statistics for port {port_id} cleared" not in clear_output: + raise InteractiveCommandExecutionError( + f"Test pmd failed to set clear forwarding stats on port {port_id}" + ) + + def clear_port_stats_all(self, verify: bool = True) -> None: + """Clear the statistics of all ports that testpmd is aware of. + + Args: + verify: If :data:`True` the output of the command will be scanned to verify that all + ports had their statistics cleared, otherwise failures will be ignored. Defaults to + :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and testpmd fails to + clear the statistics of any of its ports. + """ + clear_output = self.send_command("clear port stats all") + if verify: + if type(self._app_params.port_numa_config) is list: + for port_id in range(len(self._app_params.port_numa_config)): + if f"NIC statistics for port {port_id} cleared" not in clear_output: + raise InteractiveCommandExecutionError( + f"Test pmd failed to set clear forwarding stats on port {port_id}" + ) + + @only_active + def close(self) -> None: + """Overrides :meth:`~.interactive_shell.close`.""" + self.stop() + self.send_command("quit", "Bye...") + return super().close() + + """ + ====== Capability retrieval methods ====== + """ + + def get_capabilities_rx_offload( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + ) -> None: + """Get all rx offload capabilities and divide them into supported and unsupported. + + Args: + supported_capabilities: Supported capabilities will be added to this set. + unsupported_capabilities: Unsupported capabilities will be added to this set. + """ + self._logger.debug("Getting rx offload capabilities.") + command = f"show port {self.ports[0].id} rx_offload capabilities" + rx_offload_capabilities_out = self.send_command(command) + rx_offload_capabilities = RxOffloadCapabilities.parse(rx_offload_capabilities_out) + self._update_capabilities_from_flag( + supported_capabilities, + unsupported_capabilities, + RxOffloadCapability, + rx_offload_capabilities.per_port | rx_offload_capabilities.per_queue, + ) + + def get_port_queue_info( + self, port_id: int, queue_id: int, is_rx_queue: bool + ) -> TestPmdQueueInfo: + """Returns the current state of the specified queue.""" + command = f"show {'rxq' if is_rx_queue else 'txq'} info {port_id} {queue_id}" + queue_info = TestPmdQueueInfo.parse(self.send_command(command)) + return queue_info + + def setup_port_queue(self, port_id: int, queue_id: int, is_rx_queue: bool) -> None: + """Setup a given queue on a port. + + This functionality cannot be verified because the setup action only takes effect when the + queue is started. + + Args: + port_id: ID of the port where the queue resides. + queue_id: ID of the queue to setup. + is_rx_queue: Type of queue to setup. If :data:`True` an RX queue will be setup, + otherwise a TX queue will be setup. + """ + self.send_command(f"port {port_id} {'rxq' if is_rx_queue else 'txq'} {queue_id} setup") + + def stop_port_queue( + self, port_id: int, queue_id: int, is_rx_queue: bool, verify: bool = True + ) -> None: + """Stops a given queue on a port. + + Args: + port_id: ID of the port that the queue belongs to. + queue_id: ID of the queue to stop. + is_rx_queue: Type of queue to stop. If :data:`True` an RX queue will be stopped, + otherwise a TX queue will be stopped. + verify: If :data:`True` an additional command will be sent to verify the queue stopped. + Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the queue fails to + stop. + """ + port_type = "rxq" if is_rx_queue else "txq" + stop_cmd_output = self.send_command(f"port {port_id} {port_type} {queue_id} stop") + if verify: + queue_started = self.get_port_queue_info( + port_id, queue_id, is_rx_queue + ).is_queue_started + if queue_started: + self._logger.debug( + f"Failed to stop {port_type} {queue_id} on port {port_id}:\n{stop_cmd_output}" + ) + raise InteractiveCommandExecutionError( + f"Test pmd failed to stop {port_type} {queue_id} on port {port_id}" + ) + + def start_port_queue( + self, port_id: int, queue_id: int, is_rx_queue: bool, verify: bool = True + ) -> None: + """Starts a given queue on a port. + + First sets up the port queue, then starts it. + + Args: + port_id: ID of the port that the queue belongs to. + queue_id: ID of the queue to start. + is_rx_queue: Type of queue to start. If :data:`True` an RX queue will be started, + otherwise a TX queue will be started. + verify: if :data:`True` an additional command will be sent to verify that the queue was + started. Defaults to :data:`True`. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and the queue fails to + start. + """ + port_type = "rxq" if is_rx_queue else "txq" + self.setup_port_queue(port_id, queue_id, is_rx_queue) + start_cmd_output = self.send_command(f"port {port_id} {port_type} {queue_id} start") + if verify: + queue_started = self.get_port_queue_info( + port_id, queue_id, is_rx_queue + ).is_queue_started + if not queue_started: + self._logger.debug( + f"Failed to start {port_type} {queue_id} on port {port_id}:\n{start_cmd_output}" + ) + raise InteractiveCommandExecutionError( + f"Test pmd failed to start {port_type} {queue_id} on port {port_id}" + ) + + def get_queue_ring_size(self, port_id: int, queue_id: int, is_rx_queue: bool) -> int: + """Returns the current size of the ring on the specified queue.""" + command = f"show {'rxq' if is_rx_queue else 'txq'} info {port_id} {queue_id}" + queue_info = TestPmdQueueInfo.parse(self.send_command(command)) + return queue_info.ring_size + + def set_queue_ring_size( + self, + port_id: int, + queue_id: int, + size: int, + is_rx_queue: bool, + verify: bool = True, + ) -> None: + """Update the ring size of an Rx/Tx queue on a given port. + + Queue is setup after setting the ring size so that the queue info reflects this change and + it can be verified. + + Args: + port_id: The port that the queue resides on. + queue_id: The ID of the queue on the port. + size: The size to update the ring size to. + is_rx_queue: Whether to modify an RX or TX queue. If :data:`True` an RX queue will be + updated, otherwise a TX queue will be updated. + verify: If :data:`True` an additional command will be sent to check the ring size of + the queue in an attempt to validate that the size was changes properly. + + Raises: + InteractiveCommandExecutionError: If `verify` is :data:`True` and there is a failure + when updating ring size. + """ + queue_type = "rxq" if is_rx_queue else "txq" + self.send_command(f"port config {port_id} {queue_type} {queue_id} ring_size {size}") + self.setup_port_queue(port_id, queue_id, is_rx_queue) + if verify: + curr_ring_size = self.get_queue_ring_size(port_id, queue_id, is_rx_queue) + if curr_ring_size != size: + self._logger.debug( + f"Failed up update ring size of queue {queue_id} on port {port_id}. Current" + f" ring size is {curr_ring_size}." + ) + raise InteractiveCommandExecutionError( + f"Failed to update ring size of queue {queue_id} on port {port_id}" + ) + + @_requires_stopped_ports + def set_queue_deferred_start( + self, port_id: int, queue_id: int, is_rx_queue: bool, on: bool + ) -> None: + """Set the deferred start attribute of the specified queue on/off. + + Args: + port_id: The port that the queue resides on. + queue_id: The ID of the queue on the port. + is_rx_queue: Whether to modify an RX or TX queue. If :data:`True` an RX queue will be + updated, otherwise a TX queue will be updated. + on: Whether to set deferred start mode on or off. If :data:`True` deferred start will + be turned on, otherwise it will be turned off. + """ + queue_type = "rxq" if is_rx_queue else "txq" + action = "on" if on else "off" + self.send_command(f"port {port_id} {queue_type} {queue_id} deferred_start {action}") + + def _update_capabilities_from_flag( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + flag_class: type[Flag], + supported_flags: Flag, + ) -> None: + """Divide all flags from `flag_class` into supported and unsupported.""" + for flag in flag_class: + if flag in supported_flags: + supported_capabilities.add(NicCapability[str(<a href="http://flag.name" rel="noreferrer" target="_blank">flag.name</a>)]) + else: + unsupported_capabilities.add(NicCapability[str(<a href="http://flag.name" rel="noreferrer" target="_blank">flag.name</a>)]) + + @_requires_started_ports + def get_capabilities_rxq_info( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + ) -> None: + """Get all rxq capabilities and divide them into supported and unsupported. + + Args: + supported_capabilities: Supported capabilities will be added to this set. + unsupported_capabilities: Unsupported capabilities will be added to this set. + """ + self._logger.debug("Getting rxq capabilities.") + command = f"show rxq info {self.ports[0].id} 0" + rxq_info = TestPmdRxqInfo.parse(self.send_command(command)) + if rxq_info.scattered_packets: + supported_capabilities.add(NicCapability.SCATTERED_RX_ENABLED) + else: + unsupported_capabilities.add(NicCapability.SCATTERED_RX_ENABLED) + + def get_capabilities_show_port_info( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + ) -> None: + """Get all capabilities from show port info and divide them into supported and unsupported. + + Args: + supported_capabilities: Supported capabilities will be added to this set. + unsupported_capabilities: Unsupported capabilities will be added to this set. + """ + self._update_capabilities_from_flag( + supported_capabilities, + unsupported_capabilities, + DeviceCapabilitiesFlag, + self.ports[0].device_capabilities, + ) + + def get_capabilities_mcast_filtering( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + ) -> None: + """Get multicast filtering capability from mcast_addr add and check for testpmd error code. + + Args: + supported_capabilities: Supported capabilities will be added to this set. + unsupported_capabilities: Unsupported capabilities will be added to this set. + """ + self._logger.debug("Getting mcast filter capabilities.") + command = f"mcast_addr add {self.ports[0].id} 01:00:5E:00:00:00" + output = self.send_command(command) + if "diag=-95" in output: + unsupported_capabilities.add(NicCapability.MCAST_FILTERING) + else: + supported_capabilities.add(NicCapability.MCAST_FILTERING) + command = str.replace(command, "add", "remove", 1) + self.send_command(command) + + def get_capabilities_flow_ctrl( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + ) -> None: + """Get flow control capability and check for testpmd failure. + + Args: + supported_capabilities: Supported capabilities will be added to this set. + unsupported_capabilities: Unsupported capabilities will be added to this set. + """ + self._logger.debug("Getting flow ctrl capabilities.") + command = f"show port {self.ports[0].id} flow_ctrl" + output = self.send_command(command) + if "Flow control infos" in output: + supported_capabilities.add(NicCapability.FLOW_CTRL) + else: + unsupported_capabilities.add(NicCapability.FLOW_CTRL) + + def get_capabilities_physical_function( + self, + supported_capabilities: MutableSet["NicCapability"], + unsupported_capabilities: MutableSet["NicCapability"], + ) -> None: + """Store capability representing a physical function test run. + + Args: + supported_capabilities: Supported capabilities will be added to this set. + unsupported_capabilities: Unsupported capabilities will be added to this set. + """ + ctx = get_ctx() + if ctx.topology.vf_ports == []: + supported_capabilities.add(NicCapability.PHYSICAL_FUNCTION) + else: + unsupported_capabilities.add(NicCapability.PHYSICAL_FUNCTION) diff --git a/dts/framework/params/testpmd.py b/dts/api/testpmd/config.py similarity index 98% rename from dts/framework/params/testpmd.py rename to dts/api/testpmd/config.py index 1913bd0fa2..e71a3e1ef0 100644 --- a/dts/framework/params/testpmd.py +++ b/dts/api/testpmd/config.py @@ -1,7 +1,12 @@ # SPDX-License-Identifier: BSD-3-Clause # Copyright(c) 2024 Arm Limited -"""Module containing all the TestPmd-related parameter classes.""" +"""Module containing all classes needed to configure :class:`TestPmd`. + +This module defines the :class:`TestPmdParams` class which is used to configure the +TestPmd shell. It also includes various data classes and enums that are used +to represent different configurations and settings. +""" from dataclasses import dataclass, field from enum import EnumMeta, Flag, auto, unique @@ -146,7 +151,7 @@ class RSSSetting(EnumMeta): class SimpleForwardingModes(StrEnum): - r"""The supported packet forwarding modes for :class:`~TestPmdShell`\s.""" + r"""The supported packet forwarding modes for :class:`~TestPmd`\s.""" #: io = auto() diff --git a/dts/api/testpmd/types.py b/dts/api/testpmd/types.py new file mode 100644 index 0000000000..553438c904 --- /dev/null +++ b/dts/api/testpmd/types.py @@ -0,0 +1,1406 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2023 University of New Hampshire +# Copyright(c) 2023 PANTHEON.tech s.r.o. +# Copyright(c) 2024 Arm Limited + +"""TestPmd types module. + +Exposes types used in the TestPmd API. +""" + +import re +from dataclasses import dataclass, field +from enum import Flag, auto +from typing import Literal + +from typing_extensions import Self + +from framework.parser import ParserFn, TextParser +from framework.utils import REGEX_FOR_MAC_ADDRESS, StrEnum + + +class TestPmdDevice: + """The data of a device that testpmd can recognize. + + Attributes: + pci_address: The PCI address of the device. + """ + + pci_address: str + + def __init__(self, pci_address_line: str): + """Initialize the device from the testpmd output line string. + + Args: + pci_address_line: A line of testpmd output that contains a device. + """ + self.pci_address = pci_address_line.strip().split(": ")[1].strip() + + def __str__(self) -> str: + """The PCI address captures what the device is.""" + return self.pci_address + + +class VLANOffloadFlag(Flag): + """Flag representing the VLAN offload settings of a NIC port.""" + + #: + STRIP = auto() + #: + FILTER = auto() + #: + EXTEND = auto() + #: + QINQ_STRIP = auto() + + @classmethod + def from_str_dict(cls, d): + """Makes an instance from a dict containing the flag member names with an "on" value. + + Args: + d: A dictionary containing the flag members as keys and any string value. + + Returns: + A new instance of the flag. + """ + flag = cls(0) + for name in cls.__members__: + if d.get(name) == "on": + flag |= cls[name] + return flag + + @classmethod + def make_parser(cls) -> ParserFn: + """Makes a parser function. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this flag from text. + """ + return TextParser.wrap( + TextParser.find( + r"VLAN offload:\s+" + r"strip (?P<STRIP>on|off), " + r"filter (?P<FILTER>on|off), " + r"extend (?P<EXTEND>on|off), " + r"qinq strip (?P<QINQ_STRIP>on|off)", + re.MULTILINE, + named=True, + ), + cls.from_str_dict, + ) + + +class ChecksumOffloadOptions(Flag): + """Flag representing checksum hardware offload layer options.""" + + #: + ip = auto() + #: + udp = auto() + #: + tcp = auto() + #: + sctp = auto() + #: + outer_ip = auto() + #: + outer_udp = auto() + + +class RSSOffloadTypesFlag(Flag): + """Flag representing the RSS offload flow types supported by the NIC port.""" + + #: + ipv4 = auto() + #: + ipv4_frag = auto() + #: + ipv4_tcp = auto() + #: + ipv4_udp = auto() + #: + ipv4_sctp = auto() + #: + ipv4_other = auto() + #: + ipv6 = auto() + #: + ipv6_frag = auto() + #: + ipv6_tcp = auto() + #: + ipv6_udp = auto() + #: + ipv6_sctp = auto() + #: + ipv6_other = auto() + #: + l2_payload = auto() + #: + ipv6_ex = auto() + #: + ipv6_tcp_ex = auto() + #: + ipv6_udp_ex = auto() + #: + port = auto() + #: + vxlan = auto() + #: + geneve = auto() + #: + nvgre = auto() + #: + user_defined_22 = auto() + #: + gtpu = auto() + #: + eth = auto() + #: + s_vlan = auto() + #: + c_vlan = auto() + #: + esp = auto() + #: + ah = auto() + #: + l2tpv3 = auto() + #: + pfcp = auto() + #: + pppoe = auto() + #: + ecpri = auto() + #: + mpls = auto() + #: + ipv4_chksum = auto() + #: + l4_chksum = auto() + #: + l2tpv2 = auto() + #: + ipv6_flow_label = auto() + #: + user_defined_38 = auto() + #: + user_defined_39 = auto() + #: + user_defined_40 = auto() + #: + user_defined_41 = auto() + #: + user_defined_42 = auto() + #: + user_defined_43 = auto() + #: + user_defined_44 = auto() + #: + user_defined_45 = auto() + #: + user_defined_46 = auto() + #: + user_defined_47 = auto() + #: + user_defined_48 = auto() + #: + user_defined_49 = auto() + #: + user_defined_50 = auto() + #: + user_defined_51 = auto() + #: + l3_pre96 = auto() + #: + l3_pre64 = auto() + #: + l3_pre56 = auto() + #: + l3_pre48 = auto() + #: + l3_pre40 = auto() + #: + l3_pre32 = auto() + #: + l2_dst_only = auto() + #: + l2_src_only = auto() + #: + l4_dst_only = auto() + #: + l4_src_only = auto() + #: + l3_dst_only = auto() + #: + l3_src_only = auto() + + #: + ip = ipv4 | ipv4_frag | ipv4_other | ipv6 | ipv6_frag | ipv6_other | ipv6_ex + #: + udp = ipv4_udp | ipv6_udp | ipv6_udp_ex + #: + tcp = ipv4_tcp | ipv6_tcp | ipv6_tcp_ex + #: + sctp = ipv4_sctp | ipv6_sctp + #: + tunnel = vxlan | geneve | nvgre + #: + vlan = s_vlan | c_vlan + #: + all = ( + eth + | vlan + | ip + | tcp + | udp + | sctp + | l2_payload + | l2tpv3 + | esp + | ah + | pfcp + | gtpu + | ecpri + | mpls + | l2tpv2 + ) + + @classmethod + def from_list_string(cls, names: str) -> Self: + """Makes a flag from a whitespace-separated list of names. + + Args: + names: a whitespace-separated list containing the members of this flag. + + Returns: + An instance of this flag. + """ + flag = cls(0) + for name in names.split(): + flag |= cls.from_str(name) + return flag + + @classmethod + def from_str(cls, name: str) -> Self: + """Makes a flag matching the supplied name. + + Args: + name: a valid member of this flag in text + Returns: + An instance of this flag. + """ + member_name = name.strip().replace("-", "_") + return cls[member_name] + + @classmethod + def make_parser(cls) -> ParserFn: + """Makes a parser function. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this flag from text. + """ + return TextParser.wrap( + TextParser.find(r"Supported RSS offload flow types:((?:\r?\n? \S+)+)", re.MULTILINE), + RSSOffloadTypesFlag.from_list_string, + ) + + +class DeviceCapabilitiesFlag(Flag): + """Flag representing the device capabilities.""" + + #: Device supports Rx queue setup after device started. + RUNTIME_RX_QUEUE_SETUP = auto() + #: Device supports Tx queue setup after device started. + RUNTIME_TX_QUEUE_SETUP = auto() + #: Device supports shared Rx queue among ports within Rx domain and switch domain. + RXQ_SHARE = auto() + #: Device supports keeping flow rules across restart. + FLOW_RULE_KEEP = auto() + #: Device supports keeping shared flow objects across restart. + FLOW_SHARED_OBJECT_KEEP = auto() + + @classmethod + def make_parser(cls) -> ParserFn: + """Makes a parser function. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this flag from text. + """ + return TextParser.wrap( + TextParser.find_int(r"Device capabilities: (0x[A-Fa-f\d]+)"), + cls, + ) + + +class DeviceErrorHandlingMode(StrEnum): + """Enum representing the device error handling mode.""" + + #: + none = auto() + #: + passive = auto() + #: + proactive = auto() + #: + unknown = auto() + + @classmethod + def make_parser(cls) -> ParserFn: + """Makes a parser function. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this enum from text. + """ + return TextParser.wrap(TextParser.find(r"Device error handling mode: (\w+)"), cls) + + +class RxQueueState(StrEnum): + """RX queue states. + + References: + DPDK lib: ``lib/ethdev/rte_ethdev.h`` + testpmd display function: ``app/test-pmd/config.c:get_queue_state_name()`` + """ + + #: + stopped = auto() + #: + started = auto() + #: + hairpin = auto() + #: + unknown = auto() + + @classmethod + def make_parser(cls) -> ParserFn: + """Makes a parser function. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this enum from text. + """ + return TextParser.wrap(TextParser.find(r"Rx queue state: ([^\r\n]+)"), cls) + + +@dataclass +class TestPmdQueueInfo(TextParser): + """Dataclass representation of the common parts of the testpmd `show rxq/txq info` commands.""" + + #: + prefetch_threshold: int = field(metadata=TextParser.find_int(r"prefetch threshold: (\d+)")) + #: + host_threshold: int = field(metadata=TextParser.find_int(r"host threshold: (\d+)")) + #: + writeback_threshold: int = field(metadata=TextParser.find_int(r"writeback threshold: (\d+)")) + #: + free_threshold: int = field(metadata=TextParser.find_int(r"free threshold: (\d+)")) + #: + deferred_start: bool = field(metadata=TextParser.find("deferred start: on")) + #: The number of RXD/TXDs is just the ring size of the queue. + ring_size: int = field(metadata=TextParser.find_int(r"Number of (?:RXDs|TXDs): (\d+)")) + #: + is_queue_started: bool = field(metadata=TextParser.find("queue state: started")) + #: + burst_mode: str | None = field( + default=None, metadata=TextParser.find(r"Burst mode: ([^\r\n]+)") + ) + + +@dataclass +class TestPmdTxqInfo(TestPmdQueueInfo): + """Representation of testpmd's ``show txq info <port_id> <queue_id>`` command. + + References: + testpmd command function: ``app/test-pmd/cmdline.c:cmd_showqueue()`` + testpmd display function: ``app/test-pmd/config.c:rx_queue_infos_display()`` + """ + + #: Ring size threshold + rs_threshold: int | None = field( + default=None, metadata=TextParser.find_int(r"TX RS threshold: (\d+)\b") + ) + + +@dataclass +class TestPmdRxqInfo(TestPmdQueueInfo): + """Representation of testpmd's ``show rxq info <port_id> <queue_id>`` command. + + References: + testpmd command function: ``app/test-pmd/cmdline.c:cmd_showqueue()`` + testpmd display function: ``app/test-pmd/config.c:rx_queue_infos_display()`` + """ + + #: Mempool used by that queue + mempool: str | None = field(default=None, metadata=TextParser.find(r"Mempool: ([^\r\n]+)")) + #: Drop packets if no descriptors are available + drop_packets: bool | None = field( + default=None, metadata=TextParser.find(r"RX drop packets: on") + ) + #: Scattered packets Rx enabled + scattered_packets: bool | None = field( + default=None, metadata=TextParser.find(r"RX scattered packets: on") + ) + #: The state of the queue + queue_state: str | None = field(default=None, metadata=RxQueueState.make_parser()) + + +@dataclass +class TestPmdPort(TextParser): + """Dataclass representing the result of testpmd's ``show port info`` command.""" + + @staticmethod + def _make_device_private_info_parser() -> ParserFn: + """Device private information parser. + + Ensures that we are not parsing invalid device private info output. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a parser + function that parses the device private info from the TestPmd port info output. + """ + + def _validate(info: str): + info = info.strip() + if ( + info == "none" + or info.startswith("Invalid file") + or info.startswith("Failed to dump") + ): + return None + return info + + return TextParser.wrap(TextParser.find(r"Device private info:\s+([\s\S]+)"), _validate) + + #: + id: int = field(metadata=TextParser.find_int(r"Infos for port (\d+)\b")) + #: + device_name: str = field(metadata=TextParser.find(r"Device name: ([^\r\n]+)")) + #: + driver_name: str = field(metadata=TextParser.find(r"Driver name: ([^\r\n]+)")) + #: + socket_id: int = field(metadata=TextParser.find_int(r"Connect to socket: (\d+)")) + #: + is_link_up: bool = field(metadata=TextParser.find("Link status: up")) + #: + link_speed: str = field(metadata=TextParser.find(r"Link speed: ([^\r\n]+)")) + #: + is_link_full_duplex: bool = field(metadata=TextParser.find("Link duplex: full-duplex")) + #: + is_link_autonegotiated: bool = field(metadata=TextParser.find("Autoneg status: On")) + #: + is_promiscuous_mode_enabled: bool = field(metadata=TextParser.find("Promiscuous mode: enabled")) + #: + is_allmulticast_mode_enabled: bool = field( + metadata=TextParser.find("Allmulticast mode: enabled") + ) + #: Maximum number of MAC addresses + max_mac_addresses_num: int = field( + metadata=TextParser.find_int(r"Maximum number of MAC addresses: (\d+)") + ) + #: Maximum configurable length of RX packet + max_hash_mac_addresses_num: int = field( + metadata=TextParser.find_int(r"Maximum number of MAC addresses of hash filtering: (\d+)") + ) + #: Minimum size of RX buffer + min_rx_bufsize: int = field(metadata=TextParser.find_int(r"Minimum size of RX buffer: (\d+)")) + #: Maximum configurable length of RX packet + max_rx_packet_length: int = field( + metadata=TextParser.find_int(r"Maximum configurable length of RX packet: (\d+)") + ) + #: Maximum configurable size of LRO aggregated packet + max_lro_packet_size: int = field( + metadata=TextParser.find_int(r"Maximum configurable size of LRO aggregated packet: (\d+)") + ) + + #: Current number of RX queues + rx_queues_num: int = field(metadata=TextParser.find_int(r"Current number of RX queues: (\d+)")) + #: Max possible RX queues + max_rx_queues_num: int = field(metadata=TextParser.find_int(r"Max possible RX queues: (\d+)")) + #: Max possible number of RXDs per queue + max_queue_rxd_num: int = field( + metadata=TextParser.find_int(r"Max possible number of RXDs per queue: (\d+)") + ) + #: Min possible number of RXDs per queue + min_queue_rxd_num: int = field( + metadata=TextParser.find_int(r"Min possible number of RXDs per queue: (\d+)") + ) + #: RXDs number alignment + rxd_alignment_num: int = field(metadata=TextParser.find_int(r"RXDs number alignment: (\d+)")) + + #: Current number of TX queues + tx_queues_num: int = field(metadata=TextParser.find_int(r"Current number of TX queues: (\d+)")) + #: Max possible TX queues + max_tx_queues_num: int = field(metadata=TextParser.find_int(r"Max possible TX queues: (\d+)")) + #: Max possible number of TXDs per queue + max_queue_txd_num: int = field( + metadata=TextParser.find_int(r"Max possible number of TXDs per queue: (\d+)") + ) + #: Min possible number of TXDs per queue + min_queue_txd_num: int = field( + metadata=TextParser.find_int(r"Min possible number of TXDs per queue: (\d+)") + ) + #: TXDs number alignment + txd_alignment_num: int = field(metadata=TextParser.find_int(r"TXDs number alignment: (\d+)")) + #: Max segment number per packet + max_packet_segment_num: int = field( + metadata=TextParser.find_int(r"Max segment number per packet: (\d+)") + ) + #: Max segment number per MTU/TSO + max_mtu_segment_num: int = field( + metadata=TextParser.find_int(r"Max segment number per MTU\/TSO: (\d+)") + ) + + #: + device_capabilities: DeviceCapabilitiesFlag = field( + metadata=DeviceCapabilitiesFlag.make_parser(), + ) + #: + device_error_handling_mode: DeviceErrorHandlingMode | None = field( + default=None, metadata=DeviceErrorHandlingMode.make_parser() + ) + #: + device_private_info: str | None = field( + default=None, + metadata=_make_device_private_info_parser(), + ) + + #: + hash_key_size: int | None = field( + default=None, metadata=TextParser.find_int(r"Hash key size in bytes: (\d+)") + ) + #: + redirection_table_size: int | None = field( + default=None, metadata=TextParser.find_int(r"Redirection table size: (\d+)") + ) + #: + supported_rss_offload_flow_types: RSSOffloadTypesFlag = field( + default=RSSOffloadTypesFlag(0), metadata=RSSOffloadTypesFlag.make_parser() + ) + + #: + mac_address: str | None = field( + default=None, metadata=TextParser.find(r"MAC address: ([A-Fa-f0-9:]+)") + ) + #: + fw_version: str | None = field( + default=None, metadata=TextParser.find(r"Firmware-version: ([^\r\n]+)") + ) + #: + dev_args: str | None = field(default=None, metadata=TextParser.find(r"Devargs: ([^\r\n]+)")) + #: Socket id of the memory allocation + mem_alloc_socket_id: int | None = field( + default=None, + metadata=TextParser.find_int(r"memory allocation on the socket: (\d+)"), + ) + #: + mtu: int | None = field(default=None, metadata=TextParser.find_int(r"MTU: (\d+)")) + + #: + vlan_offload: VLANOffloadFlag | None = field( + default=None, + metadata=VLANOffloadFlag.make_parser(), + ) + + #: Maximum size of RX buffer + max_rx_bufsize: int | None = field( + default=None, metadata=TextParser.find_int(r"Maximum size of RX buffer: (\d+)") + ) + #: Maximum number of VFs + max_vfs_num: int | None = field( + default=None, metadata=TextParser.find_int(r"Maximum number of VFs: (\d+)") + ) + #: Maximum number of VMDq pools + max_vmdq_pools_num: int | None = field( + default=None, metadata=TextParser.find_int(r"Maximum number of VMDq pools: (\d+)") + ) + + #: + switch_name: str | None = field( + default=None, metadata=TextParser.find(r"Switch name: ([\r\n]+)") + ) + #: + switch_domain_id: int | None = field( + default=None, metadata=TextParser.find_int(r"Switch domain Id: (\d+)") + ) + #: + switch_port_id: int | None = field( + default=None, metadata=TextParser.find_int(r"Switch Port Id: (\d+)") + ) + #: + switch_rx_domain: int | None = field( + default=None, metadata=TextParser.find_int(r"Switch Rx domain: (\d+)") + ) + + +@dataclass +class TestPmdPortStats(TextParser): + """Port statistics.""" + + #: + port_id: int = field(metadata=TextParser.find_int(r"NIC statistics for port (\d+)")) + + #: + rx_packets: int = field(metadata=TextParser.find_int(r"RX-packets:\s+(\d+)")) + #: + rx_missed: int = field(metadata=TextParser.find_int(r"RX-missed:\s+(\d+)")) + #: + rx_bytes: int = field(metadata=TextParser.find_int(r"RX-bytes:\s+(\d+)")) + #: + rx_errors: int = field(metadata=TextParser.find_int(r"RX-errors:\s+(\d+)")) + #: + rx_nombuf: int = field(metadata=TextParser.find_int(r"RX-nombuf:\s+(\d+)")) + + #: + tx_packets: int = field(metadata=TextParser.find_int(r"TX-packets:\s+(\d+)")) + #: + tx_errors: int = field(metadata=TextParser.find_int(r"TX-errors:\s+(\d+)")) + #: + tx_bytes: int = field(metadata=TextParser.find_int(r"TX-bytes:\s+(\d+)")) + + #: + rx_pps: int = field(metadata=TextParser.find_int(r"Rx-pps:\s+(\d+)")) + #: + rx_bps: int = field(metadata=TextParser.find_int(r"Rx-bps:\s+(\d+)")) + + #: + tx_pps: int = field(metadata=TextParser.find_int(r"Tx-pps:\s+(\d+)")) + #: + tx_bps: int = field(metadata=TextParser.find_int(r"Tx-bps:\s+(\d+)")) + + +@dataclass(kw_only=True) +class FlowRule: + """Class representation of flow rule parameters. + + This class represents the parameters of any flow rule as per the + following pattern: + + [group {group_id}] [priority {level}] [ingress] [egress] + [user_id {user_id}] pattern {item} [/ {item} [...]] / end + actions {action} [/ {action} [...]] / end + """ + + #: + group_id: int | None = None + #: + priority_level: int | None = None + #: + direction: Literal["ingress", "egress"] + #: + user_id: int | None = None + #: + pattern: list[str] + #: + actions: list[str] + + def __str__(self) -> str: + """Returns the string representation of this instance.""" + ret = "" + pattern = " / ".join(self.pattern) + action = " / ".join(self.actions) + if self.group_id is not None: + ret += f"group {self.group_id} " + if self.priority_level is not None: + ret += f"priority {self.priority_level} " + ret += f"{self.direction} " + if self.user_id is not None: + ret += f"user_id {self.user_id} " + ret += f"pattern {pattern} / end " + ret += f"actions {action} / end" + return ret + + +class PacketOffloadFlag(Flag): + """Flag representing the Packet Offload Features Flags in DPDK. + + Values in this class are taken from the definitions in the RTE MBUF core library in DPDK + located in ``lib/mbuf/rte_mbuf_core.h``. It is expected that flag values in this class will + match the values they are set to in said DPDK library with one exception; all values must be + unique. For example, the definitions for unknown checksum flags in ``rte_mbuf_core.h`` are all + set to :data:`0`, but it is valuable to distinguish between them in this framework. For this + reason flags that are not unique in the DPDK library are set either to values within the + RTE_MBUF_F_FIRST_FREE-RTE_MBUF_F_LAST_FREE range for Rx or shifted 61+ bits for Tx. + + References: + DPDK lib: ``lib/mbuf/rte_mbuf_core.h`` + """ + + # RX flags + + #: The RX packet is a 802.1q VLAN packet, and the tci has been saved in mbuf->vlan_tci. If the + #: flag RTE_MBUF_F_RX_VLAN_STRIPPED is also present, the VLAN header has been stripped from + #: mbuf data, else it is still present. + RTE_MBUF_F_RX_VLAN = auto() + + #: RX packet with RSS hash result. + RTE_MBUF_F_RX_RSS_HASH = auto() + + #: RX packet with FDIR match indicate. + RTE_MBUF_F_RX_FDIR = auto() + + #: This flag is set when the outermost IP header checksum is detected as wrong by the hardware. + RTE_MBUF_F_RX_OUTER_IP_CKSUM_BAD = 1 << 5 + + #: A vlan has been stripped by the hardware and its tci is saved in mbuf->vlan_tci. This can + #: only happen if vlan stripping is enabled in the RX configuration of the PMD. When + #: RTE_MBUF_F_RX_VLAN_STRIPPED is set, RTE_MBUF_F_RX_VLAN must also be set. + RTE_MBUF_F_RX_VLAN_STRIPPED = auto() + + #: No information about the RX IP checksum. Value is 0 in the DPDK library. + RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN = 1 << 23 + #: The IP checksum in the packet is wrong. + RTE_MBUF_F_RX_IP_CKSUM_BAD = 1 << 4 + #: The IP checksum in the packet is valid. + RTE_MBUF_F_RX_IP_CKSUM_GOOD = 1 << 7 + #: The IP checksum is not correct in the packet data, but the integrity of the IP header is + #: verified. Value is RTE_MBUF_F_RX_IP_CKSUM_BAD | RTE_MBUF_F_RX_IP_CKSUM_GOOD in the DPDK + #: library. + RTE_MBUF_F_RX_IP_CKSUM_NONE = 1 << 24 + + #: No information about the RX L4 checksum. Value is 0 in the DPDK library. + RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN = 1 << 25 + #: The L4 checksum in the packet is wrong. + RTE_MBUF_F_RX_L4_CKSUM_BAD = 1 << 3 + #: The L4 checksum in the packet is valid. + RTE_MBUF_F_RX_L4_CKSUM_GOOD = 1 << 8 + #: The L4 checksum is not correct in the packet data, but the integrity of the L4 data is + #: verified. Value is RTE_MBUF_F_RX_L4_CKSUM_BAD | RTE_MBUF_F_RX_L4_CKSUM_GOOD in the DPDK + #: library. + RTE_MBUF_F_RX_L4_CKSUM_NONE = 1 << 26 + + #: RX IEEE1588 L2 Ethernet PT Packet. + RTE_MBUF_F_RX_IEEE1588_PTP = 1 << 9 + #: RX IEEE1588 L2/L4 timestamped packet. + RTE_MBUF_F_RX_IEEE1588_TMST = 1 << 10 + + #: FD id reported if FDIR match. + RTE_MBUF_F_RX_FDIR_ID = 1 << 13 + #: Flexible bytes reported if FDIR match. + RTE_MBUF_F_RX_FDIR_FLX = 1 << 14 + + #: If both RTE_MBUF_F_RX_QINQ_STRIPPED and RTE_MBUF_F_RX_VLAN_STRIPPED are set, the 2 VLANs + #: have been stripped by the hardware. If RTE_MBUF_F_RX_QINQ_STRIPPED is set and + #: RTE_MBUF_F_RX_VLAN_STRIPPED is unset, only the outer VLAN is removed from packet data. + RTE_MBUF_F_RX_QINQ_STRIPPED = auto() + + #: When packets are coalesced by a hardware or virtual driver, this flag can be set in the RX + #: mbuf, meaning that the m->tso_segsz field is valid and is set to the segment size of + #: original packets. + RTE_MBUF_F_RX_LRO = auto() + + #: Indicate that security offload processing was applied on the RX packet. + RTE_MBUF_F_RX_SEC_OFFLOAD = 1 << 18 + #: Indicate that security offload processing failed on the RX packet. + RTE_MBUF_F_RX_SEC_OFFLOAD_FAILED = auto() + + #: The RX packet is a double VLAN. If this flag is set, RTE_MBUF_F_RX_VLAN must also be set. If + #: the flag RTE_MBUF_F_RX_QINQ_STRIPPED is also present, both VLANs headers have been stripped + #: from mbuf data, else they are still present. + RTE_MBUF_F_RX_QINQ = auto() + + #: No info about the outer RX L4 checksum. Value is 0 in the DPDK library. + RTE_MBUF_F_RX_OUTER_L4_CKSUM_UNKNOWN = 1 << 27 + #: The outer L4 checksum in the packet is wrong + RTE_MBUF_F_RX_OUTER_L4_CKSUM_BAD = 1 << 21 + #: The outer L4 checksum in the packet is valid + RTE_MBUF_F_RX_OUTER_L4_CKSUM_GOOD = 1 << 22 + #: Invalid outer L4 checksum state. Value is + #: RTE_MBUF_F_RX_OUTER_L4_CKSUM_BAD | RTE_MBUF_F_RX_OUTER_L4_CKSUM_GOOD in the DPDK library. + RTE_MBUF_F_RX_OUTER_L4_CKSUM_INVALID = 1 << 28 + + # TX flags + + #: Outer UDP checksum offload flag. This flag is used for enabling outer UDP checksum in PMD. + #: To use outer UDP checksum, the user either needs to enable the following in mbuf: + #: + #: a) Fill outer_l2_len and outer_l3_len in mbuf. + #: b) Set the RTE_MBUF_F_TX_OUTER_UDP_CKSUM flag. + #: c) Set the RTE_MBUF_F_TX_OUTER_IPV4 or RTE_MBUF_F_TX_OUTER_IPV6 flag. + #: + #: Or configure RTE_ETH_TX_OFFLOAD_OUTER_UDP_CKSUM offload flag. + RTE_MBUF_F_TX_OUTER_UDP_CKSUM = 1 << 41 + + #: UDP Fragmentation Offload flag. This flag is used for enabling UDP fragmentation in SW or in + #: HW. + RTE_MBUF_F_TX_UDP_SEG = auto() + + #: Request security offload processing on the TX packet. To use Tx security offload, the user + #: needs to fill l2_len in mbuf indicating L2 header size and where L3 header starts. + #: Similarly, l3_len should also be filled along with ol_flags reflecting current L3 type. + RTE_MBUF_F_TX_SEC_OFFLOAD = auto() + + #: Offload the MACsec. This flag must be set by the application to enable this offload feature + #: for a packet to be transmitted. + RTE_MBUF_F_TX_MACSEC = auto() + + # Bits 45:48 are used for the tunnel type in ``lib/mbuf/rte_mbuf_core.h``, but some are modified + # in this Flag to maintain uniqueness. The tunnel type must be specified for TSO or checksum on + # the inner part of tunnel packets. These flags can be used with RTE_MBUF_F_TX_TCP_SEG for TSO, + # or RTE_MBUF_F_TX_xxx_CKSUM. The mbuf fields for inner and outer header lengths are required: + # outer_l2_len, outer_l3_len, l2_len, l3_len, l4_len and tso_segsz for TSO. + + #: + RTE_MBUF_F_TX_TUNNEL_VXLAN = 1 << 45 + #: + RTE_MBUF_F_TX_TUNNEL_GRE = 1 << 46 + #: Value is 3 << 45 in the DPDK library. + RTE_MBUF_F_TX_TUNNEL_IPIP = 1 << 61 + #: + RTE_MBUF_F_TX_TUNNEL_GENEVE = 1 << 47 + #: TX packet with MPLS-in-UDP RFC 7510 header. Value is 5 << 45 in the DPDK library. + RTE_MBUF_F_TX_TUNNEL_MPLSINUDP = 1 << 62 + #: Value is 6 << 45 in the DPDK library. + RTE_MBUF_F_TX_TUNNEL_VXLAN_GPE = 1 << 63 + #: Value is 7 << 45 in the DPDK library. + RTE_MBUF_F_TX_TUNNEL_GTP = 1 << 64 + #: + RTE_MBUF_F_TX_TUNNEL_ESP = 1 << 48 + #: Generic IP encapsulated tunnel type, used for TSO and checksum offload. This can be used for + #: tunnels which are not standards or listed above. It is preferred to use specific tunnel + #: flags like RTE_MBUF_F_TX_TUNNEL_GRE or RTE_MBUF_F_TX_TUNNEL_IPIP if possible. The ethdev + #: must be configured with RTE_ETH_TX_OFFLOAD_IP_TNL_TSO. Outer and inner checksums are done + #: according to the existing flags like RTE_MBUF_F_TX_xxx_CKSUM. Specific tunnel headers that + #: contain payload length, sequence id or checksum are not expected to be updated. Value is + #: 0xD << 45 in the DPDK library. + RTE_MBUF_F_TX_TUNNEL_IP = 1 << 65 + #: Generic UDP encapsulated tunnel type, used for TSO and checksum offload. UDP tunnel type + #: implies outer IP layer. It can be used for tunnels which are not standards or listed above. + #: It is preferred to use specific tunnel flags like RTE_MBUF_F_TX_TUNNEL_VXLAN if possible. + #: The ethdev must be configured with RTE_ETH_TX_OFFLOAD_UDP_TNL_TSO. Outer and inner checksums + #: are done according to the existing flags like RTE_MBUF_F_TX_xxx_CKSUM. Specific tunnel + #: headers that contain payload length, sequence id or checksum are not expected to be updated. + #: value is 0xE << 45 in the DPDK library. + RTE_MBUF_F_TX_TUNNEL_UDP = 1 << 66 + + #: Double VLAN insertion (QinQ) request to driver, driver may offload the insertion based on + #: device capability. Mbuf 'vlan_tci' & 'vlan_tci_outer' must be valid when this flag is set. + RTE_MBUF_F_TX_QINQ = 1 << 49 + + #: TCP segmentation offload. To enable this offload feature for a packet to be transmitted on + #: hardware supporting TSO: + #: + #: - set the RTE_MBUF_F_TX_TCP_SEG flag in mbuf->ol_flags (this flag implies + #: RTE_MBUF_F_TX_TCP_CKSUM) + #: - set the flag RTE_MBUF_F_TX_IPV4 or RTE_MBUF_F_TX_IPV6 + #: * if it's IPv4, set the RTE_MBUF_F_TX_IP_CKSUM flag + #: - fill the mbuf offload information: l2_len, l3_len, l4_len, tso_segsz + RTE_MBUF_F_TX_TCP_SEG = auto() + + #: TX IEEE1588 packet to timestamp. + RTE_MBUF_F_TX_IEEE1588_TMST = auto() + + # Bits 52+53 used for L4 packet type with checksum enabled in ``lib/mbuf/rte_mbuf_core.h`` but + # some values must be modified in this framework to maintain uniqueness. To use hardware + # L4 checksum offload, the user needs to: + # + # - fill l2_len and l3_len in mbuf + # - set the flags RTE_MBUF_F_TX_TCP_CKSUM, RTE_MBUF_F_TX_SCTP_CKSUM or + # RTE_MBUF_F_TX_UDP_CKSUM + # - set the flag RTE_MBUF_F_TX_IPV4 or RTE_MBUF_F_TX_IPV6 + + #: Disable L4 cksum of TX pkt. Value is 0 in the DPDK library. + RTE_MBUF_F_TX_L4_NO_CKSUM = 1 << 67 + #: TCP cksum of TX pkt. Computed by NIC. + RTE_MBUF_F_TX_TCP_CKSUM = 1 << 52 + #: SCTP cksum of TX pkt. Computed by NIC. + RTE_MBUF_F_TX_SCTP_CKSUM = 1 << 53 + #: UDP cksum of TX pkt. Computed by NIC. Value is 3 << 52 in the DPDK library. + RTE_MBUF_F_TX_UDP_CKSUM = 1 << 68 + + #: Offload the IP checksum in the hardware. The flag RTE_MBUF_F_TX_IPV4 should also be set by + #: the application, although a PMD will only check RTE_MBUF_F_TX_IP_CKSUM. + RTE_MBUF_F_TX_IP_CKSUM = 1 << 54 + + #: Packet is IPv4. This flag must be set when using any offload feature (TSO, L3 or L4 + #: checksum) to tell the NIC that the packet is an IPv4 packet. If the packet is a tunneled + #: packet, this flag is related to the inner headers. + RTE_MBUF_F_TX_IPV4 = auto() + #: Packet is IPv6. This flag must be set when using an offload feature (TSO or L4 checksum) to + #: tell the NIC that the packet is an IPv6 packet. If the packet is a tunneled packet, this + #: flag is related to the inner headers. + RTE_MBUF_F_TX_IPV6 = auto() + #: VLAN tag insertion request to driver, driver may offload the insertion based on the device + #: capability. mbuf 'vlan_tci' field must be valid when this flag is set. + RTE_MBUF_F_TX_VLAN = auto() + + #: Offload the IP checksum of an external header in the hardware. The flag + #: RTE_MBUF_F_TX_OUTER_IPV4 should also be set by the application, although a PMD will only + #: check RTE_MBUF_F_TX_OUTER_IP_CKSUM. + RTE_MBUF_F_TX_OUTER_IP_CKSUM = auto() + #: Packet outer header is IPv4. This flag must be set when using any outer offload feature (L3 + #: or L4 checksum) to tell the NIC that the outer header of the tunneled packet is an IPv4 + #: packet. + RTE_MBUF_F_TX_OUTER_IPV4 = auto() + #: Packet outer header is IPv6. This flag must be set when using any outer offload feature (L4 + #: checksum) to tell the NIC that the outer header of the tunneled packet is an IPv6 packet. + RTE_MBUF_F_TX_OUTER_IPV6 = auto() + + @classmethod + def from_list_string(cls, names: str) -> Self: + """Makes a flag from a whitespace-separated list of names. + + Args: + names: a whitespace-separated list containing the members of this flag. + + Returns: + An instance of this flag. + """ + flag = cls(0) + for name in names.split(): + flag |= cls.from_str(name) + return flag + + @classmethod + def from_str(cls, name: str) -> Self: + """Makes a flag matching the supplied name. + + Args: + name: a valid member of this flag in text + Returns: + An instance of this flag. + """ + member_name = name.strip().replace("-", "_") + return cls[member_name] + + @classmethod + def make_parser(cls) -> ParserFn: + """Makes a parser function. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this flag from text. + """ + return TextParser.wrap( + TextParser.find(r"ol_flags: ([^\n]+)"), + cls.from_list_string, + ) + + +class RtePTypes(Flag): + """Flag representing possible packet types in DPDK verbose output. + + Values in this class are derived from definitions in the RTE MBUF ptype library in DPDK located + in ``lib/mbuf/rte_mbuf_ptype.h``. Specifically, the names of values in this class should match + the possible return options from the functions ``rte_get_ptype_*_name`` in ``rte_mbuf_ptype.c``. + + References: + DPDK lib: ``lib/mbuf/rte_mbuf_ptype.h`` + DPDK ptype name formatting functions: ``lib/mbuf/rte_mbuf_ptype.c:rte_get_ptype_*_name()`` + """ + + # L2 + #: Ethernet packet type. This is used for outer packet for tunneling cases. + L2_ETHER = auto() + #: Ethernet packet type for time sync. + L2_ETHER_TIMESYNC = auto() + #: ARP (Address Resolution Protocol) packet type. + L2_ETHER_ARP = auto() + #: LLDP (Link Layer Discovery Protocol) packet type. + L2_ETHER_LLDP = auto() + #: NSH (Network Service Header) packet type. + L2_ETHER_NSH = auto() + #: VLAN packet type. + L2_ETHER_VLAN = auto() + #: QinQ packet type. + L2_ETHER_QINQ = auto() + #: PPPOE packet type. + L2_ETHER_PPPOE = auto() + #: FCoE packet type.. + L2_ETHER_FCOE = auto() + #: MPLS packet type. + L2_ETHER_MPLS = auto() + #: No L2 packet information. + L2_UNKNOWN = auto() + + # L3 + #: IP (Internet Protocol) version 4 packet type. This is used for outer packet for tunneling + #: cases, and does not contain any header option. + L3_IPV4 = auto() + #: IP (Internet Protocol) version 4 packet type. This is used for outer packet for tunneling + #: cases, and contains header options. + L3_IPV4_EXT = auto() + #: IP (Internet Protocol) version 6 packet type. This is used for outer packet for tunneling + #: cases, and does not contain any extension header. + L3_IPV6 = auto() + #: IP (Internet Protocol) version 4 packet type. This is used for outer packet for tunneling + #: cases, and may or maynot contain header options. + L3_IPV4_EXT_UNKNOWN = auto() + #: IP (Internet Protocol) version 6 packet type. This is used for outer packet for tunneling + #: cases, and contains extension headers. + L3_IPV6_EXT = auto() + #: IP (Internet Protocol) version 6 packet type. This is used for outer packet for tunneling + #: cases, and may or maynot contain extension headers. + L3_IPV6_EXT_UNKNOWN = auto() + #: No L3 packet information. + L3_UNKNOWN = auto() + + # L4 + #: TCP (Transmission Control Protocol) packet type. This is used for outer packet for tunneling + #: cases. + L4_TCP = auto() + #: UDP (User Datagram Protocol) packet type. This is used for outer packet for tunneling cases. + L4_UDP = auto() + #: Fragmented IP (Internet Protocol) packet type. This is used for outer packet for tunneling + #: cases and refers to those packets of any IP types which can be recognized as fragmented. A + #: fragmented packet cannot be recognized as any other L4 types (RTE_PTYPE_L4_TCP, + #: RTE_PTYPE_L4_UDP, RTE_PTYPE_L4_SCTP, RTE_PTYPE_L4_ICMP, RTE_PTYPE_L4_NONFRAG). + L4_FRAG = auto() + #: SCTP (Stream Control Transmission Protocol) packet type. This is used for outer packet for + #: tunneling cases. + L4_SCTP = auto() + #: ICMP (Internet Control Message Protocol) packet type. This is used for outer packet for + #: tunneling cases. + L4_ICMP = auto() + #: Non-fragmented IP (Internet Protocol) packet type. This is used for outer packet for + #: tunneling cases and refers to those packets of any IP types, that cannot be recognized as + #: any of the above L4 types (RTE_PTYPE_L4_TCP, RTE_PTYPE_L4_UDP, RTE_PTYPE_L4_FRAG, + #: RTE_PTYPE_L4_SCTP, RTE_PTYPE_L4_ICMP). + L4_NONFRAG = auto() + #: IGMP (Internet Group Management Protocol) packet type. + L4_IGMP = auto() + #: No L4 packet information. + L4_UNKNOWN = auto() + + # Tunnel + #: IP (Internet Protocol) in IP (Internet Protocol) tunneling packet type. + TUNNEL_IP = auto() + #: GRE (Generic Routing Encapsulation) tunneling packet type. + TUNNEL_GRE = auto() + #: VXLAN (Virtual eXtensible Local Area Network) tunneling packet type. + TUNNEL_VXLAN = auto() + #: NVGRE (Network Virtualization using Generic Routing Encapsulation) tunneling packet type. + TUNNEL_NVGRE = auto() + #: GENEVE (Generic Network Virtualization Encapsulation) tunneling packet type. + TUNNEL_GENEVE = auto() + #: Tunneling packet type of Teredo, VXLAN (Virtual eXtensible Local Area Network) or GRE + #: (Generic Routing Encapsulation) could be recognized as this packet type, if they can not be + #: recognized independently as of hardware capability. + TUNNEL_GRENAT = auto() + #: GTP-C (GPRS Tunnelling Protocol) control tunneling packet type. + TUNNEL_GTPC = auto() + #: GTP-U (GPRS Tunnelling Protocol) user data tunneling packet type. + TUNNEL_GTPU = auto() + #: ESP (IP Encapsulating Security Payload) tunneling packet type. + TUNNEL_ESP = auto() + #: L2TP (Layer 2 Tunneling Protocol) tunneling packet type. + TUNNEL_L2TP = auto() + #: VXLAN-GPE (VXLAN Generic Protocol Extension) tunneling packet type. + TUNNEL_VXLAN_GPE = auto() + #: MPLS-in-UDP tunneling packet type (RFC 7510). + TUNNEL_MPLS_IN_UDP = auto() + #: MPLS-in-GRE tunneling packet type (RFC 4023). + TUNNEL_MPLS_IN_GRE = auto() + #: No tunnel information found on the packet. + TUNNEL_UNKNOWN = auto() + + # Inner L2 + #: Ethernet packet type. This is used for inner packet type only. + INNER_L2_ETHER = auto() + #: Ethernet packet type with VLAN (Virtual Local Area Network) tag. + INNER_L2_ETHER_VLAN = auto() + #: QinQ packet type. + INNER_L2_ETHER_QINQ = auto() + #: No inner L2 information found on the packet. + INNER_L2_UNKNOWN = auto() + + # Inner L3 + #: IP (Internet Protocol) version 4 packet type. This is used for inner packet only, and does + #: not contain any header option. + INNER_L3_IPV4 = auto() + #: IP (Internet Protocol) version 4 packet type. This is used for inner packet only, and + #: contains header options. + INNER_L3_IPV4_EXT = auto() + #: IP (Internet Protocol) version 6 packet type. This is used for inner packet only, and does + #: not contain any extension header. + INNER_L3_IPV6 = auto() + #: IP (Internet Protocol) version 4 packet type. This is used for inner packet only, and may or + #: may not contain header options. + INNER_L3_IPV4_EXT_UNKNOWN = auto() + #: IP (Internet Protocol) version 6 packet type. This is used for inner packet only, and + #: contains extension headers. + INNER_L3_IPV6_EXT = auto() + #: IP (Internet Protocol) version 6 packet type. This is used for inner packet only, and may or + #: may not contain extension headers. + INNER_L3_IPV6_EXT_UNKNOWN = auto() + #: No inner L3 information found on the packet. + INNER_L3_UNKNOWN = auto() + + # Inner L4 + #: TCP (Transmission Control Protocol) packet type. This is used for inner packet only. + INNER_L4_TCP = auto() + #: UDP (User Datagram Protocol) packet type. This is used for inner packet only. + INNER_L4_UDP = auto() + #: Fragmented IP (Internet Protocol) packet type. This is used for inner packet only, and may + #: or maynot have a layer 4 packet. + INNER_L4_FRAG = auto() + #: SCTP (Stream Control Transmission Protocol) packet type. This is used for inner packet only. + INNER_L4_SCTP = auto() + #: ICMP (Internet Control Message Protocol) packet type. This is used for inner packet only. + INNER_L4_ICMP = auto() + #: Non-fragmented IP (Internet Protocol) packet type. It is used for inner packet only, and may + #: or may not have other unknown layer 4 packet types. + INNER_L4_NONFRAG = auto() + #: No inner L4 information found on the packet. + INNER_L4_UNKNOWN = auto() + + @classmethod + def from_list_string(cls, names: str) -> Self: + """Makes a flag from a whitespace-separated list of names. + + Args: + names: a whitespace-separated list containing the members of this flag. + + Returns: + An instance of this flag. + """ + flag = cls(0) + for name in names.split(): + flag |= cls.from_str(name) + return flag + + @classmethod + def from_str(cls, name: str) -> Self: + """Makes a flag matching the supplied name. + + Args: + name: a valid member of this flag in text + Returns: + An instance of this flag. + """ + member_name = name.strip().replace("-", "_") + return cls[member_name] + + @classmethod + def make_parser(cls, hw: bool) -> ParserFn: + """Makes a parser function. + + Args: + hw: Whether to make a parser for hardware ptypes or software ptypes. If :data:`True`, + hardware ptypes will be collected, otherwise software pytpes will. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this flag from text. + """ + return TextParser.wrap( + TextParser.find(f"{'hw' if hw else 'sw'} ptype: ([^-]+)"), + cls.from_list_string, + ) + + +@dataclass +class TestPmdVerbosePacket(TextParser): + """Packet information provided by verbose output in Testpmd. + + This dataclass expects that packet information be prepended with the starting line of packet + bursts. Specifically, the line that reads "port X/queue Y: sent/received Z packets". + """ + + #: ID of the port that handled the packet. + port_id: int = field(metadata=TextParser.find_int(r"port (\d+)/queue \d+")) + #: ID of the queue that handled the packet. + queue_id: int = field(metadata=TextParser.find_int(r"port \d+/queue (\d+)")) + #: Whether the packet was received or sent by the queue/port. + was_received: bool = field(metadata=TextParser.find(r"received \d+ packets")) + #: + src_mac: str = field(metadata=TextParser.find(f"src=({REGEX_FOR_MAC_ADDRESS})")) + #: + dst_mac: str = field(metadata=TextParser.find(f"dst=({REGEX_FOR_MAC_ADDRESS})")) + #: Memory pool the packet was handled on. + pool: str = field(metadata=TextParser.find(r"pool=(\S+)")) + #: Packet type in hex. + p_type: int = field(metadata=TextParser.find_int(r"type=(0x[a-fA-F\d]+)")) + #: + length: int = field(metadata=TextParser.find_int(r"length=(\d+)")) + #: Number of segments in the packet. + nb_segs: int = field(metadata=TextParser.find_int(r"nb_segs=(\d+)")) + #: Hardware packet type. + hw_ptype: RtePTypes = field(metadata=RtePTypes.make_parser(hw=True)) + #: Software packet type. + sw_ptype: RtePTypes = field(metadata=RtePTypes.make_parser(hw=False)) + #: + l2_len: int = field(metadata=TextParser.find_int(r"l2_len=(\d+)")) + #: + ol_flags: PacketOffloadFlag = field(metadata=PacketOffloadFlag.make_parser()) + #: RSS hash of the packet in hex. + rss_hash: int | None = field( + default=None, metadata=TextParser.find_int(r"RSS hash=(0x[a-fA-F\d]+)") + ) + #: RSS queue that handled the packet in hex. + rss_queue: int | None = field( + default=None, metadata=TextParser.find_int(r"RSS queue=(0x[a-fA-F\d]+)") + ) + #: + l3_len: int | None = field(default=None, metadata=TextParser.find_int(r"l3_len=(\d+)")) + #: + l4_len: int | None = field(default=None, metadata=TextParser.find_int(r"l4_len=(\d+)")) + #: + l4_dport: int | None = field( + default=None, + metadata=TextParser.find_int(r"Destination (?:TCP|UDP) port=(\d+)"), + ) + + +class RxOffloadCapability(Flag): + """Rx offload capabilities of a device. + + The flags are taken from ``lib/ethdev/rte_ethdev.h``. + They're prefixed with ``RTE_ETH_RX_OFFLOAD`` in ``lib/ethdev/rte_ethdev.h`` + instead of ``RX_OFFLOAD``, which is what testpmd changes the prefix to. + The values are not contiguous, so the correspondence is preserved + by specifying concrete values interspersed between auto() values. + + The ``RX_OFFLOAD`` prefix has been preserved so that the same flag names can be used + in :class:`NicCapability`. The prefix is needed in :class:`NicCapability` since there's + no other qualifier which would sufficiently distinguish it from other capabilities. + + References: + DPDK lib: ``lib/ethdev/rte_ethdev.h`` + testpmd display function: ``app/test-pmd/cmdline.c:print_rx_offloads()`` + """ + + #: + RX_OFFLOAD_VLAN_STRIP = auto() + #: Device supports L3 checksum offload. + RX_OFFLOAD_IPV4_CKSUM = auto() + #: Device supports L4 checksum offload. + RX_OFFLOAD_UDP_CKSUM = auto() + #: Device supports L4 checksum offload. + RX_OFFLOAD_TCP_CKSUM = auto() + #: Device supports Large Receive Offload. + RX_OFFLOAD_TCP_LRO = auto() + #: Device supports QinQ (queue in queue) offload. + RX_OFFLOAD_QINQ_STRIP = auto() + #: Device supports inner packet L3 checksum. + RX_OFFLOAD_OUTER_IPV4_CKSUM = auto() + #: Device supports MACsec. + RX_OFFLOAD_MACSEC_STRIP = auto() + #: Device supports filtering of a VLAN Tag identifier. + RX_OFFLOAD_VLAN_FILTER = 1 << 9 + #: Device supports VLAN offload. + RX_OFFLOAD_VLAN_EXTEND = auto() + #: Device supports receiving segmented mbufs. + RX_OFFLOAD_SCATTER = 1 << 13 + #: Device supports Timestamp. + RX_OFFLOAD_TIMESTAMP = auto() + #: Device supports crypto processing while packet is received in NIC. + RX_OFFLOAD_SECURITY = auto() + #: Device supports CRC stripping. + RX_OFFLOAD_KEEP_CRC = auto() + #: Device supports L4 checksum offload. + RX_OFFLOAD_SCTP_CKSUM = auto() + #: Device supports inner packet L4 checksum. + RX_OFFLOAD_OUTER_UDP_CKSUM = auto() + #: Device supports RSS hashing. + RX_OFFLOAD_RSS_HASH = auto() + #: Device supports + RX_OFFLOAD_BUFFER_SPLIT = auto() + #: Device supports all checksum capabilities. + RX_OFFLOAD_CHECKSUM = RX_OFFLOAD_IPV4_CKSUM | RX_OFFLOAD_UDP_CKSUM | RX_OFFLOAD_TCP_CKSUM + #: Device supports all VLAN capabilities. + RX_OFFLOAD_VLAN = ( + RX_OFFLOAD_VLAN_STRIP + | RX_OFFLOAD_VLAN_FILTER + | RX_OFFLOAD_VLAN_EXTEND + | RX_OFFLOAD_QINQ_STRIP + ) + + @classmethod + def from_string(cls, line: str) -> Self: + """Make an instance from a string containing the flag names separated with a space. + + Args: + line: The line to parse. + + Returns: + A new instance containing all found flags. + """ + flag = cls(0) + for flag_name in line.split(): + flag |= cls[f"RX_OFFLOAD_{flag_name}"] + return flag + + @classmethod + def make_parser(cls, per_port: bool) -> ParserFn: + """Make a parser function. + + Args: + per_port: If :data:`True`, will return capabilities per port. If :data:`False`, + will return capabilities per queue. + + Returns: + ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a + parser function that makes an instance of this flag from text. + """ + granularity = "Port" if per_port else "Queue" + return TextParser.wrap( + TextParser.find(rf"Per {granularity}\s+:(.*)$", re.MULTILINE), + cls.from_string, + ) + + +@dataclass +class RxOffloadCapabilities(TextParser): + """The result of testpmd's ``show port <port_id> rx_offload capabilities`` command. + + References: + testpmd command function: ``app/test-pmd/cmdline.c:cmd_rx_offload_get_capa()`` + testpmd display function: ``app/test-pmd/cmdline.c:cmd_rx_offload_get_capa_parsed()`` + """ + + #: + port_id: int = field( + metadata=TextParser.find_int(r"Rx Offloading Capabilities of port (\d+) :") + ) + #: Per-queue Rx offload capabilities. + per_queue: RxOffloadCapability = field(metadata=RxOffloadCapability.make_parser(False)) + #: Capabilities other than per-queue Rx offload capabilities. + per_port: RxOffloadCapability = field(metadata=RxOffloadCapability.make_parser(True)) + + +@dataclass +class TestPmdPortFlowCtrl(TextParser): + """Class representing a port's flow control parameters. + + The parameters can also be parsed from the output of ``show port <port_id> flow_ctrl``. + """ + + #: Enable Reactive Extensions. + rx: bool = field(default=False, metadata=TextParser.find(r"Rx pause: on")) + #: Enable Transmit. + tx: bool = field(default=False, metadata=TextParser.find(r"Tx pause: on")) + #: High threshold value to trigger XOFF. + high_water: int = field( + default=0, metadata=TextParser.find_int(r"High waterline: (0x[a-fA-F\d]+)") + ) + #: Low threshold value to trigger XON. + low_water: int = field( + default=0, metadata=TextParser.find_int(r"Low waterline: (0x[a-fA-F\d]+)") + ) + #: Pause quota in the Pause frame. + pause_time: int = field(default=0, metadata=TextParser.find_int(r"Pause time: (0x[a-fA-F\d]+)")) + #: Send XON frame. + send_xon: bool = field(default=False, metadata=TextParser.find(r"Tx pause: on")) + #: Enable receiving MAC control frames. + mac_ctrl_frame_fwd: bool = field(default=False, metadata=TextParser.find(r"Tx pause: on")) + #: Change the auto-negotiation parameter. + autoneg: bool = field(default=False, metadata=TextParser.find(r"Autoneg: on")) + + def __str__(self) -> str: + """Returns the string representation of this instance.""" + ret = ( + f"rx {'on' if self.rx else 'off'} " + f"tx {'on' if self.tx else 'off'} " + f"{self.high_water} " + f"{self.low_water} " + f"{self.pause_time} " + f"{1 if self.send_xon else 0} " + f"mac_ctrl_frame_fwd {'on' if self.mac_ctrl_frame_fwd else 'off'} " + f"autoneg {'on' if self.autoneg else 'off'}" + ) + return ret diff --git a/dts/framework/config/__init__.py b/dts/framework/config/__init__.py index 1ec744d1d4..d2f0138e4a 100644 --- a/dts/framework/config/__init__.py +++ b/dts/framework/config/__init__.py @@ -28,7 +28,6 @@ and makes it thread safe should we ever want to move in that direction. """ -import os from pathlib import Path from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeVar, cast @@ -43,7 +42,7 @@ from .test_run import TestRunConfiguration, create_test_suites_config_model # Import only if type checking or building docs, to prevent circular imports. -if TYPE_CHECKING or os.environ.get("DTS_DOC_BUILD"): +if TYPE_CHECKING: from framework.test_suite import BaseConfig NodesConfig = Annotated[list[NodeConfiguration], Field(min_length=1)] diff --git a/dts/framework/params/eal.py b/dts/framework/params/eal.py index b90ff33dcf..e84a20f02f 100644 --- a/dts/framework/params/eal.py +++ b/dts/framework/params/eal.py @@ -4,15 +4,17 @@ """Module representing the DPDK EAL-related parameters.""" from dataclasses import dataclass, field -from typing import Literal +from typing import TYPE_CHECKING, Literal from framework.params import Params, Switch from framework.testbed_model.cpu import LogicalCoreList -from framework.testbed_model.port import Port from framework.testbed_model.virtual_device import VirtualDevice +if TYPE_CHECKING: + from framework.testbed_model.port import Port -def _port_to_pci(port: Port) -> str: + +def _port_to_pci(port: "Port") -> str: return port.pci @@ -42,11 +44,11 @@ class EalParams(Params): vdevs: list[VirtualDevice] | None = field( default=None, metadata=Params.multiple() | Params.long("vdev") ) - allowed_ports: list[Port] | None = field( + allowed_ports: list["Port"] | None = field( default=None, metadata=Params.convert_value(_port_to_pci) | Params.multiple() | Params.short("a"), ) - blocked_ports: list[Port] | None = field( + blocked_ports: list["Port"] | None = field( default=None, metadata=Params.convert_value(_port_to_pci) | Params.multiple() | Params.short("b"), ) diff --git a/dts/framework/params/types.py b/dts/framework/params/types.py index 87d11502e8..5bc4bd37d9 100644 --- a/dts/framework/params/types.py +++ b/dts/framework/params/types.py @@ -15,8 +15,7 @@ def create_testpmd(**kwargs: Unpack[TestPmdParamsDict]): from pathlib import PurePath from typing import TypedDict -from framework.params import Switch, YesNoSwitch -from framework.params.testpmd import ( +from api.testpmd.config import ( AnonMempoolAllocationMode, EthPeer, Event, @@ -37,6 +36,7 @@ def create_testpmd(**kwargs: Unpack[TestPmdParamsDict]): TXRingParams, TxUDPPortPair, ) +from framework.params import Switch, YesNoSwitch from framework.testbed_model.cpu import LogicalCoreList from framework.testbed_model.port import Port from framework.testbed_model.virtual_device import VirtualDevice diff --git a/dts/framework/remote_session/__init__.py b/dts/framework/remote_session/__init__.py index 1a5cf6abd3..394c88b25e 100644 --- a/dts/framework/remote_session/__init__.py +++ b/dts/framework/remote_session/__init__.py @@ -11,47 +11,3 @@ The interactive sessions open an interactive shell which is continuously open, allowing it to send and receive data within that particular shell. """ - -from framework.config.node import NodeConfiguration -from framework.logger import DTSLogger - -from .interactive_remote_session import InteractiveRemoteSession -from .remote_session import RemoteSession -from .ssh_session import SSHSession - - -def create_remote_session( - node_config: NodeConfiguration, name: str, logger: DTSLogger -) -> RemoteSession: - """Factory for non-interactive remote sessions. - - The function returns an SSH session, but will be extended if support - for other protocols is added. - - Args: - node_config: The test run configuration of the node to connect to. - name: The name of the session. - logger: The logger instance this session will use. - - Returns: - The SSH remote session. - """ - return SSHSession(node_config, name, logger) - - -def create_interactive_session( - node_config: NodeConfiguration, logger: DTSLogger -) -> InteractiveRemoteSession: - """Factory for interactive remote sessions. - - The function returns an interactive SSH session, but will be extended if support - for other protocols is added. - - Args: - node_config: The test run configuration of the node to connect to. - logger: The logger instance this session will use. - - Returns: - The interactive SSH remote session. - """ - return InteractiveRemoteSession(node_config, logger) diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py deleted file mode 100644 index ad8cb273dc..0000000000 --- a/dts/framework/remote_session/testpmd_shell.py +++ /dev/null @@ -1,2844 +0,0 @@ -# SPDX-License-Identifier: BSD-3-Clause -# Copyright(c) 2023 University of New Hampshire -# Copyright(c) 2023 PANTHEON.tech s.r.o. -# Copyright(c) 2024 Arm Limited - -"""Testpmd interactive shell. - -Typical usage example in a TestSuite:: - - testpmd_shell = TestPmdShell(self.sut_node) - devices = testpmd_shell.get_devices() - for device in devices: - print(device) - testpmd_shell.close() -""" - -import functools -import re -import time -from collections.abc import Callable, MutableSet -from dataclasses import dataclass, field -from enum import Flag, auto -from os import environ -from pathlib import PurePath -from typing import TYPE_CHECKING, Any, ClassVar, Concatenate, Literal, ParamSpec, Tuple, TypeAlias - -from framework.context import get_ctx -from framework.remote_session.interactive_shell import only_active -from framework.testbed_model.topology import TopologyType - -if TYPE_CHECKING or environ.get("DTS_DOC_BUILD"): - from enum import Enum as NoAliasEnum -else: - from aenum import NoAliasEnum - -from typing_extensions import Self, Unpack - -from framework.exception import InteractiveCommandExecutionError, InternalError -from framework.params.testpmd import PortTopology, SimpleForwardingModes, TestPmdParams -from framework.params.types import TestPmdParamsDict -from framework.parser import ParserFn, TextParser -from framework.remote_session.dpdk_shell import DPDKShell -from framework.settings import SETTINGS -from framework.utils import REGEX_FOR_MAC_ADDRESS, StrEnum - -P = ParamSpec("P") -TestPmdShellMethod = Callable[Concatenate["TestPmdShell", P], Any] - -TestPmdShellCapabilityMethod: TypeAlias = Callable[ - ["TestPmdShell", MutableSet["NicCapability"], MutableSet["NicCapability"]], None -] - -TestPmdShellDecorator: TypeAlias = Callable[[TestPmdShellMethod], TestPmdShellMethod] - -TestPmdShellNicCapability = tuple[TestPmdShellCapabilityMethod, TestPmdShellDecorator | None] - - -class TestPmdDevice: - """The data of a device that testpmd can recognize. - - Attributes: - pci_address: The PCI address of the device. - """ - - pci_address: str - - def __init__(self, pci_address_line: str): - """Initialize the device from the testpmd output line string. - - Args: - pci_address_line: A line of testpmd output that contains a device. - """ - self.pci_address = pci_address_line.strip().split(": ")[1].strip() - - def __str__(self) -> str: - """The PCI address captures what the device is.""" - return self.pci_address - - -class VLANOffloadFlag(Flag): - """Flag representing the VLAN offload settings of a NIC port.""" - - #: - STRIP = auto() - #: - FILTER = auto() - #: - EXTEND = auto() - #: - QINQ_STRIP = auto() - - @classmethod - def from_str_dict(cls, d): - """Makes an instance from a dict containing the flag member names with an "on" value. - - Args: - d: A dictionary containing the flag members as keys and any string value. - - Returns: - A new instance of the flag. - """ - flag = cls(0) - for name in cls.__members__: - if d.get(name) == "on": - flag |= cls[name] - return flag - - @classmethod - def make_parser(cls) -> ParserFn: - """Makes a parser function. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this flag from text. - """ - return TextParser.wrap( - TextParser.find( - r"VLAN offload:\s+" - r"strip (?P<STRIP>on|off), " - r"filter (?P<FILTER>on|off), " - r"extend (?P<EXTEND>on|off), " - r"qinq strip (?P<QINQ_STRIP>on|off)", - re.MULTILINE, - named=True, - ), - cls.from_str_dict, - ) - - -class ChecksumOffloadOptions(Flag): - """Flag representing checksum hardware offload layer options.""" - - #: - ip = auto() - #: - udp = auto() - #: - tcp = auto() - #: - sctp = auto() - #: - outer_ip = auto() - #: - outer_udp = auto() - - -class RSSOffloadTypesFlag(Flag): - """Flag representing the RSS offload flow types supported by the NIC port.""" - - #: - ipv4 = auto() - #: - ipv4_frag = auto() - #: - ipv4_tcp = auto() - #: - ipv4_udp = auto() - #: - ipv4_sctp = auto() - #: - ipv4_other = auto() - #: - ipv6 = auto() - #: - ipv6_frag = auto() - #: - ipv6_tcp = auto() - #: - ipv6_udp = auto() - #: - ipv6_sctp = auto() - #: - ipv6_other = auto() - #: - l2_payload = auto() - #: - ipv6_ex = auto() - #: - ipv6_tcp_ex = auto() - #: - ipv6_udp_ex = auto() - #: - port = auto() - #: - vxlan = auto() - #: - geneve = auto() - #: - nvgre = auto() - #: - user_defined_22 = auto() - #: - gtpu = auto() - #: - eth = auto() - #: - s_vlan = auto() - #: - c_vlan = auto() - #: - esp = auto() - #: - ah = auto() - #: - l2tpv3 = auto() - #: - pfcp = auto() - #: - pppoe = auto() - #: - ecpri = auto() - #: - mpls = auto() - #: - ipv4_chksum = auto() - #: - l4_chksum = auto() - #: - l2tpv2 = auto() - #: - ipv6_flow_label = auto() - #: - user_defined_38 = auto() - #: - user_defined_39 = auto() - #: - user_defined_40 = auto() - #: - user_defined_41 = auto() - #: - user_defined_42 = auto() - #: - user_defined_43 = auto() - #: - user_defined_44 = auto() - #: - user_defined_45 = auto() - #: - user_defined_46 = auto() - #: - user_defined_47 = auto() - #: - user_defined_48 = auto() - #: - user_defined_49 = auto() - #: - user_defined_50 = auto() - #: - user_defined_51 = auto() - #: - l3_pre96 = auto() - #: - l3_pre64 = auto() - #: - l3_pre56 = auto() - #: - l3_pre48 = auto() - #: - l3_pre40 = auto() - #: - l3_pre32 = auto() - #: - l2_dst_only = auto() - #: - l2_src_only = auto() - #: - l4_dst_only = auto() - #: - l4_src_only = auto() - #: - l3_dst_only = auto() - #: - l3_src_only = auto() - - #: - ip = ipv4 | ipv4_frag | ipv4_other | ipv6 | ipv6_frag | ipv6_other | ipv6_ex - #: - udp = ipv4_udp | ipv6_udp | ipv6_udp_ex - #: - tcp = ipv4_tcp | ipv6_tcp | ipv6_tcp_ex - #: - sctp = ipv4_sctp | ipv6_sctp - #: - tunnel = vxlan | geneve | nvgre - #: - vlan = s_vlan | c_vlan - #: - all = ( - eth - | vlan - | ip - | tcp - | udp - | sctp - | l2_payload - | l2tpv3 - | esp - | ah - | pfcp - | gtpu - | ecpri - | mpls - | l2tpv2 - ) - - @classmethod - def from_list_string(cls, names: str) -> Self: - """Makes a flag from a whitespace-separated list of names. - - Args: - names: a whitespace-separated list containing the members of this flag. - - Returns: - An instance of this flag. - """ - flag = cls(0) - for name in names.split(): - flag |= cls.from_str(name) - return flag - - @classmethod - def from_str(cls, name: str) -> Self: - """Makes a flag matching the supplied name. - - Args: - name: a valid member of this flag in text - Returns: - An instance of this flag. - """ - member_name = name.strip().replace("-", "_") - return cls[member_name] - - @classmethod - def make_parser(cls) -> ParserFn: - """Makes a parser function. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this flag from text. - """ - return TextParser.wrap( - TextParser.find(r"Supported RSS offload flow types:((?:\r?\n? \S+)+)", re.MULTILINE), - RSSOffloadTypesFlag.from_list_string, - ) - - -class DeviceCapabilitiesFlag(Flag): - """Flag representing the device capabilities.""" - - #: Device supports Rx queue setup after device started. - RUNTIME_RX_QUEUE_SETUP = auto() - #: Device supports Tx queue setup after device started. - RUNTIME_TX_QUEUE_SETUP = auto() - #: Device supports shared Rx queue among ports within Rx domain and switch domain. - RXQ_SHARE = auto() - #: Device supports keeping flow rules across restart. - FLOW_RULE_KEEP = auto() - #: Device supports keeping shared flow objects across restart. - FLOW_SHARED_OBJECT_KEEP = auto() - - @classmethod - def make_parser(cls) -> ParserFn: - """Makes a parser function. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this flag from text. - """ - return TextParser.wrap( - TextParser.find_int(r"Device capabilities: (0x[A-Fa-f\d]+)"), - cls, - ) - - -class DeviceErrorHandlingMode(StrEnum): - """Enum representing the device error handling mode.""" - - #: - none = auto() - #: - passive = auto() - #: - proactive = auto() - #: - unknown = auto() - - @classmethod - def make_parser(cls) -> ParserFn: - """Makes a parser function. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this enum from text. - """ - return TextParser.wrap(TextParser.find(r"Device error handling mode: (\w+)"), cls) - - -def make_device_private_info_parser() -> ParserFn: - """Device private information parser. - - Ensures that we are not parsing invalid device private info output. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a parser - function that parses the device private info from the TestPmd port info output. - """ - - def _validate(info: str): - info = info.strip() - if info == "none" or info.startswith("Invalid file") or info.startswith("Failed to dump"): - return None - return info - - return TextParser.wrap(TextParser.find(r"Device private info:\s+([\s\S]+)"), _validate) - - -class RxQueueState(StrEnum): - """RX queue states. - - References: - DPDK lib: ``lib/ethdev/rte_ethdev.h`` - testpmd display function: ``app/test-pmd/config.c:get_queue_state_name()`` - """ - - #: - stopped = auto() - #: - started = auto() - #: - hairpin = auto() - #: - unknown = auto() - - @classmethod - def make_parser(cls) -> ParserFn: - """Makes a parser function. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this enum from text. - """ - return TextParser.wrap(TextParser.find(r"Rx queue state: ([^\r\n]+)"), cls) - - -@dataclass -class TestPmdQueueInfo(TextParser): - """Dataclass representation of the common parts of the testpmd `show rxq/txq info` commands.""" - - #: - prefetch_threshold: int = field(metadata=TextParser.find_int(r"prefetch threshold: (\d+)")) - #: - host_threshold: int = field(metadata=TextParser.find_int(r"host threshold: (\d+)")) - #: - writeback_threshold: int = field(metadata=TextParser.find_int(r"writeback threshold: (\d+)")) - #: - free_threshold: int = field(metadata=TextParser.find_int(r"free threshold: (\d+)")) - #: - deferred_start: bool = field(metadata=TextParser.find("deferred start: on")) - #: The number of RXD/TXDs is just the ring size of the queue. - ring_size: int = field(metadata=TextParser.find_int(r"Number of (?:RXDs|TXDs): (\d+)")) - #: - is_queue_started: bool = field(metadata=TextParser.find("queue state: started")) - #: - burst_mode: str | None = field( - default=None, metadata=TextParser.find(r"Burst mode: ([^\r\n]+)") - ) - - -@dataclass -class TestPmdTxqInfo(TestPmdQueueInfo): - """Representation of testpmd's ``show txq info <port_id> <queue_id>`` command. - - References: - testpmd command function: ``app/test-pmd/cmdline.c:cmd_showqueue()`` - testpmd display function: ``app/test-pmd/config.c:rx_queue_infos_display()`` - """ - - #: Ring size threshold - rs_threshold: int | None = field( - default=None, metadata=TextParser.find_int(r"TX RS threshold: (\d+)\b") - ) - - -@dataclass -class TestPmdRxqInfo(TestPmdQueueInfo): - """Representation of testpmd's ``show rxq info <port_id> <queue_id>`` command. - - References: - testpmd command function: ``app/test-pmd/cmdline.c:cmd_showqueue()`` - testpmd display function: ``app/test-pmd/config.c:rx_queue_infos_display()`` - """ - - #: Mempool used by that queue - mempool: str | None = field(default=None, metadata=TextParser.find(r"Mempool: ([^\r\n]+)")) - #: Drop packets if no descriptors are available - drop_packets: bool | None = field( - default=None, metadata=TextParser.find(r"RX drop packets: on") - ) - #: Scattered packets Rx enabled - scattered_packets: bool | None = field( - default=None, metadata=TextParser.find(r"RX scattered packets: on") - ) - #: The state of the queue - queue_state: str | None = field(default=None, metadata=RxQueueState.make_parser()) - - -@dataclass -class TestPmdPort(TextParser): - """Dataclass representing the result of testpmd's ``show port info`` command.""" - - #: - id: int = field(metadata=TextParser.find_int(r"Infos for port (\d+)\b")) - #: - device_name: str = field(metadata=TextParser.find(r"Device name: ([^\r\n]+)")) - #: - driver_name: str = field(metadata=TextParser.find(r"Driver name: ([^\r\n]+)")) - #: - socket_id: int = field(metadata=TextParser.find_int(r"Connect to socket: (\d+)")) - #: - is_link_up: bool = field(metadata=TextParser.find("Link status: up")) - #: - link_speed: str = field(metadata=TextParser.find(r"Link speed: ([^\r\n]+)")) - #: - is_link_full_duplex: bool = field(metadata=TextParser.find("Link duplex: full-duplex")) - #: - is_link_autonegotiated: bool = field(metadata=TextParser.find("Autoneg status: On")) - #: - is_promiscuous_mode_enabled: bool = field(metadata=TextParser.find("Promiscuous mode: enabled")) - #: - is_allmulticast_mode_enabled: bool = field( - metadata=TextParser.find("Allmulticast mode: enabled") - ) - #: Maximum number of MAC addresses - max_mac_addresses_num: int = field( - metadata=TextParser.find_int(r"Maximum number of MAC addresses: (\d+)") - ) - #: Maximum configurable length of RX packet - max_hash_mac_addresses_num: int = field( - metadata=TextParser.find_int(r"Maximum number of MAC addresses of hash filtering: (\d+)") - ) - #: Minimum size of RX buffer - min_rx_bufsize: int = field(metadata=TextParser.find_int(r"Minimum size of RX buffer: (\d+)")) - #: Maximum configurable length of RX packet - max_rx_packet_length: int = field( - metadata=TextParser.find_int(r"Maximum configurable length of RX packet: (\d+)") - ) - #: Maximum configurable size of LRO aggregated packet - max_lro_packet_size: int = field( - metadata=TextParser.find_int(r"Maximum configurable size of LRO aggregated packet: (\d+)") - ) - - #: Current number of RX queues - rx_queues_num: int = field(metadata=TextParser.find_int(r"Current number of RX queues: (\d+)")) - #: Max possible RX queues - max_rx_queues_num: int = field(metadata=TextParser.find_int(r"Max possible RX queues: (\d+)")) - #: Max possible number of RXDs per queue - max_queue_rxd_num: int = field( - metadata=TextParser.find_int(r"Max possible number of RXDs per queue: (\d+)") - ) - #: Min possible number of RXDs per queue - min_queue_rxd_num: int = field( - metadata=TextParser.find_int(r"Min possible number of RXDs per queue: (\d+)") - ) - #: RXDs number alignment - rxd_alignment_num: int = field(metadata=TextParser.find_int(r"RXDs number alignment: (\d+)")) - - #: Current number of TX queues - tx_queues_num: int = field(metadata=TextParser.find_int(r"Current number of TX queues: (\d+)")) - #: Max possible TX queues - max_tx_queues_num: int = field(metadata=TextParser.find_int(r"Max possible TX queues: (\d+)")) - #: Max possible number of TXDs per queue - max_queue_txd_num: int = field( - metadata=TextParser.find_int(r"Max possible number of TXDs per queue: (\d+)") - ) - #: Min possible number of TXDs per queue - min_queue_txd_num: int = field( - metadata=TextParser.find_int(r"Min possible number of TXDs per queue: (\d+)") - ) - #: TXDs number alignment - txd_alignment_num: int = field(metadata=TextParser.find_int(r"TXDs number alignment: (\d+)")) - #: Max segment number per packet - max_packet_segment_num: int = field( - metadata=TextParser.find_int(r"Max segment number per packet: (\d+)") - ) - #: Max segment number per MTU/TSO - max_mtu_segment_num: int = field( - metadata=TextParser.find_int(r"Max segment number per MTU\/TSO: (\d+)") - ) - - #: - device_capabilities: DeviceCapabilitiesFlag = field( - metadata=DeviceCapabilitiesFlag.make_parser(), - ) - #: - device_error_handling_mode: DeviceErrorHandlingMode | None = field( - default=None, metadata=DeviceErrorHandlingMode.make_parser() - ) - #: - device_private_info: str | None = field( - default=None, - metadata=make_device_private_info_parser(), - ) - - #: - hash_key_size: int | None = field( - default=None, metadata=TextParser.find_int(r"Hash key size in bytes: (\d+)") - ) - #: - redirection_table_size: int | None = field( - default=None, metadata=TextParser.find_int(r"Redirection table size: (\d+)") - ) - #: - supported_rss_offload_flow_types: RSSOffloadTypesFlag = field( - default=RSSOffloadTypesFlag(0), metadata=RSSOffloadTypesFlag.make_parser() - ) - - #: - mac_address: str | None = field( - default=None, metadata=TextParser.find(r"MAC address: ([A-Fa-f0-9:]+)") - ) - #: - fw_version: str | None = field( - default=None, metadata=TextParser.find(r"Firmware-version: ([^\r\n]+)") - ) - #: - dev_args: str | None = field(default=None, metadata=TextParser.find(r"Devargs: ([^\r\n]+)")) - #: Socket id of the memory allocation - mem_alloc_socket_id: int | None = field( - default=None, - metadata=TextParser.find_int(r"memory allocation on the socket: (\d+)"), - ) - #: - mtu: int | None = field(default=None, metadata=TextParser.find_int(r"MTU: (\d+)")) - - #: - vlan_offload: VLANOffloadFlag | None = field( - default=None, - metadata=VLANOffloadFlag.make_parser(), - ) - - #: Maximum size of RX buffer - max_rx_bufsize: int | None = field( - default=None, metadata=TextParser.find_int(r"Maximum size of RX buffer: (\d+)") - ) - #: Maximum number of VFs - max_vfs_num: int | None = field( - default=None, metadata=TextParser.find_int(r"Maximum number of VFs: (\d+)") - ) - #: Maximum number of VMDq pools - max_vmdq_pools_num: int | None = field( - default=None, metadata=TextParser.find_int(r"Maximum number of VMDq pools: (\d+)") - ) - - #: - switch_name: str | None = field( - default=None, metadata=TextParser.find(r"Switch name: ([\r\n]+)") - ) - #: - switch_domain_id: int | None = field( - default=None, metadata=TextParser.find_int(r"Switch domain Id: (\d+)") - ) - #: - switch_port_id: int | None = field( - default=None, metadata=TextParser.find_int(r"Switch Port Id: (\d+)") - ) - #: - switch_rx_domain: int | None = field( - default=None, metadata=TextParser.find_int(r"Switch Rx domain: (\d+)") - ) - - -@dataclass -class TestPmdPortStats(TextParser): - """Port statistics.""" - - #: - port_id: int = field(metadata=TextParser.find_int(r"NIC statistics for port (\d+)")) - - #: - rx_packets: int = field(metadata=TextParser.find_int(r"RX-packets:\s+(\d+)")) - #: - rx_missed: int = field(metadata=TextParser.find_int(r"RX-missed:\s+(\d+)")) - #: - rx_bytes: int = field(metadata=TextParser.find_int(r"RX-bytes:\s+(\d+)")) - #: - rx_errors: int = field(metadata=TextParser.find_int(r"RX-errors:\s+(\d+)")) - #: - rx_nombuf: int = field(metadata=TextParser.find_int(r"RX-nombuf:\s+(\d+)")) - - #: - tx_packets: int = field(metadata=TextParser.find_int(r"TX-packets:\s+(\d+)")) - #: - tx_errors: int = field(metadata=TextParser.find_int(r"TX-errors:\s+(\d+)")) - #: - tx_bytes: int = field(metadata=TextParser.find_int(r"TX-bytes:\s+(\d+)")) - - #: - rx_pps: int = field(metadata=TextParser.find_int(r"Rx-pps:\s+(\d+)")) - #: - rx_bps: int = field(metadata=TextParser.find_int(r"Rx-bps:\s+(\d+)")) - - #: - tx_pps: int = field(metadata=TextParser.find_int(r"Tx-pps:\s+(\d+)")) - #: - tx_bps: int = field(metadata=TextParser.find_int(r"Tx-bps:\s+(\d+)")) - - -@dataclass(kw_only=True) -class FlowRule: - """Class representation of flow rule parameters. - - This class represents the parameters of any flow rule as per the - following pattern: - - [group {group_id}] [priority {level}] [ingress] [egress] - [user_id {user_id}] pattern {item} [/ {item} [...]] / end - actions {action} [/ {action} [...]] / end - """ - - #: - group_id: int | None = None - #: - priority_level: int | None = None - #: - direction: Literal["ingress", "egress"] - #: - user_id: int | None = None - #: - pattern: list[str] - #: - actions: list[str] - - def __str__(self) -> str: - """Returns the string representation of this instance.""" - ret = "" - pattern = " / ".join(self.pattern) - action = " / ".join(self.actions) - if self.group_id is not None: - ret += f"group {self.group_id} " - if self.priority_level is not None: - ret += f"priority {self.priority_level} " - ret += f"{self.direction} " - if self.user_id is not None: - ret += f"user_id {self.user_id} " - ret += f"pattern {pattern} / end " - ret += f"actions {action} / end" - return ret - - -class PacketOffloadFlag(Flag): - """Flag representing the Packet Offload Features Flags in DPDK. - - Values in this class are taken from the definitions in the RTE MBUF core library in DPDK - located in ``lib/mbuf/rte_mbuf_core.h``. It is expected that flag values in this class will - match the values they are set to in said DPDK library with one exception; all values must be - unique. For example, the definitions for unknown checksum flags in ``rte_mbuf_core.h`` are all - set to :data:`0`, but it is valuable to distinguish between them in this framework. For this - reason flags that are not unique in the DPDK library are set either to values within the - RTE_MBUF_F_FIRST_FREE-RTE_MBUF_F_LAST_FREE range for Rx or shifted 61+ bits for Tx. - - References: - DPDK lib: ``lib/mbuf/rte_mbuf_core.h`` - """ - - # RX flags - - #: The RX packet is a 802.1q VLAN packet, and the tci has been saved in mbuf->vlan_tci. If the - #: flag RTE_MBUF_F_RX_VLAN_STRIPPED is also present, the VLAN header has been stripped from - #: mbuf data, else it is still present. - RTE_MBUF_F_RX_VLAN = auto() - - #: RX packet with RSS hash result. - RTE_MBUF_F_RX_RSS_HASH = auto() - - #: RX packet with FDIR match indicate. - RTE_MBUF_F_RX_FDIR = auto() - - #: This flag is set when the outermost IP header checksum is detected as wrong by the hardware. - RTE_MBUF_F_RX_OUTER_IP_CKSUM_BAD = 1 << 5 - - #: A vlan has been stripped by the hardware and its tci is saved in mbuf->vlan_tci. This can - #: only happen if vlan stripping is enabled in the RX configuration of the PMD. When - #: RTE_MBUF_F_RX_VLAN_STRIPPED is set, RTE_MBUF_F_RX_VLAN must also be set. - RTE_MBUF_F_RX_VLAN_STRIPPED = auto() - - #: No information about the RX IP checksum. Value is 0 in the DPDK library. - RTE_MBUF_F_RX_IP_CKSUM_UNKNOWN = 1 << 23 - #: The IP checksum in the packet is wrong. - RTE_MBUF_F_RX_IP_CKSUM_BAD = 1 << 4 - #: The IP checksum in the packet is valid. - RTE_MBUF_F_RX_IP_CKSUM_GOOD = 1 << 7 - #: The IP checksum is not correct in the packet data, but the integrity of the IP header is - #: verified. Value is RTE_MBUF_F_RX_IP_CKSUM_BAD | RTE_MBUF_F_RX_IP_CKSUM_GOOD in the DPDK - #: library. - RTE_MBUF_F_RX_IP_CKSUM_NONE = 1 << 24 - - #: No information about the RX L4 checksum. Value is 0 in the DPDK library. - RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN = 1 << 25 - #: The L4 checksum in the packet is wrong. - RTE_MBUF_F_RX_L4_CKSUM_BAD = 1 << 3 - #: The L4 checksum in the packet is valid. - RTE_MBUF_F_RX_L4_CKSUM_GOOD = 1 << 8 - #: The L4 checksum is not correct in the packet data, but the integrity of the L4 data is - #: verified. Value is RTE_MBUF_F_RX_L4_CKSUM_BAD | RTE_MBUF_F_RX_L4_CKSUM_GOOD in the DPDK - #: library. - RTE_MBUF_F_RX_L4_CKSUM_NONE = 1 << 26 - - #: RX IEEE1588 L2 Ethernet PT Packet. - RTE_MBUF_F_RX_IEEE1588_PTP = 1 << 9 - #: RX IEEE1588 L2/L4 timestamped packet. - RTE_MBUF_F_RX_IEEE1588_TMST = 1 << 10 - - #: FD id reported if FDIR match. - RTE_MBUF_F_RX_FDIR_ID = 1 << 13 - #: Flexible bytes reported if FDIR match. - RTE_MBUF_F_RX_FDIR_FLX = 1 << 14 - - #: If both RTE_MBUF_F_RX_QINQ_STRIPPED and RTE_MBUF_F_RX_VLAN_STRIPPED are set, the 2 VLANs - #: have been stripped by the hardware. If RTE_MBUF_F_RX_QINQ_STRIPPED is set and - #: RTE_MBUF_F_RX_VLAN_STRIPPED is unset, only the outer VLAN is removed from packet data. - RTE_MBUF_F_RX_QINQ_STRIPPED = auto() - - #: When packets are coalesced by a hardware or virtual driver, this flag can be set in the RX - #: mbuf, meaning that the m->tso_segsz field is valid and is set to the segment size of - #: original packets. - RTE_MBUF_F_RX_LRO = auto() - - #: Indicate that security offload processing was applied on the RX packet. - RTE_MBUF_F_RX_SEC_OFFLOAD = 1 << 18 - #: Indicate that security offload processing failed on the RX packet. - RTE_MBUF_F_RX_SEC_OFFLOAD_FAILED = auto() - - #: The RX packet is a double VLAN. If this flag is set, RTE_MBUF_F_RX_VLAN must also be set. If - #: the flag RTE_MBUF_F_RX_QINQ_STRIPPED is also present, both VLANs headers have been stripped - #: from mbuf data, else they are still present. - RTE_MBUF_F_RX_QINQ = auto() - - #: No info about the outer RX L4 checksum. Value is 0 in the DPDK library. - RTE_MBUF_F_RX_OUTER_L4_CKSUM_UNKNOWN = 1 << 27 - #: The outer L4 checksum in the packet is wrong - RTE_MBUF_F_RX_OUTER_L4_CKSUM_BAD = 1 << 21 - #: The outer L4 checksum in the packet is valid - RTE_MBUF_F_RX_OUTER_L4_CKSUM_GOOD = 1 << 22 - #: Invalid outer L4 checksum state. Value is - #: RTE_MBUF_F_RX_OUTER_L4_CKSUM_BAD | RTE_MBUF_F_RX_OUTER_L4_CKSUM_GOOD in the DPDK library. - RTE_MBUF_F_RX_OUTER_L4_CKSUM_INVALID = 1 << 28 - - # TX flags - - #: Outer UDP checksum offload flag. This flag is used for enabling outer UDP checksum in PMD. - #: To use outer UDP checksum, the user either needs to enable the following in mbuf: - #: - #: a) Fill outer_l2_len and outer_l3_len in mbuf. - #: b) Set the RTE_MBUF_F_TX_OUTER_UDP_CKSUM flag. - #: c) Set the RTE_MBUF_F_TX_OUTER_IPV4 or RTE_MBUF_F_TX_OUTER_IPV6 flag. - #: - #: Or configure RTE_ETH_TX_OFFLOAD_OUTER_UDP_CKSUM offload flag. - RTE_MBUF_F_TX_OUTER_UDP_CKSUM = 1 << 41 - - #: UDP Fragmentation Offload flag. This flag is used for enabling UDP fragmentation in SW or in - #: HW. - RTE_MBUF_F_TX_UDP_SEG = auto() - - #: Request security offload processing on the TX packet. To use Tx security offload, the user - #: needs to fill l2_len in mbuf indicating L2 header size and where L3 header starts. - #: Similarly, l3_len should also be filled along with ol_flags reflecting current L3 type. - RTE_MBUF_F_TX_SEC_OFFLOAD = auto() - - #: Offload the MACsec. This flag must be set by the application to enable this offload feature - #: for a packet to be transmitted. - RTE_MBUF_F_TX_MACSEC = auto() - - # Bits 45:48 are used for the tunnel type in ``lib/mbuf/rte_mbuf_core.h``, but some are modified - # in this Flag to maintain uniqueness. The tunnel type must be specified for TSO or checksum on - # the inner part of tunnel packets. These flags can be used with RTE_MBUF_F_TX_TCP_SEG for TSO, - # or RTE_MBUF_F_TX_xxx_CKSUM. The mbuf fields for inner and outer header lengths are required: - # outer_l2_len, outer_l3_len, l2_len, l3_len, l4_len and tso_segsz for TSO. - - #: - RTE_MBUF_F_TX_TUNNEL_VXLAN = 1 << 45 - #: - RTE_MBUF_F_TX_TUNNEL_GRE = 1 << 46 - #: Value is 3 << 45 in the DPDK library. - RTE_MBUF_F_TX_TUNNEL_IPIP = 1 << 61 - #: - RTE_MBUF_F_TX_TUNNEL_GENEVE = 1 << 47 - #: TX packet with MPLS-in-UDP RFC 7510 header. Value is 5 << 45 in the DPDK library. - RTE_MBUF_F_TX_TUNNEL_MPLSINUDP = 1 << 62 - #: Value is 6 << 45 in the DPDK library. - RTE_MBUF_F_TX_TUNNEL_VXLAN_GPE = 1 << 63 - #: Value is 7 << 45 in the DPDK library. - RTE_MBUF_F_TX_TUNNEL_GTP = 1 << 64 - #: - RTE_MBUF_F_TX_TUNNEL_ESP = 1 << 48 - #: Generic IP encapsulated tunnel type, used for TSO and checksum offload. This can be used for - #: tunnels which are not standards or listed above. It is preferred to use specific tunnel - #: flags like RTE_MBUF_F_TX_TUNNEL_GRE or RTE_MBUF_F_TX_TUNNEL_IPIP if possible. The ethdev - #: must be configured with RTE_ETH_TX_OFFLOAD_IP_TNL_TSO. Outer and inner checksums are done - #: according to the existing flags like RTE_MBUF_F_TX_xxx_CKSUM. Specific tunnel headers that - #: contain payload length, sequence id or checksum are not expected to be updated. Value is - #: 0xD << 45 in the DPDK library. - RTE_MBUF_F_TX_TUNNEL_IP = 1 << 65 - #: Generic UDP encapsulated tunnel type, used for TSO and checksum offload. UDP tunnel type - #: implies outer IP layer. It can be used for tunnels which are not standards or listed above. - #: It is preferred to use specific tunnel flags like RTE_MBUF_F_TX_TUNNEL_VXLAN if possible. - #: The ethdev must be configured with RTE_ETH_TX_OFFLOAD_UDP_TNL_TSO. Outer and inner checksums - #: are done according to the existing flags like RTE_MBUF_F_TX_xxx_CKSUM. Specific tunnel - #: headers that contain payload length, sequence id or checksum are not expected to be updated. - #: value is 0xE << 45 in the DPDK library. - RTE_MBUF_F_TX_TUNNEL_UDP = 1 << 66 - - #: Double VLAN insertion (QinQ) request to driver, driver may offload the insertion based on - #: device capability. Mbuf 'vlan_tci' & 'vlan_tci_outer' must be valid when this flag is set. - RTE_MBUF_F_TX_QINQ = 1 << 49 - - #: TCP segmentation offload. To enable this offload feature for a packet to be transmitted on - #: hardware supporting TSO: - #: - #: - set the RTE_MBUF_F_TX_TCP_SEG flag in mbuf->ol_flags (this flag implies - #: RTE_MBUF_F_TX_TCP_CKSUM) - #: - set the flag RTE_MBUF_F_TX_IPV4 or RTE_MBUF_F_TX_IPV6 - #: * if it's IPv4, set the RTE_MBUF_F_TX_IP_CKSUM flag - #: - fill the mbuf offload information: l2_len, l3_len, l4_len, tso_segsz - RTE_MBUF_F_TX_TCP_SEG = auto() - - #: TX IEEE1588 packet to timestamp. - RTE_MBUF_F_TX_IEEE1588_TMST = auto() - - # Bits 52+53 used for L4 packet type with checksum enabled in ``lib/mbuf/rte_mbuf_core.h`` but - # some values must be modified in this framework to maintain uniqueness. To use hardware - # L4 checksum offload, the user needs to: - # - # - fill l2_len and l3_len in mbuf - # - set the flags RTE_MBUF_F_TX_TCP_CKSUM, RTE_MBUF_F_TX_SCTP_CKSUM or - # RTE_MBUF_F_TX_UDP_CKSUM - # - set the flag RTE_MBUF_F_TX_IPV4 or RTE_MBUF_F_TX_IPV6 - - #: Disable L4 cksum of TX pkt. Value is 0 in the DPDK library. - RTE_MBUF_F_TX_L4_NO_CKSUM = 1 << 67 - #: TCP cksum of TX pkt. Computed by NIC. - RTE_MBUF_F_TX_TCP_CKSUM = 1 << 52 - #: SCTP cksum of TX pkt. Computed by NIC. - RTE_MBUF_F_TX_SCTP_CKSUM = 1 << 53 - #: UDP cksum of TX pkt. Computed by NIC. Value is 3 << 52 in the DPDK library. - RTE_MBUF_F_TX_UDP_CKSUM = 1 << 68 - - #: Offload the IP checksum in the hardware. The flag RTE_MBUF_F_TX_IPV4 should also be set by - #: the application, although a PMD will only check RTE_MBUF_F_TX_IP_CKSUM. - RTE_MBUF_F_TX_IP_CKSUM = 1 << 54 - - #: Packet is IPv4. This flag must be set when using any offload feature (TSO, L3 or L4 - #: checksum) to tell the NIC that the packet is an IPv4 packet. If the packet is a tunneled - #: packet, this flag is related to the inner headers. - RTE_MBUF_F_TX_IPV4 = auto() - #: Packet is IPv6. This flag must be set when using an offload feature (TSO or L4 checksum) to - #: tell the NIC that the packet is an IPv6 packet. If the packet is a tunneled packet, this - #: flag is related to the inner headers. - RTE_MBUF_F_TX_IPV6 = auto() - #: VLAN tag insertion request to driver, driver may offload the insertion based on the device - #: capability. mbuf 'vlan_tci' field must be valid when this flag is set. - RTE_MBUF_F_TX_VLAN = auto() - - #: Offload the IP checksum of an external header in the hardware. The flag - #: RTE_MBUF_F_TX_OUTER_IPV4 should also be set by the application, although a PMD will only - #: check RTE_MBUF_F_TX_OUTER_IP_CKSUM. - RTE_MBUF_F_TX_OUTER_IP_CKSUM = auto() - #: Packet outer header is IPv4. This flag must be set when using any outer offload feature (L3 - #: or L4 checksum) to tell the NIC that the outer header of the tunneled packet is an IPv4 - #: packet. - RTE_MBUF_F_TX_OUTER_IPV4 = auto() - #: Packet outer header is IPv6. This flag must be set when using any outer offload feature (L4 - #: checksum) to tell the NIC that the outer header of the tunneled packet is an IPv6 packet. - RTE_MBUF_F_TX_OUTER_IPV6 = auto() - - @classmethod - def from_list_string(cls, names: str) -> Self: - """Makes a flag from a whitespace-separated list of names. - - Args: - names: a whitespace-separated list containing the members of this flag. - - Returns: - An instance of this flag. - """ - flag = cls(0) - for name in names.split(): - flag |= cls.from_str(name) - return flag - - @classmethod - def from_str(cls, name: str) -> Self: - """Makes a flag matching the supplied name. - - Args: - name: a valid member of this flag in text - Returns: - An instance of this flag. - """ - member_name = name.strip().replace("-", "_") - return cls[member_name] - - @classmethod - def make_parser(cls) -> ParserFn: - """Makes a parser function. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this flag from text. - """ - return TextParser.wrap( - TextParser.find(r"ol_flags: ([^\n]+)"), - cls.from_list_string, - ) - - -class RtePTypes(Flag): - """Flag representing possible packet types in DPDK verbose output. - - Values in this class are derived from definitions in the RTE MBUF ptype library in DPDK located - in ``lib/mbuf/rte_mbuf_ptype.h``. Specifically, the names of values in this class should match - the possible return options from the functions ``rte_get_ptype_*_name`` in ``rte_mbuf_ptype.c``. - - References: - DPDK lib: ``lib/mbuf/rte_mbuf_ptype.h`` - DPDK ptype name formatting functions: ``lib/mbuf/rte_mbuf_ptype.c:rte_get_ptype_*_name()`` - """ - - # L2 - #: Ethernet packet type. This is used for outer packet for tunneling cases. - L2_ETHER = auto() - #: Ethernet packet type for time sync. - L2_ETHER_TIMESYNC = auto() - #: ARP (Address Resolution Protocol) packet type. - L2_ETHER_ARP = auto() - #: LLDP (Link Layer Discovery Protocol) packet type. - L2_ETHER_LLDP = auto() - #: NSH (Network Service Header) packet type. - L2_ETHER_NSH = auto() - #: VLAN packet type. - L2_ETHER_VLAN = auto() - #: QinQ packet type. - L2_ETHER_QINQ = auto() - #: PPPOE packet type. - L2_ETHER_PPPOE = auto() - #: FCoE packet type.. - L2_ETHER_FCOE = auto() - #: MPLS packet type. - L2_ETHER_MPLS = auto() - #: No L2 packet information. - L2_UNKNOWN = auto() - - # L3 - #: IP (Internet Protocol) version 4 packet type. This is used for outer packet for tunneling - #: cases, and does not contain any header option. - L3_IPV4 = auto() - #: IP (Internet Protocol) version 4 packet type. This is used for outer packet for tunneling - #: cases, and contains header options. - L3_IPV4_EXT = auto() - #: IP (Internet Protocol) version 6 packet type. This is used for outer packet for tunneling - #: cases, and does not contain any extension header. - L3_IPV6 = auto() - #: IP (Internet Protocol) version 4 packet type. This is used for outer packet for tunneling - #: cases, and may or maynot contain header options. - L3_IPV4_EXT_UNKNOWN = auto() - #: IP (Internet Protocol) version 6 packet type. This is used for outer packet for tunneling - #: cases, and contains extension headers. - L3_IPV6_EXT = auto() - #: IP (Internet Protocol) version 6 packet type. This is used for outer packet for tunneling - #: cases, and may or maynot contain extension headers. - L3_IPV6_EXT_UNKNOWN = auto() - #: No L3 packet information. - L3_UNKNOWN = auto() - - # L4 - #: TCP (Transmission Control Protocol) packet type. This is used for outer packet for tunneling - #: cases. - L4_TCP = auto() - #: UDP (User Datagram Protocol) packet type. This is used for outer packet for tunneling cases. - L4_UDP = auto() - #: Fragmented IP (Internet Protocol) packet type. This is used for outer packet for tunneling - #: cases and refers to those packets of any IP types which can be recognized as fragmented. A - #: fragmented packet cannot be recognized as any other L4 types (RTE_PTYPE_L4_TCP, - #: RTE_PTYPE_L4_UDP, RTE_PTYPE_L4_SCTP, RTE_PTYPE_L4_ICMP, RTE_PTYPE_L4_NONFRAG). - L4_FRAG = auto() - #: SCTP (Stream Control Transmission Protocol) packet type. This is used for outer packet for - #: tunneling cases. - L4_SCTP = auto() - #: ICMP (Internet Control Message Protocol) packet type. This is used for outer packet for - #: tunneling cases. - L4_ICMP = auto() - #: Non-fragmented IP (Internet Protocol) packet type. This is used for outer packet for - #: tunneling cases and refers to those packets of any IP types, that cannot be recognized as - #: any of the above L4 types (RTE_PTYPE_L4_TCP, RTE_PTYPE_L4_UDP, RTE_PTYPE_L4_FRAG, - #: RTE_PTYPE_L4_SCTP, RTE_PTYPE_L4_ICMP). - L4_NONFRAG = auto() - #: IGMP (Internet Group Management Protocol) packet type. - L4_IGMP = auto() - #: No L4 packet information. - L4_UNKNOWN = auto() - - # Tunnel - #: IP (Internet Protocol) in IP (Internet Protocol) tunneling packet type. - TUNNEL_IP = auto() - #: GRE (Generic Routing Encapsulation) tunneling packet type. - TUNNEL_GRE = auto() - #: VXLAN (Virtual eXtensible Local Area Network) tunneling packet type. - TUNNEL_VXLAN = auto() - #: NVGRE (Network Virtualization using Generic Routing Encapsulation) tunneling packet type. - TUNNEL_NVGRE = auto() - #: GENEVE (Generic Network Virtualization Encapsulation) tunneling packet type. - TUNNEL_GENEVE = auto() - #: Tunneling packet type of Teredo, VXLAN (Virtual eXtensible Local Area Network) or GRE - #: (Generic Routing Encapsulation) could be recognized as this packet type, if they can not be - #: recognized independently as of hardware capability. - TUNNEL_GRENAT = auto() - #: GTP-C (GPRS Tunnelling Protocol) control tunneling packet type. - TUNNEL_GTPC = auto() - #: GTP-U (GPRS Tunnelling Protocol) user data tunneling packet type. - TUNNEL_GTPU = auto() - #: ESP (IP Encapsulating Security Payload) tunneling packet type. - TUNNEL_ESP = auto() - #: L2TP (Layer 2 Tunneling Protocol) tunneling packet type. - TUNNEL_L2TP = auto() - #: VXLAN-GPE (VXLAN Generic Protocol Extension) tunneling packet type. - TUNNEL_VXLAN_GPE = auto() - #: MPLS-in-UDP tunneling packet type (RFC 7510). - TUNNEL_MPLS_IN_UDP = auto() - #: MPLS-in-GRE tunneling packet type (RFC 4023). - TUNNEL_MPLS_IN_GRE = auto() - #: No tunnel information found on the packet. - TUNNEL_UNKNOWN = auto() - - # Inner L2 - #: Ethernet packet type. This is used for inner packet type only. - INNER_L2_ETHER = auto() - #: Ethernet packet type with VLAN (Virtual Local Area Network) tag. - INNER_L2_ETHER_VLAN = auto() - #: QinQ packet type. - INNER_L2_ETHER_QINQ = auto() - #: No inner L2 information found on the packet. - INNER_L2_UNKNOWN = auto() - - # Inner L3 - #: IP (Internet Protocol) version 4 packet type. This is used for inner packet only, and does - #: not contain any header option. - INNER_L3_IPV4 = auto() - #: IP (Internet Protocol) version 4 packet type. This is used for inner packet only, and - #: contains header options. - INNER_L3_IPV4_EXT = auto() - #: IP (Internet Protocol) version 6 packet type. This is used for inner packet only, and does - #: not contain any extension header. - INNER_L3_IPV6 = auto() - #: IP (Internet Protocol) version 4 packet type. This is used for inner packet only, and may or - #: may not contain header options. - INNER_L3_IPV4_EXT_UNKNOWN = auto() - #: IP (Internet Protocol) version 6 packet type. This is used for inner packet only, and - #: contains extension headers. - INNER_L3_IPV6_EXT = auto() - #: IP (Internet Protocol) version 6 packet type. This is used for inner packet only, and may or - #: may not contain extension headers. - INNER_L3_IPV6_EXT_UNKNOWN = auto() - #: No inner L3 information found on the packet. - INNER_L3_UNKNOWN = auto() - - # Inner L4 - #: TCP (Transmission Control Protocol) packet type. This is used for inner packet only. - INNER_L4_TCP = auto() - #: UDP (User Datagram Protocol) packet type. This is used for inner packet only. - INNER_L4_UDP = auto() - #: Fragmented IP (Internet Protocol) packet type. This is used for inner packet only, and may - #: or maynot have a layer 4 packet. - INNER_L4_FRAG = auto() - #: SCTP (Stream Control Transmission Protocol) packet type. This is used for inner packet only. - INNER_L4_SCTP = auto() - #: ICMP (Internet Control Message Protocol) packet type. This is used for inner packet only. - INNER_L4_ICMP = auto() - #: Non-fragmented IP (Internet Protocol) packet type. It is used for inner packet only, and may - #: or may not have other unknown layer 4 packet types. - INNER_L4_NONFRAG = auto() - #: No inner L4 information found on the packet. - INNER_L4_UNKNOWN = auto() - - @classmethod - def from_list_string(cls, names: str) -> Self: - """Makes a flag from a whitespace-separated list of names. - - Args: - names: a whitespace-separated list containing the members of this flag. - - Returns: - An instance of this flag. - """ - flag = cls(0) - for name in names.split(): - flag |= cls.from_str(name) - return flag - - @classmethod - def from_str(cls, name: str) -> Self: - """Makes a flag matching the supplied name. - - Args: - name: a valid member of this flag in text - Returns: - An instance of this flag. - """ - member_name = name.strip().replace("-", "_") - return cls[member_name] - - @classmethod - def make_parser(cls, hw: bool) -> ParserFn: - """Makes a parser function. - - Args: - hw: Whether to make a parser for hardware ptypes or software ptypes. If :data:`True`, - hardware ptypes will be collected, otherwise software pytpes will. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this flag from text. - """ - return TextParser.wrap( - TextParser.find(f"{'hw' if hw else 'sw'} ptype: ([^-]+)"), - cls.from_list_string, - ) - - -@dataclass -class TestPmdVerbosePacket(TextParser): - """Packet information provided by verbose output in Testpmd. - - This dataclass expects that packet information be prepended with the starting line of packet - bursts. Specifically, the line that reads "port X/queue Y: sent/received Z packets". - """ - - #: ID of the port that handled the packet. - port_id: int = field(metadata=TextParser.find_int(r"port (\d+)/queue \d+")) - #: ID of the queue that handled the packet. - queue_id: int = field(metadata=TextParser.find_int(r"port \d+/queue (\d+)")) - #: Whether the packet was received or sent by the queue/port. - was_received: bool = field(metadata=TextParser.find(r"received \d+ packets")) - #: - src_mac: str = field(metadata=TextParser.find(f"src=({REGEX_FOR_MAC_ADDRESS})")) - #: - dst_mac: str = field(metadata=TextParser.find(f"dst=({REGEX_FOR_MAC_ADDRESS})")) - #: Memory pool the packet was handled on. - pool: str = field(metadata=TextParser.find(r"pool=(\S+)")) - #: Packet type in hex. - p_type: int = field(metadata=TextParser.find_int(r"type=(0x[a-fA-F\d]+)")) - #: - length: int = field(metadata=TextParser.find_int(r"length=(\d+)")) - #: Number of segments in the packet. - nb_segs: int = field(metadata=TextParser.find_int(r"nb_segs=(\d+)")) - #: Hardware packet type. - hw_ptype: RtePTypes = field(metadata=RtePTypes.make_parser(hw=True)) - #: Software packet type. - sw_ptype: RtePTypes = field(metadata=RtePTypes.make_parser(hw=False)) - #: - l2_len: int = field(metadata=TextParser.find_int(r"l2_len=(\d+)")) - #: - ol_flags: PacketOffloadFlag = field(metadata=PacketOffloadFlag.make_parser()) - #: RSS hash of the packet in hex. - rss_hash: int | None = field( - default=None, metadata=TextParser.find_int(r"RSS hash=(0x[a-fA-F\d]+)") - ) - #: RSS queue that handled the packet in hex. - rss_queue: int | None = field( - default=None, metadata=TextParser.find_int(r"RSS queue=(0x[a-fA-F\d]+)") - ) - #: - l3_len: int | None = field(default=None, metadata=TextParser.find_int(r"l3_len=(\d+)")) - #: - l4_len: int | None = field(default=None, metadata=TextParser.find_int(r"l4_len=(\d+)")) - #: - l4_dport: int | None = field( - default=None, - metadata=TextParser.find_int(r"Destination (?:TCP|UDP) port=(\d+)"), - ) - - -class RxOffloadCapability(Flag): - """Rx offload capabilities of a device. - - The flags are taken from ``lib/ethdev/rte_ethdev.h``. - They're prefixed with ``RTE_ETH_RX_OFFLOAD`` in ``lib/ethdev/rte_ethdev.h`` - instead of ``RX_OFFLOAD``, which is what testpmd changes the prefix to. - The values are not contiguous, so the correspondence is preserved - by specifying concrete values interspersed between auto() values. - - The ``RX_OFFLOAD`` prefix has been preserved so that the same flag names can be used - in :class:`NicCapability`. The prefix is needed in :class:`NicCapability` since there's - no other qualifier which would sufficiently distinguish it from other capabilities. - - References: - DPDK lib: ``lib/ethdev/rte_ethdev.h`` - testpmd display function: ``app/test-pmd/cmdline.c:print_rx_offloads()`` - """ - - #: - RX_OFFLOAD_VLAN_STRIP = auto() - #: Device supports L3 checksum offload. - RX_OFFLOAD_IPV4_CKSUM = auto() - #: Device supports L4 checksum offload. - RX_OFFLOAD_UDP_CKSUM = auto() - #: Device supports L4 checksum offload. - RX_OFFLOAD_TCP_CKSUM = auto() - #: Device supports Large Receive Offload. - RX_OFFLOAD_TCP_LRO = auto() - #: Device supports QinQ (queue in queue) offload. - RX_OFFLOAD_QINQ_STRIP = auto() - #: Device supports inner packet L3 checksum. - RX_OFFLOAD_OUTER_IPV4_CKSUM = auto() - #: Device supports MACsec. - RX_OFFLOAD_MACSEC_STRIP = auto() - #: Device supports filtering of a VLAN Tag identifier. - RX_OFFLOAD_VLAN_FILTER = 1 << 9 - #: Device supports VLAN offload. - RX_OFFLOAD_VLAN_EXTEND = auto() - #: Device supports receiving segmented mbufs. - RX_OFFLOAD_SCATTER = 1 << 13 - #: Device supports Timestamp. - RX_OFFLOAD_TIMESTAMP = auto() - #: Device supports crypto processing while packet is received in NIC. - RX_OFFLOAD_SECURITY = auto() - #: Device supports CRC stripping. - RX_OFFLOAD_KEEP_CRC = auto() - #: Device supports L4 checksum offload. - RX_OFFLOAD_SCTP_CKSUM = auto() - #: Device supports inner packet L4 checksum. - RX_OFFLOAD_OUTER_UDP_CKSUM = auto() - #: Device supports RSS hashing. - RX_OFFLOAD_RSS_HASH = auto() - #: Device supports - RX_OFFLOAD_BUFFER_SPLIT = auto() - #: Device supports all checksum capabilities. - RX_OFFLOAD_CHECKSUM = RX_OFFLOAD_IPV4_CKSUM | RX_OFFLOAD_UDP_CKSUM | RX_OFFLOAD_TCP_CKSUM - #: Device supports all VLAN capabilities. - RX_OFFLOAD_VLAN = ( - RX_OFFLOAD_VLAN_STRIP - | RX_OFFLOAD_VLAN_FILTER - | RX_OFFLOAD_VLAN_EXTEND - | RX_OFFLOAD_QINQ_STRIP - ) - - @classmethod - def from_string(cls, line: str) -> Self: - """Make an instance from a string containing the flag names separated with a space. - - Args: - line: The line to parse. - - Returns: - A new instance containing all found flags. - """ - flag = cls(0) - for flag_name in line.split(): - flag |= cls[f"RX_OFFLOAD_{flag_name}"] - return flag - - @classmethod - def make_parser(cls, per_port: bool) -> ParserFn: - """Make a parser function. - - Args: - per_port: If :data:`True`, will return capabilities per port. If :data:`False`, - will return capabilities per queue. - - Returns: - ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a - parser function that makes an instance of this flag from text. - """ - granularity = "Port" if per_port else "Queue" - return TextParser.wrap( - TextParser.find(rf"Per {granularity}\s+:(.*)$", re.MULTILINE), - cls.from_string, - ) - - -@dataclass -class RxOffloadCapabilities(TextParser): - """The result of testpmd's ``show port <port_id> rx_offload capabilities`` command. - - References: - testpmd command function: ``app/test-pmd/cmdline.c:cmd_rx_offload_get_capa()`` - testpmd display function: ``app/test-pmd/cmdline.c:cmd_rx_offload_get_capa_parsed()`` - """ - - #: - port_id: int = field( - metadata=TextParser.find_int(r"Rx Offloading Capabilities of port (\d+) :") - ) - #: Per-queue Rx offload capabilities. - per_queue: RxOffloadCapability = field(metadata=RxOffloadCapability.make_parser(False)) - #: Capabilities other than per-queue Rx offload capabilities. - per_port: RxOffloadCapability = field(metadata=RxOffloadCapability.make_parser(True)) - - -@dataclass -class TestPmdPortFlowCtrl(TextParser): - """Class representing a port's flow control parameters. - - The parameters can also be parsed from the output of ``show port <port_id> flow_ctrl``. - """ - - #: Enable Reactive Extensions. - rx: bool = field(default=False, metadata=TextParser.find(r"Rx pause: on")) - #: Enable Transmit. - tx: bool = field(default=False, metadata=TextParser.find(r"Tx pause: on")) - #: High threshold value to trigger XOFF. - high_water: int = field( - default=0, metadata=TextParser.find_int(r"High waterline: (0x[a-fA-F\d]+)") - ) - #: Low threshold value to trigger XON. - low_water: int = field( - default=0, metadata=TextParser.find_int(r"Low waterline: (0x[a-fA-F\d]+)") - ) - #: Pause quota in the Pause frame. - pause_time: int = field(default=0, metadata=TextParser.find_int(r"Pause time: (0x[a-fA-F\d]+)")) - #: Send XON frame. - send_xon: bool = field(default=False, metadata=TextParser.find(r"Tx pause: on")) - #: Enable receiving MAC control frames. - mac_ctrl_frame_fwd: bool = field(default=False, metadata=TextParser.find(r"Tx pause: on")) - #: Change the auto-negotiation parameter. - autoneg: bool = field(default=False, metadata=TextParser.find(r"Autoneg: on")) - - def __str__(self) -> str: - """Returns the string representation of this instance.""" - ret = ( - f"rx {'on' if self.rx else 'off'} " - f"tx {'on' if self.tx else 'off'} " - f"{self.high_water} " - f"{self.low_water} " - f"{self.pause_time} " - f"{1 if self.send_xon else 0} " - f"mac_ctrl_frame_fwd {'on' if self.mac_ctrl_frame_fwd else 'off'} " - f"autoneg {'on' if self.autoneg else 'off'}" - ) - return ret - - -def requires_stopped_ports(func: TestPmdShellMethod) -> TestPmdShellMethod: - """Decorator for :class:`TestPmdShell` commands methods that require stopped ports. - - If the decorated method is called while the ports are started, then these are stopped before - continuing. - - Args: - func: The :class:`TestPmdShell` method to decorate. - """ - - @functools.wraps(func) - def _wrapper(self: "TestPmdShell", *args: P.args, **kwargs: P.kwargs): - if self.ports_started: - self._logger.debug("Ports need to be stopped to continue.") - self.stop_all_ports() - - return func(self, *args, **kwargs) - - return _wrapper - - -def requires_started_ports(func: TestPmdShellMethod) -> TestPmdShellMethod: - """Decorator for :class:`TestPmdShell` commands methods that require started ports. - - If the decorated method is called while the ports are stopped, then these are started before - continuing. - - Args: - func: The :class:`TestPmdShell` method to decorate. - """ - - @functools.wraps(func) - def _wrapper(self: "TestPmdShell", *args: P.args, **kwargs: P.kwargs): - if not self.ports_started: - self._logger.debug("Ports need to be started to continue.") - self.start_all_ports() - - return func(self, *args, **kwargs) - - return _wrapper - - -def add_remove_mtu(mtu: int = 1500) -> Callable[[TestPmdShellMethod], TestPmdShellMethod]: - """Configure MTU to `mtu` on all ports, run the decorated function, then revert. - - Args: - mtu: The MTU to configure all ports on. - - Returns: - The method decorated with setting and reverting MTU. - """ - - def decorator(func: TestPmdShellMethod) -> TestPmdShellMethod: - @functools.wraps(func) - def wrapper(self: "TestPmdShell", *args: P.args, **kwargs: P.kwargs): - original_mtu = self.ports[0].mtu - self.set_port_mtu_all(mtu=mtu, verify=False) - retval = func(self, *args, **kwargs) - self.set_port_mtu_all(original_mtu if original_mtu else 1500, verify=False) - return retval - - return wrapper - - return decorator - - -class TestPmdShell(DPDKShell): - """Testpmd interactive shell. - - The testpmd shell users should never use - the :meth:`~.interactive_shell.InteractiveShell.send_command` method directly, but rather - call specialized methods. If there isn't one that satisfies a need, it should be added. - - Attributes: - ports_started: Indicates whether the ports are started. - """ - - _app_params: TestPmdParams - _ports: list[TestPmdPort] | None - - #: The testpmd's prompt. - _default_prompt: ClassVar[str] = "testpmd>" - - #: This forces the prompt to appear after sending a command. - _command_extra_chars: ClassVar[str] = "\n" - - ports_started: bool - - def __init__( - self, - name: str | None = None, - privileged: bool = True, - **app_params: Unpack[TestPmdParamsDict], - ) -> None: - """Overrides :meth:`~.dpdk_shell.DPDKShell.__init__`. Changes app_params to kwargs.""" - if "port_topology" not in app_params and get_ctx().topology.type is TopologyType.one_link: - app_params["port_topology"] = PortTopology.loop - super().__init__(name, privileged, app_params=TestPmdParams(**app_params)) - self.ports_started = not self._app_params.disable_device_start - self._ports = None - - @property - def path(self) -> PurePath: - """The path to the testpmd executable.""" - return PurePath("app/dpdk-testpmd") - - @property - def ports(self) -> list[TestPmdPort]: - """The ports of the instance. - - This caches the ports returned by :meth:`show_port_info_all`. - To force an update of port information, execute :meth:`show_port_info_all` or - :meth:`show_port_info`. - - Returns: The list of known testpmd ports. - """ - if self._ports is None: - return self.show_port_info_all() - return self._ports - - @requires_started_ports - def start(self, verify: bool = True) -> None: - """Start packet forwarding with the current configuration. - - Args: - verify: If :data:`True` , a second start command will be sent in an attempt to verify - packet forwarding started as expected. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and forwarding fails to - start or ports fail to come up. - """ - self.send_command("start") - if verify: - # If forwarding was already started, sending "start" again should tell us - start_cmd_output = self.send_command("start") - if "Packet forwarding already started" not in start_cmd_output: - self._logger.debug(f"Failed to start packet forwarding: \n{start_cmd_output}") - raise InteractiveCommandExecutionError("Testpmd failed to start packet forwarding.") - - def stop(self, verify: bool = True) -> str: - """Stop packet forwarding. - - Args: - verify: If :data:`True` , the output of the stop command is scanned to verify that - forwarding was stopped successfully or not started. If neither is found, it is - considered an error. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the command to stop - forwarding results in an error. - - Returns: - Output gathered from the stop command and all other preceding logs in the buffer. This - output is most often used to view forwarding statistics that are displayed when this - command is sent as well as any verbose packet information that hasn't been consumed - prior to calling this method. - """ - stop_cmd_output = self.send_command("stop") - if verify: - if ( - "Done." not in stop_cmd_output - and "Packet forwarding not started" not in stop_cmd_output - ): - self._logger.debug(f"Failed to stop packet forwarding: \n{stop_cmd_output}") - raise InteractiveCommandExecutionError("Testpmd failed to stop packet forwarding.") - return stop_cmd_output - - def get_devices(self) -> list[TestPmdDevice]: - """Get a list of device names that are known to testpmd. - - Uses the device info listed in testpmd and then parses the output. - - Returns: - A list of devices. - """ - dev_info: str = self.send_command("show device info all") - dev_list: list[TestPmdDevice] = [] - for line in dev_info.split("\n"): - if "device name:" in line.lower(): - dev_list.append(TestPmdDevice(line)) - return dev_list - - def wait_link_status_up(self, port_id: int, timeout=SETTINGS.timeout) -> bool: - """Wait until the link status on the given port is "up". - - Arguments: - port_id: Port to check the link status on. - timeout: Time to wait for the link to come up. The default value for this - argument may be modified using the :option:`--timeout` command-line argument - or the :envvar:`DTS_TIMEOUT` environment variable. - - Returns: - Whether the link came up in time or not. - """ - time_to_stop = time.time() + timeout - port_info: str = "" - while time.time() < time_to_stop: - port_info = self.send_command(f"show port info {port_id}") - if "Link status: up" in port_info: - break - time.sleep(0.5) - else: - self._logger.error(f"The link for port {port_id} did not come up in the given timeout.") - return "Link status: up" in port_info - - def set_forward_mode(self, mode: SimpleForwardingModes, verify: bool = True): - """Set packet forwarding mode. - - Args: - mode: The forwarding mode to use. - verify: If :data:`True` the output of the command will be scanned in an attempt to - verify that the forwarding mode was set to `mode` properly. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the forwarding mode - fails to update. - """ - set_fwd_output = self.send_command(f"set fwd {mode.value}") - if verify: - if f"Set {mode.value} packet forwarding mode" not in set_fwd_output: - self._logger.debug(f"Failed to set fwd mode to {mode.value}:\n{set_fwd_output}") - raise InteractiveCommandExecutionError( - f"Test pmd failed to set fwd mode to {mode.value}" - ) - - def stop_all_ports(self, verify: bool = True) -> None: - """Stops all the ports. - - Args: - verify: If :data:`True`, the output of the command will be checked for a successful - execution. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the ports were not - stopped successfully. - """ - self._logger.debug("Stopping all the ports...") - output = self.send_command("port stop all") - if verify and not output.strip().endswith("Done"): - raise InteractiveCommandExecutionError("Ports were not stopped successfully.") - - self.ports_started = False - - def start_all_ports(self, verify: bool = True) -> None: - """Starts all the ports. - - Args: - verify: If :data:`True`, the output of the command will be checked for a successful - execution. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the ports were not - started successfully. - """ - self._logger.debug("Starting all the ports...") - output = self.send_command("port start all") - if verify and not output.strip().endswith("Done"): - raise InteractiveCommandExecutionError("Ports were not started successfully.") - - self.ports_started = True - - @requires_stopped_ports - def set_ports_queues(self, number_of: int) -> None: - """Sets the number of queues per port. - - Args: - number_of: The number of RX/TX queues to create per port. - - Raises: - InternalError: If `number_of` is invalid. - """ - if number_of < 1: - raise InternalError("The number of queues must be positive and non-zero.") - - self.send_command(f"port config all rxq {number_of}") - self.send_command(f"port config all txq {number_of}") - - @requires_stopped_ports - def close_all_ports(self, verify: bool = True) -> None: - """Close all ports. - - Args: - verify: If :data:`True` the output of the close command will be scanned in an attempt - to verify that all ports were stopped successfully. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and at lease one port - failed to close. - """ - port_close_output = self.send_command("port close all") - if verify: - num_ports = len(self.ports) - if not all(f"Port {p_id} is closed" in port_close_output for p_id in range(num_ports)): - raise InteractiveCommandExecutionError("Ports were not closed successfully.") - - def show_port_info_all(self) -> list[TestPmdPort]: - """Returns the information of all the ports. - - Returns: - list[TestPmdPort]: A list containing all the ports information as `TestPmdPort`. - """ - output = self.send_command("show port info all") - - # Sample output of the "all" command looks like: - # - # <start> - # - # ********************* Infos for port 0 ********************* - # Key: value - # - # ********************* Infos for port 1 ********************* - # Key: value - # <end> - # - # Takes advantage of the double new line in between ports as end delimiter. But we need to - # artificially add a new line at the end to pick up the last port. Because commands are - # executed on a pseudo-terminal created by paramiko on the remote node, lines end with CRLF. - # Therefore we also need to take the carriage return into account. - iter = re.finditer(r"\*{21}.*?[\r\n]{4}", output + "\r\n", re.S) - self._ports = [TestPmdPort.parse(block.group(0)) for block in iter] - return self._ports - - def show_port_info(self, port_id: int) -> TestPmdPort: - """Returns the given port information. - - Args: - port_id: The port ID to gather information for. - - Raises: - InteractiveCommandExecutionError: If `port_id` is invalid. - - Returns: - TestPmdPort: An instance of `TestPmdPort` containing the given port's information. - """ - output = self.send_command(f"show port info {port_id}", skip_first_line=True) - if output.startswith("Invalid port"): - raise InteractiveCommandExecutionError("invalid port given") - - port = TestPmdPort.parse(output) - self._update_port(port) - return port - - def _update_port(self, port: TestPmdPort) -> None: - if self._ports: - self._ports = [ - existing_port if <a href="http://port.id" rel="noreferrer" target="_blank">port.id</a> != <a href="http://existing_port.id" rel="noreferrer" target="_blank">existing_port.id</a> else port - for existing_port in self._ports - ] - - def set_mac_addr(self, port_id: int, mac_address: str, add: bool, verify: bool = True) -> None: - """Add or remove a mac address on a given port's Allowlist. - - Args: - port_id: The port ID the mac address is set on. - mac_address: The mac address to be added to or removed from the specified port. - add: If :data:`True`, add the specified mac address. If :data:`False`, remove specified - mac address. - verify: If :data:'True', assert that the 'mac_addr' operation was successful. If - :data:'False', run the command and skip this assertion. - - Raises: - InteractiveCommandExecutionError: If the set mac address operation fails. - """ - mac_cmd = "add" if add else "remove" - output = self.send_command(f"mac_addr {mac_cmd} {port_id} {mac_address}") - if "Bad arguments" in output: - self._logger.debug("Invalid argument provided to mac_addr") - raise InteractiveCommandExecutionError("Invalid argument provided") - - if verify: - if "mac_addr_cmd error:" in output: - self._logger.debug(f"Failed to {mac_cmd} {mac_address} on port {port_id}") - raise InteractiveCommandExecutionError( - f"Failed to {mac_cmd} {mac_address} on port {port_id} \n{output}" - ) - - def set_multicast_mac_addr( - self, port_id: int, multi_addr: str, add: bool, verify: bool = True - ) -> None: - """Add or remove multicast mac address to a specified port's allow list. - - Args: - port_id: The port ID the multicast address is set on. - multi_addr: The multicast address to be added or removed from the filter. - add: If :data:'True', add the specified multicast address to the port filter. - If :data:'False', remove the specified multicast address from the port filter. - verify: If :data:'True', assert that the 'mcast_addr' operations was successful. - If :data:'False', execute the 'mcast_addr' operation and skip the assertion. - - Raises: - InteractiveCommandExecutionError: If either the 'add' or 'remove' operations fails. - """ - mcast_cmd = "add" if add else "remove" - output = self.send_command(f"mcast_addr {mcast_cmd} {port_id} {multi_addr}") - if "Bad arguments" in output: - self._logger.debug("Invalid arguments provided to mcast_addr") - raise InteractiveCommandExecutionError("Invalid argument provided") - - if verify: - if ( - "Invalid multicast_addr" in output - or f"multicast address {'already' if add else 'not'} filtered by port" in output - ): - self._logger.debug(f"Failed to {mcast_cmd} {multi_addr} on port {port_id}") - raise InteractiveCommandExecutionError( - f"Failed to {mcast_cmd} {multi_addr} on port {port_id} \n{output}" - ) - - def show_port_stats_all(self) -> Tuple[list[TestPmdPortStats], str]: - """Returns the statistics of all the ports. - - Returns: - Tuple[str, list[TestPmdPortStats]]: A tuple where the first element is the stats of all - ports as `TestPmdPortStats` and second is the raw testpmd output that was collected - from the sent command. - """ - output = self.send_command("show port stats all") - - # Sample output of the "all" command looks like: - # - # ########### NIC statistics for port 0 ########### - # values... - # ################################################# - # - # ########### NIC statistics for port 1 ########### - # values... - # ################################################# - # - iter = re.finditer(r"(^ #*.+#*$[^#]+)^ #*\r$", output, re.MULTILINE) - return ([TestPmdPortStats.parse(block.group(1)) for block in iter], output) - - def show_port_stats(self, port_id: int) -> TestPmdPortStats: - """Returns the given port statistics. - - Args: - port_id: The port ID to gather information for. - - Raises: - InteractiveCommandExecutionError: If `port_id` is invalid. - - Returns: - TestPmdPortStats: An instance of `TestPmdPortStats` containing the given port's stats. - """ - output = self.send_command(f"show port stats {port_id}", skip_first_line=True) - if output.startswith("Invalid port"): - raise InteractiveCommandExecutionError("invalid port given") - - return TestPmdPortStats.parse(output) - - def set_multicast_all(self, on: bool, verify: bool = True) -> None: - """Turns multicast mode on/off for the specified port. - - Args: - on: If :data:`True`, turns multicast mode on, otherwise turns off. - verify: If :data:`True` an additional command will be sent to verify - that multicast mode is properly set. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and multicast - mode is not properly set. - """ - multicast_cmd_output = self.send_command(f"set allmulti all {'on' if on else 'off'}") - if verify: - port_stats = self.show_port_info_all() - if on ^ all(stats.is_allmulticast_mode_enabled for stats in port_stats): - self._logger.debug( - f"Failed to set multicast mode on all ports.: \n{multicast_cmd_output}" - ) - raise InteractiveCommandExecutionError( - "Testpmd failed to set multicast mode on all ports." - ) - - @requires_stopped_ports - def csum_set_hw( - self, layers: ChecksumOffloadOptions, port_id: int, verify: bool = True - ) -> None: - """Enables hardware checksum offloading on the specified layer. - - Args: - layers: The layer/layers that checksum offloading should be enabled on. - port_id: The port number to enable checksum offloading on, should be within 0-32. - verify: If :data:`True` the output of the command will be scanned in an attempt to - verify that checksum offloading was enabled on the port. - - Raises: - InteractiveCommandExecutionError: If checksum offload is not enabled successfully. - """ - for name, offload in ChecksumOffloadOptions.__members__.items(): - if offload in layers: - name = name.replace("_", "-") - csum_output = self.send_command(f"csum set {name} hw {port_id}") - if verify: - if ( - "Bad arguments" in csum_output - or f"Please stop port {port_id} first" in csum_output - or f"checksum offload is not supported by port {port_id}" in csum_output - ): - self._logger.debug(f"Csum set hw error:\n{csum_output}") - raise InteractiveCommandExecutionError( - f"Failed to set csum hw {name} mode on port {port_id}" - ) - success = False - if f"{name} checksum offload is hw" in csum_output.lower(): - success = True - if not success and verify: - self._logger.debug( - f"Failed to set csum hw mode on port {port_id}:\n{csum_output}" - ) - raise InteractiveCommandExecutionError( - f"""Failed to set csum hw mode on port - {port_id}:\n{csum_output}""" - ) - - def flow_create(self, flow_rule: FlowRule, port_id: int) -> int: - """Creates a flow rule in the testpmd session. - - This command is implicitly verified as needed to return the created flow rule id. - - Args: - flow_rule: :class:`FlowRule` object used for creating testpmd flow rule. - port_id: Integer representing the port to use. - - Raises: - InteractiveCommandExecutionError: If flow rule is invalid. - - Returns: - Id of created flow rule. - """ - flow_output = self.send_command(f"flow create {port_id} {flow_rule}") - match = re.search(r"#(\d+)", flow_output) - if match is not None: - match_str = match.group(1) - flow_id = int(match_str) - return flow_id - else: - self._logger.debug(f"Failed to create flow rule:\n{flow_output}") - raise InteractiveCommandExecutionError(f"Failed to create flow rule:\n{flow_output}") - - def flow_validate(self, flow_rule: FlowRule, port_id: int) -> bool: - """Validates a flow rule in the testpmd session. - - Args: - flow_rule: :class:`FlowRule` object used for validating testpmd flow rule. - port_id: Integer representing the port to use. - - Returns: - Boolean representing whether rule is valid or not. - """ - flow_output = self.send_command(f"flow validate {port_id} {flow_rule}") - if "Flow rule validated" in flow_output: - return True - return False - - def flow_delete(self, flow_id: int, port_id: int, verify: bool = True) -> None: - """Deletes the specified flow rule from the testpmd session. - - Args: - flow_id: ID of the flow to remove. - port_id: Integer representing the port to use. - verify: If :data:`True`, the output of the command is scanned - to ensure the flow rule was deleted successfully. - - Raises: - InteractiveCommandExecutionError: If flow rule is invalid. - """ - flow_output = self.send_command(f"flow destroy {port_id} rule {flow_id}") - if verify: - if "destroyed" not in flow_output: - self._logger.debug(f"Failed to delete flow rule:\n{flow_output}") - raise InteractiveCommandExecutionError( - f"Failed to delete flow rule:\n{flow_output}" - ) - - @requires_started_ports - @requires_stopped_ports - def set_port_mtu(self, port_id: int, mtu: int, verify: bool = True) -> None: - """Change the MTU of a port using testpmd. - - Some PMDs require that the port be stopped before changing the MTU, and it does no harm to - stop the port before configuring in cases where it isn't required, so ports are stopped - prior to changing their MTU. On the other hand, some PMDs require that the port had already - been started once since testpmd startup. Therefore, ports are also started before stopping - them to ensure this has happened. - - Args: - port_id: ID of the port to adjust the MTU on. - mtu: Desired value for the MTU to be set to. - verify: If `verify` is :data:`True` then the output will be scanned in an attempt to - verify that the mtu was properly set on the port. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the MTU was not - properly updated on the port matching `port_id`. - """ - set_mtu_output = self.send_command(f"port config mtu {port_id} {mtu}") - if verify and (f"MTU: {mtu}" not in self.send_command(f"show port info {port_id}")): - self._logger.debug( - f"Failed to set mtu to {mtu} on port {port_id}. Output was:\n{set_mtu_output}" - ) - raise InteractiveCommandExecutionError( - f"Test pmd failed to update mtu of port {port_id} to {mtu}" - ) - - def set_port_mtu_all(self, mtu: int, verify: bool = True) -> None: - """Change the MTU of all ports using testpmd. - - Runs :meth:`set_port_mtu` for every port that testpmd is aware of. - - Args: - mtu: Desired value for the MTU to be set to. - verify: Whether to verify that setting the MTU on each port was successful or not. - Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the MTU was not - properly updated on at least one port. - """ - for port in self.ports: - self.set_port_mtu(<a href="http://port.id" rel="noreferrer" target="_blank">port.id</a>, mtu, verify) - - @staticmethod - def extract_verbose_output(output: str) -> list[TestPmdVerbosePacket]: - """Extract the verbose information present in given testpmd output. - - This method extracts sections of verbose output that begin with the line - "port X/queue Y: sent/received Z packets" and end with the ol_flags of a packet. - - Args: - output: Testpmd output that contains verbose information - - Returns: - List of parsed packet information gathered from verbose information in `output`. - """ - out: list[TestPmdVerbosePacket] = [] - prev_header: str = "" - iter = re.finditer( - r"(?P<HEADER>(?:port \d+/queue \d+: (?:received|sent) \d+ packets)?)\s*" - r"(?P<PACKET>src=[\w\s=:-]+?ol_flags: [\w ]+)", - output, - ) - for match in iter: - if match.group("HEADER"): - prev_header = match.group("HEADER") - out.append(TestPmdVerbosePacket.parse(f"{prev_header}\n{match.group('PACKET')}")) - return out - - @requires_stopped_ports - def set_vlan_filter(self, port: int, enable: bool, verify: bool = True) -> None: - """Set vlan filter on. - - Args: - port: The port number to enable VLAN filter on. - enable: Enable the filter on `port` if :data:`True`, otherwise disable it. - verify: If :data:`True`, the output of the command and show port info - is scanned to verify that vlan filtering was set successfully. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the filter - fails to update. - """ - filter_cmd_output = self.send_command(f"vlan set filter {'on' if enable else 'off'} {port}") - if verify: - vlan_settings = self.show_port_info(port_id=port).vlan_offload - if enable ^ (vlan_settings is not None and VLANOffloadFlag.FILTER in vlan_settings): - self._logger.debug( - f"""Failed to {"enable" if enable else "disable"} - filter on port {port}: \n{filter_cmd_output}""" - ) - raise InteractiveCommandExecutionError( - f"""Failed to {"enable" if enable else "disable"} - filter on port {port}""" - ) - - def set_mac_address(self, port: int, mac_address: str, verify: bool = True) -> None: - """Set port's MAC address. - - Args: - port: The number of the requested port. - mac_address: The MAC address to set. - verify: If :data:`True`, the output of the command is scanned to verify that - the mac address is set in the specified port. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the command - fails to execute. - """ - output = self.send_command(f"mac_addr set {port} {mac_address}", skip_first_line=True) - if verify: - if output.strip(): - self._logger.debug( - f"Testpmd failed to set MAC address {mac_address} on port {port}." - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to set MAC address {mac_address} on port {port}." - ) - - def set_flow_control( - self, port: int, flow_ctrl: TestPmdPortFlowCtrl, verify: bool = True - ) -> None: - """Set the given `port`'s flow control. - - Args: - port: The number of the requested port. - flow_ctrl: The requested flow control parameters. - verify: If :data:`True`, the output of the command is scanned to verify that - the flow control in the specified port is set. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the command - fails to execute. - """ - output = self.send_command(f"set flow_ctrl {flow_ctrl} {port}", skip_first_line=True) - if verify: - if output.strip(): - self._logger.debug(f"Testpmd failed to set the {flow_ctrl} in port {port}.") - raise InteractiveCommandExecutionError( - f"Testpmd failed to set the {flow_ctrl} in port {port}." - ) - - def show_port_flow_info(self, port: int) -> TestPmdPortFlowCtrl | None: - """Show port info flow. - - Args: - port: The number of the requested port. - - Returns: - The current port flow control parameters if supported, otherwise :data:`None`. - """ - output = self.send_command(f"show port {port} flow_ctrl") - if "Flow control infos" in output: - return TestPmdPortFlowCtrl.parse(output) - return None - - @requires_stopped_ports - def rx_vlan(self, vlan: int, port: int, add: bool, verify: bool = True) -> None: - """Add specified vlan tag to the filter list on a port. Requires vlan filter to be on. - - Args: - vlan: The vlan tag to add, should be within 1-1005. - port: The port number to add the tag on. - add: Adds the tag if :data:`True`, otherwise removes the tag. - verify: If :data:`True`, the output of the command is scanned to verify that - the vlan tag was added to the filter list on the specified port. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the tag - is not added. - """ - rx_cmd_output = self.send_command(f"rx_vlan {'add' if add else 'rm'} {vlan} {port}") - if verify: - if ( - "VLAN-filtering disabled" in rx_cmd_output - or "Invalid vlan_id" in rx_cmd_output - or "Bad arguments" in rx_cmd_output - ): - self._logger.debug( - f"""Failed to {"add" if add else "remove"} tag {vlan} - port {port}: \n{rx_cmd_output}""" - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to {'add' if add else 'remove'} tag {vlan} on port {port}." - ) - - @requires_stopped_ports - def set_vlan_strip(self, port: int, enable: bool, verify: bool = True) -> None: - """Enable or disable vlan stripping on the specified port. - - Args: - port: The port number to use. - enable: If :data:`True`, will turn vlan stripping on, otherwise will turn off. - verify: If :data:`True`, the output of the command and show port info - is scanned to verify that vlan stripping was enabled on the specified port. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and stripping - fails to update. - """ - strip_cmd_output = self.send_command(f"vlan set strip {'on' if enable else 'off'} {port}") - if verify: - vlan_settings = self.show_port_info(port_id=port).vlan_offload - if enable ^ (vlan_settings is not None and VLANOffloadFlag.STRIP in vlan_settings): - self._logger.debug( - f"""Failed to set strip {"on" if enable else "off"} - port {port}: \n{strip_cmd_output}""" - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to set strip {'on' if enable else 'off'} port {port}." - ) - - @requires_stopped_ports - def tx_vlan_set( - self, port: int, enable: bool, vlan: int | None = None, verify: bool = True - ) -> None: - """Set hardware insertion of vlan tags in packets sent on a port. - - Args: - port: The port number to use. - enable: Sets vlan tag insertion if :data:`True`, and resets if :data:`False`. - vlan: The vlan tag to insert if enable is :data:`True`. - verify: If :data:`True`, the output of the command is scanned to verify that - vlan insertion was enabled on the specified port. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the insertion - tag is not set. - """ - if enable: - tx_vlan_cmd_output = self.send_command(f"tx_vlan set {port} {vlan}") - if verify: - if ( - "Please stop port" in tx_vlan_cmd_output - or "Invalid vlan_id" in tx_vlan_cmd_output - or "Invalid port" in tx_vlan_cmd_output - ): - self._logger.debug( - f"Failed to set vlan tag {vlan} on port {port}:\n{tx_vlan_cmd_output}" - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to set vlan insertion tag {vlan} on port {port}." - ) - else: - tx_vlan_cmd_output = self.send_command(f"tx_vlan reset {port}") - if verify: - if "Please stop port" in tx_vlan_cmd_output or "Invalid port" in tx_vlan_cmd_output: - self._logger.debug( - f"Failed to reset vlan insertion on port {port}: \n{tx_vlan_cmd_output}" - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to reset vlan insertion on port {port}." - ) - - def set_promisc(self, port: int, enable: bool, verify: bool = True) -> None: - """Enable or disable promiscuous mode for the specified port. - - Args: - port: Port number to use. - enable: If :data:`True`, turn promiscuous mode on, otherwise turn off. - verify: If :data:`True` an additional command will be sent to verify that - promiscuous mode is properly set. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and promiscuous mode - is not correctly set. - """ - promisc_cmd_output = self.send_command(f"set promisc {port} {'on' if enable else 'off'}") - if verify: - stats = self.show_port_info(port_id=port) - if enable ^ stats.is_promiscuous_mode_enabled: - self._logger.debug( - f"Failed to set promiscuous mode on port {port}: \n{promisc_cmd_output}" - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to set promiscuous mode on port {port}." - ) - - def set_verbose(self, level: int, verify: bool = True) -> None: - """Set debug verbosity level. - - Args: - level: 0 - silent except for error - 1 - fully verbose except for Tx packets - 2 - fully verbose except for Rx packets - >2 - fully verbose - verify: If :data:`True` the command output will be scanned to verify that verbose level - is properly set. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and verbose level - is not correctly set. - """ - verbose_cmd_output = self.send_command(f"set verbose {level}") - if verify: - if "Change verbose level" not in verbose_cmd_output: - self._logger.debug( - f"Failed to set verbose level to {level}: \n{verbose_cmd_output}" - ) - raise InteractiveCommandExecutionError( - f"Testpmd failed to set verbose level to {level}." - ) - - def rx_vxlan(self, vxlan_id: int, port_id: int, enable: bool, verify: bool = True) -> None: - """Add or remove vxlan id to/from filter list. - - Args: - vxlan_id: VXLAN ID to add to port filter list. - port_id: ID of the port to modify VXLAN filter of. - enable: If :data:`True`, adds specified VXLAN ID, otherwise removes it. - verify: If :data:`True`, the output of the command is checked to verify - the VXLAN ID was successfully added/removed from the port. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and VXLAN ID - is not successfully added or removed. - """ - action = "add" if enable else "rm" - vxlan_output = self.send_command(f"rx_vxlan_port {action} {vxlan_id} {port_id}") - if verify: - if "udp tunneling add error" in vxlan_output: - self._logger.debug(f"Failed to set VXLAN:\n{vxlan_output}") - raise InteractiveCommandExecutionError(f"Failed to set VXLAN:\n{vxlan_output}") - - def clear_port_stats(self, port_id: int, verify: bool = True) -> None: - """Clear statistics of a given port. - - Args: - port_id: ID of the port to clear the statistics on. - verify: If :data:`True` the output of the command will be scanned to verify that it was - successful, otherwise failures will be ignored. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and testpmd fails to - clear the statistics of the given port. - """ - clear_output = self.send_command(f"clear port stats {port_id}") - if verify and f"NIC statistics for port {port_id} cleared" not in clear_output: - raise InteractiveCommandExecutionError( - f"Test pmd failed to set clear forwarding stats on port {port_id}" - ) - - def clear_port_stats_all(self, verify: bool = True) -> None: - """Clear the statistics of all ports that testpmd is aware of. - - Args: - verify: If :data:`True` the output of the command will be scanned to verify that all - ports had their statistics cleared, otherwise failures will be ignored. Defaults to - :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and testpmd fails to - clear the statistics of any of its ports. - """ - clear_output = self.send_command("clear port stats all") - if verify: - if type(self._app_params.port_numa_config) is list: - for port_id in range(len(self._app_params.port_numa_config)): - if f"NIC statistics for port {port_id} cleared" not in clear_output: - raise InteractiveCommandExecutionError( - f"Test pmd failed to set clear forwarding stats on port {port_id}" - ) - - @only_active - def close(self) -> None: - """Overrides :meth:`~.interactive_shell.close`.""" - self.stop() - self.send_command("quit", "Bye...") - return super().close() - - """ - ====== Capability retrieval methods ====== - """ - - def get_capabilities_rx_offload( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - ) -> None: - """Get all rx offload capabilities and divide them into supported and unsupported. - - Args: - supported_capabilities: Supported capabilities will be added to this set. - unsupported_capabilities: Unsupported capabilities will be added to this set. - """ - self._logger.debug("Getting rx offload capabilities.") - command = f"show port {self.ports[0].id} rx_offload capabilities" - rx_offload_capabilities_out = self.send_command(command) - rx_offload_capabilities = RxOffloadCapabilities.parse(rx_offload_capabilities_out) - self._update_capabilities_from_flag( - supported_capabilities, - unsupported_capabilities, - RxOffloadCapability, - rx_offload_capabilities.per_port | rx_offload_capabilities.per_queue, - ) - - def get_port_queue_info( - self, port_id: int, queue_id: int, is_rx_queue: bool - ) -> TestPmdQueueInfo: - """Returns the current state of the specified queue.""" - command = f"show {'rxq' if is_rx_queue else 'txq'} info {port_id} {queue_id}" - queue_info = TestPmdQueueInfo.parse(self.send_command(command)) - return queue_info - - def setup_port_queue(self, port_id: int, queue_id: int, is_rx_queue: bool) -> None: - """Setup a given queue on a port. - - This functionality cannot be verified because the setup action only takes effect when the - queue is started. - - Args: - port_id: ID of the port where the queue resides. - queue_id: ID of the queue to setup. - is_rx_queue: Type of queue to setup. If :data:`True` an RX queue will be setup, - otherwise a TX queue will be setup. - """ - self.send_command(f"port {port_id} {'rxq' if is_rx_queue else 'txq'} {queue_id} setup") - - def stop_port_queue( - self, port_id: int, queue_id: int, is_rx_queue: bool, verify: bool = True - ) -> None: - """Stops a given queue on a port. - - Args: - port_id: ID of the port that the queue belongs to. - queue_id: ID of the queue to stop. - is_rx_queue: Type of queue to stop. If :data:`True` an RX queue will be stopped, - otherwise a TX queue will be stopped. - verify: If :data:`True` an additional command will be sent to verify the queue stopped. - Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the queue fails to - stop. - """ - port_type = "rxq" if is_rx_queue else "txq" - stop_cmd_output = self.send_command(f"port {port_id} {port_type} {queue_id} stop") - if verify: - queue_started = self.get_port_queue_info( - port_id, queue_id, is_rx_queue - ).is_queue_started - if queue_started: - self._logger.debug( - f"Failed to stop {port_type} {queue_id} on port {port_id}:\n{stop_cmd_output}" - ) - raise InteractiveCommandExecutionError( - f"Test pmd failed to stop {port_type} {queue_id} on port {port_id}" - ) - - def start_port_queue( - self, port_id: int, queue_id: int, is_rx_queue: bool, verify: bool = True - ) -> None: - """Starts a given queue on a port. - - First sets up the port queue, then starts it. - - Args: - port_id: ID of the port that the queue belongs to. - queue_id: ID of the queue to start. - is_rx_queue: Type of queue to start. If :data:`True` an RX queue will be started, - otherwise a TX queue will be started. - verify: if :data:`True` an additional command will be sent to verify that the queue was - started. Defaults to :data:`True`. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and the queue fails to - start. - """ - port_type = "rxq" if is_rx_queue else "txq" - self.setup_port_queue(port_id, queue_id, is_rx_queue) - start_cmd_output = self.send_command(f"port {port_id} {port_type} {queue_id} start") - if verify: - queue_started = self.get_port_queue_info( - port_id, queue_id, is_rx_queue - ).is_queue_started - if not queue_started: - self._logger.debug( - f"Failed to start {port_type} {queue_id} on port {port_id}:\n{start_cmd_output}" - ) - raise InteractiveCommandExecutionError( - f"Test pmd failed to start {port_type} {queue_id} on port {port_id}" - ) - - def get_queue_ring_size(self, port_id: int, queue_id: int, is_rx_queue: bool) -> int: - """Returns the current size of the ring on the specified queue.""" - command = f"show {'rxq' if is_rx_queue else 'txq'} info {port_id} {queue_id}" - queue_info = TestPmdQueueInfo.parse(self.send_command(command)) - return queue_info.ring_size - - def set_queue_ring_size( - self, - port_id: int, - queue_id: int, - size: int, - is_rx_queue: bool, - verify: bool = True, - ) -> None: - """Update the ring size of an Rx/Tx queue on a given port. - - Queue is setup after setting the ring size so that the queue info reflects this change and - it can be verified. - - Args: - port_id: The port that the queue resides on. - queue_id: The ID of the queue on the port. - size: The size to update the ring size to. - is_rx_queue: Whether to modify an RX or TX queue. If :data:`True` an RX queue will be - updated, otherwise a TX queue will be updated. - verify: If :data:`True` an additional command will be sent to check the ring size of - the queue in an attempt to validate that the size was changes properly. - - Raises: - InteractiveCommandExecutionError: If `verify` is :data:`True` and there is a failure - when updating ring size. - """ - queue_type = "rxq" if is_rx_queue else "txq" - self.send_command(f"port config {port_id} {queue_type} {queue_id} ring_size {size}") - self.setup_port_queue(port_id, queue_id, is_rx_queue) - if verify: - curr_ring_size = self.get_queue_ring_size(port_id, queue_id, is_rx_queue) - if curr_ring_size != size: - self._logger.debug( - f"Failed up update ring size of queue {queue_id} on port {port_id}. Current" - f" ring size is {curr_ring_size}." - ) - raise InteractiveCommandExecutionError( - f"Failed to update ring size of queue {queue_id} on port {port_id}" - ) - - @requires_stopped_ports - def set_queue_deferred_start( - self, port_id: int, queue_id: int, is_rx_queue: bool, on: bool - ) -> None: - """Set the deferred start attribute of the specified queue on/off. - - Args: - port_id: The port that the queue resides on. - queue_id: The ID of the queue on the port. - is_rx_queue: Whether to modify an RX or TX queue. If :data:`True` an RX queue will be - updated, otherwise a TX queue will be updated. - on: Whether to set deferred start mode on or off. If :data:`True` deferred start will - be turned on, otherwise it will be turned off. - """ - queue_type = "rxq" if is_rx_queue else "txq" - action = "on" if on else "off" - self.send_command(f"port {port_id} {queue_type} {queue_id} deferred_start {action}") - - def _update_capabilities_from_flag( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - flag_class: type[Flag], - supported_flags: Flag, - ) -> None: - """Divide all flags from `flag_class` into supported and unsupported.""" - for flag in flag_class: - if flag in supported_flags: - supported_capabilities.add(NicCapability[str(<a href="http://flag.name" rel="noreferrer" target="_blank">flag.name</a>)]) - else: - unsupported_capabilities.add(NicCapability[str(<a href="http://flag.name" rel="noreferrer" target="_blank">flag.name</a>)]) - - @requires_started_ports - def get_capabilities_rxq_info( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - ) -> None: - """Get all rxq capabilities and divide them into supported and unsupported. - - Args: - supported_capabilities: Supported capabilities will be added to this set. - unsupported_capabilities: Unsupported capabilities will be added to this set. - """ - self._logger.debug("Getting rxq capabilities.") - command = f"show rxq info {self.ports[0].id} 0" - rxq_info = TestPmdRxqInfo.parse(self.send_command(command)) - if rxq_info.scattered_packets: - supported_capabilities.add(NicCapability.SCATTERED_RX_ENABLED) - else: - unsupported_capabilities.add(NicCapability.SCATTERED_RX_ENABLED) - - def get_capabilities_show_port_info( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - ) -> None: - """Get all capabilities from show port info and divide them into supported and unsupported. - - Args: - supported_capabilities: Supported capabilities will be added to this set. - unsupported_capabilities: Unsupported capabilities will be added to this set. - """ - self._update_capabilities_from_flag( - supported_capabilities, - unsupported_capabilities, - DeviceCapabilitiesFlag, - self.ports[0].device_capabilities, - ) - - def get_capabilities_mcast_filtering( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - ) -> None: - """Get multicast filtering capability from mcast_addr add and check for testpmd error code. - - Args: - supported_capabilities: Supported capabilities will be added to this set. - unsupported_capabilities: Unsupported capabilities will be added to this set. - """ - self._logger.debug("Getting mcast filter capabilities.") - command = f"mcast_addr add {self.ports[0].id} 01:00:5E:00:00:00" - output = self.send_command(command) - if "diag=-95" in output: - unsupported_capabilities.add(NicCapability.MCAST_FILTERING) - else: - supported_capabilities.add(NicCapability.MCAST_FILTERING) - command = str.replace(command, "add", "remove", 1) - self.send_command(command) - - def get_capabilities_flow_ctrl( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - ) -> None: - """Get flow control capability and check for testpmd failure. - - Args: - supported_capabilities: Supported capabilities will be added to this set. - unsupported_capabilities: Unsupported capabilities will be added to this set. - """ - self._logger.debug("Getting flow ctrl capabilities.") - command = f"show port {self.ports[0].id} flow_ctrl" - output = self.send_command(command) - if "Flow control infos" in output: - supported_capabilities.add(NicCapability.FLOW_CTRL) - else: - unsupported_capabilities.add(NicCapability.FLOW_CTRL) - - def get_capabilities_physical_function( - self, - supported_capabilities: MutableSet["NicCapability"], - unsupported_capabilities: MutableSet["NicCapability"], - ) -> None: - """Store capability representing a physical function test run. - - Args: - supported_capabilities: Supported capabilities will be added to this set. - unsupported_capabilities: Unsupported capabilities will be added to this set. - """ - ctx = get_ctx() - if ctx.topology.vf_ports == []: - supported_capabilities.add(NicCapability.PHYSICAL_FUNCTION) - else: - unsupported_capabilities.add(NicCapability.PHYSICAL_FUNCTION) - - -class NicCapability(NoAliasEnum): - """A mapping between capability names and the associated :class:`TestPmdShell` methods. - - The :class:`TestPmdShell` capability checking method executes the command that checks - whether the capability is supported. - A decorator may optionally be added to the method that will add and remove configuration - that's necessary to retrieve the capability support status. - The Enum members may be assigned the method itself or a tuple of - (capability_checking_method, decorator_function). - - The signature of each :class:`TestPmdShell` capability checking method must be:: - - fn(self, supported_capabilities: MutableSet, unsupported_capabilities: MutableSet) -> None - - The capability checking method must get whether a capability is supported or not - from a testpmd command. If multiple capabilities can be obtained from a testpmd command, - each should be obtained in the method. These capabilities should then - be added to `supported_capabilities` or `unsupported_capabilities` based on their support. - - The two dictionaries are shared across all capability discovery function calls in a given - test run so that we don't call the same function multiple times. For example, when we find - :attr:`SCATTERED_RX_ENABLED` in :meth:`TestPmdShell.get_capabilities_rxq_info`, - we don't go looking for it again if a different test case also needs it. - """ - - #: Scattered packets Rx enabled - SCATTERED_RX_ENABLED: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rxq_info, - add_remove_mtu(9000), - ) - #: - RX_OFFLOAD_VLAN_STRIP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports L3 checksum offload. - RX_OFFLOAD_IPV4_CKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports L4 checksum offload. - RX_OFFLOAD_UDP_CKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports L4 checksum offload. - RX_OFFLOAD_TCP_CKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports Large Receive Offload. - RX_OFFLOAD_TCP_LRO: TestPmdShellNicCapability = (TestPmdShell.get_capabilities_rx_offload, None) - #: Device supports QinQ (queue in queue) offload. - RX_OFFLOAD_QINQ_STRIP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports inner packet L3 checksum. - RX_OFFLOAD_OUTER_IPV4_CKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports MACsec. - RX_OFFLOAD_MACSEC_STRIP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports filtering of a VLAN Tag identifier. - RX_OFFLOAD_VLAN_FILTER: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports VLAN offload. - RX_OFFLOAD_VLAN_EXTEND: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports receiving segmented mbufs. - RX_OFFLOAD_SCATTER: TestPmdShellNicCapability = (TestPmdShell.get_capabilities_rx_offload, None) - #: Device supports Timestamp. - RX_OFFLOAD_TIMESTAMP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports crypto processing while packet is received in NIC. - RX_OFFLOAD_SECURITY: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports CRC stripping. - RX_OFFLOAD_KEEP_CRC: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports L4 checksum offload. - RX_OFFLOAD_SCTP_CKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports inner packet L4 checksum. - RX_OFFLOAD_OUTER_UDP_CKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports RSS hashing. - RX_OFFLOAD_RSS_HASH: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports scatter Rx packets to segmented mbufs. - RX_OFFLOAD_BUFFER_SPLIT: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports all checksum capabilities. - RX_OFFLOAD_CHECKSUM: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_rx_offload, - None, - ) - #: Device supports all VLAN capabilities. - RX_OFFLOAD_VLAN: TestPmdShellNicCapability = (TestPmdShell.get_capabilities_rx_offload, None) - #: Device supports Rx queue setup after device started. - RUNTIME_RX_QUEUE_SETUP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_show_port_info, - None, - ) - #: Device supports Tx queue setup after device started. - RUNTIME_TX_QUEUE_SETUP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_show_port_info, - None, - ) - #: Device supports shared Rx queue among ports within Rx domain and switch domain. - RXQ_SHARE: TestPmdShellNicCapability = (TestPmdShell.get_capabilities_show_port_info, None) - #: Device supports keeping flow rules across restart. - FLOW_RULE_KEEP: TestPmdShellNicCapability = (TestPmdShell.get_capabilities_show_port_info, None) - #: Device supports keeping shared flow objects across restart. - FLOW_SHARED_OBJECT_KEEP: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_show_port_info, - None, - ) - #: Device supports multicast address filtering. - MCAST_FILTERING: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_mcast_filtering, - None, - ) - #: Device supports flow ctrl. - FLOW_CTRL: TestPmdShellNicCapability = (TestPmdShell.get_capabilities_flow_ctrl, None) - #: Device is running on a physical function. - PHYSICAL_FUNCTION: TestPmdShellNicCapability = ( - TestPmdShell.get_capabilities_physical_function, - None, - ) - - def __call__( - self, - testpmd_shell: TestPmdShell, - supported_capabilities: MutableSet[Self], - unsupported_capabilities: MutableSet[Self], - ) -> None: - """Execute the associated capability retrieval function. - - Args: - testpmd_shell: :class:`TestPmdShell` object to which the function will be bound to. - supported_capabilities: The dictionary storing the supported capabilities - of a given test run. - unsupported_capabilities: The dictionary storing the unsupported capabilities - of a given test run. - """ - self.value(testpmd_shell, supported_capabilities, unsupported_capabilities) diff --git a/dts/framework/testbed_model/capability.py b/dts/framework/testbed_model/capability.py index f895b22bb3..a4a8d9b7b4 100644 --- a/dts/framework/testbed_model/capability.py +++ b/dts/framework/testbed_model/capability.py @@ -26,9 +26,9 @@ .. code:: python from framework.test_suite import TestSuite, func_test - from framework.testbed_model.capability import TopologyType, requires + from framework.testbed_model.capability import LinkTopology, requires # The whole test suite (each test case within) doesn't require any links. - @requires(topology_type=TopologyType.no_link) + @requires_link_topology(LinkTopology.NO_LINK) @func_test class TestHelloWorld(TestSuite): def hello_world_single_core(self): @@ -41,7 +41,7 @@ def hello_world_single_core(self): class TestPmdBufferScatter(TestSuite): # only the test case requires the SCATTERED_RX_ENABLED capability # other test cases may not require it - @requires(NicCapability.SCATTERED_RX_ENABLED) + @requires_nic_capability(NicCapability.SCATTERED_RX_ENABLED) @func_test def test_scatter_mbuf_2048(self): """ @@ -50,27 +50,38 @@ def test_scatter_mbuf_2048(self): from abc import ABC, abstractmethod from collections.abc import MutableSet from dataclasses import dataclass -from typing import TYPE_CHECKING, Callable, ClassVar, Protocol +from typing import ( + TYPE_CHECKING, + Any, + Callable, + ClassVar, + Concatenate, + ParamSpec, + Protocol, + TypeAlias, +) from typing_extensions import Self +from api.capabilities import LinkTopology, NicCapability from framework.exception import ConfigurationError, InternalError, SkippedTestException from framework.logger import get_dts_logger -from framework.remote_session.testpmd_shell import ( - NicCapability, - TestPmdShell, - TestPmdShellCapabilityMethod, - TestPmdShellDecorator, - TestPmdShellMethod, -) from framework.testbed_model.node import Node from framework.testbed_model.port import DriverKind - -from .topology import Topology, TopologyType +from framework.testbed_model.topology import Topology if TYPE_CHECKING: + from api.testpmd import TestPmd from framework.test_suite import TestCase +P = ParamSpec("P") +TestPmdMethod = Callable[Concatenate["TestPmd", P], Any] +TestPmdCapabilityMethod: TypeAlias = Callable[ + ["TestPmd", MutableSet["NicCapability"], MutableSet["NicCapability"]], None +] +TestPmdDecorator: TypeAlias = Callable[[TestPmdMethod], TestPmdMethod] +TestPmdNicCapability = tuple[TestPmdCapabilityMethod, TestPmdDecorator | None] + class Capability(ABC): """The base class for various capabilities. @@ -153,7 +164,7 @@ def __hash__(self) -> int: @dataclass class DecoratedNicCapability(Capability): - """A wrapper around :class:`~framework.remote_session.testpmd_shell.NicCapability`. + """A wrapper around :class:`~api.testpmd.NicCapability`. New instances should be created with the :meth:`create_unique` class method to ensure there are no duplicate instances. @@ -166,10 +177,69 @@ class DecoratedNicCapability(Capability): """ nic_capability: NicCapability - capability_fn: TestPmdShellCapabilityMethod - capability_decorator: TestPmdShellDecorator | None + capability_fn: TestPmdCapabilityMethod + capability_decorator: TestPmdDecorator | None _unique_capabilities: ClassVar[dict[NicCapability, Self]] = {} + @classmethod + def _get_nic_capability_check(cls) -> list[TestPmdNicCapability]: + """A mapping between capability names and the associated :class:`TestPmd` methods. + + The :class:`TestPmd` capability checking method executes the command that checks + whether the capability is supported. + A decorator may optionally be added to the method that will add and remove configuration + that's necessary to retrieve the capability support status. + The Enum members may be assigned the method itself or a tuple of + (capability_checking_method, decorator_function). + + The signature of each :class:`TestPmd` capability checking method must be:: + + fn(self, supported_capabilities: MutableSet, unsupported_capabilities: MutableSet) + + The capability checking method must get whether a capability is supported or not + from a testpmd command. If multiple capabilities can be obtained from a testpmd command, + each should be obtained in the method. These capabilities should then + be added to `supported_capabilities` or `unsupported_capabilities` based on their support. + + The two dictionaries are shared across all capability discovery function calls in a given + test run so that we don't call the same function multiple times. For example, when we find + :attr:`SCATTERED_RX_ENABLED` in :meth:`TestPmd.get_capabilities_rxq_info`, + we don't go looking for it again if a different test case also needs it. + """ + from api.testpmd import TestPmd, _add_remove_mtu + + return [ + (TestPmd.get_capabilities_rxq_info, _add_remove_mtu(9000)), + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_VLAN_STRIP + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_IPV4_CKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_UDP_CKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_TCP_CKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_TCP_LRO + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_QINQ_STRIP + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_OUTER_IPV4_CKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_MACSEC_STRIP + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_VLAN_FILTER + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_VLAN_EXTEND + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_SCATTER + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_TIMESTAMP + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_SECURITY + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_KEEP_CRC + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_SCTP_CKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_OUTER_UDP_CKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_RSS_HASH + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_BUFFER_SPLIT + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_CHECKSUM + (TestPmd.get_capabilities_rx_offload, None), # RX_OFFLOAD_VLAN + (TestPmd.get_capabilities_show_port_info, None), # RUNTIME_RX_QUEUE_SETUP + (TestPmd.get_capabilities_show_port_info, None), # RUNTIME_TX_QUEUE_SETUP + (TestPmd.get_capabilities_show_port_info, None), # RXQ_SHARE + (TestPmd.get_capabilities_show_port_info, None), # FLOW_RULE_KEEP + (TestPmd.get_capabilities_show_port_info, None), # FLOW_SHARED_OBJECT_KEEP + (TestPmd.get_capabilities_mcast_filtering, None), # MCAST_FILTERING + (TestPmd.get_capabilities_flow_ctrl, None), # FLOW_CTRL + (TestPmd.get_capabilities_physical_function, None), # PHYSICAL_FUNCTION + ] + @classmethod def get_unique(cls, nic_capability: NicCapability) -> Self: """Get the capability uniquely identified by `nic_capability`. @@ -188,7 +258,7 @@ def get_unique(cls, nic_capability: NicCapability) -> Self: Returns: The capability uniquely identified by `nic_capability`. """ - capability_fn, decorator_fn = nic_capability.value + capability_fn, decorator_fn = cls._get_nic_capability_check()[nic_capability.value] if nic_capability not in cls._unique_capabilities: cls._unique_capabilities[nic_capability] = cls( @@ -207,9 +277,11 @@ def get_supported_capabilities( Each capability is first checked whether it's supported/unsupported before executing its `capability_fn` so that each capability is retrieved only once. """ + from api.testpmd import TestPmd + supported_conditional_capabilities: set["DecoratedNicCapability"] = set() logger = get_dts_logger(f"{<a href="http://node.name" rel="noreferrer" target="_blank">node.name</a>}.{cls.__name__}") - if topology.type is topology.type.no_link: + if topology.type is topology.type.NO_LINK: logger.debug( "No links available in the current topology, not getting NIC capabilities." ) @@ -219,7 +291,7 @@ def get_supported_capabilities( ) if cls.capabilities_to_check: capabilities_to_check_map = cls._get_decorated_capabilities_map() - with TestPmdShell() as testpmd_shell: + with TestPmd() as testpmd: for ( conditional_capability_fn, capabilities, @@ -231,7 +303,7 @@ def get_supported_capabilities( ) if conditional_capability_fn: capability_fn = conditional_capability_fn(capability_fn) - capability_fn(testpmd_shell) + capability_fn(testpmd) for capability in capabilities: if capability.nic_capability in supported_capabilities: supported_conditional_capabilities.add(capability) @@ -242,8 +314,8 @@ def get_supported_capabilities( @classmethod def _get_decorated_capabilities_map( cls, - ) -> dict[TestPmdShellDecorator | None, set["DecoratedNicCapability"]]: - capabilities_map: dict[TestPmdShellDecorator | None, set["DecoratedNicCapability"]] = {} + ) -> dict[TestPmdDecorator | None, set["DecoratedNicCapability"]]: + capabilities_map: dict[TestPmdDecorator | None, set["DecoratedNicCapability"]] = {} for capability in cls.capabilities_to_check: if capability.capability_decorator not in capabilities_map: capabilities_map[capability.capability_decorator] = set() @@ -257,12 +329,12 @@ def _reduce_capabilities( capabilities: set["DecoratedNicCapability"], supported_capabilities: MutableSet, unsupported_capabilities: MutableSet, - ) -> TestPmdShellMethod: - def reduced_fn(testpmd_shell: TestPmdShell) -> None: + ) -> TestPmdMethod: + def reduced_fn(testpmd: "TestPmd") -> None: for capability in capabilities: if capability not in supported_capabilities | unsupported_capabilities: capability.capability_fn( - testpmd_shell, supported_capabilities, unsupported_capabilities + testpmd, supported_capabilities, unsupported_capabilities ) return reduced_fn @@ -278,11 +350,11 @@ def __repr__(self) -> str: @dataclass class TopologyCapability(Capability): - """A wrapper around :class:`~.topology.TopologyType`. + """A wrapper around :class:`~.topology.LinkTopology`. Each test case must be assigned a topology. It could be done explicitly; - the implicit default is given by :meth:`~.topology.TopologyType.default`, which this class - returns :attr:`~.topology.TopologyType.two_links`. + the implicit default is given by :meth:`~.topology.LinkTopology.default`, which this class + returns :attr:`~.topology.LinkTopology.TWO_LINKS`. Test case topology may be set by setting the topology for the whole suite. The priority in which topology is set is as follows: @@ -301,7 +373,7 @@ class TopologyCapability(Capability): topology_type: The topology type that defines each instance. """ - topology_type: TopologyType + topology_type: LinkTopology _unique_capabilities: ClassVar[dict[str, Self]] = {} @@ -310,7 +382,7 @@ def _preprocess_required(self, test_case_or_suite: type["TestProtocol"]) -> None test_case_or_suite.topology_type = self @classmethod - def get_unique(cls, topology_type: TopologyType) -> Self: + def get_unique(cls, topology_type: LinkTopology) -> Self: """Get the capability uniquely identified by `topology_type`. This is a factory method that implements a quasi-enum pattern. @@ -338,7 +410,7 @@ def get_supported_capabilities( """Overrides :meth:`~Capability.get_supported_capabilities`.""" supported_capabilities = set() topology_capability = cls.get_unique(topology.type) - for topology_type in TopologyType: + for topology_type in LinkTopology: candidate_topology_type = cls.get_unique(topology_type) if candidate_topology_type <= topology_capability: supported_capabilities.add(candidate_topology_type) @@ -351,17 +423,17 @@ def set_required(self, test_case_or_suite: type["TestProtocol"]) -> None: This means we have to modify test case topologies when processing the test suite topologies. At that point, the test case topologies have been set by the :func:`requires` decorator. The test suite topology only affects the test case topologies - if not :attr:`~.topology.TopologyType.default`. + if not :attr:`~.topology.LinkTopology.default`. Raises: ConfigurationError: If the topology type requested by the test case is more complex than the test suite's. """ if inspect.isclass(test_case_or_suite): - if self.topology_type is not TopologyType.default(): + if self.topology_type is not LinkTopology.default(): self.add_to_required(test_case_or_suite) for test_case in test_case_or_suite.get_test_cases(): - if test_case.topology_type.topology_type is TopologyType.default(): + if test_case.topology_type.topology_type is LinkTopology.default(): # test case topology has not been set, use the one set by the test suite self.add_to_required(test_case) elif test_case.topology_type > test_case_or_suite.topology_type: @@ -440,7 +512,7 @@ class TestProtocol(Protocol): #: The reason for skipping the test case or suite. skip_reason: ClassVar[str] = "" #: The topology type of the test case or suite. - topology_type: ClassVar[TopologyCapability] = TopologyCapability(TopologyType.default()) + topology_type: ClassVar[TopologyCapability] = TopologyCapability(LinkTopology.default()) #: The capabilities the test case or suite requires in order to be executed. required_capabilities: ClassVar[set[Capability]] = set() #: The SUT ports topology configuration of the test case or suite. @@ -481,7 +553,7 @@ def _decorator(func: type[TestProtocol]) -> type[TestProtocol]: def requires( *nic_capabilities: NicCapability, - topology_type: TopologyType = TopologyType.default(), + topology_type: LinkTopology = LinkTopology.default(), ) -> Callable[[type[TestProtocol]], type[TestProtocol]]: """A decorator that adds the required capabilities to a test case or test suite. diff --git a/dts/framework/testbed_model/linux_session.py b/dts/framework/testbed_model/linux_session.py index 604245d855..1f11c3e740 100644 --- a/dts/framework/testbed_model/linux_session.py +++ b/dts/framework/testbed_model/linux_session.py @@ -22,7 +22,7 @@ InternalError, RemoteCommandExecutionError, ) -from framework.testbed_model.os_session import PortInfo +from framework.testbed_model.port import PortInfo from framework.utils import expand_range from .cpu import LogicalCore diff --git a/dts/framework/testbed_model/os_session.py b/dts/framework/testbed_model/os_session.py index b6e03aa83d..c1872880da 100644 --- a/dts/framework/testbed_model/os_session.py +++ b/dts/framework/testbed_model/os_session.py @@ -31,13 +31,9 @@ from framework.config.node import NodeConfiguration from framework.logger import DTSLogger -from framework.remote_session import ( - InteractiveRemoteSession, - RemoteSession, - create_interactive_session, - create_remote_session, -) -from framework.remote_session.remote_session import CommandResult +from framework.remote_session.interactive_remote_session import InteractiveRemoteSession +from framework.remote_session.remote_session import CommandResult, RemoteSession +from framework.remote_session.ssh_session import SSHSession from framework.settings import SETTINGS from framework.utils import MesonArgs, TarCompressionFormat @@ -129,8 +125,8 @@ def __init__( self._config = node_config <a href="http://self.name" rel="noreferrer" target="_blank">self.name</a> = name self._logger = logger - self.remote_session = create_remote_session(node_config, name, logger) - self.interactive_session = create_interactive_session(node_config, logger) + self.remote_session = SSHSession(node_config, name, logger) + self.interactive_session = InteractiveRemoteSession(node_config, logger) def is_alive(self) -> bool: """Check whether the underlying remote session is still responding.""" diff --git a/dts/framework/testbed_model/topology.py b/dts/framework/testbed_model/topology.py index 899ea0ad3a..09303a1f08 100644 --- a/dts/framework/testbed_model/topology.py +++ b/dts/framework/testbed_model/topology.py @@ -11,33 +11,17 @@ from collections import defaultdict from collections.abc import Iterator from dataclasses import dataclass -from enum import Enum from typing import Literal, NamedTuple from typing_extensions import Self +from api.capabilities import LinkTopology from framework.exception import ConfigurationError, InternalError from framework.testbed_model.node import Node from .port import DriverKind, Port, PortConfig -class TopologyType(int, Enum): - """Supported topology types.""" - - #: A topology with no Traffic Generator. - no_link = 0 - #: A topology with one physical link between the SUT node and the TG node. - one_link = 1 - #: A topology with two physical links between the Sut node and the TG node. - two_links = 2 - - @classmethod - def default(cls) -> "TopologyType": - """The default topology required by test cases if not specified otherwise.""" - return cls.two_links - - class PortLink(NamedTuple): """The physical, cabled connection between the ports.""" @@ -71,7 +55,7 @@ class Topology: tg_ports: The TG ports. """ - type: TopologyType + type: LinkTopology sut_ports: list[Port] tg_ports: list[Port] pf_ports: list[Port] @@ -87,15 +71,15 @@ def from_port_links(cls, port_links: Iterator[PortLink]) -> Self: Raises: ConfigurationError: If an unsupported link topology is supplied. """ - type = TopologyType.no_link + type = LinkTopology.NO_LINK if port_link := next(port_links, None): - type = TopologyType.one_link + type = LinkTopology.ONE_LINK sut_ports = [port_link.sut_port] tg_ports = [port_link.tg_port] if port_link := next(port_links, None): - type = TopologyType.two_links + type = LinkTopology.TWO_LINKS sut_ports.append(port_link.sut_port) tg_ports.append(port_link.tg_port) @@ -268,9 +252,9 @@ def sut_port_ingress(self) -> Port: @property def sut_port_egress(self) -> Port: """The egress port of the SUT node.""" - return self.sut_ports[1 if self.type is TopologyType.two_links else 0] + return self.sut_ports[1 if self.type is LinkTopology.TWO_LINKS else 0] @property def tg_port_ingress(self) -> Port: """The ingress port of the TG node.""" - return self.tg_ports[1 if self.type is TopologyType.two_links else 0] + return self.tg_ports[1 if self.type is LinkTopology.TWO_LINKS else 0] -- 2.39.5 </blockquote></div>