[RFC Patch v1 0/5] Add TREX Traffic Generator to DTS Framework

[RFC Patch v1 0/5] Add TREX Traffic Generator to DTS Framework

[Nicholas Pratte]

Included in a semi-complete, RFC implementation for a would-be
implementation of the TREX traffic generator, in addition to a
mock implementation of a single core performance test suite,
leveraging newly added performance test API functionality.

Code is incomplete with a handful of typing issues, but does execute
without errors on my system.

Nicholas Pratte (5):
  dts: rework config module to support perf TGs
  dts: rework traffic generator inheritance structure.
  dts: add asychronous support to ssh sessions.
  dts: add trex traffic generator to dts framework
  dts: add performance test functions to test suite api

 dts/{ => configurations}/nodes.example.yaml   |   0
 .../test_run.example.yaml                     |   8 +-
 .../tests_config.example.yaml                 |   0
 .../trex_configs/intel_40g.yaml               |  18 ++
 dts/framework/config/test_run.py              |  20 +-
 dts/framework/context.py                      |  11 +-
 dts/framework/remote_session/ssh_session.py   |  17 ++
 dts/framework/settings.py                     |   6 +-
 dts/framework/test_run.py                     |  28 +-
 dts/framework/test_suite.py                   |  33 +-
 .../traffic_generator/__init__.py             |  22 +-
 .../capturing_traffic_generator.py            |  34 +++
 .../performance_traffic_generator.py          |  62 ++++
 .../traffic_generator/traffic_generator.py    |  43 +--
 .../testbed_model/traffic_generator/trex.py   | 287 ++++++++++++++++++
 dts/tests/TestSuite_single_core_perf.py       |  24 ++
 16 files changed, 550 insertions(+), 63 deletions(-)
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (82%)
 rename dts/{ => configurations}/tests_config.example.yaml (100%)
 create mode 100644 dts/configurations/trex_configs/intel_40g.yaml
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py
 create mode 100644 dts/tests/TestSuite_single_core_perf.py

-- 
2.47.1

[RFC Patch v1 1/5] dts: rework config module to support perf TGs

[Nicholas Pratte]

Rework test run configuration file for TGs to support both application
directory location and any necessary configuration files; an example
TREX configuration file is provided. Configuration files have been moved
to a configurations directory, requiring a slight modification to the
settings module.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/{ => configurations}/nodes.example.yaml    |  0
 dts/{ => configurations}/test_run.example.yaml |  8 +++++++-
 .../tests_config.example.yaml                  |  0
 dts/configurations/trex_configs/intel_40g.yaml | 18 ++++++++++++++++++
 dts/framework/settings.py                      |  6 ++++--
 5 files changed, 29 insertions(+), 3 deletions(-)
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (82%)
 rename dts/{ => configurations}/tests_config.example.yaml (100%)
 create mode 100644 dts/configurations/trex_configs/intel_40g.yaml

diff --git a/dts/nodes.example.yaml b/dts/configurations/nodes.example.yaml
similarity index 100%
rename from dts/nodes.example.yaml
rename to dts/configurations/nodes.example.yaml
diff --git a/dts/test_run.example.yaml b/dts/configurations/test_run.example.yaml
similarity index 82%
rename from dts/test_run.example.yaml
rename to dts/configurations/test_run.example.yaml
index 330a31bb18..fdfff4a879 100644
--- a/dts/test_run.example.yaml
+++ b/dts/configurations/test_run.example.yaml
@@ -23,8 +23,14 @@ dpdk:
     # in a subdirectory of DPDK tree root directory. Otherwise, will be using the `build_options`
     # to build the DPDK from source. Either `precompiled_build_dir` or `build_options` can be
     # defined, but not both.
-traffic_generator:
+func_traffic_generator:
   type: SCAPY
+  remote_path: "" # The remote path of the traffic generator application. (Leave blank for SCAPY)
+  config: "" # Additional configuration files. (Leave blank if not required)
+perf_traffic_generator:
+  type: TREX
+  remote_path: "/opt/trex/v3.03" # The remote path of the traffic generator application. (Leave blank for SCAPY)
+  config: "trex_config.yaml" # Additional configuration files. (Leave blank if not required)
 perf: false # disable performance testing
 func: true # enable functional testing
 skip_smoke_tests: false # optional
diff --git a/dts/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
similarity index 100%
rename from dts/tests_config.example.yaml
rename to dts/configurations/tests_config.example.yaml
diff --git a/dts/configurations/trex_configs/intel_40g.yaml b/dts/configurations/trex_configs/intel_40g.yaml
new file mode 100644
index 0000000000..dae003795b
--- /dev/null
+++ b/dts/configurations/trex_configs/intel_40g.yaml
@@ -0,0 +1,18 @@
+### Config file generated by dpdk_setup_ports.py ###
+
+- version: 2
+  interfaces: ['11:00.0', '11:00.1']
+  port_bandwidth_gb: 40
+  port_info:
+      - dest_mac: 3c:fd:fe:d5:e5:f9 # MAC OF LOOPBACK TO IT'S DUAL INTERFACE
+        src_mac:  3c:fd:fe:d5:e5:f8
+      - dest_mac: 3c:fd:fe:d5:e5:f8 # MAC OF LOOPBACK TO IT'S DUAL INTERFACE
+        src_mac:  3c:fd:fe:d5:e5:f9
+
+  platform:
+      master_thread_id: 0
+      latency_thread_id: 7
+      dual_if:
+        - socket: 0
+          threads: [1,2,3,4,5,6]
+
diff --git a/dts/framework/settings.py b/dts/framework/settings.py
index 3f21615223..ccf4df25b0 100644
--- a/dts/framework/settings.py
+++ b/dts/framework/settings.py
@@ -130,9 +130,11 @@ class Settings:
     """
 
     #:
-    test_run_config_path: Path = Path(__file__).parent.parent.joinpath("test_run.yaml")
+    test_run_config_path: Path = Path(__file__).parent.parent.joinpath(
+        "configurations/test_run.yaml"
+    )
     #:
-    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("nodes.yaml")
+    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("configurations/nodes.yaml")
     #:
     tests_config_path: Path | None = None
     #:
-- 
2.47.1

[RFC Patch v1 2/5] dts: rework traffic generator inheritance structure.

[Nicholas Pratte]

Rework TG class hierarchy to include performance traffic generators, in
addition to capturing traffic generators. As such, methods garnered
to capturing traffic have been moved to the CapturingTrafficGenerator
subclass.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 .../capturing_traffic_generator.py            | 34 ++++++++++
 .../performance_traffic_generator.py          | 62 +++++++++++++++++++
 .../traffic_generator/traffic_generator.py    | 43 +------------
 3 files changed, 97 insertions(+), 42 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py

diff --git a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
index e31ba2a9b7..41b70f7f48 100644
--- a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
@@ -63,6 +63,40 @@ def is_capturing(self) -> bool:
         """This traffic generator can capture traffic."""
         return True
 
+    def send_packet(self, packet: Packet, port: Port) -> None:
+        """Send `packet` and block until it is fully sent.
+
+        Send `packet` on `port`, then wait until `packet` is fully sent.
+
+        Args:
+            packet: The packet to send.
+            port: The egress port on the TG node.
+        """
+        self.send_packets([packet], port)
+
+    def send_packets(self, packets: list[Packet], port: Port) -> None:
+        """Send `packets` and block until they are fully sent.
+
+        Send `packets` on `port`, then wait until `packets` are fully sent.
+
+        Args:
+            packets: The packets to send.
+            port: The egress port on the TG node.
+        """
+        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
+        self._logger.debug(get_packet_summaries(packets))
+        self._send_packets(packets, port)
+
+    @abstractmethod
+    def _send_packets(self, packets: list[Packet], port: Port) -> None:
+        """The implementation of :method:`send_packets`.
+
+        The subclasses must implement this method which sends `packets` on `port`.
+        The method should block until all `packets` are fully sent.
+
+        What fully sent means is defined by the traffic generator.
+        """
+
     def send_packets_and_capture(
         self,
         packets: list[Packet],
diff --git a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
new file mode 100644
index 0000000000..7a384cf6e0
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
@@ -0,0 +1,62 @@
+"""Performance testing capable traffic generatiors."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Callable
+
+from scapy.packet import Packet
+
+from framework.testbed_model.traffic_generator.traffic_generator import TrafficGenerator
+
+
+@dataclass(slots=True)
+class PerformanceTrafficStats(ABC):
+    """Data structure for stats offered by a given traffic generator."""
+
+    frame_size: int
+
+
+class PerformanceTrafficGenerator(TrafficGenerator):
+    """An Abstract Base Class for all performance-oriented traffic generators.
+
+    Provides an intermediary interface for performance-based traffic generator.
+    """
+
+    _test_stats: list[PerformanceTrafficStats]
+
+    @property
+    def is_capturing(self) -> bool:
+        """Used for synchronization."""
+        return False
+
+    @property
+    def last_results(self) -> PerformanceTrafficStats | None:
+        """Get the latest set of results from TG instance.
+
+        Returns:
+            The most recent set of traffic statistics.
+        """
+        return self._test_stats.pop(0)
+
+    def generate_traffic_and_stats(
+        self,
+        packet: Packet,
+        duration: float,  # Default of 60 (in seconds).
+    ) -> PerformanceTrafficStats:
+        """Send packet traffic and acquire associated statistics."""
+        return self._calculate_traffic_stats(packet, duration, self._generate_traffic)
+
+    def setup(self, ports):
+        """Preliminary port setup prior to TG execution."""
+        for port in self._tg_node.ports:
+            self._tg_node.main_session.configure_port_mtu(2000, port)
+
+    @abstractmethod
+    def _calculate_traffic_stats(
+        self, packet: Packet, duration: float, traffic_gen_callback: Callable[[Packet, float], str]
+    ) -> PerformanceTrafficStats:
+        """Calculate packet traffic stats based on TG output."""
+
+    @abstractmethod
+    def _generate_traffic(self, packet: Packet, duration: float) -> str:
+        """Implementation for :method:`generate_traffic_and_stats`."""
diff --git a/dts/framework/testbed_model/traffic_generator/traffic_generator.py b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
index 804662e114..12b9568d1a 100644
--- a/dts/framework/testbed_model/traffic_generator/traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
@@ -11,13 +11,11 @@
 from abc import ABC, abstractmethod
 from typing import Iterable
 
-from scapy.packet import Packet
-
 from framework.config.test_run import TrafficGeneratorConfig
 from framework.logger import DTSLogger, get_dts_logger
 from framework.testbed_model.node import Node
 from framework.testbed_model.port import Port
-from framework.utils import MultiInheritanceBaseClass, get_packet_summaries
+from framework.utils import MultiInheritanceBaseClass
 
 
 class TrafficGenerator(MultiInheritanceBaseClass, ABC):
@@ -56,45 +54,6 @@ def setup(self, ports: Iterable[Port]):
     def teardown(self, ports: Iterable[Port]):
         """Teardown the traffic generator."""
 
-    def send_packet(self, packet: Packet, port: Port) -> None:
-        """Send `packet` and block until it is fully sent.
-
-        Send `packet` on `port`, then wait until `packet` is fully sent.
-
-        Args:
-            packet: The packet to send.
-            port: The egress port on the TG node.
-        """
-        self.send_packets([packet], port)
-
-    def send_packets(self, packets: list[Packet], port: Port) -> None:
-        """Send `packets` and block until they are fully sent.
-
-        Send `packets` on `port`, then wait until `packets` are fully sent.
-
-        Args:
-            packets: The packets to send.
-            port: The egress port on the TG node.
-        """
-        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
-        self._logger.debug(get_packet_summaries(packets))
-        self._send_packets(packets, port)
-
-    @abstractmethod
-    def _send_packets(self, packets: list[Packet], port: Port) -> None:
-        """The implementation of :method:`send_packets`.
-
-        The subclasses must implement this method which sends `packets` on `port`.
-        The method should block until all `packets` are fully sent.
-
-        What fully sent means is defined by the traffic generator.
-        """
-
-    @property
-    def is_capturing(self) -> bool:
-        """This traffic generator can't capture traffic."""
-        return False
-
     @abstractmethod
     def close(self) -> None:
         """Free all resources used by the traffic generator."""
-- 
2.47.1

[RFC Patch v1 3/5] dts: add asychronous support to ssh sessions.

[Nicholas Pratte]

Execution of the TREX server process requires an SSH session rework to
support asynchronous process management. Allowing access to asynchronous
functionality allows developers to execute processes without hijacking
the SSH session being used. In doing so, both timeout and runtime errors
may be avoided. This functionality leverages Fabric's Promise class,
which provides a join method to terminate the process when the process
is done being used, providing more secure process control.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/framework/remote_session/ssh_session.py | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/dts/framework/remote_session/ssh_session.py b/dts/framework/remote_session/ssh_session.py
index e6e4704bc2..185905f701 100644
--- a/dts/framework/remote_session/ssh_session.py
+++ b/dts/framework/remote_session/ssh_session.py
@@ -13,6 +13,7 @@
     ThreadException,
     UnexpectedExit,
 )
+from invoke.runners import Promise
 from paramiko.ssh_exception import (
     AuthenticationException,
     BadHostKeyException,
@@ -99,6 +100,22 @@ def _send_command(self, command: str, timeout: float, env: dict | None) -> Comma
 
         return CommandResult(self.name, command, output.stdout, output.stderr, output.return_code)
 
+    def _send_async_command(self, command: str, timeout: float, env: dict | None) -> Promise:
+        try:
+            promise = self.session.run(
+                command, env=env, warn=True, hide=True, timeout=timeout, asynchronous=True
+            )
+
+        except (UnexpectedExit, ThreadException) as e:
+            self._logger.exception(e)
+            raise SSHSessionDeadError(self.hostname) from e
+
+        except CommandTimedOut as e:
+            self._logger.exception(e)
+            raise SSHTimeoutError(command) from e
+
+        return promise
+
     def is_alive(self) -> bool:
         """Overrides :meth:`~.remote_session.RemoteSession.is_alive`."""
         return self.session.is_connected
-- 
2.47.1

[RFC Patch v1 4/5] dts: add trex traffic generator to dts framework

[Nicholas Pratte]

Implement the TREX traffic generator for use in the DTS framework. The
provided implementation leverages TREX's stateless API automation
library, via use of a Python shell. As such, version control of TREX may
be needed. The DTS context has been modified to include a performance
traffic generator in addition to a functional traffic generator.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/framework/config/test_run.py              |  20 +-
 dts/framework/context.py                      |  11 +-
 dts/framework/test_run.py                     |  28 +-
 dts/framework/test_suite.py                   |   6 +-
 .../traffic_generator/__init__.py             |  22 +-
 .../testbed_model/traffic_generator/trex.py   | 287 ++++++++++++++++++
 6 files changed, 356 insertions(+), 18 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py

diff --git a/dts/framework/config/test_run.py b/dts/framework/config/test_run.py
index 06fe28143c..5cdd391b49 100644
--- a/dts/framework/config/test_run.py
+++ b/dts/framework/config/test_run.py
@@ -393,6 +393,8 @@ class TrafficGeneratorType(str, Enum):
 
     #:
     SCAPY = "SCAPY"
+    #:
+    TREX = "TREX"
 
 
 class TrafficGeneratorConfig(FrozenModel):
@@ -401,6 +403,8 @@ class TrafficGeneratorConfig(FrozenModel):
     #: The traffic generator type the child class is required to define to be distinguished among
     #: others.
     type: TrafficGeneratorType
+    remote_path: PurePath
+    config: PurePath
 
 
 class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
@@ -409,8 +413,16 @@ class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
     type: Literal[TrafficGeneratorType.SCAPY]
 
 
+class TrexTrafficGeneratorConfig(TrafficGeneratorConfig):
+    """TREX traffic generator specific configuration."""
+
+    type: Literal[TrafficGeneratorType.TREX]
+
+
 #: A union type discriminating traffic generators by the `type` field.
-TrafficGeneratorConfigTypes = Annotated[ScapyTrafficGeneratorConfig, Field(discriminator="type")]
+TrafficGeneratorConfigTypes = Annotated[
+    TrexTrafficGeneratorConfig, ScapyTrafficGeneratorConfig, Field(discriminator="type")
+]
 
 #: Comma-separated list of logical cores to use. An empty string or ```any``` means use all lcores.
 LogicalCores = Annotated[
@@ -458,8 +470,10 @@ class TestRunConfiguration(FrozenModel):
 
     #: The DPDK configuration used to test.
     dpdk: DPDKConfiguration
-    #: The traffic generator configuration used to test.
-    traffic_generator: TrafficGeneratorConfigTypes
+    #: The traffic generator configuration used for functional tests.
+    func_traffic_generator: TrafficGeneratorConfig
+    #: The traffic generator configuration used for performance tests.
+    perf_traffic_generator: TrafficGeneratorConfig
     #: Whether to run performance tests.
     perf: bool
     #: Whether to run functional tests.
diff --git a/dts/framework/context.py b/dts/framework/context.py
index ddd7ed4d36..0a83e2b269 100644
--- a/dts/framework/context.py
+++ b/dts/framework/context.py
@@ -15,7 +15,12 @@
 
 if TYPE_CHECKING:
     from framework.remote_session.dpdk import DPDKBuildEnvironment, DPDKRuntimeEnvironment
-    from framework.testbed_model.traffic_generator.traffic_generator import TrafficGenerator
+    from framework.testbed_model.traffic_generator.capturing_traffic_generator import (
+        CapturingTrafficGenerator,
+    )
+    from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+        PerformanceTrafficGenerator,
+    )
 
 P = ParamSpec("P")
 
@@ -68,7 +73,9 @@ class Context:
     topology: Topology
     dpdk_build: "DPDKBuildEnvironment"
     dpdk: "DPDKRuntimeEnvironment"
-    tg: "TrafficGenerator"
+    tg_dpdk: "DPDKRuntimeEnvironment" | None
+    func_tg: "CapturingTrafficGenerator"
+    perf_tg: "PerformanceTrafficGenerator"
     local: LocalContext = field(default_factory=LocalContext)
 
 
diff --git a/dts/framework/test_run.py b/dts/framework/test_run.py
index f9cfe5e908..81dc553b00 100644
--- a/dts/framework/test_run.py
+++ b/dts/framework/test_run.py
@@ -107,7 +107,7 @@
 from types import MethodType
 from typing import ClassVar, Protocol, Union
 
-from framework.config.test_run import TestRunConfiguration
+from framework.config.test_run import TestRunConfiguration, TrafficGeneratorType
 from framework.context import Context, init_ctx
 from framework.exception import (
     InternalError,
@@ -204,10 +204,25 @@ def __init__(
 
         dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
         dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node, dpdk_build_env)
-        traffic_generator = create_traffic_generator(config.traffic_generator, tg_node)
+        # There is definitely a better way to do this.
+        tg_dpdk_runtime_env = None
+        if (
+            config.perf_traffic_generator.type == TrafficGeneratorType.TREX
+            or config.func_traffic_generator.type == TrafficGeneratorType.TREX
+        ):
+            tg_dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, tg_node, None)
+        func_traffic_generator = create_traffic_generator(config.func_traffic_generator, tg_node)
+        perf_traffic_generator = create_traffic_generator(config.perf_traffic_generator, tg_node)
 
         self.ctx = Context(
-            sut_node, tg_node, topology, dpdk_build_env, dpdk_runtime_env, traffic_generator
+            sut_node,
+            tg_node,
+            topology,
+            dpdk_build_env,
+            dpdk_runtime_env,
+            tg_dpdk_runtime_env,
+            func_traffic_generator,
+            perf_traffic_generator,
         )
         self.result = result
         self.selected_tests = list(self.config.filter_tests(tests_config))
@@ -345,7 +360,9 @@ def next(self) -> State | None:
         test_run.ctx.sut_node.setup()
         test_run.ctx.tg_node.setup()
         test_run.ctx.dpdk.setup(test_run.ctx.topology.sut_ports)
-        test_run.ctx.tg.setup(test_run.ctx.topology.tg_ports)
+        test_run.ctx.tg_dpdk.setup(test_run.ctx.topology.tg_ports)
+        test_run.ctx.func_tg.setup(test_run.ctx.topology.tg_ports)
+        test_run.ctx.perf_tg.setup(test_run.ctx.topology.tg_ports)
 
         self.result.ports = test_run.ctx.topology.sut_ports + test_run.ctx.topology.tg_ports
         self.result.sut_info = test_run.ctx.sut_node.node_info
@@ -430,7 +447,8 @@ def description(self) -> str:
 
     def next(self) -> State | None:
         """Next state."""
-        self.test_run.ctx.tg.teardown(self.test_run.ctx.topology.tg_ports)
+        self.test_run.ctx.func_tg.teardown(self.test_run.ctx.topology.tg_ports)
+        self.test_run.ctx.perf_tg.teardown(self.test_run.ctx.topology.tg_ports)
         self.test_run.ctx.dpdk.teardown(self.test_run.ctx.topology.sut_ports)
         self.test_run.ctx.tg_node.teardown()
         self.test_run.ctx.sut_node.teardown()
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index e07c327b77..507df508cb 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -254,11 +254,11 @@ def send_packets_and_capture(
             A list of received packets.
         """
         assert isinstance(
-            self._ctx.tg, CapturingTrafficGenerator
+            self._ctx.func_tg, CapturingTrafficGenerator
         ), "Cannot capture with a non-capturing traffic generator"
         # TODO: implement @requires for types of traffic generator
         packets = self._adjust_addresses(packets)
-        return self._ctx.tg.send_packets_and_capture(
+        return self._ctx.func_tg.send_packets_and_capture(
             packets,
             self._ctx.topology.tg_port_egress,
             self._ctx.topology.tg_port_ingress,
@@ -276,7 +276,7 @@ def send_packets(
             packets: Packets to send.
         """
         packets = self._adjust_addresses(packets)
-        self._ctx.tg.send_packets(packets, self._ctx.topology.tg_port_egress)
+        self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
     def get_expected_packets(
         self,
diff --git a/dts/framework/testbed_model/traffic_generator/__init__.py b/dts/framework/testbed_model/traffic_generator/__init__.py
index 2a259a6e6c..53125995cd 100644
--- a/dts/framework/testbed_model/traffic_generator/__init__.py
+++ b/dts/framework/testbed_model/traffic_generator/__init__.py
@@ -14,17 +14,27 @@
 and a capturing traffic generator is required.
 """
 
-from framework.config.test_run import ScapyTrafficGeneratorConfig, TrafficGeneratorConfig
+from framework.config.test_run import (
+    ScapyTrafficGeneratorConfig as ScapyTrafficGeneratorConfig,
+)
+from framework.config.test_run import (
+    TrafficGeneratorConfig,
+    TrafficGeneratorType,
+)
+from framework.config.test_run import (
+    TrexTrafficGeneratorConfig as TrexTrafficGeneratorConfig,
+)
 from framework.exception import ConfigurationError
 from framework.testbed_model.node import Node
 
-from .capturing_traffic_generator import CapturingTrafficGenerator
 from .scapy import ScapyTrafficGenerator
+from .traffic_generator import TrafficGenerator
+from .trex import TrexTrafficGenerator
 
 
 def create_traffic_generator(
     traffic_generator_config: TrafficGeneratorConfig, node: Node
-) -> CapturingTrafficGenerator:
+) -> TrafficGenerator:
     """The factory function for creating traffic generator objects from the test run configuration.
 
     Args:
@@ -37,8 +47,10 @@ def create_traffic_generator(
     Raises:
         ConfigurationError: If an unknown traffic generator has been setup.
     """
-    match traffic_generator_config:
-        case ScapyTrafficGeneratorConfig():
+    match traffic_generator_config.type:
+        case TrafficGeneratorType.SCAPY:
             return ScapyTrafficGenerator(node, traffic_generator_config, privileged=True)
+        case TrafficGeneratorType.TREX:
+            return TrexTrafficGenerator(node, traffic_generator_config, privileged=True)
         case _:
             raise ConfigurationError(f"Unknown traffic generator: {traffic_generator_config.type}")
diff --git a/dts/framework/testbed_model/traffic_generator/trex.py b/dts/framework/testbed_model/traffic_generator/trex.py
new file mode 100644
index 0000000000..0053174ede
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/trex.py
@@ -0,0 +1,287 @@
+"""Implementation for TREX performance traffic generator."""
+
+import time
+from dataclasses import dataclass
+from enum import Flag, auto
+from typing import Callable, ClassVar
+
+from invoke.runners import Promise
+from scapy.packet import Packet
+
+from framework.config.node import NodeConfiguration
+from framework.config.test_run import TrafficGeneratorConfig
+from framework.exception import SSHTimeoutError
+from framework.remote_session.python_shell import PythonShell
+from framework.remote_session.ssh_session import SSHSession
+from framework.testbed_model.linux_session import LinuxSession
+from framework.testbed_model.node import Node, create_session
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
+
+
+@dataclass
+class TrexPerPortStats:
+    """Performance statistics on a per port basis.
+
+    Attributes:
+        opackets: Number of packets sent.
+        obytes: Number of egress bytes sent.
+        tx_bps: Maximum bits per second transmitted.
+        tx_pps: Number of transmitted packets sent.
+    """
+
+    opackets: float
+    obytes: float
+    tx_bps: float
+    tx_pps: float
+
+
+@dataclass
+class TrexPerformanceStats(PerformanceTrafficStats):
+    """Data structure to store performance statistics for a given test run.
+
+    Attributes:
+        packet: The packet that was sent in the test run.
+        frame_size: The total length of the frame. (L2 downward)
+        tx_expected_bps: The expected bits per second on a given NIC.
+        tx_expected_cps: ...
+        tx_expected_pps: The expected packets per second of a given NIC.
+        tx_pps: The recorded maximum packets per second of the tested NIC.
+        tx_cps: The recorded maximum cps of the tested NIC
+        tx_bps: The recorded maximum bits per second of the tested NIC.
+        obytes: Total bytes output during test run.
+        port_stats: A list of :class:`TrexPerPortStats` provided by TREX.
+    """
+
+    packet: Packet
+    frame_size: int
+
+    tx_expected_bps: float
+    tx_expected_cps: float
+    tx_expected_pps: float
+
+    tx_pps: float
+    tx_cps: float
+    tx_bps: float
+
+    obytes: float
+
+    port_stats: list[TrexPerPortStats] | None
+
+
+class TrexStatelessTXModes(Flag):
+    """Flags indicating TREX instance's current trasmission mode."""
+
+    CONTINUOUS = auto()
+    SINGLE_BURST = auto()
+    MULTI_BURST = auto()
+
+
+class TrexTrafficGenerator(PythonShell, PerformanceTrafficGenerator):
+    """TREX traffic generator.
+
+    This implementation leverages the stateless API library provided in the TREX installation.
+
+    Attributes:
+        stl_client_name: The name of the stateless client used in the stateless API.
+        packet_stream_name: The name of the stateless packet stream used in the stateless API.
+        timeout_duration: Internal timeout for connection to the TREX server.
+    """
+
+    _os_session: LinuxSession
+    _server_remote_session: SSHSession
+    _trex_server_process: Promise
+
+    _tg_config: TrafficGeneratorConfig
+    _node_config: NodeConfiguration
+
+    _python_indentation: ClassVar[str] = " " * 4
+
+    stl_client_name: ClassVar[str] = "client"
+    packet_stream_name: ClassVar[str] = "stream"
+
+    _streaming_mode: TrexStatelessTXModes = TrexStatelessTXModes.CONTINUOUS
+
+    timeout_duration: int
+
+    def __init__(
+        self, tg_node: Node, config: TrafficGeneratorConfig, timeout_duration: int = 5, **kwargs
+    ) -> None:
+        """Initialize the TREX server.
+
+        Initializes needed OS sessions for the creation of the TREX server process.
+
+        Attributes:
+            tg_node: TG node the TREX instance is operating on.
+            config: Traffic generator config provided for TREX instance.
+            timeout_duration: Internal timeout for connection to the TREX server.
+        """
+        super().__init__(node=tg_node, config=config, tg_node=tg_node, **kwargs)
+        self._node_config = tg_node.config
+        self._tg_config = config
+        self.timeout_duration = timeout_duration
+
+        # Create TREX server session.
+        self._tg_node._other_sessions.append(
+            create_session(self._tg_node.config, "TREX Server.", self._logger)
+        )
+        self._os_session = self._tg_node._other_sessions[0]
+        self._server_remote_session = self._os_session.remote_session
+
+    def setup(self, ports):
+        """Initialize and start a TREX server process.
+
+        Binds TG ports to vfio-pci and starts the trex process.
+
+        Attributes:
+            ports: Related ports utilized in TG instance.
+        """
+        super().setup(ports)
+        # Start TREX server process.
+        try:
+            self._logger.info("Starting TREX server process: sending 45 second sleep.")
+            privileged_command = self._os_session._get_privileged_command(
+                f"""
+                    cd /opt/v3.03/; {self._tg_config.remote_path}/t-rex-64
+                     --cfg {self._tg_config.config} -i
+                """
+            )
+            self._server_remote_session = self._server_remote_session._send_async_command(
+                privileged_command, timeout=None, env=None
+            )
+            time.sleep(45)
+        except SSHTimeoutError as e:
+            self._logger.exception("Failed to start TREX server process.", e)
+
+        # Start Python shell.
+        self.start_application()
+        self.send_command("import os")
+        # Parent directory: /opt/v3.03/automation/trex_control_plane/interactive
+        self.send_command(
+            f"os.chdir('{self._tg_config.remote_path}/automation/trex_control_plane/interactive')"
+        )
+
+        # Import stateless API components.
+        imports = [
+            "import trex",
+            "import trex.stl",
+            "import trex.stl.trex_stl_client",
+            "import trex.stl.trex_stl_streams",
+            "import trex.stl.trex_stl_packet_builder_scapy",
+            "from scapy.layers.l2 import Ether",
+            "from scapy.layers.inet import IP",
+            "from scapy.packet import Raw",
+        ]
+        self.send_command("\n".join(imports))
+
+        stateless_client = [
+            f"{self.stl_client_name} = trex.stl.trex_stl_client.STLClient(",
+            f"username='{self._node_config.user}',",
+            "server='127.0.0.1',",
+            f"sync_timeout={self.timeout_duration}",
+            ")",
+        ]
+        self.send_command(f"\n{self._python_indentation}".join(stateless_client))
+        self.send_command(f"{self.stl_client_name}.connect()")
+
+    def teardown(self, ports):
+        """Teardown the TREX server and stateless implementation.
+
+        close the TREX server process, and stop the Python shell.
+
+        Attributes:
+            ports: Associated ports used by the TREX instance.
+        """
+        super().teardown(ports)
+        self.send_command(f"{self.stl_client_name}.disconnect()")
+        self.close()
+        self._trex_server_process.join()
+
+    def _calculate_traffic_stats(
+        self, packet: Packet, duration: float, callback: Callable[[Packet, float], str]
+    ) -> PerformanceTrafficStats:
+        """Calculate the traffic statistics, using provided TG output.
+
+        Takes in the statistics output provided by the stateless API implementation, and collects
+        them into a performance statistics data structure.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+            duration: The duration of the test.
+            callback: The callback function used to generate the traffic.
+        """
+        # Convert to a dictionary.
+        stats_output = eval(callback(packet, duration))
+        return TrexPerformanceStats(
+            len(packet),
+            packet,
+            stats_output.get("tx_expected_bps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_expected_cps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_expected_pps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_pps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_cps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_bps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("obytes", "ERROR - DATA NOT FOUND"),
+            None,
+        )
+
+    def set_streaming_mode(self, streaming_mode: TrexStatelessTXModes) -> None:
+        """Set the streaming mode of the TREX instance."""
+        # Streaming modes are mutually exclusive.
+        self._streaming_mode = self._streaming_mode & streaming_mode
+
+    def _generate_traffic(self, packet: Packet, duration: float) -> str:
+        """Generate traffic using provided packet.
+
+        Uses the provided packet to generate traffic for the provided duration.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+            duration: The duration of the test being performed.
+
+        Returns:
+            a string output of statistics provided by the traffic generator.
+        """
+        """Implementation for :method:`generate_traffic_and_stats`."""
+        streaming_mode = ""
+        if self._streaming_mode == TrexStatelessTXModes.CONTINUOUS:
+            streaming_mode = "STLTXCont"
+        elif self._streaming_mode == TrexStatelessTXModes.SINGLE_BURST:
+            streaming_mode = "STLTXSingleBurst"
+        elif self._streaming_mode == TrexStatelessTXModes.MULTI_BURST:
+            streaming_mode = "STLTXMultiBurst"
+
+        packet_stream = [
+            f"{self.packet_stream_name} = trex.stl.trex_stl_streams.STLStream(",
+            f"name='Test_{len(packet)}_bytes',",
+            f"packet=trex.stl.trex_stl_packet_builder_scapy.STLPktBuilder(pkt={packet.command()}),",
+            f"mode=trex.stl.trex_stl_streams.{streaming_mode}(),",
+            ")",
+        ]
+        self.send_command("\n".join(packet_stream))
+
+        # Prepare TREX console for next performance test.
+        procedure = [
+            f"{self.stl_client_name}.connect()",
+            f"{self.stl_client_name}.reset(ports = [0, 1])",
+            f"{self.stl_client_name}.add_streams({self.packet_stream_name}, ports=[0, 1])",
+            f"{self.stl_client_name}.clear_stats()",
+            ")",
+        ]
+        self.send_command("\n".join(procedure))
+
+        start_test = [
+            f"{self.stl_client_name}.start(ports=[0, 1], duration={duration})",
+            f"{self.stl_client_name}.wait_on_traffic(ports=[0, 1])",
+        ]
+        self.send_command("\n".join(start_test))
+        import time
+
+        time.sleep(duration + 1)
+
+        # Gather statistics output for parsing.
+        return self.send_command(
+            f"{self.stl_client_name}.get_stats(ports=[0, 1])", skip_first_line=True
+        )
-- 
2.47.1

[RFC Patch v1 5/5] dts: add performance test functions to test suite api

[Nicholas Pratte]

Provide functional performance method to run performance tests using a
user-supplied performance traffic generator. The single core performance
test is included, with some basic statistics checks verifying TG packet
transmission rates.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/framework/test_suite.py             | 27 +++++++++++++++++++++++++
 dts/tests/TestSuite_single_core_perf.py | 24 ++++++++++++++++++++++
 2 files changed, 51 insertions(+)
 create mode 100644 dts/tests/TestSuite_single_core_perf.py

diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index 507df508cb..a89faac2d5 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -38,6 +38,10 @@
     CapturingTrafficGenerator,
     PacketFilteringConfig,
 )
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
 
 from .exception import ConfigurationError, InternalError, TestCaseVerifyError
 from .logger import DTSLogger, get_dts_logger
@@ -266,6 +270,26 @@ def send_packets_and_capture(
             duration,
         )
 
+    def assess_performance_by_packet(
+        self, packet: Packet, duration: int = 60
+    ) -> PerformanceTrafficStats:
+        """Send a given packet for a given duration and assess basic performance statistics.
+
+        Send `packet` and assess NIC performance for a given duration, corresponding to the test
+        suite's given topology.
+
+        Args:
+            packet: The packet to send.
+            duration: Performance test duration (in seconds)
+
+        Returns:
+            Performance statistics of the generated test.
+        """
+        assert isinstance(
+            self._ctx.perf_tg, PerformanceTrafficGenerator
+        ), "Cannot run performance tests on non-performance traffic generator."
+        return self._ctx.perf_tg.generate_traffic_and_stats(packet, duration)
+
     def send_packets(
         self,
         packets: list[Packet],
@@ -275,6 +299,9 @@ def send_packets(
         Args:
             packets: Packets to send.
         """
+        assert isinstance(
+            self._ctx.perf_tg, CapturingTrafficGenerator
+        ), "Cannot run performance tests on non-capturing traffic generator."
         packets = self._adjust_addresses(packets)
         self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
diff --git a/dts/tests/TestSuite_single_core_perf.py b/dts/tests/TestSuite_single_core_perf.py
new file mode 100644
index 0000000000..31a8c8608f
--- /dev/null
+++ b/dts/tests/TestSuite_single_core_perf.py
@@ -0,0 +1,24 @@
+"""Single core performance test suite."""
+
+from scapy.layers.inet import IP
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
+
+from framework.remote_session.testpmd_shell import TestPmdShell
+from framework.test_suite import TestSuite, perf_test
+
+
+class TestSingleCorePerf(TestSuite):
+    """Single core performance test suite."""
+
+    @perf_test
+    def test_perf_test(self) -> None:
+        """Prototype test case."""
+        with TestPmdShell() as testpmd:
+            packet = Ether() / IP() / Raw(load="x" * 1484)  # 1518 byte packet.
+
+            testpmd.start()
+            stats = self.assess_performance_by_packet(packet, duration=5)
+            self.verify(
+                stats.tx_expected_bps == 40, "Expected output does not patch recorded output."
+            )
-- 
2.47.1

Re: [RFC Patch v1 2/5] dts: rework traffic generator inheritance structure.

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 3057 bytes --]

On Wed, Apr 23, 2025 at 3:40 PM Nicholas Pratte <npratte@iol.unh.edu> wrote:

>
> +
>      def send_packets_and_capture(
>          self,
>          packets: list[Packet],
> diff --git
> a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
> b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
> new file mode 100644
> index 0000000000..7a384cf6e0
> --- /dev/null
> +++
> b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
> @@ -0,0 +1,62 @@
> +"""Performance testing capable traffic generatiors."""
> +
> +from abc import ABC, abstractmethod
> +from dataclasses import dataclass
> +from typing import Callable
> +
> +from scapy.packet import Packet
> +
> +from framework.testbed_model.traffic_generator.traffic_generator import
> TrafficGenerator
>

I think this can become:

from .traffic_generator import TrafficGenerator


> +
> +
> +@dataclass(slots=True)
> +class PerformanceTrafficStats(ABC):
> +    """Data structure for stats offered by a given traffic generator."""
> +
> +    frame_size: int
>

Do we need to add an optional number of packet descriptors attribute? I
realize that is a SUT testpmd param, not a TREX param, but presumably when
we gather stats, we want to store the descriptor count with it.


> +
> +
> +class PerformanceTrafficGenerator(TrafficGenerator):
> +    """An Abstract Base Class for all performance-oriented traffic
> generators.
> +
> +    Provides an intermediary interface for performance-based traffic
> generator.
> +    """
> +
> +    _test_stats: list[PerformanceTrafficStats]
> +
> +    @property
> +    def is_capturing(self) -> bool:
> +        """Used for synchronization."""
> +        return False
> +
> +    @property
> +    def last_results(self) -> PerformanceTrafficStats | None:
> +        """Get the latest set of results from TG instance.
> +
> +        Returns:
> +            The most recent set of traffic statistics.
> +        """
> +        return self._test_stats.pop(0)
> +
> +    def generate_traffic_and_stats(
> +        self,
> +        packet: Packet,
> +        duration: float,  # Default of 60 (in seconds).
>

Should it be float = 60, so the default is coming from the function, and
not a "default" which is coming from the testsuite? If not, I guess the
"default" comment belongs in the testsuite?


> +    ) -> PerformanceTrafficStats:
> +        """Send packet traffic and acquire associated statistics."""
> +        return self._calculate_traffic_stats(packet, duration,
> self._generate_traffic)
> +
> +    def setup(self, ports):
> +        """Preliminary port setup prior to TG execution."""
> +        for port in self._tg_node.ports:
> +            self._tg_node.main_session.configure_port_mtu(2000, port)
>

Okay just checking... so if we do not do this, even after binding the port
to vfio-pci we cannot send traffic up to 2000 bytes?

Reviewed-by: Patrick Robb <probb@iol.unh.edu>

[-- Attachment #2: Type: text/html, Size: 4419 bytes --]

Re: [RFC Patch v1 3/5] dts: add asychronous support to ssh sessions.

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 46 bytes --]

Reviewed-by: Patrick Robb <probb@iol.unh.edu>

[-- Attachment #2: Type: text/html, Size: 112 bytes --]

Re: [RFC Patch v1 4/5] dts: add trex traffic generator to dts framework

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 2921 bytes --]

On Wed, Apr 23, 2025 at 3:40 PM Nicholas Pratte <npratte@iol.unh.edu> wrote:

> Implement the TREX traffic generator for use in the DTS framework. The
> provided implementation leverages TREX's stateless API automation
> library, via use of a Python shell. As such, version control of TREX may
> be needed. The DTS context has been modified to include a performance
> traffic generator in addition to a functional traffic generator.
>

I think the statement is that only certain versions are confirmed to work
off of your implementation? I think usage of the term version control is
confusing since we are just talking about using particular versions of
TREX, and not using a "version control" tool like git. This is more about
listing approved versions of the dependency for the TG.

>
>
>          dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
>          dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node,
> dpdk_build_env)
> -        traffic_generator =
> create_traffic_generator(config.traffic_generator, tg_node)
> +        # There is definitely a better way to do this.
> +        tg_dpdk_runtime_env = None
> +        if (
> +            config.perf_traffic_generator.type ==
> TrafficGeneratorType.TREX
> +            or config.func_traffic_generator.type ==
> TrafficGeneratorType.TREX
> +        ):
>

We have this from the testrun config

perf: false # disable performance testing
func: true # enable functional testing

So, use if testrunconfig.perf == true?

>
> +
> +@dataclass
> +class TrexPerformanceStats(PerformanceTrafficStats):
> +    """Data structure to store performance statistics for a given test
> run.
> +
> +    Attributes:
> +        packet: The packet that was sent in the test run.
> +        frame_size: The total length of the frame. (L2 downward)
> +        tx_expected_bps: The expected bits per second on a given NIC.
> +        tx_expected_cps: ...
>

"The expected connections per second" ? I think you just missed this one.


> +
> +        Attributes:
> +            ports: Related ports utilized in TG instance.
> +        """
> +        super().setup(ports)
> +        # Start TREX server process.
> +        try:
> +            self._logger.info("Starting TREX server process: sending 45
> second sleep.")
> +            privileged_command = self._os_session._get_privileged_command(
> +                f"""
> +                    cd /opt/v3.03/; {self._tg_config.remote_path}/t-rex-64
> +                     --cfg {self._tg_config.config} -i
> +                """
>

I know this was just a work in progress for RFC with some hardcoding, but
you already have tg_config.remote_path, right? So, the leading hardcoded cd
does not do anything, and even if you had to cd there you could use the
tg_config.remote_path?

Reviewed-by: Patrick Robb <probb@iol.unh.edu>

[-- Attachment #2: Type: text/html, Size: 4200 bytes --]

Re: [RFC Patch v1 5/5] dts: add performance test functions to test suite api

[Patrick Robb]

[-- Attachment #1: Type: text/plain, Size: 491 bytes --]

On Wed, Apr 23, 2025 at 3:40 PM Nicholas Pratte <npratte@iol.unh.edu> wrote:

> Provide functional performance method to run performance tests using a
> user-supplied performance traffic generator. The single core performance
> test is included, with some basic statistics checks verifying TG packet
> transmission rates.
>
>
>
Obviously the testsuite is a placeholder to demonstrate the TG usage, as
you say, which is all good.

Reviewed-by: Patrick Robb <probb@iol.unh.edu>

[-- Attachment #2: Type: text/html, Size: 918 bytes --]

Re: [RFC Patch v1 2/5] dts: rework traffic generator inheritance structure.

[Nicholas Pratte]

Hi Patick, see below!

>> +
>> +from framework.testbed_model.traffic_generator.traffic_generator import TrafficGenerator
>
>
> I think this can become:
>
> from .traffic_generator import TrafficGenerator

Ack.
>
>>
>> +
>> +
>> +@dataclass(slots=True)
>> +class PerformanceTrafficStats(ABC):
>> +    """Data structure for stats offered by a given traffic generator."""
>> +
>> +    frame_size: int
>
>
> Do we need to add an optional number of packet descriptors attribute? I realize that is a SUT testpmd param, not a TREX param, but presumably when we gather stats, we want to store the descriptor count with it.'

It could be done by passing another parameter when the function is
called. Maybe adding these parameters is helpful, maybe it isn't. It
might just be easier to strip the frame size and descriptors away
entirely from the statistics data structure, under the assumption that
these fields would be managed in the test suites themselves. I think
either/or would be fine.


>
>>
>> +
>> +
>> +class PerformanceTrafficGenerator(TrafficGenerator):
>> +    """An Abstract Base Class for all performance-oriented traffic generators.
>> +
>> +    Provides an intermediary interface for performance-based traffic generator.
>> +    """
>> +
>> +    _test_stats: list[PerformanceTrafficStats]
>> +
>> +    @property
>> +    def is_capturing(self) -> bool:
>> +        """Used for synchronization."""
>> +        return False
>> +
>> +    @property
>> +    def last_results(self) -> PerformanceTrafficStats | None:
>> +        """Get the latest set of results from TG instance.
>> +
>> +        Returns:
>> +            The most recent set of traffic statistics.
>> +        """
>> +        return self._test_stats.pop(0)
>> +
>> +    def generate_traffic_and_stats(
>> +        self,
>> +        packet: Packet,
>> +        duration: float,  # Default of 60 (in seconds).
>
>
> Should it be float = 60, so the default is coming from the function, and not a "default" which is coming from the testsuite? If not, I guess the "default" comment belongs in the testsuite?

I would say the comment can be removed. Right now, as you mentioned,
it is set up so that the default value is asserted at the test suite
level. It might just make sense to do away with this default value and
just insist the user explicitly provide a duration. That would make
sense in its own right.

Re: [RFC Patch v1 4/5] dts: add trex traffic generator to dts framework

[Nicholas Pratte]

>> Implement the TREX traffic generator for use in the DTS framework. The
>> provided implementation leverages TREX's stateless API automation
>> library, via use of a Python shell. As such, version control of TREX may
>> be needed. The DTS context has been modified to include a performance
>> traffic generator in addition to a functional traffic generator.
>
>
> I think the statement is that only certain versions are confirmed to work off of your implementation? I think usage of the term version control is confusing since we are just talking about using particular versions of TREX, and not using a "version control" tool like git. This is more about listing approved versions of the dependency for the TG.

I can fix that!
>>
>>
>>
>>          dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
>>          dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node, dpdk_build_env)
>> -        traffic_generator = create_traffic_generator(config.traffic_generator, tg_node)
>> +        # There is definitely a better way to do this.
>> +        tg_dpdk_runtime_env = None
>> +        if (
>> +            config.perf_traffic_generator.type == TrafficGeneratorType.TREX
>> +            or config.func_traffic_generator.type == TrafficGeneratorType.TREX
>> +        ):
>
>
> We have this from the testrun config
>
> perf: false # disable performance testing
> func: true # enable functional testing
>
> So, use if testrunconfig.perf == true?

I see what you're saying, I can make that change for the time being.
My reason for explicitly checking traffic generators types was that
any future implementations may not need a DPDK runtime environment.

>>
>>
>> +
>> +@dataclass
>> +class TrexPerformanceStats(PerformanceTrafficStats):
>> +    """Data structure to store performance statistics for a given test run.
>> +
>> +    Attributes:
>> +        packet: The packet that was sent in the test run.
>> +        frame_size: The total length of the frame. (L2 downward)
>> +        tx_expected_bps: The expected bits per second on a given NIC.
>> +        tx_expected_cps: ...
>
>
> "The expected connections per second" ? I think you just missed this one.

This was ultimately removed from the final version, incidentally. But
the way this was implemented does allow us to check this data.

>
>>
>> +
>> +        Attributes:
>> +            ports: Related ports utilized in TG instance.
>> +        """
>> +        super().setup(ports)
>> +        # Start TREX server process.
>> +        try:
>> +            self._logger.info("Starting TREX server process: sending 45 second sleep.")
>> +            privileged_command = self._os_session._get_privileged_command(
>> +                f"""
>> +                    cd /opt/v3.03/; {self._tg_config.remote_path}/t-rex-64
>> +                     --cfg {self._tg_config.config} -i
>> +                """
>
>
> I know this was just a work in progress for RFC with some hardcoding, but you already have tg_config.remote_path, right? So, the leading hardcoded cd does not do anything, and even if you had to cd there you could use the tg_config.remote_path?

Yes, that was hard coded as a temporary measure. The cd is needed to
ensure that the TREX executable actually runs properly. I must have
just forgot to switch that back as I was working with haste.

[RFC v2 0/6] Add TREX Traffic Generator to DTS Framework

[Nicholas Pratte]

v2:
  * Still some formatting issues that need clean up.
  * Several issues have been addressed
     - Includes some of Patrick's comments.
     - Personally identified bug fixes.
  * Single core perf test has been fleshed out.

Nicholas Pratte (6):
  dts: rework config module to support perf TGs
  dts: rework traffic generator inheritance structure.
  dts: add asynchronous support to ssh sessions.
  dts: add extended timeout option to interactive shells.
  dts: add trex traffic generator to dts framework
  dts: add performance test functions to test suite api

 dts/{ => configurations}/nodes.example.yaml   |   0
 .../test_run.example.yaml                     |   8 +-
 dts/configurations/tests_config.example.yaml  |   9 +
 .../trex_configs/intel_40g.yaml               |  18 ++
 dts/framework/config/test_run.py              |  20 +-
 dts/framework/context.py                      |  11 +-
 .../remote_session/interactive_shell.py       |   9 +-
 dts/framework/remote_session/ssh_session.py   |  14 +
 dts/framework/settings.py                     |   6 +-
 dts/framework/test_run.py                     |  27 +-
 dts/framework/test_suite.py                   |  33 +-
 .../traffic_generator/__init__.py             |  22 +-
 .../capturing_traffic_generator.py            |  34 ++
 .../performance_traffic_generator.py          |  69 +++++
 .../traffic_generator/traffic_generator.py    |  43 ---
 .../testbed_model/traffic_generator/trex.py   | 292 ++++++++++++++++++
 dts/tests/TestSuite_single_core_perf.py       |  56 ++++
 dts/tests_config.example.yaml                 |   4 -
 18 files changed, 605 insertions(+), 70 deletions(-)
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (82%)
 create mode 100644 dts/configurations/tests_config.example.yaml
 create mode 100644 dts/configurations/trex_configs/intel_40g.yaml
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py
 create mode 100644 dts/tests/TestSuite_single_core_perf.py
 delete mode 100644 dts/tests_config.example.yaml

-- 
2.47.1

[RFC v2 1/6] dts: rework config module to support perf TGs

[Nicholas Pratte]

Rework test run configuration file for TGs to support both application
directory location and any necessary configuration files; an example
TREX configuration file is provided. Configuration files have been moved
to a configurations directory, requiring a slight modification to the
settings module.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/{ => configurations}/nodes.example.yaml    |  0
 dts/{ => configurations}/test_run.example.yaml |  8 +++++++-
 .../tests_config.example.yaml                  |  0
 dts/configurations/trex_configs/intel_40g.yaml | 18 ++++++++++++++++++
 dts/framework/settings.py                      |  6 ++++--
 5 files changed, 29 insertions(+), 3 deletions(-)
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (82%)
 rename dts/{ => configurations}/tests_config.example.yaml (100%)
 create mode 100644 dts/configurations/trex_configs/intel_40g.yaml

diff --git a/dts/nodes.example.yaml b/dts/configurations/nodes.example.yaml
similarity index 100%
rename from dts/nodes.example.yaml
rename to dts/configurations/nodes.example.yaml
diff --git a/dts/test_run.example.yaml b/dts/configurations/test_run.example.yaml
similarity index 82%
rename from dts/test_run.example.yaml
rename to dts/configurations/test_run.example.yaml
index 1bc436eed1..49088b44b6 100644
--- a/dts/test_run.example.yaml
+++ b/dts/configurations/test_run.example.yaml
@@ -23,8 +23,14 @@ dpdk:
     # in a subdirectory of DPDK tree root directory. Otherwise, will be using the `build_options`
     # to build the DPDK from source. Either `precompiled_build_dir` or `build_options` can be
     # defined, but not both.
-traffic_generator:
+func_traffic_generator:
   type: SCAPY
+  remote_path: "" # The remote path of the traffic generator application. (Leave blank for SCAPY)
+  config: "" # Additional configuration files. (Leave blank if not required)
+perf_traffic_generator:
+  type: TREX
+  remote_path: "/opt/trex/v3.03" # The remote path of the traffic generator application. (Leave blank for SCAPY)
+  config: "trex_config.yaml" # Additional configuration files. (Leave blank if not required)
 perf: false # disable performance testing
 func: true # enable functional testing
 skip_smoke_tests: true # optional
diff --git a/dts/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
similarity index 100%
rename from dts/tests_config.example.yaml
rename to dts/configurations/tests_config.example.yaml
diff --git a/dts/configurations/trex_configs/intel_40g.yaml b/dts/configurations/trex_configs/intel_40g.yaml
new file mode 100644
index 0000000000..dae003795b
--- /dev/null
+++ b/dts/configurations/trex_configs/intel_40g.yaml
@@ -0,0 +1,18 @@
+### Config file generated by dpdk_setup_ports.py ###
+
+- version: 2
+  interfaces: ['11:00.0', '11:00.1']
+  port_bandwidth_gb: 40
+  port_info:
+      - dest_mac: 3c:fd:fe:d5:e5:f9 # MAC OF LOOPBACK TO IT'S DUAL INTERFACE
+        src_mac:  3c:fd:fe:d5:e5:f8
+      - dest_mac: 3c:fd:fe:d5:e5:f8 # MAC OF LOOPBACK TO IT'S DUAL INTERFACE
+        src_mac:  3c:fd:fe:d5:e5:f9
+
+  platform:
+      master_thread_id: 0
+      latency_thread_id: 7
+      dual_if:
+        - socket: 0
+          threads: [1,2,3,4,5,6]
+
diff --git a/dts/framework/settings.py b/dts/framework/settings.py
index 3f21615223..ccf4df25b0 100644
--- a/dts/framework/settings.py
+++ b/dts/framework/settings.py
@@ -130,9 +130,11 @@ class Settings:
     """
 
     #:
-    test_run_config_path: Path = Path(__file__).parent.parent.joinpath("test_run.yaml")
+    test_run_config_path: Path = Path(__file__).parent.parent.joinpath(
+        "configurations/test_run.yaml"
+    )
     #:
-    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("nodes.yaml")
+    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("configurations/nodes.yaml")
     #:
     tests_config_path: Path | None = None
     #:
-- 
2.47.1

[RFC v2 2/6] dts: rework traffic generator inheritance structure.

[Nicholas Pratte]

Rework TG class hierarchy to include performance traffic generators, in
addition to capturing traffic generators. As such, methods garnered
to capturing traffic have been moved to the CapturingTrafficGenerator
subclass.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 .../capturing_traffic_generator.py            | 34 +++++++++
 .../performance_traffic_generator.py          | 69 +++++++++++++++++++
 .../traffic_generator/traffic_generator.py    | 43 ------------
 3 files changed, 103 insertions(+), 43 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py

diff --git a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
index 61e5033f0b..124a1e5b86 100644
--- a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
@@ -65,6 +65,40 @@ def is_capturing(self) -> bool:
         """This traffic generator can capture traffic."""
         return True
 
+    def send_packet(self, packet: Packet, port: Port) -> None:
+        """Send `packet` and block until it is fully sent.
+
+        Send `packet` on `port`, then wait until `packet` is fully sent.
+
+        Args:
+            packet: The packet to send.
+            port: The egress port on the TG node.
+        """
+        self.send_packets([packet], port)
+
+    def send_packets(self, packets: list[Packet], port: Port) -> None:
+        """Send `packets` and block until they are fully sent.
+
+        Send `packets` on `port`, then wait until `packets` are fully sent.
+
+        Args:
+            packets: The packets to send.
+            port: The egress port on the TG node.
+        """
+        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
+        self._logger.debug(get_packet_summaries(packets))
+        self._send_packets(packets, port)
+
+    @abstractmethod
+    def _send_packets(self, packets: list[Packet], port: Port) -> None:
+        """The implementation of :method:`send_packets`.
+
+        The subclasses must implement this method which sends `packets` on `port`.
+        The method should block until all `packets` are fully sent.
+
+        What fully sent means is defined by the traffic generator.
+        """
+
     def send_packets_and_capture(
         self,
         packets: list[Packet],
diff --git a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
new file mode 100644
index 0000000000..54458325e2
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
@@ -0,0 +1,69 @@
+"""Performance testing capable traffic generatiors."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Callable
+
+from scapy.packet import Packet
+
+from .traffic_generator import TrafficGenerator
+
+
+@dataclass(slots=True)
+class PerformanceTrafficStats(ABC):
+    """Data structure for stats offered by a given traffic generator."""
+
+    frame_size: int
+    tx_expected_bps: float
+    tx_recorded_bps: float
+
+
+class PerformanceTrafficGenerator(TrafficGenerator):
+    """An Abstract Base Class for all performance-oriented traffic generators.
+
+    Provides an intermediary interface for performance-based traffic generator.
+    """
+
+    _test_stats: list[PerformanceTrafficStats]
+
+    @property
+    def is_capturing(self) -> bool:
+        """Used for synchronization."""
+        return False
+
+    @property
+    def last_results(self) -> PerformanceTrafficStats | None:
+        """Get the latest set of results from TG instance.
+
+        Returns:
+            The most recent set of traffic statistics.
+        """
+        return self._test_stats.pop(0)
+
+    def generate_traffic_and_stats(
+        self,
+        packet: Packet,
+        duration: float,
+    ) -> PerformanceTrafficStats:
+        """Send packet traffic and acquire associated statistics."""
+        return self._calculate_traffic_stats(packet, duration, self._generate_traffic)
+
+    def setup(self, ports):
+        """Preliminary port setup prior to TG execution."""
+        for port in self._tg_node.ports:
+            self._tg_node.main_session.configure_port_mtu(2000, port)
+
+    def teardown(self, ports):
+        """Port teardown after TG execution."""
+        for port in self._tg_node.ports:
+            self._tg_node.main_session.configure_port_mtu(1500, port)
+
+    @abstractmethod
+    def _calculate_traffic_stats(
+        self, packet: Packet, duration: float, traffic_gen_callback: Callable[[Packet, float], str]
+    ) -> PerformanceTrafficStats:
+        """Calculate packet traffic stats based on TG output."""
+
+    @abstractmethod
+    def _generate_traffic(self, packet: Packet, duration: float) -> str:
+        """Implementation for :method:`generate_traffic_and_stats`."""
diff --git a/dts/framework/testbed_model/traffic_generator/traffic_generator.py b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
index 6b9705d025..e6154ef1df 100644
--- a/dts/framework/testbed_model/traffic_generator/traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
@@ -11,13 +11,10 @@
 from abc import ABC, abstractmethod
 from typing import Iterable
 
-from scapy.packet import Packet
-
 from framework.config.test_run import TrafficGeneratorConfig
 from framework.logger import DTSLogger, get_dts_logger
 from framework.testbed_model.node import Node
 from framework.testbed_model.port import Port
-from framework.utils import get_packet_summaries
 
 
 class TrafficGenerator(ABC):
@@ -54,46 +51,6 @@ def setup(self, ports: Iterable[Port], rx_port: Port):
 
     def teardown(self, ports: Iterable[Port]):
         """Teardown the traffic generator."""
-        self.close()
-
-    def send_packet(self, packet: Packet, port: Port) -> None:
-        """Send `packet` and block until it is fully sent.
-
-        Send `packet` on `port`, then wait until `packet` is fully sent.
-
-        Args:
-            packet: The packet to send.
-            port: The egress port on the TG node.
-        """
-        self.send_packets([packet], port)
-
-    def send_packets(self, packets: list[Packet], port: Port) -> None:
-        """Send `packets` and block until they are fully sent.
-
-        Send `packets` on `port`, then wait until `packets` are fully sent.
-
-        Args:
-            packets: The packets to send.
-            port: The egress port on the TG node.
-        """
-        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
-        self._logger.debug(get_packet_summaries(packets))
-        self._send_packets(packets, port)
-
-    @abstractmethod
-    def _send_packets(self, packets: list[Packet], port: Port) -> None:
-        """The implementation of :method:`send_packets`.
-
-        The subclasses must implement this method which sends `packets` on `port`.
-        The method should block until all `packets` are fully sent.
-
-        What fully sent means is defined by the traffic generator.
-        """
-
-    @property
-    def is_capturing(self) -> bool:
-        """This traffic generator can't capture traffic."""
-        return False
 
     @abstractmethod
     def close(self) -> None:
-- 
2.47.1

[RFC v2 3/6] dts: add asynchronous support to ssh sessions.

[Nicholas Pratte]

Execution of the TREX server process requires an SSH session rework to
support asynchronous process management. Allowing access to asynchronous
functionality allows developers to execute processes without hijacking
the SSH session being used. In doing so, both timeout and runtime errors
may be avoided. This functionality leverages Fabric's Promise class,
which provides a join method to terminate the process when the process
is done being used, providing more secure process control.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/framework/remote_session/ssh_session.py | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/dts/framework/remote_session/ssh_session.py b/dts/framework/remote_session/ssh_session.py
index e6e4704bc2..e7d8b9cf53 100644
--- a/dts/framework/remote_session/ssh_session.py
+++ b/dts/framework/remote_session/ssh_session.py
@@ -13,6 +13,7 @@
     ThreadException,
     UnexpectedExit,
 )
+from invoke.runners import Promise
 from paramiko.ssh_exception import (
     AuthenticationException,
     BadHostKeyException,
@@ -99,6 +100,19 @@ def _send_command(self, command: str, timeout: float, env: dict | None) -> Comma
 
         return CommandResult(self.name, command, output.stdout, output.stderr, output.return_code)
 
+    def _send_async_command(self, command: str, timeout: float, env: dict | None) -> Promise:
+        try:
+            promise = self.session.run(
+                command, env=env, warn=True, hide=True, timeout=timeout, asynchronous=True
+            )
+            return promise
+        except (UnexpectedExit, ThreadException) as e:
+            self._logger.exception(e)
+            raise SSHSessionDeadError(self.hostname) from e
+        except CommandTimedOut as e:
+            self._logger.exception(e)
+            raise SSHTimeoutError(command) from e
+
     def is_alive(self) -> bool:
         """Overrides :meth:`~.remote_session.RemoteSession.is_alive`."""
         return self.session.is_connected
-- 
2.47.1

[RFC v2 4/6] dts: add extended timeout option to interactive shells.

[Nicholas Pratte]

stdin commands may require an explicit duration of time to
facilitate a total execution. Add an extra parameter to interactive
shells to handle these circumstances.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/framework/remote_session/interactive_shell.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/dts/framework/remote_session/interactive_shell.py b/dts/framework/remote_session/interactive_shell.py
index ba8489eafa..5331b8c7d1 100644
--- a/dts/framework/remote_session/interactive_shell.py
+++ b/dts/framework/remote_session/interactive_shell.py
@@ -177,7 +177,11 @@ def start_application(self, prompt: str | None = None) -> None:
         get_ctx().shell_pool.register_shell(self)
 
     def send_command(
-        self, command: str, prompt: str | None = None, skip_first_line: bool = False
+        self,
+        command: str,
+        prompt: str | None = None,
+        skip_first_line: bool = False,
+        added_timeout: int = 0,
     ) -> str:
         """Send `command` and get all output before the expected ending string.
 
@@ -195,6 +199,7 @@ def send_command(
             prompt: After sending the command, `send_command` will be expecting this string.
                 If :data:`None`, will use the class's default prompt.
             skip_first_line: Skip the first line when capturing the output.
+            added_timeout: additional duration for a given command, if needed.
 
         Returns:
             All output in the buffer before expected string.
@@ -213,6 +218,7 @@ def send_command(
         self._logger.info(f"Sending: '{command}'")
         if prompt is None:
             prompt = self._default_prompt
+        self._ssh_channel.settimeout(self._timeout + added_timeout)
         out: str = ""
         try:
             self._stdin.write(f"{command}{self._command_extra_chars}\n")
@@ -236,6 +242,7 @@ def send_command(
                 self._node.main_session.interactive_session.hostname
             ) from e
         finally:
+            self._ssh_channel.settimeout(self._timeout)
             self._logger.debug(f"Got output: {out}")
         return out
 
-- 
2.47.1

[RFC v2 5/6] dts: add trex traffic generator to dts framework

[Nicholas Pratte]

Implement the TREX traffic generator for use in the DTS framework. The
provided implementation leverages TREX's stateless API automation
library, via use of a Python shell. As such, limitation to specific TREX
versions may be be needed. The DTS context has been modified to include
a performance traffic generator in addition to a functional traffic
generator.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/framework/config/test_run.py              |  20 +-
 dts/framework/context.py                      |  11 +-
 dts/framework/test_run.py                     |  27 +-
 dts/framework/test_suite.py                   |   6 +-
 .../traffic_generator/__init__.py             |  22 +-
 .../testbed_model/traffic_generator/trex.py   | 292 ++++++++++++++++++
 6 files changed, 359 insertions(+), 19 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py

diff --git a/dts/framework/config/test_run.py b/dts/framework/config/test_run.py
index b6e4099eeb..3e09005338 100644
--- a/dts/framework/config/test_run.py
+++ b/dts/framework/config/test_run.py
@@ -396,6 +396,8 @@ class TrafficGeneratorType(str, Enum):
 
     #:
     SCAPY = "SCAPY"
+    #:
+    TREX = "TREX"
 
 
 class TrafficGeneratorConfig(FrozenModel):
@@ -404,6 +406,8 @@ class TrafficGeneratorConfig(FrozenModel):
     #: The traffic generator type the child class is required to define to be distinguished among
     #: others.
     type: TrafficGeneratorType
+    remote_path: PurePath
+    config: PurePath
 
 
 class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
@@ -412,8 +416,16 @@ class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
     type: Literal[TrafficGeneratorType.SCAPY]
 
 
+class TrexTrafficGeneratorConfig(TrafficGeneratorConfig):
+    """TREX traffic generator specific configuration."""
+
+    type: Literal[TrafficGeneratorType.TREX]
+
+
 #: A union type discriminating traffic generators by the `type` field.
-TrafficGeneratorConfigTypes = Annotated[ScapyTrafficGeneratorConfig, Field(discriminator="type")]
+TrafficGeneratorConfigTypes = Annotated[
+    TrexTrafficGeneratorConfig, ScapyTrafficGeneratorConfig, Field(discriminator="type")
+]
 
 #: Comma-separated list of logical cores to use. An empty string or ```any``` means use all lcores.
 LogicalCores = Annotated[
@@ -461,8 +473,10 @@ class TestRunConfiguration(FrozenModel):
 
     #: The DPDK configuration used to test.
     dpdk: DPDKConfiguration
-    #: The traffic generator configuration used to test.
-    traffic_generator: TrafficGeneratorConfigTypes
+    #: The traffic generator configuration used for functional tests.
+    func_traffic_generator: TrafficGeneratorConfig
+    #: The traffic generator configuration used for performance tests.
+    perf_traffic_generator: TrafficGeneratorConfig
     #: Whether to run performance tests.
     perf: bool
     #: Whether to run functional tests.
diff --git a/dts/framework/context.py b/dts/framework/context.py
index 4360bc8699..fb7ded2a10 100644
--- a/dts/framework/context.py
+++ b/dts/framework/context.py
@@ -16,7 +16,12 @@
 
 if TYPE_CHECKING:
     from framework.remote_session.dpdk import DPDKBuildEnvironment, DPDKRuntimeEnvironment
-    from framework.testbed_model.traffic_generator.traffic_generator import TrafficGenerator
+    from framework.testbed_model.traffic_generator.capturing_traffic_generator import (
+        CapturingTrafficGenerator,
+    )
+    from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+        PerformanceTrafficGenerator,
+    )
 
 P = ParamSpec("P")
 
@@ -69,7 +74,9 @@ class Context:
     topology: Topology
     dpdk_build: "DPDKBuildEnvironment"
     dpdk: "DPDKRuntimeEnvironment"
-    tg: "TrafficGenerator"
+    tg_dpdk: "DPDKRuntimeEnvironment"
+    func_tg: "CapturingTrafficGenerator"
+    perf_tg: "PerformanceTrafficGenerator"
     local: LocalContext = field(default_factory=LocalContext)
     shell_pool: ShellPool = field(default_factory=ShellPool)
 
diff --git a/dts/framework/test_run.py b/dts/framework/test_run.py
index 0fdc57ea9c..be1476081a 100644
--- a/dts/framework/test_run.py
+++ b/dts/framework/test_run.py
@@ -107,7 +107,7 @@
 from types import MethodType
 from typing import ClassVar, Protocol, Union
 
-from framework.config.test_run import TestRunConfiguration
+from framework.config.test_run import TestRunConfiguration, TrafficGeneratorType
 from framework.context import Context, init_ctx
 from framework.exception import (
     InternalError,
@@ -204,10 +204,22 @@ def __init__(
 
         dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
         dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node, dpdk_build_env)
-        traffic_generator = create_traffic_generator(config.traffic_generator, tg_node)
+        # There is definitely a better way to do this.
+        tg_dpdk_runtime_env = None
+        if config.perf:
+            tg_dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, tg_node, None)
+        func_traffic_generator = create_traffic_generator(config.func_traffic_generator, tg_node)
+        perf_traffic_generator = create_traffic_generator(config.perf_traffic_generator, tg_node)
 
         self.ctx = Context(
-            sut_node, tg_node, topology, dpdk_build_env, dpdk_runtime_env, traffic_generator
+            sut_node,
+            tg_node,
+            topology,
+            dpdk_build_env,
+            dpdk_runtime_env,
+            tg_dpdk_runtime_env,
+            func_traffic_generator,
+            perf_traffic_generator,
         )
         self.result = result
         self.selected_tests = list(self.config.filter_tests(tests_config))
@@ -345,7 +357,9 @@ def next(self) -> State | None:
         test_run.ctx.sut_node.setup()
         test_run.ctx.tg_node.setup()
         test_run.ctx.dpdk.setup(test_run.ctx.topology.sut_ports)
-        test_run.ctx.tg.setup(test_run.ctx.topology.tg_ports, test_run.ctx.topology.tg_port_ingress)
+        test_run.ctx.tg_dpdk.setup(test_run.ctx.topology.tg_ports)
+        test_run.ctx.func_tg.setup(test_run.ctx.topology.tg_ports)
+        test_run.ctx.perf_tg.setup(test_run.ctx.topology.tg_ports)
 
         self.result.ports = test_run.ctx.topology.sut_ports + test_run.ctx.topology.tg_ports
         self.result.sut_info = test_run.ctx.sut_node.node_info
@@ -430,8 +444,9 @@ def description(self) -> str:
 
     def next(self) -> State | None:
         """Next state."""
-        self.test_run.ctx.shell_pool.terminate_current_pool()
-        self.test_run.ctx.tg.teardown(self.test_run.ctx.topology.tg_ports)
+        self.test_run.ctx.tg_dpdk.teardown(self.test_run.ctx.topology.tg_ports)
+        self.test_run.ctx.func_tg.teardown(self.test_run.ctx.topology.tg_ports)
+        self.test_run.ctx.perf_tg.teardown(self.test_run.ctx.topology.tg_ports)
         self.test_run.ctx.dpdk.teardown(self.test_run.ctx.topology.sut_ports)
         self.test_run.ctx.tg_node.teardown()
         self.test_run.ctx.sut_node.teardown()
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index e07c327b77..507df508cb 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -254,11 +254,11 @@ def send_packets_and_capture(
             A list of received packets.
         """
         assert isinstance(
-            self._ctx.tg, CapturingTrafficGenerator
+            self._ctx.func_tg, CapturingTrafficGenerator
         ), "Cannot capture with a non-capturing traffic generator"
         # TODO: implement @requires for types of traffic generator
         packets = self._adjust_addresses(packets)
-        return self._ctx.tg.send_packets_and_capture(
+        return self._ctx.func_tg.send_packets_and_capture(
             packets,
             self._ctx.topology.tg_port_egress,
             self._ctx.topology.tg_port_ingress,
@@ -276,7 +276,7 @@ def send_packets(
             packets: Packets to send.
         """
         packets = self._adjust_addresses(packets)
-        self._ctx.tg.send_packets(packets, self._ctx.topology.tg_port_egress)
+        self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
     def get_expected_packets(
         self,
diff --git a/dts/framework/testbed_model/traffic_generator/__init__.py b/dts/framework/testbed_model/traffic_generator/__init__.py
index 2a259a6e6c..53125995cd 100644
--- a/dts/framework/testbed_model/traffic_generator/__init__.py
+++ b/dts/framework/testbed_model/traffic_generator/__init__.py
@@ -14,17 +14,27 @@
 and a capturing traffic generator is required.
 """
 
-from framework.config.test_run import ScapyTrafficGeneratorConfig, TrafficGeneratorConfig
+from framework.config.test_run import (
+    ScapyTrafficGeneratorConfig as ScapyTrafficGeneratorConfig,
+)
+from framework.config.test_run import (
+    TrafficGeneratorConfig,
+    TrafficGeneratorType,
+)
+from framework.config.test_run import (
+    TrexTrafficGeneratorConfig as TrexTrafficGeneratorConfig,
+)
 from framework.exception import ConfigurationError
 from framework.testbed_model.node import Node
 
-from .capturing_traffic_generator import CapturingTrafficGenerator
 from .scapy import ScapyTrafficGenerator
+from .traffic_generator import TrafficGenerator
+from .trex import TrexTrafficGenerator
 
 
 def create_traffic_generator(
     traffic_generator_config: TrafficGeneratorConfig, node: Node
-) -> CapturingTrafficGenerator:
+) -> TrafficGenerator:
     """The factory function for creating traffic generator objects from the test run configuration.
 
     Args:
@@ -37,8 +47,10 @@ def create_traffic_generator(
     Raises:
         ConfigurationError: If an unknown traffic generator has been setup.
     """
-    match traffic_generator_config:
-        case ScapyTrafficGeneratorConfig():
+    match traffic_generator_config.type:
+        case TrafficGeneratorType.SCAPY:
             return ScapyTrafficGenerator(node, traffic_generator_config, privileged=True)
+        case TrafficGeneratorType.TREX:
+            return TrexTrafficGenerator(node, traffic_generator_config, privileged=True)
         case _:
             raise ConfigurationError(f"Unknown traffic generator: {traffic_generator_config.type}")
diff --git a/dts/framework/testbed_model/traffic_generator/trex.py b/dts/framework/testbed_model/traffic_generator/trex.py
new file mode 100644
index 0000000000..b517333ab0
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/trex.py
@@ -0,0 +1,292 @@
+"""Implementation for TREX performance traffic generator."""
+
+import time
+from dataclasses import dataclass
+from enum import Flag, auto
+from typing import Callable, ClassVar
+
+from invoke.runners import Promise
+from scapy.packet import Packet
+
+from framework.config.node import NodeConfiguration
+from framework.config.test_run import TrafficGeneratorConfig
+from framework.exception import SSHTimeoutError
+from framework.remote_session.python_shell import PythonShell
+from framework.remote_session.ssh_session import SSHSession
+from framework.testbed_model.linux_session import LinuxSession
+from framework.testbed_model.node import Node, create_session
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
+
+
+@dataclass
+class TrexPerPortStats:
+    """Performance statistics on a per port basis.
+
+    Attributes:
+        opackets: Number of packets sent.
+        obytes: Number of egress bytes sent.
+        tx_bps: Maximum bits per second transmitted.
+        tx_pps: Number of transmitted packets sent.
+    """
+
+    tx_bps: float
+    tx_pps: float
+
+
+@dataclass
+class TrexPerformanceStats(PerformanceTrafficStats):
+    """Data structure to store performance statistics for a given test run.
+
+    Attributes:
+        packet: The packet that was sent in the test run.
+        frame_size: The total length of the frame. (L2 downward)
+        tx_expected_bps: The expected bits per second on a given NIC.
+        tx_expected_cps: ...
+        tx_expected_pps: The expected packets per second of a given NIC.
+        tx_pps: The recorded maximum packets per second of the tested NIC.
+        tx_cps: The recorded maximum cps of the tested NIC
+        tx_bps: The recorded maximum bits per second of the tested NIC.
+        obytes: Total bytes output during test run.
+        port_stats: A list of :class:`TrexPerPortStats` provided by TREX.
+    """
+
+    packet: Packet
+    frame_size: int
+
+    tx_expected_bps: float
+    tx_expected_cps: float
+    tx_expected_pps: float
+
+    tx_pps: float
+    tx_cps: float
+    tx_bps: float
+
+    port_stats: list[TrexPerPortStats] | None
+
+
+class TrexStatelessTXModes(Flag):
+    """Flags indicating TREX instance's current trasmission mode."""
+
+    CONTINUOUS = auto()
+    SINGLE_BURST = auto()
+    MULTI_BURST = auto()
+
+
+class TrexTrafficGenerator(PythonShell, PerformanceTrafficGenerator):
+    """TREX traffic generator.
+
+    This implementation leverages the stateless API library provided in the TREX installation.
+
+    Attributes:
+        stl_client_name: The name of the stateless client used in the stateless API.
+        packet_stream_name: The name of the stateless packet stream used in the stateless API.
+        timeout_duration: Internal timeout for connection to the TREX server.
+    """
+
+    _os_session: LinuxSession
+    _server_remote_session: SSHSession
+    _trex_server_process: Promise
+
+    _tg_config: TrafficGeneratorConfig
+    _node_config: NodeConfiguration
+
+    _python_indentation: ClassVar[str] = " " * 4
+
+    stl_client_name: ClassVar[str] = "client"
+    packet_stream_name: ClassVar[str] = "stream"
+
+    _streaming_mode: TrexStatelessTXModes = TrexStatelessTXModes.CONTINUOUS
+
+    timeout_duration: int
+
+    def __init__(
+        self, tg_node: Node, config: TrafficGeneratorConfig, timeout_duration: int = 5, **kwargs
+    ) -> None:
+        """Initialize the TREX server.
+
+        Initializes needed OS sessions for the creation of the TREX server process.
+
+        Attributes:
+            tg_node: TG node the TREX instance is operating on.
+            config: Traffic generator config provided for TREX instance.
+            timeout_duration: Internal timeout for connection to the TREX server.
+        """
+        super().__init__(node=tg_node, config=config, tg_node=tg_node, **kwargs)
+        self._node_config = tg_node.config
+        self._tg_config = config
+        self.timeout_duration = timeout_duration
+
+        # Create TREX server session.
+        self._tg_node._other_sessions.append(
+            create_session(self._tg_node.config, "TREX Server.", self._logger)
+        )
+        self._os_session = self._tg_node._other_sessions[0]
+        self._server_remote_session = self._os_session.remote_session
+
+    def setup(self, ports):
+        """Initialize and start a TREX server process.
+
+        Binds TG ports to vfio-pci and starts the trex process.
+
+        Attributes:
+            ports: Related ports utilized in TG instance.
+        """
+        super().setup(ports)
+        # Start TREX server process.
+        try:
+            self._logger.info("Starting TREX server process: sending 45 second sleep.")
+            server_command = [
+                f"cd {self._tg_config.remote_path}; {self._tg_config.remote_path}/t-rex-64",
+                f"--cfg {self._tg_config.config} -i"
+            ]
+            privileged_command = self._os_session._get_privileged_command(" ".join(server_command))
+            self._trex_server_process = self._server_remote_session._send_async_command(
+                privileged_command, timeout=None, env=None
+            )
+            self._logger.info(f"Sending: '{privileged_command}")
+            time.sleep(45)
+        except SSHTimeoutError as e:
+            self._logger.exception("Failed to start TREX server process.", e)
+
+        # Start Python shell.
+        self.start_application()
+        self.send_command("import os")
+        # Parent directory: /opt/v3.03/automation/trex_control_plane/interactive
+        self.send_command(
+            f"os.chdir('{self._tg_config.remote_path}/automation/trex_control_plane/interactive')"
+        )
+
+        # Import stateless API components.
+        imports = [
+            "import trex",
+            "import trex.stl",
+            "import trex.stl.trex_stl_client",
+            "import trex.stl.trex_stl_streams",
+            "import trex.stl.trex_stl_packet_builder_scapy",
+            "from scapy.layers.l2 import Ether",
+            "from scapy.layers.inet import IP",
+            "from scapy.packet import Raw",
+        ]
+        self.send_command("\n".join(imports))
+
+        stateless_client = [
+            f"{self.stl_client_name} = trex.stl.trex_stl_client.STLClient(",
+            f"username='{self._node_config.user}',",
+            "server='127.0.0.1',",
+            f"sync_timeout={self.timeout_duration}",
+            ")",
+        ]
+        self.send_command(f"\n{self._python_indentation}".join(stateless_client))
+        self.send_command(f"{self.stl_client_name}.connect()")
+
+    def teardown(self, ports) -> None:
+        """Teardown the TREX server and stateless implementation.
+
+        close the TREX server process, and stop the Python shell.
+
+        Attributes:
+            ports: Associated ports used by the TREX instance.
+        """
+        super().teardown(ports)
+        self.send_command(f"{self.stl_client_name}.disconnect()")
+        self._os_session.send_command(f"pkill 't-rex-64'", privileged=True)
+        self._trex_server_process.join()
+        self.close()
+
+    def _calculate_traffic_stats(
+        self, packet: Packet, duration: float, callback: Callable[[Packet, float], str]
+    ) -> PerformanceTrafficStats:
+        """Calculate the traffic statistics, using provided TG output.
+
+        Takes in the statistics output provided by the stateless API implementation, and collects
+        them into a performance statistics data structure.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+            duration: The duration of the test.
+            callback: The callback function used to generate the traffic.
+        """
+        # Convert to a dictionary.
+        stats_output = eval(callback(packet, duration))
+        global_output = stats_output.get("global", "ERROR - DATA NOT FOUND")
+
+        print(type(stats_output))
+        print(stats_output)
+
+        print(type(global_output))
+        print(global_output)
+        return TrexPerformanceStats(
+            len(packet),
+            packet,
+            stats_output.get("tx_expected_bps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_expected_cps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_expected_pps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_pps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_cps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("tx_bps", "ERROR - DATA NOT FOUND"),
+            stats_output.get("obytes", "ERROR - DATA NOT FOUND"),
+            None,
+        )
+
+    def set_streaming_mode(self, streaming_mode: TrexStatelessTXModes) -> None:
+        """Set the streaming mode of the TREX instance."""
+        # Streaming modes are mutually exclusive.
+        self._streaming_mode = self._streaming_mode & streaming_mode
+
+    def _generate_traffic(self, packet: Packet, duration: float) -> str:
+        """Generate traffic using provided packet.
+
+        Uses the provided packet to generate traffic for the provided duration.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+            duration: The duration of the test being performed.
+
+        Returns:
+            a string output of statistics provided by the traffic generator.
+        """
+        """Implementation for :method:`generate_traffic_and_stats`."""
+        streaming_mode = ""
+        if self._streaming_mode == TrexStatelessTXModes.CONTINUOUS:
+            streaming_mode = "STLTXCont"
+        elif self._streaming_mode == TrexStatelessTXModes.SINGLE_BURST:
+            streaming_mode = "STLTXSingleBurst"
+        elif self._streaming_mode == TrexStatelessTXModes.MULTI_BURST:
+            streaming_mode = "STLTXMultiBurst"
+
+        packet_stream = [
+            f"{self.packet_stream_name} = trex.stl.trex_stl_streams.STLStream(",
+            f"name='Test_{len(packet)}_bytes',",
+            f"packet=trex.stl.trex_stl_packet_builder_scapy.STLPktBuilder(pkt={packet.command()}),",
+            f"mode=trex.stl.trex_stl_streams.{streaming_mode}(),",
+            ")",
+        ]
+        self.send_command("\n".join(packet_stream))
+
+        # Prepare TREX console for next performance test.
+        procedure = [
+            f"{self.stl_client_name}.connect()",
+            f"{self.stl_client_name}.reset(ports = [0, 1])",
+            f"{self.stl_client_name}.add_streams({self.packet_stream_name}, ports=[0, 1])",
+            f"{self.stl_client_name}.clear_stats()",
+            ")",
+        ]
+        self.send_command("\n".join(procedure))
+
+        start_test = [
+            f"{self.stl_client_name}.start(ports=[0, 1], duration={duration})",
+            f"{self.stl_client_name}.wait_on_traffic(ports=[0, 1])",
+        ]
+        self._timeout = duration
+        self.send_command("\n".join(start_test), added_timeout=duration)
+        import time
+
+        time.sleep(duration + 1)
+
+        # Gather statistics output for parsing.
+        return self.send_command(
+            f"{self.stl_client_name}.get_stats(ports=[0, 1])", skip_first_line=True
+        )
-- 
2.47.1

[RFC v2 6/6] dts: add performance test functions to test suite api

[Nicholas Pratte]

Provide functional performance method to run performance tests using a
user-supplied performance traffic generator. The single core performance
test is included, with some basic statistics checks verifying TG packet
transmission rates.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
---
 dts/configurations/tests_config.example.yaml |  5 ++
 dts/framework/test_suite.py                  | 27 ++++++++++
 dts/tests/TestSuite_single_core_perf.py      | 56 ++++++++++++++++++++
 3 files changed, 88 insertions(+)
 create mode 100644 dts/tests/TestSuite_single_core_perf.py

diff --git a/dts/configurations/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
index 38cf7a0dce..d3d867ae18 100644
--- a/dts/configurations/tests_config.example.yaml
+++ b/dts/configurations/tests_config.example.yaml
@@ -2,3 +2,8 @@
 
 hello_world:
   msg: A custom hello world to you!
+single_core_perf:
+  tx_rx_descriptors: [128, 512, 2048]
+  frame_sizes: [64, 128]
+  expected_throughput: 40
+  # rate: gbps | mbps
\ No newline at end of file
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index 507df508cb..a89faac2d5 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -38,6 +38,10 @@
     CapturingTrafficGenerator,
     PacketFilteringConfig,
 )
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
 
 from .exception import ConfigurationError, InternalError, TestCaseVerifyError
 from .logger import DTSLogger, get_dts_logger
@@ -266,6 +270,26 @@ def send_packets_and_capture(
             duration,
         )
 
+    def assess_performance_by_packet(
+        self, packet: Packet, duration: int = 60
+    ) -> PerformanceTrafficStats:
+        """Send a given packet for a given duration and assess basic performance statistics.
+
+        Send `packet` and assess NIC performance for a given duration, corresponding to the test
+        suite's given topology.
+
+        Args:
+            packet: The packet to send.
+            duration: Performance test duration (in seconds)
+
+        Returns:
+            Performance statistics of the generated test.
+        """
+        assert isinstance(
+            self._ctx.perf_tg, PerformanceTrafficGenerator
+        ), "Cannot run performance tests on non-performance traffic generator."
+        return self._ctx.perf_tg.generate_traffic_and_stats(packet, duration)
+
     def send_packets(
         self,
         packets: list[Packet],
@@ -275,6 +299,9 @@ def send_packets(
         Args:
             packets: Packets to send.
         """
+        assert isinstance(
+            self._ctx.perf_tg, CapturingTrafficGenerator
+        ), "Cannot run performance tests on non-capturing traffic generator."
         packets = self._adjust_addresses(packets)
         self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
diff --git a/dts/tests/TestSuite_single_core_perf.py b/dts/tests/TestSuite_single_core_perf.py
new file mode 100644
index 0000000000..2e1d3c1ae8
--- /dev/null
+++ b/dts/tests/TestSuite_single_core_perf.py
@@ -0,0 +1,56 @@
+"""Single core performance test suite."""
+
+
+from framework.params.testpmd import RXRingParams, TXRingParams
+from framework.remote_session.testpmd_shell import TestPmdShell
+from framework.test_suite import TestSuite, perf_test, BaseConfig
+
+from scapy.layers.inet import IP
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
+
+class Config(BaseConfig):
+    """Performance test metrics to be compared by real-world results."""
+
+    #: Expected results
+    tx_rx_descriptors: list[int]
+    frame_sizes: list[int]
+    expected_throughput: int
+    
+
+class TestSingleCorePerf(TestSuite):
+    """Single core performance test suite."""
+
+    config: Config
+
+    frame_sizes: list[int]
+    tx_rx_descriptors: list[int]
+    expected_throughput: int
+
+
+    def set_up_suite(self):
+        self.frame_sizes = self.config.frame_sizes
+        for frame_size in self.frame_sizes:
+            self.verify(frame_size >= 34,
+                "Provided frame size is too small. (Space needed for Ether()/IP())"
+            )
+        self.tx_rx_descriptors = self.config.tx_rx_descriptors
+        self.expected_throughput = self.config.expected_throughput
+
+    @perf_test
+    def test_perf_nic_single_core(self) -> None:
+        """Prototype test case."""
+        for frame_size in self.frame_sizes:
+            for descriptor_size in self.tx_rx_descriptors:
+                with TestPmdShell(
+                    tx_ring=TXRingParams(descriptors=descriptor_size),
+                    rx_ring=RXRingParams(descriptors=descriptor_size)
+                ) as testpmd:
+                    packet = Ether() / IP() / Raw(load="x" * (frame_size - 14 - 20))
+
+                    testpmd.start()
+                    stats = self.assess_performance_by_packet(packet, duration=60)
+                    print(stats)
+                    self.verify(
+                        stats.tx_expected_bps == 40, "Expected output does not match recorded output."
+                    )
\ No newline at end of file
-- 
2.47.1

Re: [RFC v2 1/6] dts: rework config module to support perf TGs

[Dean Marx]

<snip>
> +      - dest_mac: 3c:fd:fe:d5:e5:f9 # MAC OF LOOPBACK TO IT'S DUAL INTERFACE
> +        src_mac:  3c:fd:fe:d5:e5:f8
> +      - dest_mac: 3c:fd:fe:d5:e5:f8 # MAC OF LOOPBACK TO IT'S DUAL INTERFACE
> +        src_mac:  3c:fd:fe:d5:e5:f9

Should be its not it's

I like the concept of having a configurations directory within DTS,
those files were starting to add up anyways.

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

Re: [RFC v2 2/6] dts: rework traffic generator inheritance structure.

[Dean Marx]

On Fri, May 16, 2025 at 4:19 PM Nicholas Pratte <npratte@iol.unh.edu> wrote:
>
> Rework TG class hierarchy to include performance traffic generators, in
> addition to capturing traffic generators. As such, methods garnered
> to capturing traffic have been moved to the CapturingTrafficGenerator
> subclass.

This commit message is worded a bit strangely to me, not sure garnered
is the right word to use here. Also not sure what "as such" is
referring to exactly

<snip>
> +"""Performance testing capable traffic generatiors."""

*generators

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

Re: [RFC v2 3/6] dts: add asynchronous support to ssh sessions.

[Dean Marx]

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

Re: [RFC v2 4/6] dts: add extended timeout option to interactive shells.

[Dean Marx]

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

Re: [RFC v2 5/6] dts: add trex traffic generator to dts framework

[Dean Marx]

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

Re: [RFC v2 6/6] dts: add performance test functions to test suite api

[Dean Marx]

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

[PATCH v3 1/5] dts: rework config module to support perf TGs

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Rework test run configuration file for TGs to support both application
directory location and any necessary configuration files; an example
TREX configuration file is provided. Configuration files have been moved
to a configurations directory, requiring a slight modification to the
settings module.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/{ => configurations}/nodes.example.yaml        |  0
 dts/{ => configurations}/test_run.example.yaml     | 13 +++++++------
 dts/{ => configurations}/tests_config.example.yaml |  0
 dts/framework/settings.py                          |  6 ++++--
 4 files changed, 11 insertions(+), 8 deletions(-)
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (80%)
 rename dts/{ => configurations}/tests_config.example.yaml (100%)

diff --git a/dts/nodes.example.yaml b/dts/configurations/nodes.example.yaml
similarity index 100%
rename from dts/nodes.example.yaml
rename to dts/configurations/nodes.example.yaml
diff --git a/dts/test_run.example.yaml b/dts/configurations/test_run.example.yaml
similarity index 80%
rename from dts/test_run.example.yaml
rename to dts/configurations/test_run.example.yaml
index 1bc436eed1..e9baf10035 100644
--- a/dts/test_run.example.yaml
+++ b/dts/configurations/test_run.example.yaml
@@ -1,8 +1,3 @@
-# SPDX-License-Identifier: BSD-3-Clause
-# Copyright 2022-2023 The DPDK contributors
-# Copyright 2023 Arm Limited
-
-# Define the test run environment
 dpdk:
   lcores: "" # use all available logical cores (Skips first core)
   memory_channels: 4 # tells DPDK to use 4 memory channels
@@ -23,8 +18,14 @@ dpdk:
     # in a subdirectory of DPDK tree root directory. Otherwise, will be using the `build_options`
     # to build the DPDK from source. Either `precompiled_build_dir` or `build_options` can be
     # defined, but not both.
-traffic_generator:
+func_traffic_generator:
   type: SCAPY
+  remote_path: "" # The remote path of the traffic generator application. (Leave blank for SCAPY)
+  config: "" # Additional configuration files. (Leave blank if not required)
+perf_traffic_generator:
+  type: TREX
+  remote_path: "/opt/trex/v3.03" # The remote path of the traffic generator application. (Leave blank for SCAPY)
+  config: "/opt/trex_config/trex_config.yaml" # Additional configuration files. (Leave blank if not required)
 perf: false # disable performance testing
 func: true # enable functional testing
 skip_smoke_tests: true # optional
diff --git a/dts/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
similarity index 100%
rename from dts/tests_config.example.yaml
rename to dts/configurations/tests_config.example.yaml
diff --git a/dts/framework/settings.py b/dts/framework/settings.py
index 3f21615223..ccf4df25b0 100644
--- a/dts/framework/settings.py
+++ b/dts/framework/settings.py
@@ -130,9 +130,11 @@ class Settings:
     """
 
     #:
-    test_run_config_path: Path = Path(__file__).parent.parent.joinpath("test_run.yaml")
+    test_run_config_path: Path = Path(__file__).parent.parent.joinpath(
+        "configurations/test_run.yaml"
+    )
     #:
-    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("nodes.yaml")
+    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("configurations/nodes.yaml")
     #:
     tests_config_path: Path | None = None
     #:
-- 
2.49.0

[PATCH v3 2/5] dts: rework traffic generator inheritance structure

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Rework TG class hierarchy to include performance traffic generators.
As such, methods specific to capturing traffic have been moved to the
CapturingTrafficGenerator subclass.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 .../capturing_traffic_generator.py            | 34 ++++++++++
 .../performance_traffic_generator.py          | 63 +++++++++++++++++++
 .../testbed_model/traffic_generator/scapy.py  | 10 ++-
 .../traffic_generator/traffic_generator.py    | 44 -------------
 4 files changed, 104 insertions(+), 47 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py

diff --git a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
index a85858ba07..110244a7b0 100644
--- a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
@@ -65,6 +65,40 @@ def is_capturing(self) -> bool:
         """This traffic generator can capture traffic."""
         return True
 
+    def send_packet(self, packet: Packet, port: Port) -> None:
+        """Send `packet` and block until it is fully sent.
+
+        Send `packet` on `port`, then wait until `packet` is fully sent.
+
+        Args:
+            packet: The packet to send.
+            port: The egress port on the TG node.
+        """
+        self.send_packets([packet], port)
+
+    def send_packets(self, packets: list[Packet], port: Port) -> None:
+        """Send `packets` and block until they are fully sent.
+
+        Send `packets` on `port`, then wait until `packets` are fully sent.
+
+        Args:
+            packets: The packets to send.
+            port: The egress port on the TG node.
+        """
+        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
+        self._logger.debug(get_packet_summaries(packets))
+        self._send_packets(packets, port)
+
+    @abstractmethod
+    def _send_packets(self, packets: list[Packet], port: Port) -> None:
+        """The implementation of :method:`send_packets`.
+
+        The subclasses must implement this method which sends `packets` on `port`.
+        The method should block until all `packets` are fully sent.
+
+        What fully sent means is defined by the traffic generator.
+        """
+
     def send_packets_and_capture(
         self,
         packets: list[Packet],
diff --git a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
new file mode 100644
index 0000000000..59dac6a075
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
@@ -0,0 +1,63 @@
+"""Traffic generators for performance tests which can generate a high number of packets."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+from scapy.packet import Packet
+
+from framework.testbed_model.topology import Topology
+
+from .traffic_generator import TrafficGenerator
+
+
+@dataclass(slots=True)
+class PerformanceTrafficStats(ABC):
+    """Data structure to store performance statistics for a given test run.
+
+    Attributes:
+        frame_size: The total length of the frame
+        tx_pps: Recorded tx packets per second
+        tx_cps: Recorded tx connections per second
+        tx_bps: Recorded tx bytes per second
+        rx_pps: Recorded rx packets per second
+        rx_bps: Recorded rx bytes per second
+    """
+
+    frame_size: int
+
+    tx_pps: float
+    tx_cps: float
+    tx_bps: float
+
+    rx_pps: float
+    rx_bps: float
+
+
+class PerformanceTrafficGenerator(TrafficGenerator):
+    """An abstract base class for all performance-oriented traffic generators.
+
+    Provides an intermediary interface for performance-based traffic generator.
+    """
+
+    @abstractmethod
+    def calculate_traffic_and_stats(
+        self,
+        packet: Packet,
+        send_mpps: int,
+        duration: float,
+    ) -> PerformanceTrafficStats:
+        """Send packet traffic and acquire associated statistics.
+
+        Args:
+        packet: The packet to send.
+        send_mpps: The millions packets per second send rate.
+        duration: Performance test duration (in seconds).
+
+        Returns:
+            Performance statistics of the generated test.
+        """
+
+    def setup(self, topology: Topology) -> None:
+        """Overrides :meth:`.traffic_generator.TrafficGenerator.setup`."""
+        for port in self._tg_node.ports:
+            self._tg_node.main_session.configure_port_mtu(2000, port)
diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py b/dts/framework/testbed_model/traffic_generator/scapy.py
index e21ba4ed96..602b93d473 100644
--- a/dts/framework/testbed_model/traffic_generator/scapy.py
+++ b/dts/framework/testbed_model/traffic_generator/scapy.py
@@ -26,7 +26,7 @@
 from scapy.packet import Packet
 
 from framework.config.node import OS
-from framework.config.test_run import ScapyTrafficGeneratorConfig
+from framework.config.test_run import TrafficGeneratorConfig
 from framework.exception import InteractiveSSHSessionDeadError, InternalError
 from framework.remote_session.python_shell import PythonShell
 from framework.testbed_model.node import Node
@@ -285,7 +285,7 @@ class also extends :class:`.capturing_traffic_generator.CapturingTrafficGenerato
     first.
     """
 
-    _config: ScapyTrafficGeneratorConfig
+    _config: TrafficGeneratorConfig
     _shell: PythonShell
     _sniffer: ScapyAsyncSniffer
 
@@ -296,7 +296,7 @@ class also extends :class:`.capturing_traffic_generator.CapturingTrafficGenerato
     #: Padding to add to the start of a line for python syntax compliance.
     _python_indentation: ClassVar[str] = " " * 4
 
-    def __init__(self, tg_node: Node, config: ScapyTrafficGeneratorConfig, **kwargs):
+    def __init__(self, tg_node: Node, config: TrafficGeneratorConfig, **kwargs):
         """Extend the constructor with Scapy TG specifics.
 
         Initializes both the traffic generator and the interactive shell used to handle Scapy
@@ -332,6 +332,10 @@ def setup(self, topology: Topology):
         self._shell.send_command("from scapy.all import *")
         self._shell.send_command("from scapy.contrib.lldp import *")
 
+    def teardown(self):
+        """Overrides :meth:`.traffic_generator.TrafficGenerator.teardown`."""
+        pass
+
     def close(self):
         """Overrides :meth:`.traffic_generator.TrafficGenerator.close`.
 
diff --git a/dts/framework/testbed_model/traffic_generator/traffic_generator.py b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
index 8f53b07daf..ea3075989d 100644
--- a/dts/framework/testbed_model/traffic_generator/traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
@@ -10,14 +10,10 @@
 
 from abc import ABC, abstractmethod
 
-from scapy.packet import Packet
-
 from framework.config.test_run import TrafficGeneratorConfig
 from framework.logger import DTSLogger, get_dts_logger
 from framework.testbed_model.node import Node
-from framework.testbed_model.port import Port
 from framework.testbed_model.topology import Topology
-from framework.utils import get_packet_summaries
 
 
 class TrafficGenerator(ABC):
@@ -54,46 +50,6 @@ def setup(self, topology: Topology):
 
     def teardown(self):
         """Teardown the traffic generator."""
-        self.close()
-
-    def send_packet(self, packet: Packet, port: Port) -> None:
-        """Send `packet` and block until it is fully sent.
-
-        Send `packet` on `port`, then wait until `packet` is fully sent.
-
-        Args:
-            packet: The packet to send.
-            port: The egress port on the TG node.
-        """
-        self.send_packets([packet], port)
-
-    def send_packets(self, packets: list[Packet], port: Port) -> None:
-        """Send `packets` and block until they are fully sent.
-
-        Send `packets` on `port`, then wait until `packets` are fully sent.
-
-        Args:
-            packets: The packets to send.
-            port: The egress port on the TG node.
-        """
-        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
-        self._logger.debug(get_packet_summaries(packets))
-        self._send_packets(packets, port)
-
-    @abstractmethod
-    def _send_packets(self, packets: list[Packet], port: Port) -> None:
-        """The implementation of :method:`send_packets`.
-
-        The subclasses must implement this method which sends `packets` on `port`.
-        The method should block until all `packets` are fully sent.
-
-        What fully sent means is defined by the traffic generator.
-        """
-
-    @property
-    def is_capturing(self) -> bool:
-        """This traffic generator can't capture traffic."""
-        return False
 
     @abstractmethod
     def close(self) -> None:
-- 
2.49.0

[PATCH v3 3/5] dts: add timeout override option to interactive shells

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Add an extra parameter to the interactive shell send command
to handle function to allow for a 1 time override of the send
command timeout.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/remote_session/interactive_shell.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/dts/framework/remote_session/interactive_shell.py b/dts/framework/remote_session/interactive_shell.py
index ba8489eafa..5331b8c7d1 100644
--- a/dts/framework/remote_session/interactive_shell.py
+++ b/dts/framework/remote_session/interactive_shell.py
@@ -177,7 +177,11 @@ def start_application(self, prompt: str | None = None) -> None:
         get_ctx().shell_pool.register_shell(self)
 
     def send_command(
-        self, command: str, prompt: str | None = None, skip_first_line: bool = False
+        self,
+        command: str,
+        prompt: str | None = None,
+        skip_first_line: bool = False,
+        added_timeout: int = 0,
     ) -> str:
         """Send `command` and get all output before the expected ending string.
 
@@ -195,6 +199,7 @@ def send_command(
             prompt: After sending the command, `send_command` will be expecting this string.
                 If :data:`None`, will use the class's default prompt.
             skip_first_line: Skip the first line when capturing the output.
+            added_timeout: additional duration for a given command, if needed.
 
         Returns:
             All output in the buffer before expected string.
@@ -213,6 +218,7 @@ def send_command(
         self._logger.info(f"Sending: '{command}'")
         if prompt is None:
             prompt = self._default_prompt
+        self._ssh_channel.settimeout(self._timeout + added_timeout)
         out: str = ""
         try:
             self._stdin.write(f"{command}{self._command_extra_chars}\n")
@@ -236,6 +242,7 @@ def send_command(
                 self._node.main_session.interactive_session.hostname
             ) from e
         finally:
+            self._ssh_channel.settimeout(self._timeout)
             self._logger.debug(f"Got output: {out}")
         return out
 
-- 
2.49.0

[PATCH v3 4/5] dts: add trex traffic generator to dts framework

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Implement the TREX traffic generator for use in the DTS framework. The
provided implementation leverages TREX's stateless API automation
library, via use of a Python shell. As such, limitation to specific TREX
versions may be needed. The DTS context has been modified to include
a performance traffic generator in addition to a functional traffic
generator.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/config/test_run.py              |  20 +-
 dts/framework/context.py                      |   2 +-
 dts/framework/test_run.py                     |  12 +-
 dts/framework/test_suite.py                   |   6 +-
 .../traffic_generator/__init__.py             |  22 +-
 .../testbed_model/traffic_generator/trex.py   | 258 ++++++++++++++++++
 6 files changed, 306 insertions(+), 14 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py

diff --git a/dts/framework/config/test_run.py b/dts/framework/config/test_run.py
index b6e4099eeb..3e09005338 100644
--- a/dts/framework/config/test_run.py
+++ b/dts/framework/config/test_run.py
@@ -396,6 +396,8 @@ class TrafficGeneratorType(str, Enum):
 
     #:
     SCAPY = "SCAPY"
+    #:
+    TREX = "TREX"
 
 
 class TrafficGeneratorConfig(FrozenModel):
@@ -404,6 +406,8 @@ class TrafficGeneratorConfig(FrozenModel):
     #: The traffic generator type the child class is required to define to be distinguished among
     #: others.
     type: TrafficGeneratorType
+    remote_path: PurePath
+    config: PurePath
 
 
 class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
@@ -412,8 +416,16 @@ class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
     type: Literal[TrafficGeneratorType.SCAPY]
 
 
+class TrexTrafficGeneratorConfig(TrafficGeneratorConfig):
+    """TREX traffic generator specific configuration."""
+
+    type: Literal[TrafficGeneratorType.TREX]
+
+
 #: A union type discriminating traffic generators by the `type` field.
-TrafficGeneratorConfigTypes = Annotated[ScapyTrafficGeneratorConfig, Field(discriminator="type")]
+TrafficGeneratorConfigTypes = Annotated[
+    TrexTrafficGeneratorConfig, ScapyTrafficGeneratorConfig, Field(discriminator="type")
+]
 
 #: Comma-separated list of logical cores to use. An empty string or ```any``` means use all lcores.
 LogicalCores = Annotated[
@@ -461,8 +473,10 @@ class TestRunConfiguration(FrozenModel):
 
     #: The DPDK configuration used to test.
     dpdk: DPDKConfiguration
-    #: The traffic generator configuration used to test.
-    traffic_generator: TrafficGeneratorConfigTypes
+    #: The traffic generator configuration used for functional tests.
+    func_traffic_generator: TrafficGeneratorConfig
+    #: The traffic generator configuration used for performance tests.
+    perf_traffic_generator: TrafficGeneratorConfig
     #: Whether to run performance tests.
     perf: bool
     #: Whether to run functional tests.
diff --git a/dts/framework/context.py b/dts/framework/context.py
index 4360bc8699..8caac935a5 100644
--- a/dts/framework/context.py
+++ b/dts/framework/context.py
@@ -16,7 +16,7 @@
 
 if TYPE_CHECKING:
     from framework.remote_session.dpdk import DPDKBuildEnvironment, DPDKRuntimeEnvironment
-    from framework.testbed_model.traffic_generator.traffic_generator import TrafficGenerator
+    from framework.testbed_model.traffic_generator import TrafficGenerator
 
 P = ParamSpec("P")
 
diff --git a/dts/framework/test_run.py b/dts/framework/test_run.py
index 10a5e1a6b8..ea8ca41414 100644
--- a/dts/framework/test_run.py
+++ b/dts/framework/test_run.py
@@ -204,10 +204,18 @@ def __init__(
 
         dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
         dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node, dpdk_build_env)
-        traffic_generator = create_traffic_generator(config.traffic_generator, tg_node)
+        if config.func:
+            traffic_generator = create_traffic_generator(config.func_traffic_generator, tg_node)
+        if config.perf:
+            traffic_generator = create_traffic_generator(config.perf_traffic_generator, tg_node)
 
         self.ctx = Context(
-            sut_node, tg_node, topology, dpdk_build_env, dpdk_runtime_env, traffic_generator
+            sut_node,
+            tg_node,
+            topology,
+            dpdk_build_env,
+            dpdk_runtime_env,
+            traffic_generator,
         )
         self.result = result
         self.selected_tests = list(self.config.filter_tests(tests_config))
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index e5fbadd1a1..75d7a6eb6c 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -254,11 +254,11 @@ def send_packets_and_capture(
             A list of received packets.
         """
         assert isinstance(
-            self._ctx.tg, CapturingTrafficGenerator
+            self._ctx.func_tg, CapturingTrafficGenerator
         ), "Cannot capture with a non-capturing traffic generator"
         # TODO: implement @requires for types of traffic generator
         packets = self._adjust_addresses(packets)
-        return self._ctx.tg.send_packets_and_capture(
+        return self._ctx.func_tg.send_packets_and_capture(
             packets,
             self._ctx.topology.tg_port_egress,
             self._ctx.topology.tg_port_ingress,
@@ -276,7 +276,7 @@ def send_packets(
             packets: Packets to send.
         """
         packets = self._adjust_addresses(packets)
-        self._ctx.tg.send_packets(packets, self._ctx.topology.tg_port_egress)
+        self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
     def get_expected_packets(
         self,
diff --git a/dts/framework/testbed_model/traffic_generator/__init__.py b/dts/framework/testbed_model/traffic_generator/__init__.py
index 2a259a6e6c..53125995cd 100644
--- a/dts/framework/testbed_model/traffic_generator/__init__.py
+++ b/dts/framework/testbed_model/traffic_generator/__init__.py
@@ -14,17 +14,27 @@
 and a capturing traffic generator is required.
 """
 
-from framework.config.test_run import ScapyTrafficGeneratorConfig, TrafficGeneratorConfig
+from framework.config.test_run import (
+    ScapyTrafficGeneratorConfig as ScapyTrafficGeneratorConfig,
+)
+from framework.config.test_run import (
+    TrafficGeneratorConfig,
+    TrafficGeneratorType,
+)
+from framework.config.test_run import (
+    TrexTrafficGeneratorConfig as TrexTrafficGeneratorConfig,
+)
 from framework.exception import ConfigurationError
 from framework.testbed_model.node import Node
 
-from .capturing_traffic_generator import CapturingTrafficGenerator
 from .scapy import ScapyTrafficGenerator
+from .traffic_generator import TrafficGenerator
+from .trex import TrexTrafficGenerator
 
 
 def create_traffic_generator(
     traffic_generator_config: TrafficGeneratorConfig, node: Node
-) -> CapturingTrafficGenerator:
+) -> TrafficGenerator:
     """The factory function for creating traffic generator objects from the test run configuration.
 
     Args:
@@ -37,8 +47,10 @@ def create_traffic_generator(
     Raises:
         ConfigurationError: If an unknown traffic generator has been setup.
     """
-    match traffic_generator_config:
-        case ScapyTrafficGeneratorConfig():
+    match traffic_generator_config.type:
+        case TrafficGeneratorType.SCAPY:
             return ScapyTrafficGenerator(node, traffic_generator_config, privileged=True)
+        case TrafficGeneratorType.TREX:
+            return TrexTrafficGenerator(node, traffic_generator_config, privileged=True)
         case _:
             raise ConfigurationError(f"Unknown traffic generator: {traffic_generator_config.type}")
diff --git a/dts/framework/testbed_model/traffic_generator/trex.py b/dts/framework/testbed_model/traffic_generator/trex.py
new file mode 100644
index 0000000000..1ba7d6bbbd
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/trex.py
@@ -0,0 +1,258 @@
+"""Implementation for TREX performance traffic generator."""
+
+import time
+from enum import Flag, auto
+from typing import Any, ClassVar
+
+from scapy.packet import Packet
+
+from framework.config.node import NodeConfiguration
+from framework.config.test_run import TrafficGeneratorConfig
+from framework.context import get_ctx
+from framework.exception import SSHTimeoutError
+from framework.remote_session.python_shell import PythonShell
+from framework.testbed_model.node import Node, create_session
+from framework.testbed_model.os_session import OSSession
+from framework.testbed_model.topology import Topology
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
+
+
+class TrexStatelessTXModes(Flag):
+    """Flags indicating TREX instance's current transmission mode."""
+
+    CONTINUOUS = auto()
+    SINGLE_BURST = auto()
+    MULTI_BURST = auto()
+
+
+class TrexTrafficGenerator(PerformanceTrafficGenerator):
+    """TREX traffic generator.
+
+    This implementation leverages the stateless API library provided in the TREX installation.
+
+    Attributes:
+        stl_client_name: The name of the stateless client used in the stateless API.
+        packet_stream_name: The name of the stateless packet stream used in the stateless API.
+    """
+
+    _os_session: OSSession
+    _server_remote_session: Any
+
+    _tg_config: TrafficGeneratorConfig
+    _node_config: NodeConfiguration
+
+    _shell: PythonShell
+    _python_indentation: ClassVar[str] = " " * 4
+
+    stl_client_name: ClassVar[str] = "client"
+    packet_stream_name: ClassVar[str] = "stream"
+
+    _streaming_mode: TrexStatelessTXModes = TrexStatelessTXModes.CONTINUOUS
+
+    tg_cores: int = 10
+
+    def __init__(self, tg_node: Node, config: TrafficGeneratorConfig, **kwargs) -> None:
+        """Initialize the TREX server.
+
+        Initializes needed OS sessions for the creation of the TREX server process.
+
+        Attributes:
+            tg_node: TG node the TREX instance is operating on.
+            config: Traffic generator config provided for TREX instance.
+        """
+        super().__init__(tg_node=tg_node, config=config, **kwargs)
+        self._tg_node_config = tg_node.config
+        self._tg_config = config
+
+        self._os_session = create_session(self._tg_node.config, "TREX", self._logger)
+        self._server_remote_session = self._os_session.remote_session
+
+        self._shell = PythonShell(self._tg_node, "TREX-client", privileged=True)
+
+    def setup(self, topology: Topology):
+        """Initialize and start a TREX server process."""
+        super().setup(get_ctx().topology)
+        # Start TREX server process.
+        try:
+            self._logger.info("Starting TREX server process: sending 20 second sleep.")
+            server_command = [
+                f"cd {self._tg_config.remote_path};",
+                self._os_session._get_privileged_command(
+                    f"screen -d -m ./t-rex-64 --cfg {self._tg_config.config} -c {self.tg_cores} -i"
+                ),
+            ]
+            privileged_command = " ".join(server_command)
+            self._logger.info(f"Sending: '{privileged_command}")
+            self._server_remote_session.session.run(privileged_command)
+            time.sleep(20)
+
+        except SSHTimeoutError as e:
+            self._logger.exception("Failed to start TREX server process.", e)
+
+        # Start Python shell.
+        self._shell.start_application()
+        self._shell.send_command("import os")
+        self._shell.send_command(
+            f"os.chdir('{self._tg_config.remote_path}/automation/trex_control_plane/interactive')"
+        )
+
+        # Import stateless API components.
+        imports = [
+            "import trex",
+            "import trex.stl",
+            "import trex.stl.trex_stl_client",
+            "import trex.stl.trex_stl_streams",
+            "import trex.stl.trex_stl_packet_builder_scapy",
+            "from scapy.layers.l2 import Ether",
+            "from scapy.layers.inet import IP",
+            "from scapy.packet import Raw",
+        ]
+        self._shell.send_command("\n".join(imports))
+
+        stateless_client = [
+            f"{self.stl_client_name} = trex.stl.trex_stl_client.STLClient(",
+            f"username='{self._tg_node_config.user}',",
+            "server='127.0.0.1',",
+            ")",
+        ]
+
+        self._shell.send_command(f"\n{self._python_indentation}".join(stateless_client))
+        self._shell.send_command(f"{self.stl_client_name}.connect()")
+
+    def teardown(self) -> None:
+        """Teardown the TREX server and stateless implementation.
+
+        close the TREX server process, and stop the Python shell.
+
+        Attributes:
+            ports: Associated ports used by the TREX instance.
+        """
+        super().teardown()
+        self._os_session.send_command("pkill t-rex-64", privileged=True)
+        self.close()
+
+    def calculate_traffic_and_stats(
+        self, packet: Packet, send_mpps: int, duration: float
+    ) -> PerformanceTrafficStats:
+        """Calculate the traffic statistics, using provided TG output.
+
+        Takes in the statistics output provided by the stateless API implementation, and collects
+        them into a performance statistics data structure.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+            send_mpps: the MPPS send rate.
+            duration: The duration of the test.
+        """
+        # Convert to a dictionary.
+        stats_output = eval(self._generate_traffic(packet, send_mpps, duration))
+
+        global_output = stats_output.get("global", "ERROR - DATA NOT FOUND")
+
+        self._logger.info(f"The global stats for the current set of params are: {global_output}")
+
+        return PerformanceTrafficStats(
+            frame_size=len(packet),
+            tx_pps=global_output.get("tx_pps", "ERROR - tx_pps NOT FOUND"),
+            tx_cps=global_output.get("tx_cps", "ERROR - tx_cps NOT FOUND"),
+            tx_bps=global_output.get("tx_bps", "ERROR - tx_bps NOT FOUND"),
+            rx_pps=global_output.get("rx_pps", "ERROR - rx_pps NOT FOUND"),
+            rx_bps=global_output.get("rx_bps", "ERROR - rx_bps NOT FOUND"),
+        )
+
+    def _generate_traffic(self, packet: Packet, send_mpps: int, duration: float) -> str:
+        """Generate traffic using provided packet.
+
+        Uses the provided packet to generate traffic for the provided duration.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+            send_mpps: MPPS send rate.
+            duration: The duration of the test being performed.
+
+        Returns:
+            a string output of statistics provided by the traffic generator.
+        """
+        self._create_packet_stream(packet)
+        self._setup_trex_client()
+
+        stats = self._send_traffic_and_get_stats(send_mpps, duration)
+
+        return stats
+
+    def _setup_trex_client(self) -> None:
+        """Create trex client and connect to the server process."""
+        # Prepare TREX client for next performance test.
+        procedure = [
+            f"{self.stl_client_name}.connect()",
+            f"{self.stl_client_name}.reset(ports = [0, 1])",
+            f"{self.stl_client_name}.clear_stats()",
+            f"{self.stl_client_name}.add_streams({self.packet_stream_name}, ports=[0, 1])",
+        ]
+
+        for command in procedure:
+            self._shell.send_command(command)
+
+    def _create_packet_stream(self, packet: Packet) -> None:
+        """Create TREX packet stream with the given packet.
+
+        Attributes:
+            packet: The packet being used for the performance test.
+
+        """
+        streaming_mode = ""
+        if self._streaming_mode == TrexStatelessTXModes.CONTINUOUS:
+            streaming_mode = "STLTXCont"
+        elif self._streaming_mode == TrexStatelessTXModes.SINGLE_BURST:
+            streaming_mode = "STLTXSingleBurst"
+        elif self._streaming_mode == TrexStatelessTXModes.MULTI_BURST:
+            streaming_mode = "STLTXMultiBurst"
+
+        # Create the tx packet on the TG shell
+        self._shell.send_command(f"packet={packet.command()}")
+
+        packet_stream = [
+            f"{self.packet_stream_name} = trex.stl.trex_stl_streams.STLStream(",
+            f"name='Test_{len(packet)}_bytes',",
+            "packet=trex.stl.trex_stl_packet_builder_scapy.STLPktBuilder(pkt=packet),",
+            f"mode=trex.stl.trex_stl_streams.{streaming_mode}(percentage=100),",
+            ")",
+        ]
+        self._shell.send_command("\n".join(packet_stream))
+
+    def _send_traffic_and_get_stats(self, send_mpps: float, duration: float) -> str:
+        """Send traffic and get TG Rx stats.
+
+        Sends traffic from the TREX client's ports for the given duration.
+        When the traffic sending duration has passed, collect the aggregate
+        statistics and return TREX's global stats as a string.
+
+        Attributes:
+            send_mpps: The millions of packets per second for TREX to send from each port.
+            duration: The traffic generation duration.
+        """
+        mpps_send_rate = f"{send_mpps}mpps"
+
+        self._shell.send_command(f"""{self.stl_client_name}.start(ports=[0, 1],
+        mult = '{mpps_send_rate}',
+        duration = {duration})""")
+
+        time.sleep(duration)
+
+        stats = self._shell.send_command(
+            f"{self.stl_client_name}.get_stats(ports=[0, 1])", skip_first_line=True
+        )
+
+        self._shell.send_command(f"{self.stl_client_name}.stop(ports=[0, 1])")
+
+        return stats
+
+    def close(self) -> None:
+        """Overrides :meth:`.traffic_generator.TrafficGenerator.close`.
+
+        Stops the traffic generator and sniffer shells.
+        """
+        self._shell.close()
-- 
2.49.0

[PATCH v3 5/5] dts: add performance test functions to test suite API

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Provide functional performance method to run performance tests using a
user-supplied performance traffic generator. The single core performance
test is included, with some basic statistics checks verifying TG packet
transmission rates.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/configurations/tests_config.example.yaml | 12 +++++++
 dts/framework/test_suite.py                  | 34 ++++++++++++++++++--
 2 files changed, 43 insertions(+), 3 deletions(-)

diff --git a/dts/configurations/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
index c011ac0588..c56951b2b0 100644
--- a/dts/configurations/tests_config.example.yaml
+++ b/dts/configurations/tests_config.example.yaml
@@ -3,3 +3,15 @@
 # Define the custom test suite configurations
 hello_world:
   msg: A custom hello world to you!
+single_core_forward_perf:
+  test_parameters:
+    - frame_size: 64
+      num_descriptors: 128
+      expected_mpps: 30.1234
+    - frame_size: 64
+      num_descriptors: 128
+      expected_mpps: 30.2341
+    - frame_size: 512
+      num_descriptors: 128
+      expected_mpps: 19.453376
+  delta_tolerance: 0.05
\ No newline at end of file
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index 75d7a6eb6c..4044c85d4d 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -38,6 +38,10 @@
     CapturingTrafficGenerator,
     PacketFilteringConfig,
 )
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
 
 from .exception import ConfigurationError, InternalError, TestCaseVerifyError
 from .logger import DTSLogger, get_dts_logger
@@ -254,11 +258,11 @@ def send_packets_and_capture(
             A list of received packets.
         """
         assert isinstance(
-            self._ctx.func_tg, CapturingTrafficGenerator
+            self._ctx.tg, CapturingTrafficGenerator
         ), "Cannot capture with a non-capturing traffic generator"
         # TODO: implement @requires for types of traffic generator
         packets = self._adjust_addresses(packets)
-        return self._ctx.func_tg.send_packets_and_capture(
+        return self._ctx.tg.send_packets_and_capture(
             packets,
             self._ctx.topology.tg_port_egress,
             self._ctx.topology.tg_port_ingress,
@@ -266,6 +270,27 @@ def send_packets_and_capture(
             duration,
         )
 
+    def assess_performance_by_packet(
+        self, packet: Packet, send_mpps: int, duration: int = 60
+    ) -> PerformanceTrafficStats:
+        """Send a given packet for a given duration and assess basic performance statistics.
+
+        Send `packet` and assess NIC performance for a given duration, corresponding to the test
+        suite's given topology.
+
+        Args:
+            packet: The packet to send.
+            send_mpps: The millions packets per second send rate.
+            duration: Performance test duration (in seconds).
+
+        Returns:
+            Performance statistics of the generated test.
+        """
+        assert isinstance(
+            self._ctx.tg, PerformanceTrafficGenerator
+        ), "Cannot run performance tests on non-performance traffic generator."
+        return self._ctx.tg.calculate_traffic_and_stats(packet, send_mpps, duration)
+
     def send_packets(
         self,
         packets: list[Packet],
@@ -275,8 +300,11 @@ def send_packets(
         Args:
             packets: Packets to send.
         """
+        assert isinstance(
+            self._ctx.tg, CapturingTrafficGenerator
+        ), "Cannot run performance tests on non-capturing traffic generator."
         packets = self._adjust_addresses(packets)
-        self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
+        self._ctx.tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
     def get_expected_packets(
         self,
-- 
2.49.0

Re: [PATCH v3 1/5] dts: rework config module to support perf TGs

[Luca Vizzarro]

Hi Patrick and Nick,

I am unsure what happened with this commit. The commit subject doesn't 
match what's happening here. Mentions a rework of the config module, 
which is not. Description seems ok.

Another issue I see is that some changes don't make a lot logical sense 
as they stand here. The changes in test_run.yaml, don't really belong 
here, and could be considered breaking. I am assuming they were 
improperly re-ordered between commits. Similarly I don't see appropriate 
doc changes to reflect the change in the configuration examples path.

On 02/07/2025 06:21, Patrick Robb wrote:
> --- a/dts/test_run.example.yaml
> +++ b/dts/configurations/test_run.example.yaml
> @@ -1,8 +1,3 @@
> -# SPDX-License-Identifier: BSD-3-Clause
> -# Copyright 2022-2023 The DPDK contributors
> -# Copyright 2023 Arm Limited
> -
> -# Define the test run environment

And removing this line above, would cause in sphinx to complain as the 
example rendering relies on this line.

>   dpdk:
>     lcores: "" # use all available logical cores (Skips first core)
>     memory_channels: 4 # tells DPDK to use 4 memory channels
> @@ -23,8 +18,14 @@ dpdk:
>       # in a subdirectory of DPDK tree root directory. Otherwise, will be using the `build_options`
>       # to build the DPDK from source. Either `precompiled_build_dir` or `build_options` can be
>       # defined, but not both.

Re: [PATCH v3 2/5] dts: rework traffic generator inheritance structure

[Luca Vizzarro]

On 02/07/2025 06:21, Patrick Robb wrote:
> +++ b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
> @@ -0,0 +1,63 @@
> <snip>
> +
> +@dataclass(slots=True)
> +class PerformanceTrafficStats(ABC):

Why is this inheriting ABC? I don't see why a dataclass would be 
abstract given it's even missing abstract methods.

> <snip>
> +class PerformanceTrafficGenerator(TrafficGenerator):
> +    """An abstract base class for all performance-oriented traffic generators.
> +
> +    Provides an intermediary interface for performance-based traffic generator.
> +    """
> +
> +    @abstractmethod
> +    def calculate_traffic_and_stats(
> +        self,
> +        packet: Packet,
> +        send_mpps: int,
> +        duration: float,
> +    ) -> PerformanceTrafficStats:
> +        """Send packet traffic and acquire associated statistics.
> +
> +        Args:
> +        packet: The packet to send.
> +        send_mpps: The millions packets per second send rate.
> +        duration: Performance test duration (in seconds).

bad indentation for args.

> diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py b/dts/framework/testbed_model/traffic_generator/scapy.py
> index e21ba4ed96..602b93d473 100644
> --- a/dts/framework/testbed_model/traffic_generator/scapy.py
> +++ b/dts/framework/testbed_model/traffic_generator/scapy.py
> @@ -26,7 +26,7 @@
>   from scapy.packet import Packet
>   
>   from framework.config.node import OS
> -from framework.config.test_run import ScapyTrafficGeneratorConfig
> +from framework.config.test_run import TrafficGeneratorConfig

Making the Scapy class use a more generic config class looks like 
breaking behaviour. Why is this happening?

> @@ -332,6 +332,10 @@ def setup(self, topology: Topology):
>           self._shell.send_command("from scapy.all import *")
>           self._shell.send_command("from scapy.contrib.lldp import *")
>   
> +    def teardown(self):
> +        """Overrides :meth:`.traffic_generator.TrafficGenerator.teardown`."""
> +        pass

nit: `pass` is not needed here, because the docstring already completes 
the method as noop.

> <snip>
> diff --git a/dts/framework/testbed_model/traffic_generator/traffic_generator.py b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
> index 8f53b07daf..ea3075989d 100644
> --- a/dts/framework/testbed_model/traffic_generator/traffic_generator.py
> +++ b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
> @@ -10,14 +10,10 @@
> <snip>
>   
>   class TrafficGenerator(ABC):
> @@ -54,46 +50,6 @@ def setup(self, topology: Topology):
>   
>       def teardown(self):
>           """Teardown the traffic generator."""
> -        self.close()

how come are we removing the default close behaviour? I don't see it 
being replaced appropriately.

> <snip>
> -    @property
> -    def is_capturing(self) -> bool:
> -        """This traffic generator can't capture traffic."""
> -        return False
Why is the above property being removed?
>   
>       @abstractmethod
>       def close(self) -> None:

Re: [PATCH v3 3/5] dts: add timeout override option to interactive shells

[Luca Vizzarro]

Reviewed-by: Luca Vizzarro <luca.vizzarro@arm.com>

Re: [PATCH v3 4/5] dts: add trex traffic generator to dts framework

[Luca Vizzarro]

On 02/07/2025 06:21, Patrick Robb wrote:
>   class TrafficGeneratorConfig(FrozenModel):
> @@ -404,6 +406,8 @@ class TrafficGeneratorConfig(FrozenModel):
>       #: The traffic generator type the child class is required to define to be distinguished among
>       #: others.
>       type: TrafficGeneratorType
> +    remote_path: PurePath
> +    config: PurePath

What's the reason behind these? And why are we making these mandatory 
for all of the traffic generator configurations? Given Scapy hasn't 
needed them so far, I reckon these shouldn't be here.

>       #: The DPDK configuration used to test.
>       dpdk: DPDKConfiguration
> -    #: The traffic generator configuration used to test.
> -    traffic_generator: TrafficGeneratorConfigTypes
> +    #: The traffic generator configuration used for functional tests.
> +    func_traffic_generator: TrafficGeneratorConfig
> +    #: The traffic generator configuration used for performance tests.
> +    perf_traffic_generator: TrafficGeneratorConfig

This is starting to look like it should be optional and not mandatory. 
Should have a choice not to supply a perf traffic generator, same for a 
functional one. If I care only about perf tests, I shouldn't be needing 
to supply functional ones.

> diff --git a/dts/framework/test_run.py b/dts/framework/test_run.py
> index 10a5e1a6b8..ea8ca41414 100644
> --- a/dts/framework/test_run.py
> +++ b/dts/framework/test_run.py
> @@ -204,10 +204,18 @@ def __init__(
>   
>           dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
>           dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node, dpdk_build_env)
> -        traffic_generator = create_traffic_generator(config.traffic_generator, tg_node)
> +        if config.func:
> +            traffic_generator = create_traffic_generator(config.func_traffic_generator, tg_node)
> +        if config.perf:
> +            traffic_generator = create_traffic_generator(config.perf_traffic_generator, tg_node)

This is insinuating we can only have a functional or a performance 
traffic generator, which does not respect the expectations of the 
configuration. Moreover, func and perf are not mutually exclusive, if we 
have both set to True this looks like it will wreak havoc. Looks like a 
bug to me.

> diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
> index e5fbadd1a1..75d7a6eb6c 100644
> --- a/dts/framework/test_suite.py
> +++ b/dts/framework/test_suite.py
> @@ -254,11 +254,11 @@ def send_packets_and_capture(
>               A list of received packets.
>           """
>           assert isinstance(
> -            self._ctx.tg, CapturingTrafficGenerator
> +            self._ctx.func_tg, CapturingTrafficGenerator

I am evermore confused, I don't see any func_tg in Context.

> diff --git a/dts/framework/testbed_model/traffic_generator/__init__.py b/dts/framework/testbed_model/traffic_generator/__init__.py
> index 2a259a6e6c..53125995cd 100644
> @@ -37,8 +47,10 @@ def create_traffic_generator(
>       Raises:
>           ConfigurationError: If an unknown traffic generator has been setup.
>       """
> -    match traffic_generator_config:
> -        case ScapyTrafficGeneratorConfig():
> +    match traffic_generator_config.type:
> +        case TrafficGeneratorType.SCAPY:
>               return ScapyTrafficGenerator(node, traffic_generator_config, privileged=True)
> +        case TrafficGeneratorType.TREX:
> +            return TrexTrafficGenerator(node, traffic_generator_config, privileged=True)
>           case _:
>               raise ConfigurationError(f"Unknown traffic generator: {traffic_generator_config.type}")

This change defeats the entire purpose of the match case. This was 
originally doing two things:
- matching traffic generator by their own config class
- casting the config class (this is not happening anymore)
The traffic generator classes should be able to access their own 
configurations directly, this is basically breaking the whole purpose of 
mypy.

> diff --git a/dts/framework/testbed_model/traffic_generator/trex.py b/dts/framework/testbed_model/traffic_generator/trex.py
> new file mode 100644
> index 0000000000..1ba7d6bbbd
> --- /dev/null
> +++ b/dts/framework/testbed_model/traffic_generator/trex.py
> @@ -0,0 +1,258 @@
> +"""Implementation for TREX performance traffic generator."""

Missing SPDX License Identifier and copyrights. For such a big file, I'd 
have expected a description for the implementation and how to configure 
TRex in DTS.


> +class TrexStatelessTXModes(Flag):
> +    """Flags indicating TREX instance's current transmission mode."""
> +
> +    CONTINUOUS = auto()
> +    SINGLE_BURST = auto()
> +    MULTI_BURST = auto()

Missing docstrings.

> +class TrexTrafficGenerator(PerformanceTrafficGenerator):
> +    """TREX traffic generator.
> +
> +    This implementation leverages the stateless API library provided in the TREX installation.
> +
> +    Attributes:
> +        stl_client_name: The name of the stateless client used in the stateless API.
> +        packet_stream_name: The name of the stateless packet stream used in the stateless API.
> +    """
> +
> +    _os_session: OSSession
> +    _server_remote_session: Any

Why are we resorting to Any?

> +
> +    _tg_config: TrafficGeneratorConfig
> +    _node_config: NodeConfiguration
> +
> +    _shell: PythonShell
> +    _python_indentation: ClassVar[str] = " " * 4
> +
> +    stl_client_name: ClassVar[str] = "client"
> +    packet_stream_name: ClassVar[str] = "stream"
> +
> +    _streaming_mode: TrexStatelessTXModes = TrexStatelessTXModes.CONTINUOUS
> +
> +    tg_cores: int = 10

This is not documented in the docstring.

> +
> +    def __init__(self, tg_node: Node, config: TrafficGeneratorConfig, **kwargs) -> None:
> +        """Initialize the TREX server.
> +
> +        Initializes needed OS sessions for the creation of the TREX server process.
> +
> +        Attributes:
> +            tg_node: TG node the TREX instance is operating on.
> +            config: Traffic generator config provided for TREX instance.
> +        """

I think the above is meant to be Args.

> +        super().__init__(tg_node=tg_node, config=config, **kwargs)
> +        self._tg_node_config = tg_node.config
> +        self._tg_config = config
> +
> +        self._os_session = create_session(self._tg_node.config, "TREX", self._logger)
> +        self._server_remote_session = self._os_session.remote_session
> +
> +        self._shell = PythonShell(self._tg_node, "TREX-client", privileged=True)
> +
> +    def setup(self, topology: Topology):
> +        """Initialize and start a TREX server process."""
> +        super().setup(get_ctx().topology)

Why are we calling topology from get_ctx() if we are accepting it from 
the arguments?

> +        # Start TREX server process.
> +        try:
> +            self._logger.info("Starting TREX server process: sending 20 second sleep.")
> +            server_command = [
> +                f"cd {self._tg_config.remote_path};",
> +                self._os_session._get_privileged_command(
> +                    f"screen -d -m ./t-rex-64 --cfg {self._tg_config.config} -c {self.tg_cores} -i"
> +                ),
> +            ]
> +            privileged_command = " ".join(server_command)
> +            self._logger.info(f"Sending: '{privileged_command}")
> +            self._server_remote_session.session.run(privileged_command)
> +            time.sleep(20)

Looks kind of hacky and slow, is there no way to determine if the 
process is ready? Would something like the BlockingDPDKApp class work 
here? (Of course in a generic way, like the patches I am about to send)

> +
> +        except SSHTimeoutError as e:
> +            self._logger.exception("Failed to start TREX server process.", e)
> +
> +        # Start Python shell.
> +        self._shell.start_application()
> +        self._shell.send_command("import os")
> +        self._shell.send_command(
> +            f"os.chdir('{self._tg_config.remote_path}/automation/trex_control_plane/interactive')"
> +        )

I would properly join the path first use PurePath.

> +
> +        # Import stateless API components.
> +        imports = [
> +            "import trex",
> +            "import trex.stl",
> +            "import trex.stl.trex_stl_client",
> +            "import trex.stl.trex_stl_streams",
> +            "import trex.stl.trex_stl_packet_builder_scapy",
> +            "from scapy.layers.l2 import Ether",
> +            "from scapy.layers.inet import IP",
> +            "from scapy.packet import Raw",
> +        ]
> +        self._shell.send_command("\n".join(imports))
> +
> +        stateless_client = [
> +            f"{self.stl_client_name} = trex.stl.trex_stl_client.STLClient(",
> +            f"username='{self._tg_node_config.user}',",
> +            "server='127.0.0.1',",
> +            ")",
> +        ]
> +
> +        self._shell.send_command(f"\n{self._python_indentation}".join(stateless_client))
> +        self._shell.send_command(f"{self.stl_client_name}.connect()")
> +
> +    def teardown(self) -> None:
> +        """Teardown the TREX server and stateless implementation.
> +
> +        close the TREX server process, and stop the Python shell.
Missed capitalisation.
> +
> +        Attributes:
> +            ports: Associated ports used by the TREX instance.
> +        """

What attributes?

> +        super().teardown()
> +        self._os_session.send_command("pkill t-rex-64", privileged=True)
> +        self.close()
> +
> +    def calculate_traffic_and_stats(
> +        self, packet: Packet, send_mpps: int, duration: float
> +    ) -> PerformanceTrafficStats:
> +        """Calculate the traffic statistics, using provided TG output.
> +
> +        Takes in the statistics output provided by the stateless API implementation, and collects
> +        them into a performance statistics data structure.
> +
> +        Attributes:
> +            packet: The packet being used for the performance test.
> +            send_mpps: the MPPS send rate.
> +            duration: The duration of the test.
> +        """

Args, not Attributes.

> +        # Convert to a dictionary.
> +        stats_output = eval(self._generate_traffic(packet, send_mpps, duration))

I am not really pleased with the idea of injecting arbitrary code, this 
screams security issue. Is there no better way to handle this? We want 
to parse and validate output, not just evaluate it blindly. Could we get 
this as JSON and let a Pydantic model naturally parse it and validate it 
safely?

> +
> +        global_output = stats_output.get("global", "ERROR - DATA NOT FOUND")
> +
> +        self._logger.info(f"The global stats for the current set of params are: {global_output}")
> +
> +        return PerformanceTrafficStats(
> +            frame_size=len(packet),
> +            tx_pps=global_output.get("tx_pps", "ERROR - tx_pps NOT FOUND"),
> +            tx_cps=global_output.get("tx_cps", "ERROR - tx_cps NOT FOUND"),
> +            tx_bps=global_output.get("tx_bps", "ERROR - tx_bps NOT FOUND"),
> +            rx_pps=global_output.get("rx_pps", "ERROR - rx_pps NOT FOUND"),
> +            rx_bps=global_output.get("rx_bps", "ERROR - rx_bps NOT FOUND"),
> +        )

Otherwise, PerformanceTrafficStats could just re-use the TextParser 
class for this purpose as well.

> +
> +    def _generate_traffic(self, packet: Packet, send_mpps: int, duration: float) -> str:
> +        """Generate traffic using provided packet.
> +
> +        Uses the provided packet to generate traffic for the provided duration.
> +
> +        Attributes:
> +            packet: The packet being used for the performance test.
> +            send_mpps: MPPS send rate.
> +            duration: The duration of the test being performed.

Args, not Attributes.

> +
> +        Returns:
> +            a string output of statistics provided by the traffic generator.
Missed capitalisation.
> +        """
> +        self._create_packet_stream(packet)
> +        self._setup_trex_client()
> +
> +        stats = self._send_traffic_and_get_stats(send_mpps, duration)
> +
> +        return stats
> +
> +    def _setup_trex_client(self) -> None:
> +        """Create trex client and connect to the server process."""
> +        # Prepare TREX client for next performance test.
> +        procedure = [
> +            f"{self.stl_client_name}.connect()",
> +            f"{self.stl_client_name}.reset(ports = [0, 1])",
> +            f"{self.stl_client_name}.clear_stats()",
> +            f"{self.stl_client_name}.add_streams({self.packet_stream_name}, ports=[0, 1])",
> +        ]
> +
> +        for command in procedure:
> +            self._shell.send_command(command)
> +
> +    def _create_packet_stream(self, packet: Packet) -> None:
> +        """Create TREX packet stream with the given packet.
> +
> +        Attributes:
> +            packet: The packet being used for the performance test.
> +
empty line
> +        """
> +        streaming_mode = ""
> +        if self._streaming_mode == TrexStatelessTXModes.CONTINUOUS:
> +            streaming_mode = "STLTXCont"
> +        elif self._streaming_mode == TrexStatelessTXModes.SINGLE_BURST:
> +            streaming_mode = "STLTXSingleBurst"
> +        elif self._streaming_mode == TrexStatelessTXModes.MULTI_BURST:
> +            streaming_mode = "STLTXMultiBurst"

TrexStatelessTXModes could just be a StrEnum with these as values.

> +
> +        # Create the tx packet on the TG shell
> +        self._shell.send_command(f"packet={packet.command()}")
> +
> +        packet_stream = [
> +            f"{self.packet_stream_name} = trex.stl.trex_stl_streams.STLStream(",
> +            f"name='Test_{len(packet)}_bytes',",
> +            "packet=trex.stl.trex_stl_packet_builder_scapy.STLPktBuilder(pkt=packet),",
> +            f"mode=trex.stl.trex_stl_streams.{streaming_mode}(percentage=100),",
> +            ")",
> +        ]
> +        self._shell.send_command("\n".join(packet_stream))
> +
> +    def _send_traffic_and_get_stats(self, send_mpps: float, duration: float) -> str:
> +        """Send traffic and get TG Rx stats.
> +
> +        Sends traffic from the TREX client's ports for the given duration.
> +        When the traffic sending duration has passed, collect the aggregate
> +        statistics and return TREX's global stats as a string.
> +
> +        Attributes:
> +            send_mpps: The millions of packets per second for TREX to send from each port.
> +            duration: The traffic generation duration.
Args, not Attributes.
> +        """
> +        mpps_send_rate = f"{send_mpps}mpps"
> +
> +        self._shell.send_command(f"""{self.stl_client_name}.start(ports=[0, 1],
> +        mult = '{mpps_send_rate}',
> +        duration = {duration})""")
bad formatting.
> +
> +        time.sleep(duration)
> +
> +        stats = self._shell.send_command(
> +            f"{self.stl_client_name}.get_stats(ports=[0, 1])", skip_first_line=True
> +        )
> +
> +        self._shell.send_command(f"{self.stl_client_name}.stop(ports=[0, 1])")
> +
> +        return stats
> +
> +    def close(self) -> None:
> +        """Overrides :meth:`.traffic_generator.TrafficGenerator.close`.
> +
> +        Stops the traffic generator and sniffer shells.
> +        """
> +        self._shell.close()

Re: [PATCH v3 5/5] dts: add performance test functions to test suite API

[Luca Vizzarro]

On 02/07/2025 06:21, Patrick Robb wrote:
> From: Nicholas Pratte <npratte@iol.unh.edu>
> 
> Provide functional performance method to run performance tests using a
> user-supplied performance traffic generator. The single core performance
> test is included, with some basic statistics checks verifying TG packet
is the test missing?
> transmission rates.
> 
> Bugzilla ID: 1697
> Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
> Signed-off-by: Patrick Robb <probb@iol.unh.edu>
> Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
> ---
>   dts/configurations/tests_config.example.yaml | 12 +++++++
>   dts/framework/test_suite.py                  | 34 ++++++++++++++++++--
>   2 files changed, 43 insertions(+), 3 deletions(-)
> 
> diff --git a/dts/configurations/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
> index c011ac0588..c56951b2b0 100644
> --- a/dts/configurations/tests_config.example.yaml
> +++ b/dts/configurations/tests_config.example.yaml
> @@ -3,3 +3,15 @@
>   # Define the custom test suite configurations
>   hello_world:
>     msg: A custom hello world to you!
> +single_core_forward_perf:
> +  test_parameters:
> +    - frame_size: 64
> +      num_descriptors: 128
> +      expected_mpps: 30.1234
> +    - frame_size: 64
> +      num_descriptors: 128
> +      expected_mpps: 30.2341
> +    - frame_size: 512
> +      num_descriptors: 128
> +      expected_mpps: 19.453376
> +  delta_tolerance: 0.05

where does this come from? Is it meant for the missing test suite?

> \ No newline at end of file
> diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
> index 75d7a6eb6c..4044c85d4d 100644
> --- a/dts/framework/test_suite.py
> +++ b/dts/framework/test_suite.py
> @@ -38,6 +38,10 @@
>       CapturingTrafficGenerator,
>       PacketFilteringConfig,
>   )
> +from framework.testbed_model.traffic_generator.performance_traffic_generator import (
> +    PerformanceTrafficGenerator,
> +    PerformanceTrafficStats,
> +)
>   
>   from .exception import ConfigurationError, InternalError, TestCaseVerifyError
>   from .logger import DTSLogger, get_dts_logger
> @@ -254,11 +258,11 @@ def send_packets_and_capture(
>               A list of received packets.
>           """
>           assert isinstance(
> -            self._ctx.func_tg, CapturingTrafficGenerator
> +            self._ctx.tg, CapturingTrafficGenerator

This looks like it's fixing an issue I noted earlier in the review, 
should be squashed.

>           ), "Cannot capture with a non-capturing traffic generator"
>           # TODO: implement @requires for types of traffic generator
>           packets = self._adjust_addresses(packets)
> -        return self._ctx.func_tg.send_packets_and_capture(
> +        return self._ctx.tg.send_packets_and_capture(
>               packets,
>               self._ctx.topology.tg_port_egress,
>               self._ctx.topology.tg_port_ingress,
> @@ -266,6 +270,27 @@ def send_packets_and_capture(
>               duration,
>           )
>   
> +    def assess_performance_by_packet(
> +        self, packet: Packet, send_mpps: int, duration: int = 60
> +    ) -> PerformanceTrafficStats:
> +        """Send a given packet for a given duration and assess basic performance statistics.
> +
> +        Send `packet` and assess NIC performance for a given duration, corresponding to the test
> +        suite's given topology.
> +
> +        Args:
> +            packet: The packet to send.
> +            send_mpps: The millions packets per second send rate.
> +            duration: Performance test duration (in seconds).
> +
> +        Returns:
> +            Performance statistics of the generated test.
> +        """
> +        assert isinstance(
> +            self._ctx.tg, PerformanceTrafficGenerator
> +        ), "Cannot run performance tests on non-performance traffic generator."
> +        return self._ctx.tg.calculate_traffic_and_stats(packet, send_mpps, duration)
> +
>       def send_packets(
>           self,
>           packets: list[Packet],
> @@ -275,8 +300,11 @@ def send_packets(
>           Args:
>               packets: Packets to send.
>           """
> +        assert isinstance(
> +            self._ctx.tg, CapturingTrafficGenerator
> +        ), "Cannot run performance tests on non-capturing traffic generator."

I am still unsure if we are meant to have co-existing traffic 
generators. But we should make everything more tolerant, ideally this 
should never be reached, and DTS should automatically skip tests 
appropriately. The above assertion and message don't really appear 
related, so this should also be rectified.

>           packets = self._adjust_addresses(packets)
> -        self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
> +        self._ctx.tg.send_packets(packets, self._ctx.topology.tg_port_egress)
>   
>       def get_expected_packets(
>           self,

[PATCH v4 0/3] Add TREX Traffic Generator to DTS Framework

[Patrick Robb]

This series adds support for performance traffic generators, and
includes an implementation for using the TREX traffic generator for
performance testing. It also includes a new testsuite which measures the
single core forwarding performance of a SUT in millions of packets per
second.

There are some points in this series which may warrant further
discussion. Namely:

- We want to run only the functional traffic generator OR the
  performance traffic generator at a given time. So, this series makes
DTS tear down and bring up traffic generators as needed, depending on
whether a testcase is of type functional or performance. I think this is
a good approach in general, but in order to facilitate this I have
removed the shell pool cleanup at testcase teardown and testsuite
teardown. Now, it only cleans up at testrun teardown. I think it is
possible to add these shell pool cleanups back in and keep the existing
solution, but I would like to discuss it at the next CI or DTS meeting.

- Currently, the testsuite prints out a stats table to the console when
  the single core forwarding testsuite is complete. It may make sense to
start writing these stats to an output file as well.

If you would like to test this series, please also note that you will
have to move your YAML configs to the new configurations directory.

Nicholas Pratte (3):
  dts: rework traffic generator inheritance structure
  dts: add trex traffic generator to dts framework
  dts: add performance test functions to test suite API

 ...sts.TestSuite_single_core_forward_perf.rst |   8 +
 doc/guides/tools/dts.rst                      |  35 ++-
 dts/{ => configurations}/nodes.example.yaml   |   0
 .../test_run.example.yaml                     |   6 +-
 dts/configurations/tests_config.example.yaml  |  17 ++
 dts/framework/config/test_run.py              |  22 +-
 dts/framework/context.py                      |   5 +-
 dts/framework/remote_session/blocking_app.py  |   4 +-
 .../remote_session/interactive_shell.py       |   2 +-
 dts/framework/settings.py                     |  12 +-
 dts/framework/test_run.py                     |  54 +++-
 dts/framework/test_suite.py                   |  32 ++-
 .../traffic_generator/__init__.py             |  13 +-
 .../capturing_traffic_generator.py            |  34 +++
 .../performance_traffic_generator.py          |  64 +++++
 .../testbed_model/traffic_generator/scapy.py  |   1 +
 .../traffic_generator/traffic_generator.py    |  44 +--
 .../testbed_model/traffic_generator/trex.py   | 258 ++++++++++++++++++
 .../TestSuite_single_core_forward_perf.py     | 139 ++++++++++
 dts/tests_config.example.yaml                 |   5 -
 20 files changed, 690 insertions(+), 65 deletions(-)
 create mode 100644 doc/api/dts/tests.TestSuite_single_core_forward_perf.rst
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (88%)
 create mode 100644 dts/configurations/tests_config.example.yaml
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py
 create mode 100644 dts/tests/TestSuite_single_core_forward_perf.py
 delete mode 100644 dts/tests_config.example.yaml

-- 
2.49.0

[PATCH v4 1/3] dts: rework traffic generator inheritance structure

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Rework TG class hierarchy to include performance traffic generators.
As such, methods specific to capturing traffic have been moved to the
CapturingTrafficGenerator subclass.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 .../capturing_traffic_generator.py            | 34 ++++++++++
 .../performance_traffic_generator.py          | 63 +++++++++++++++++++
 .../traffic_generator/traffic_generator.py    | 38 -----------
 3 files changed, 97 insertions(+), 38 deletions(-)
 create mode 100644 dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py

diff --git a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
index ec0993e6b7..734a66d1f3 100644
--- a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
@@ -65,6 +65,40 @@ def is_capturing(self) -> bool:
         """This traffic generator can capture traffic."""
         return True
 
+    def send_packet(self, packet: Packet, port: Port) -> None:
+        """Send `packet` and block until it is fully sent.
+
+        Send `packet` on `port`, then wait until `packet` is fully sent.
+
+        Args:
+            packet: The packet to send.
+            port: The egress port on the TG node.
+        """
+        self.send_packets([packet], port)
+
+    def send_packets(self, packets: list[Packet], port: Port) -> None:
+        """Send `packets` and block until they are fully sent.
+
+        Send `packets` on `port`, then wait until `packets` are fully sent.
+
+        Args:
+            packets: The packets to send.
+            port: The egress port on the TG node.
+        """
+        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
+        self._logger.debug(get_packet_summaries(packets))
+        self._send_packets(packets, port)
+
+    @abstractmethod
+    def _send_packets(self, packets: list[Packet], port: Port) -> None:
+        """The implementation of :method:`send_packets`.
+
+        The subclasses must implement this method which sends `packets` on `port`.
+        The method should block until all `packets` are fully sent.
+
+        What fully sent means is defined by the traffic generator.
+        """
+
     def send_packets_and_capture(
         self,
         packets: list[Packet],
diff --git a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
new file mode 100644
index 0000000000..6b23faa1a5
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
@@ -0,0 +1,63 @@
+"""Traffic generators for performance tests which can generate a high number of packets."""
+
+from abc import abstractmethod
+from dataclasses import dataclass
+
+from scapy.packet import Packet
+
+from framework.testbed_model.topology import Topology
+
+from .traffic_generator import TrafficGenerator
+
+
+@dataclass(slots=True)
+class PerformanceTrafficStats:
+    """Data structure to store performance statistics for a given test run.
+
+    Attributes:
+        tx_pps: Recorded tx packets per second
+        tx_bps: Recorded tx bytes per second
+        rx_pps: Recorded rx packets per second
+        rx_bps: Recorded rx bytes per second
+        frame_size: The total length of the frame
+    """
+
+    tx_pps: float
+    tx_bps: float
+    rx_pps: float
+    rx_bps: float
+
+    frame_size: int | None = None
+
+
+class PerformanceTrafficGenerator(TrafficGenerator):
+    """An abstract base class for all performance-oriented traffic generators.
+
+    Provides an intermediary interface for performance-based traffic generator.
+    """
+
+    @abstractmethod
+    def calculate_traffic_and_stats(
+        self,
+        packet: Packet,
+        duration: float,
+        send_mpps: int | None = None,
+    ) -> PerformanceTrafficStats:
+        """Send packet traffic and acquire associated statistics.
+
+        If `send_mpps` is provided, attempt to transmit traffic at the `send_mpps` rate.
+        Otherwise, attempt to transmit at line rate.
+
+        Args:
+            packet: The packet to send.
+            duration: Performance test duration (in seconds).
+            send_mpps: The millions packets per second send rate.
+
+        Returns:
+            Performance statistics of the generated test.
+        """
+
+    def setup(self, topology: Topology) -> None:
+        """Overrides :meth:`.traffic_generator.TrafficGenerator.setup`."""
+        for port in self._tg_node.ports:
+            self._tg_node.main_session.configure_port_mtu(2000, port)
diff --git a/dts/framework/testbed_model/traffic_generator/traffic_generator.py b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
index cac119c183..e5f246df7a 100644
--- a/dts/framework/testbed_model/traffic_generator/traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
@@ -11,14 +11,10 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
-from scapy.packet import Packet
-
 from framework.config.test_run import TrafficGeneratorConfig
 from framework.logger import DTSLogger, get_dts_logger
 from framework.testbed_model.node import Node
-from framework.testbed_model.port import Port
 from framework.testbed_model.topology import Topology
-from framework.utils import get_packet_summaries
 
 
 class TrafficGenerator(ABC):
@@ -57,40 +53,6 @@ def teardown(self) -> None:
         """Teardown the traffic generator."""
         self.close()
 
-    def send_packet(self, packet: Packet, port: Port) -> None:
-        """Send `packet` and block until it is fully sent.
-
-        Send `packet` on `port`, then wait until `packet` is fully sent.
-
-        Args:
-            packet: The packet to send.
-            port: The egress port on the TG node.
-        """
-        self.send_packets([packet], port)
-
-    def send_packets(self, packets: list[Packet], port: Port) -> None:
-        """Send `packets` and block until they are fully sent.
-
-        Send `packets` on `port`, then wait until `packets` are fully sent.
-
-        Args:
-            packets: The packets to send.
-            port: The egress port on the TG node.
-        """
-        self._logger.info(f"Sending packet{'s' if len(packets) > 1 else ''}.")
-        self._logger.debug(get_packet_summaries(packets))
-        self._send_packets(packets, port)
-
-    @abstractmethod
-    def _send_packets(self, packets: list[Packet], port: Port) -> None:
-        """The implementation of :method:`send_packets`.
-
-        The subclasses must implement this method which sends `packets` on `port`.
-        The method should block until all `packets` are fully sent.
-
-        What fully sent means is defined by the traffic generator.
-        """
-
     @property
     def is_capturing(self) -> bool:
         """This traffic generator can't capture traffic."""
-- 
2.49.0

[PATCH v4 2/3] dts: add trex traffic generator to dts framework

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Implement the TREX traffic generator for use in the DTS framework. The
provided implementation leverages TREX's stateless API automation
library, via use of a Python shell. The DTS context has been modified
to include a performance traffic generator in addition to a functional
traffic generator.

In addition, the DTS testrun state machine has been modified such that
traffic generators are brought up and down as needed, and so that only
one traffic generator application is running on the TG system at a time.
During the testcase setup stage, the testcase type (perf or func) will
be checked and the correct traffic generator brought up. For instance,
if a functional TG is running from a previous test and we start a
performance test, then the functional TG is stopped and the performance
TG started. This is an attempt to strike a balance between the concept
of having the scapy asyncsniffer always on to save on execution time,
with the competing need to bring up performance traffic generators as
needed.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 doc/guides/tools/dts.rst                      |  35 ++-
 dts/{ => configurations}/nodes.example.yaml   |   0
 .../test_run.example.yaml                     |   6 +-
 .../tests_config.example.yaml                 |   0
 dts/framework/config/test_run.py              |  22 +-
 dts/framework/context.py                      |   5 +-
 dts/framework/remote_session/blocking_app.py  |   4 +-
 .../remote_session/interactive_shell.py       |   2 +-
 dts/framework/settings.py                     |  12 +-
 dts/framework/test_run.py                     |  54 +++-
 dts/framework/test_suite.py                   |   7 +-
 .../traffic_generator/__init__.py             |  13 +-
 .../performance_traffic_generator.py          |   1 +
 .../testbed_model/traffic_generator/scapy.py  |   1 +
 .../traffic_generator/traffic_generator.py    |  22 ++
 .../testbed_model/traffic_generator/trex.py   | 258 ++++++++++++++++++
 16 files changed, 412 insertions(+), 30 deletions(-)
 rename dts/{ => configurations}/nodes.example.yaml (100%)
 rename dts/{ => configurations}/test_run.example.yaml (88%)
 rename dts/{ => configurations}/tests_config.example.yaml (100%)
 create mode 100644 dts/framework/testbed_model/traffic_generator/trex.py

diff --git a/doc/guides/tools/dts.rst b/doc/guides/tools/dts.rst
index e7af60b7a6..7f2d4cf481 100644
--- a/doc/guides/tools/dts.rst
+++ b/doc/guides/tools/dts.rst
@@ -210,7 +210,8 @@ These need to be set up on a Traffic Generator Node:
 #. **Traffic generator dependencies**
 
    The traffic generator running on the traffic generator node must be installed beforehand.
-   For Scapy traffic generator, only a few Python libraries need to be installed:
+
+   For Scapy traffic generator (functional tests), only a few Python libraries need to be installed:
 
    .. code-block:: console
 
@@ -218,6 +219,32 @@ These need to be set up on a Traffic Generator Node:
       sudo pip install --upgrade pip
       sudo pip install scapy==2.5.0
 
+   For TREX traffic generator (performance tests), TREX must be downloaded and a TREX config produced for each TG NIC. For example:
+
+   .. code-block:: console
+
+      wget https://trex-tgn.cisco.com/trex/release/v3.03.tar.gz
+      tar -xf v3.03.tar.gz
+      cd v3.03
+      sudo ./dpdk_setup_ports.py -i
+
+   Within the dpdk_setup_ports.py utility, follow these instructions:
+     - Select MAC based config
+     - Select interfaces 0 and 1 on your TG NIC
+     - Do not change assumed dest to DUT MAC (just leave the default loopback)
+     - Print preview of the config
+     - Check for device address correctness
+     - Check for socket and CPU correctness (CPU/socket NUMA node should match NIC NUMA node)
+     - Write the file to a path on your system
+
+   Then, presuming you are using the test_run.example.yaml as a template for your test_run config:
+     - Uncomment the performance_traffic_generator section, making DTS use a performance TG
+     - Update the remote_path and config fields to the remote path of your TREX directory and the path to your new TREX config file
+     - Update the "perf" field to enable performance testing
+
+   After these steps, you should be ready to run performance tests with TREX.
+
+
 #. **Hardware dependencies**
 
    The traffic generators, like DPDK, need a proper driver and firmware.
@@ -525,7 +552,7 @@ And they both have two network ports which are physically connected to each othe
 ``dts/test_run.example.yaml``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-.. literalinclude:: ../../../dts/test_run.example.yaml
+.. literalinclude:: ../../../dts/configurations/test_run.example.yaml
    :language: yaml
    :start-at: # Define
 
@@ -535,7 +562,7 @@ And they both have two network ports which are physically connected to each othe
 ``dts/nodes.example.yaml``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-.. literalinclude:: ../../../dts/nodes.example.yaml
+.. literalinclude:: ../../../dts/configurations/nodes.example.yaml
    :language: yaml
    :start-at: # Define
 
@@ -551,6 +578,6 @@ to demonstrate custom test suite configuration:
 ``dts/tests_config.example.yaml``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-.. literalinclude:: ../../../dts/tests_config.example.yaml
+.. literalinclude:: ../../../dts/configurations/tests_config.example.yaml
    :language: yaml
    :start-at: # Define
diff --git a/dts/nodes.example.yaml b/dts/configurations/nodes.example.yaml
similarity index 100%
rename from dts/nodes.example.yaml
rename to dts/configurations/nodes.example.yaml
diff --git a/dts/test_run.example.yaml b/dts/configurations/test_run.example.yaml
similarity index 88%
rename from dts/test_run.example.yaml
rename to dts/configurations/test_run.example.yaml
index c90de9d68d..c8035fccf0 100644
--- a/dts/test_run.example.yaml
+++ b/dts/configurations/test_run.example.yaml
@@ -23,8 +23,12 @@ dpdk:
     # in a subdirectory of DPDK tree root directory. Otherwise, will be using the `build_options`
     # to build the DPDK from source. Either `precompiled_build_dir` or `build_options` can be
     # defined, but not both.
-traffic_generator:
+func_traffic_generator:
   type: SCAPY
+# perf_traffic_generator:
+#   type: TREX
+#   remote_path: "/opt/trex/v3.03" # The remote path of the traffic generator application.
+#   config: "/opt/trex_config/trex_config.yaml" # Additional configuration files. (Leave blank if not required)
 perf: false # disable performance testing
 func: true # enable functional testing
 use_virtual_functions: false # use virtual functions (VFs) instead of physical functions
diff --git a/dts/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
similarity index 100%
rename from dts/tests_config.example.yaml
rename to dts/configurations/tests_config.example.yaml
diff --git a/dts/framework/config/test_run.py b/dts/framework/config/test_run.py
index 71b3755d6e..68db862cea 100644
--- a/dts/framework/config/test_run.py
+++ b/dts/framework/config/test_run.py
@@ -16,7 +16,7 @@
 from enum import Enum, auto, unique
 from functools import cached_property
 from pathlib import Path, PurePath
-from typing import Annotated, Any, Literal, NamedTuple
+from typing import Annotated, Any, Literal, NamedTuple, Optional
 
 from pydantic import (
     BaseModel,
@@ -396,6 +396,8 @@ class TrafficGeneratorType(str, Enum):
 
     #:
     SCAPY = "SCAPY"
+    #:
+    TREX = "TREX"
 
 
 class TrafficGeneratorConfig(FrozenModel):
@@ -412,8 +414,18 @@ class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
     type: Literal[TrafficGeneratorType.SCAPY]
 
 
+class TrexTrafficGeneratorConfig(TrafficGeneratorConfig):
+    """TREX traffic generator specific configuration."""
+
+    type: Literal[TrafficGeneratorType.TREX]
+    remote_path: PurePath
+    config: PurePath
+
+
 #: A union type discriminating traffic generators by the `type` field.
-TrafficGeneratorConfigTypes = Annotated[ScapyTrafficGeneratorConfig, Field(discriminator="type")]
+TrafficGeneratorConfigTypes = Annotated[
+    TrexTrafficGeneratorConfig, ScapyTrafficGeneratorConfig, Field(discriminator="type")
+]
 
 #: Comma-separated list of logical cores to use. An empty string or ```any``` means use all lcores.
 LogicalCores = Annotated[
@@ -461,8 +473,10 @@ class TestRunConfiguration(FrozenModel):
 
     #: The DPDK configuration used to test.
     dpdk: DPDKConfiguration
-    #: The traffic generator configuration used to test.
-    traffic_generator: TrafficGeneratorConfigTypes
+    #: The traffic generator configuration used for functional tests.
+    func_traffic_generator: Optional[ScapyTrafficGeneratorConfig] = None
+    #: The traffic generator configuration used for performance tests.
+    perf_traffic_generator: Optional[TrexTrafficGeneratorConfig] = None
     #: Whether to run performance tests.
     perf: bool
     #: Whether to run functional tests.
diff --git a/dts/framework/context.py b/dts/framework/context.py
index ae319d949f..8f1021dc96 100644
--- a/dts/framework/context.py
+++ b/dts/framework/context.py
@@ -6,7 +6,7 @@
 import functools
 from collections.abc import Callable
 from dataclasses import MISSING, dataclass, field, fields
-from typing import TYPE_CHECKING, Any, ParamSpec, Union
+from typing import TYPE_CHECKING, Any, Optional, ParamSpec, Union
 
 from framework.exception import InternalError
 from framework.remote_session.shell_pool import ShellPool
@@ -76,7 +76,8 @@ class Context:
     topology: Topology
     dpdk_build: "DPDKBuildEnvironment"
     dpdk: "DPDKRuntimeEnvironment"
-    tg: "TrafficGenerator"
+    func_tg: Optional["TrafficGenerator"]
+    perf_tg: Optional["TrafficGenerator"]
     local: LocalContext = field(default_factory=LocalContext)
     shell_pool: ShellPool = field(default_factory=ShellPool)
 
diff --git a/dts/framework/remote_session/blocking_app.py b/dts/framework/remote_session/blocking_app.py
index 8de536c259..b38ad1c15a 100644
--- a/dts/framework/remote_session/blocking_app.py
+++ b/dts/framework/remote_session/blocking_app.py
@@ -48,7 +48,7 @@ class BlockingApp(InteractiveShell, Generic[P]):
     def __init__(
         self,
         node: Node,
-        path: PurePath,
+        path: str | PurePath,
         name: str | None = None,
         privileged: bool = False,
         app_params: P | str = "",
@@ -73,7 +73,7 @@ def __init__(
         super().__init__(node, name, privileged, app_params)
 
     @property
-    def path(self) -> PurePath:
+    def path(self) -> str | PurePath:
         """The path of the DPDK app relative to the DPDK build folder."""
         return self._path
 
diff --git a/dts/framework/remote_session/interactive_shell.py b/dts/framework/remote_session/interactive_shell.py
index ce93247051..34803f4f7f 100644
--- a/dts/framework/remote_session/interactive_shell.py
+++ b/dts/framework/remote_session/interactive_shell.py
@@ -259,7 +259,7 @@ def close(self) -> None:
 
     @property
     @abstractmethod
-    def path(self) -> PurePath:
+    def path(self) -> str | PurePath:
         """Path to the shell executable."""
 
     def _make_real_path(self) -> PurePath:
diff --git a/dts/framework/settings.py b/dts/framework/settings.py
index 84b627a06a..b08373b7ea 100644
--- a/dts/framework/settings.py
+++ b/dts/framework/settings.py
@@ -130,11 +130,17 @@ class Settings:
     """
 
     #:
-    test_run_config_path: Path = Path(__file__).parent.parent.joinpath("test_run.yaml")
+    test_run_config_path: Path = Path(__file__).parent.parent.joinpath(
+        "configurations/test_run.yaml"
+    )
     #:
-    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("nodes.yaml")
+    nodes_config_path: Path = Path(__file__).parent.parent.joinpath("configurations/nodes.yaml")
     #:
-    tests_config_path: Path | None = None
+    tests_config_path: Path | None = (
+        Path(__file__).parent.parent.joinpath("configurations/tests_config.yaml")
+        if os.path.exists("configurations/tests_config.yaml")
+        else None
+    )
     #:
     output_dir: str = "output"
     #:
diff --git a/dts/framework/test_run.py b/dts/framework/test_run.py
index 9cf04c0b06..e1450d13cf 100644
--- a/dts/framework/test_run.py
+++ b/dts/framework/test_run.py
@@ -113,7 +113,7 @@
 from framework.remote_session.dpdk import DPDKBuildEnvironment, DPDKRuntimeEnvironment
 from framework.settings import SETTINGS
 from framework.test_result import Result, ResultNode, TestRunResult
-from framework.test_suite import BaseConfig, TestCase, TestSuite
+from framework.test_suite import BaseConfig, TestCase, TestCaseType, TestSuite
 from framework.testbed_model.capability import (
     Capability,
     get_supported_capabilities,
@@ -199,10 +199,26 @@ def __init__(
 
         dpdk_build_env = DPDKBuildEnvironment(config.dpdk.build, sut_node)
         dpdk_runtime_env = DPDKRuntimeEnvironment(config.dpdk, sut_node, dpdk_build_env)
-        traffic_generator = create_traffic_generator(config.traffic_generator, tg_node)
+
+        func_traffic_generator = (
+            create_traffic_generator(config.func_traffic_generator, tg_node)
+            if config.func
+            else None
+        )
+        perf_traffic_generator = (
+            create_traffic_generator(config.perf_traffic_generator, tg_node)
+            if config.perf
+            else None
+        )
 
         self.ctx = Context(
-            sut_node, tg_node, topology, dpdk_build_env, dpdk_runtime_env, traffic_generator
+            sut_node,
+            tg_node,
+            topology,
+            dpdk_build_env,
+            dpdk_runtime_env,
+            func_traffic_generator,
+            perf_traffic_generator,
         )
         self.result = result
         self.selected_tests = list(self.config.filter_tests(tests_config))
@@ -335,7 +351,10 @@ def next(self) -> State | None:
             test_run.ctx.topology.instantiate_vf_ports()
 
         test_run.ctx.topology.configure_ports("sut", "dpdk")
-        test_run.ctx.tg.setup(test_run.ctx.topology)
+        if test_run.ctx.func_tg:
+            test_run.ctx.func_tg.setup(test_run.ctx.topology)
+        if test_run.ctx.perf_tg:
+            test_run.ctx.perf_tg.setup(test_run.ctx.topology)
 
         self.result.ports = [
             port.to_dict()
@@ -425,7 +444,10 @@ def next(self) -> State | None:
             self.test_run.ctx.topology.delete_vf_ports()
 
         self.test_run.ctx.shell_pool.terminate_current_pool()
-        self.test_run.ctx.tg.teardown()
+        if self.test_run.ctx.func_tg and self.test_run.ctx.func_tg.is_setup:
+            self.test_run.ctx.func_tg.teardown()
+        if self.test_run.ctx.perf_tg and self.test_run.ctx.perf_tg.is_setup:
+            self.test_run.ctx.perf_tg.teardown()
         self.test_run.ctx.topology.teardown()
         self.test_run.ctx.dpdk.teardown()
         self.test_run.ctx.tg_node.teardown()
@@ -555,7 +577,6 @@ def next(self) -> State | None:
         """Next state."""
         self.test_suite.tear_down_suite()
         self.test_run.ctx.dpdk.kill_cleanup_dpdk_apps()
-        self.test_run.ctx.shell_pool.terminate_current_pool()
         self.result.mark_step_as("teardown", Result.PASS)
         return TestRunExecution(self.test_run, self.test_run.result)
 
@@ -611,6 +632,26 @@ def next(self) -> State | None:
         )
         self.test_run.ctx.topology.configure_ports("sut", sut_ports_drivers)
 
+        if (
+            self.test_run.ctx.perf_tg
+            and self.test_run.ctx.perf_tg.is_setup
+            and self.test_case.test_type is TestCaseType.FUNCTIONAL
+        ):
+            self.test_run.ctx.perf_tg.teardown()
+            self.test_run.ctx.topology.configure_ports("tg", "kernel")
+            if self.test_run.ctx.func_tg and not self.test_run.ctx.func_tg.is_setup:
+                self.test_run.ctx.func_tg.setup(self.test_run.ctx.topology)
+
+        if (
+            self.test_run.ctx.func_tg
+            and self.test_run.ctx.func_tg.is_setup
+            and self.test_case.test_type is TestCaseType.PERFORMANCE
+        ):
+            self.test_run.ctx.func_tg.teardown()
+            self.test_run.ctx.topology.configure_ports("tg", "dpdk")
+            if self.test_run.ctx.perf_tg and not self.test_run.ctx.perf_tg.is_setup:
+                self.test_run.ctx.perf_tg.setup(self.test_run.ctx.topology)
+
         self.test_suite.set_up_test_case()
         self.result.mark_step_as("setup", Result.PASS)
         return TestCaseExecution(
@@ -699,7 +740,6 @@ def description(self) -> str:
     def next(self) -> State | None:
         """Next state."""
         self.test_suite.tear_down_test_case()
-        self.test_run.ctx.shell_pool.terminate_current_pool()
         self.result.mark_step_as("teardown", Result.PASS)
         assert self.result.parent is not None
         return TestSuiteExecution(
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index 5ee5a039d7..c720106112 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -257,11 +257,11 @@ def send_packets_and_capture(
             A list of received packets.
         """
         assert isinstance(
-            self._ctx.tg, CapturingTrafficGenerator
+            self._ctx.func_tg, CapturingTrafficGenerator
         ), "Cannot capture with a non-capturing traffic generator"
         # TODO: implement @requires for types of traffic generator
         packets = self._adjust_addresses(packets)
-        return self._ctx.tg.send_packets_and_capture(
+        return self._ctx.func_tg.send_packets_and_capture(
             packets,
             self._ctx.topology.tg_port_egress,
             self._ctx.topology.tg_port_ingress,
@@ -279,7 +279,8 @@ def send_packets(
             packets: Packets to send.
         """
         packets = self._adjust_addresses(packets)
-        self._ctx.tg.send_packets(packets, self._ctx.topology.tg_port_egress)
+        if self._ctx.func_tg is not None:
+            self._ctx.func_tg.send_packets(packets, self._ctx.topology.tg_port_egress)
 
     def get_expected_packets(
         self,
diff --git a/dts/framework/testbed_model/traffic_generator/__init__.py b/dts/framework/testbed_model/traffic_generator/__init__.py
index 2a259a6e6c..fca251f534 100644
--- a/dts/framework/testbed_model/traffic_generator/__init__.py
+++ b/dts/framework/testbed_model/traffic_generator/__init__.py
@@ -14,17 +14,22 @@
 and a capturing traffic generator is required.
 """
 
-from framework.config.test_run import ScapyTrafficGeneratorConfig, TrafficGeneratorConfig
+from framework.config.test_run import (
+    ScapyTrafficGeneratorConfig,
+    TrafficGeneratorConfig,
+    TrexTrafficGeneratorConfig,
+)
 from framework.exception import ConfigurationError
 from framework.testbed_model.node import Node
 
-from .capturing_traffic_generator import CapturingTrafficGenerator
 from .scapy import ScapyTrafficGenerator
+from .traffic_generator import TrafficGenerator
+from .trex import TrexTrafficGenerator
 
 
 def create_traffic_generator(
     traffic_generator_config: TrafficGeneratorConfig, node: Node
-) -> CapturingTrafficGenerator:
+) -> TrafficGenerator:
     """The factory function for creating traffic generator objects from the test run configuration.
 
     Args:
@@ -40,5 +45,7 @@ def create_traffic_generator(
     match traffic_generator_config:
         case ScapyTrafficGeneratorConfig():
             return ScapyTrafficGenerator(node, traffic_generator_config, privileged=True)
+        case TrexTrafficGeneratorConfig():
+            return TrexTrafficGenerator(node, traffic_generator_config)
         case _:
             raise ConfigurationError(f"Unknown traffic generator: {traffic_generator_config.type}")
diff --git a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
index 6b23faa1a5..f35aad64fc 100644
--- a/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/performance_traffic_generator.py
@@ -59,5 +59,6 @@ def calculate_traffic_and_stats(
 
     def setup(self, topology: Topology) -> None:
         """Overrides :meth:`.traffic_generator.TrafficGenerator.setup`."""
+        super().setup(topology)
         for port in self._tg_node.ports:
             self._tg_node.main_session.configure_port_mtu(2000, port)
diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py b/dts/framework/testbed_model/traffic_generator/scapy.py
index a31807e8e4..58453cd7e0 100644
--- a/dts/framework/testbed_model/traffic_generator/scapy.py
+++ b/dts/framework/testbed_model/traffic_generator/scapy.py
@@ -320,6 +320,7 @@ def setup(self, topology: Topology) -> None:
 
         Binds the TG node ports to the kernel drivers and starts up the async sniffer.
         """
+        super().setup(topology)
         topology.configure_ports("tg", "kernel")
 
         self._sniffer = ScapyAsyncSniffer(
diff --git a/dts/framework/testbed_model/traffic_generator/traffic_generator.py b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
index e5f246df7a..cdda5a7c08 100644
--- a/dts/framework/testbed_model/traffic_generator/traffic_generator.py
+++ b/dts/framework/testbed_model/traffic_generator/traffic_generator.py
@@ -11,9 +11,12 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
+from scapy.packet import Packet
+
 from framework.config.test_run import TrafficGeneratorConfig
 from framework.logger import DTSLogger, get_dts_logger
 from framework.testbed_model.node import Node
+from framework.testbed_model.port import Port
 from framework.testbed_model.topology import Topology
 
 
@@ -30,6 +33,7 @@ class TrafficGenerator(ABC):
     _config: TrafficGeneratorConfig
     _tg_node: Node
     _logger: DTSLogger
+    _is_setup: bool
 
     def __init__(self, tg_node: Node, config: TrafficGeneratorConfig, **kwargs: Any) -> None:
         """Initialize the traffic generator.
@@ -45,12 +49,25 @@ def __init__(self, tg_node: Node, config: TrafficGeneratorConfig, **kwargs: Any)
         self._config = config
         self._tg_node = tg_node
         self._logger = get_dts_logger(f"{self._tg_node.name} {self._config.type}")
+        self._is_setup = False
+
+    def send_packets(self, packets: list[Packet], port: Port) -> None:
+        """Send `packets` and block until they are fully sent.
+
+        Send `packets` on `port`, then wait until `packets` are fully sent.
+
+        Args:
+            packets: The packets to send.
+            port: The egress port on the TG node.
+        """
 
     def setup(self, topology: Topology) -> None:
         """Setup the traffic generator."""
+        self._is_setup = True
 
     def teardown(self) -> None:
         """Teardown the traffic generator."""
+        self._is_setup = False
         self.close()
 
     @property
@@ -61,3 +78,8 @@ def is_capturing(self) -> bool:
     @abstractmethod
     def close(self) -> None:
         """Free all resources used by the traffic generator."""
+
+    @property
+    def is_setup(self) -> bool:
+        """Indicates whether the traffic generator application is currently running."""
+        return self._is_setup
diff --git a/dts/framework/testbed_model/traffic_generator/trex.py b/dts/framework/testbed_model/traffic_generator/trex.py
new file mode 100644
index 0000000000..08147cdde2
--- /dev/null
+++ b/dts/framework/testbed_model/traffic_generator/trex.py
@@ -0,0 +1,258 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 University of New Hampshire
+
+"""Implementation for TREX performance traffic generator."""
+
+import ast
+import time
+from dataclasses import dataclass, field
+from enum import auto
+from typing import ClassVar
+
+from scapy.packet import Packet
+
+from framework.config.node import OS, NodeConfiguration
+from framework.config.test_run import TrexTrafficGeneratorConfig
+from framework.parser import TextParser
+from framework.remote_session.blocking_app import BlockingApp
+from framework.remote_session.python_shell import PythonShell
+from framework.testbed_model.node import Node, create_session
+from framework.testbed_model.os_session import OSSession
+from framework.testbed_model.topology import Topology
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
+from framework.utils import StrEnum
+
+
+@dataclass(slots=True)
+class TrexPerformanceTrafficStats(PerformanceTrafficStats, TextParser):
+    """Data structure to store performance statistics for a given test run.
+
+    This class overrides the initialization of :class:`PerformanceTrafficStats`
+    in order to set the attribute values using the TREX stats output.
+
+    Attributes:
+        tx_pps: Recorded tx packets per second
+        tx_bps: Recorded tx bytes per second
+        rx_pps: Recorded rx packets per second
+        rx_bps: Recorded rx bytes per second
+        frame_size: The total length of the frame
+    """
+
+    tx_pps: int = field(metadata=TextParser.find_int(r"total.*'tx_pps': (\d+)"))
+    tx_bps: int = field(metadata=TextParser.find_int(r"total.*'tx_bps': (\d+)"))
+    rx_pps: int = field(metadata=TextParser.find_int(r"total.*'rx_pps': (\d+)"))
+    rx_bps: int = field(metadata=TextParser.find_int(r"total.*'rx_bps': (\d+)"))
+
+
+class TrexStatelessTXModes(StrEnum):
+    """Flags indicating TREX instance's current transmission mode."""
+
+    #: Transmit continuously
+    STLTXCont = auto()
+    #: Transmit in a single burst
+    STLTXSingleBurst = auto()
+    #: Transmit in multiple bursts
+    STLTXMultiBurst = auto()
+
+
+class TrexTrafficGenerator(PerformanceTrafficGenerator):
+    """TREX traffic generator.
+
+    This implementation leverages the stateless API library provided in the TREX installation.
+
+    Attributes:
+        stl_client_name: The name of the stateless client used in the stateless API.
+        packet_stream_name: The name of the stateless packet stream used in the stateless API.
+    """
+
+    _os_session: OSSession
+
+    _tg_config: TrexTrafficGeneratorConfig
+    _node_config: NodeConfiguration
+
+    _shell: PythonShell
+    _python_indentation: ClassVar[str] = " " * 4
+
+    stl_client_name: ClassVar[str] = "client"
+    packet_stream_name: ClassVar[str] = "stream"
+
+    _streaming_mode: TrexStatelessTXModes = TrexStatelessTXModes.STLTXCont
+
+    _tg_cores: int = 10
+
+    _trex_app: BlockingApp
+
+    def __init__(self, tg_node: Node, config: TrexTrafficGeneratorConfig) -> None:
+        """Initialize the TREX server.
+
+        Initializes needed OS sessions for the creation of the TREX server process.
+
+        Args:
+            tg_node: TG node the TREX instance is operating on.
+            config: Traffic generator config provided for TREX instance.
+        """
+        assert (
+            tg_node.config.os == OS.linux
+        ), "Linux is the only supported OS for trex traffic generation"
+
+        super().__init__(tg_node=tg_node, config=config)
+        self._tg_node_config = tg_node.config
+        self._tg_config = config
+
+        self._os_session = create_session(self._tg_node.config, "TREX", self._logger)
+
+    def setup(self, topology: Topology):
+        """Initialize and start a TREX server process."""
+        super().setup(topology)
+
+        self._shell = PythonShell(self._tg_node, "TREX-client", privileged=True)
+
+        # Start TREX server process.
+        trex_app_path = f"cd {self._tg_config.remote_path} && ./t-rex-64"
+        self._trex_app = BlockingApp(
+            node=self._tg_node,
+            path=trex_app_path,
+            name="trex-tg",
+            privileged=True,
+            app_params=f"--cfg {self._tg_config.config} -c {self._tg_cores} -i",
+        )
+        self._trex_app.wait_until_ready("-Per port stats table")
+
+        self._shell.start_application()
+        self._shell.send_command("import os")
+        self._shell.send_command(
+            f"os.chdir('{self._tg_config.remote_path}/automation/trex_control_plane/interactive')"
+        )
+
+        # Import stateless API components.
+        imports = [
+            "import trex",
+            "import trex.stl",
+            "import trex.stl.trex_stl_client",
+            "import trex.stl.trex_stl_streams",
+            "import trex.stl.trex_stl_packet_builder_scapy",
+            "from scapy.layers.l2 import Ether",
+            "from scapy.layers.inet import IP",
+            "from scapy.packet import Raw",
+        ]
+        self._shell.send_command("\n".join(imports))
+
+        stateless_client = [
+            f"{self.stl_client_name} = trex.stl.trex_stl_client.STLClient(",
+            f"username='{self._tg_node_config.user}',",
+            "server='127.0.0.1',",
+            ")",
+        ]
+
+        self._shell.send_command(f"\n{self._python_indentation}".join(stateless_client))
+        self._shell.send_command(f"{self.stl_client_name}.connect()")
+
+    def calculate_traffic_and_stats(
+        self,
+        packet: Packet,
+        duration: float,
+        send_mpps: int | None = None,
+    ) -> PerformanceTrafficStats:
+        """Send packet traffic and acquire associated statistics.
+
+        Overrides
+        :meth:`~.traffic_generator.PerformanceTrafficGenerator.calculate_traffic_and_stats`.
+        """
+        trex_stats_output = ast.literal_eval(self._generate_traffic(packet, duration, send_mpps))
+        stats = TrexPerformanceTrafficStats.parse(str(trex_stats_output))
+        stats.frame_size = len(packet)
+        return stats
+
+    def _generate_traffic(
+        self, packet: Packet, duration: float, send_mpps: int | None = None
+    ) -> str:
+        """Generate traffic using provided packet.
+
+        Uses the provided packet to generate traffic for the provided duration.
+
+        Args:
+            packet: The packet being used for the performance test.
+            duration: The duration of the test being performed.
+            send_mpps: MPPS send rate.
+
+        Returns:
+            A string output of statistics provided by the traffic generator.
+        """
+        self._create_packet_stream(packet)
+        self._setup_trex_client()
+
+        stats = self._send_traffic_and_get_stats(duration, send_mpps)
+
+        return stats
+
+    def _setup_trex_client(self) -> None:
+        """Create trex client and connect to the server process."""
+        # Prepare TREX client for next performance test.
+        procedure = [
+            f"{self.stl_client_name}.connect()",
+            f"{self.stl_client_name}.reset(ports = [0, 1])",
+            f"{self.stl_client_name}.clear_stats()",
+            f"{self.stl_client_name}.add_streams({self.packet_stream_name}, ports=[0, 1])",
+        ]
+
+        for command in procedure:
+            self._shell.send_command(command)
+
+    def _create_packet_stream(self, packet: Packet) -> None:
+        """Create TREX packet stream with the given packet.
+
+        Args:
+            packet: The packet being used for the performance test.
+        """
+        # Create the tx packet on the TG shell
+        self._shell.send_command(f"packet={packet.command()}")
+
+        packet_stream = [
+            f"{self.packet_stream_name} = trex.stl.trex_stl_streams.STLStream(",
+            f"name='Test_{len(packet)}_bytes',",
+            "packet=trex.stl.trex_stl_packet_builder_scapy.STLPktBuilder(pkt=packet),",
+            f"mode=trex.stl.trex_stl_streams.{self._streaming_mode}(percentage=100),",
+            ")",
+        ]
+        self._shell.send_command("\n".join(packet_stream))
+
+    def _send_traffic_and_get_stats(self, duration: float, send_mpps: float | None = None) -> str:
+        """Send traffic and get TG Rx stats.
+
+        Sends traffic from the TREX client's ports for the given duration.
+        When the traffic sending duration has passed, collect the aggregate
+        statistics and return TREX's global stats as a string.
+
+        Args:
+            duration: The traffic generation duration.
+            send_mpps: The millions of packets per second for TREX to send from each port.
+        """
+        if send_mpps:
+            self._shell.send_command(f"""{self.stl_client_name}.start(ports=[0, 1],
+                mult = '{send_mpps}mpps',
+                duration = {duration})""")
+        else:
+            self._shell.send_command(f"""{self.stl_client_name}.start(ports=[0, 1],
+                mult = '100%',
+                duration = {duration})""")
+
+        time.sleep(duration)
+
+        stats = self._shell.send_command(
+            f"{self.stl_client_name}.get_stats(ports=[0, 1])", skip_first_line=True
+        )
+
+        self._shell.send_command(f"{self.stl_client_name}.stop(ports=[0, 1])")
+
+        return stats
+
+    def close(self) -> None:
+        """Overrides :meth:`.traffic_generator.TrafficGenerator.close`.
+
+        Stops the traffic generator and sniffer shells.
+        """
+        self._trex_app.close()
+        self._shell.close()
-- 
2.49.0

[PATCH v4 3/3] dts: add performance test functions to test suite API

[Patrick Robb]

From: Nicholas Pratte <npratte@iol.unh.edu>

Provide functional performance method to run performance tests using a
user-supplied performance traffic generator. The single core performance
test is included, with some basic statistics checks verifying TG packet
transmission rates.

Bugzilla ID: 1697
Signed-off-by: Nicholas Pratte <npratte@iol.unh.edu>
Signed-off-by: Patrick Robb <probb@iol.unh.edu>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 ...sts.TestSuite_single_core_forward_perf.rst |   8 +
 dts/configurations/tests_config.example.yaml  |  12 ++
 dts/framework/test_suite.py                   |  25 ++++
 .../TestSuite_single_core_forward_perf.py     | 139 ++++++++++++++++++
 4 files changed, 184 insertions(+)
 create mode 100644 doc/api/dts/tests.TestSuite_single_core_forward_perf.rst
 create mode 100644 dts/tests/TestSuite_single_core_forward_perf.py

diff --git a/doc/api/dts/tests.TestSuite_single_core_forward_perf.rst b/doc/api/dts/tests.TestSuite_single_core_forward_perf.rst
new file mode 100644
index 0000000000..3651b0b041
--- /dev/null
+++ b/doc/api/dts/tests.TestSuite_single_core_forward_perf.rst
@@ -0,0 +1,8 @@
+.. SPDX-License-Identifier: BSD-3-Clause
+
+single_core_forward_perf Test Suite
+===================================
+
+.. automodule:: tests.TestSuite_single_core_forward_perf
+   :members:
+   :show-inheritance:
diff --git a/dts/configurations/tests_config.example.yaml b/dts/configurations/tests_config.example.yaml
index c011ac0588..167bc91a35 100644
--- a/dts/configurations/tests_config.example.yaml
+++ b/dts/configurations/tests_config.example.yaml
@@ -3,3 +3,15 @@
 # Define the custom test suite configurations
 hello_world:
   msg: A custom hello world to you!
+single_core_forward_perf:
+  test_parameters: # Add frame size / descriptor count combinations as needed
+    - frame_size: 64
+      num_descriptors: 512
+      expected_mpps: 1.0 # Set millions of packets per second according to your devices expected throughput for this given frame size / descriptor count
+    - frame_size: 64
+      num_descriptors: 1024
+      expected_mpps: 1.0
+    - frame_size: 512
+      num_descriptors: 1024
+      expected_mpps: 1.0
+  delta_tolerance: 0.05
\ No newline at end of file
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index c720106112..3816f4c921 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -38,6 +38,10 @@
     CapturingTrafficGenerator,
     PacketFilteringConfig,
 )
+from framework.testbed_model.traffic_generator.performance_traffic_generator import (
+    PerformanceTrafficGenerator,
+    PerformanceTrafficStats,
+)
 
 from .exception import ConfigurationError, InternalError, SkippedTestException, TestCaseVerifyError
 from .logger import DTSLogger, get_dts_logger
@@ -269,6 +273,27 @@ def send_packets_and_capture(
             duration,
         )
 
+    def assess_performance_by_packet(
+        self, packet: Packet, duration: float, send_mpps: int | None = None
+    ) -> PerformanceTrafficStats:
+        """Send a given packet for a given duration and assess basic performance statistics.
+
+        Send `packet` and assess NIC performance for a given duration, corresponding to the test
+        suite's given topology.
+
+        Args:
+            packet: The packet to send.
+            duration: Performance test duration (in seconds).
+            send_mpps: The millions packets per second send rate.
+
+        Returns:
+            Performance statistics of the generated test.
+        """
+        assert isinstance(
+            self._ctx.perf_tg, PerformanceTrafficGenerator
+        ), "Cannot run performance tests on non-performance traffic generator."
+        return self._ctx.perf_tg.calculate_traffic_and_stats(packet, duration, send_mpps)
+
     def send_packets(
         self,
         packets: list[Packet],
diff --git a/dts/tests/TestSuite_single_core_forward_perf.py b/dts/tests/TestSuite_single_core_forward_perf.py
new file mode 100644
index 0000000000..eab82443b6
--- /dev/null
+++ b/dts/tests/TestSuite_single_core_forward_perf.py
@@ -0,0 +1,139 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 University of New Hampshire
+
+"""Single core forwarding performance test suite.
+
+This suite measures the amount of packets which can be forwarded by DPDK using a single core.
+The testsuites takes in as parameters a set of parameters, each consisting of a frame size,
+Tx/Rx descriptor count, and the expected MPPS to be forwarded by the DPDK application. The
+test leverages a performance traffic generator to send traffic at two paired TestPMD interfaces
+on the SUT system, which forward to one another and then back to the traffic generator's ports.
+The aggregate packets forwarded by the two TestPMD ports are compared against the expected MPPS
+baseline which is given in the test config, in order to determine the test result.
+"""
+
+from scapy.layers.inet import IP
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
+
+from api.capabilities import (
+    LinkTopology,
+    requires_link_topology,
+)
+from api.testpmd import TestPmd
+from api.testpmd.config import RXRingParams, TXRingParams
+from framework.test_suite import BaseConfig, TestSuite, perf_test
+
+
+class Config(BaseConfig):
+    """Performance test metrics."""
+
+    test_parameters: list[dict[str, int | float]] = [
+        {"frame_size": 64, "num_descriptors": 1024, "expected_mpps": 1.00},
+        {"frame_size": 128, "num_descriptors": 1024, "expected_mpps": 1.00},
+        {"frame_size": 256, "num_descriptors": 1024, "expected_mpps": 1.00},
+        {"frame_size": 512, "num_descriptors": 1024, "expected_mpps": 1.00},
+        {"frame_size": 1024, "num_descriptors": 1024, "expected_mpps": 1.00},
+        {"frame_size": 1518, "num_descriptors": 1024, "expected_mpps": 1.00},
+    ]
+    delta_tolerance: float = 0.05
+
+
+@requires_link_topology(LinkTopology.TWO_LINKS)
+class TestSingleCoreForwardPerf(TestSuite):
+    """Single core forwarding performance test suite."""
+
+    config: Config
+
+    def set_up_suite(self):
+        """Set up the test suite."""
+        self.test_parameters = self.config.test_parameters
+        self.delta_tolerance = self.config.delta_tolerance
+
+    def _transmit(self, testpmd: TestPmd, frame_size: int) -> float:
+        """Create a testpmd session with every rule in the given list, verify jump behavior.
+
+        Args:
+            testpmd: The testpmd shell to use for forwarding packets
+            frame_size: The size of the frame to transmit
+
+        Returns:
+            The MPPS (millions of packets per second) forwarded by the SUT.
+        """
+        # Build packet with dummy values, and account for the 14B and 20B Ether and IP headers
+        packet = (
+            Ether(src="52:00:00:00:00:00")
+            / IP(src="1.2.3.4", dst="192.18.1.0")
+            / Raw(load="x" * (frame_size - 14 - 20))
+        )
+
+        testpmd.start()
+
+        # Transmit for 30 seconds.
+        stats = self.assess_performance_by_packet(packet, duration=30)
+
+        rx_mpps = stats.rx_pps / 1000000
+
+        return rx_mpps
+
+    def _show_stats_table(self, test_parameters: list[dict[str, int | float]]) -> None:
+        """Create a testpmd session with every rule in the given list, verify jump behavior.
+
+        Args:
+            test_parameters: The expected and real stats per set of test parameters.
+        """
+        header = f"{'Frame Size':>12} | {'TXD/RXD':>12} | {'Real MPPS':>12} | {'Expected MPPS':>14}"
+        print("-" * len(header))
+        print(header)
+        print("-" * len(header))
+        for params in test_parameters:
+            print(f"{params['frame_size']:>12} | {params['num_descriptors']:>12} | ", end="")
+            print(f"{params['measured_mpps']:>12.2f} | {params['expected_mpps']:>14.2f}")
+            print("-" * len(header))
+
+    @perf_test
+    def single_core_forward_perf(self) -> None:
+        """Validate expected single core forwarding performance.
+
+        Steps:
+            * Create a packet according to the frame size specified in the test config.
+            * Transmit from the traffic generator's ports 0 and 1 at above the expect
+            * Forward on TestPMD's interfaces 0 and 1 with 1 core.
+
+        Verify:
+            * The resulting MPPS forwarded is greater than expected_mpps*(1-delta_tolerance)
+        """
+        # Find SUT DPDK driver to determine driver specific performance optimization flags
+        sut_dpdk_driver = self._ctx.sut_node.config.ports[0].os_driver_for_dpdk
+
+        driver_specific_testpmd_args = {}
+
+        if sut_dpdk_driver == "mlx5_core":
+            driver_specific_testpmd_args["burst"] = 64
+            driver_specific_testpmd_args["mbcache"] = 512
+        elif sut_dpdk_driver == "i40e":
+            driver_specific_testpmd_args["rx_queues"] = 2
+            driver_specific_testpmd_args["tx_queues"] = 2
+
+        for params in self.test_parameters:
+            frame_size = params["frame_size"]
+            num_descriptors = params["num_descriptors"]
+
+            with TestPmd(
+                tx_ring=TXRingParams(descriptors=num_descriptors),
+                rx_ring=RXRingParams(descriptors=num_descriptors),
+                nb_cores=1,
+                **driver_specific_testpmd_args,
+            ) as testpmd:
+                params["measured_mpps"] = self._transmit(testpmd, frame_size)
+
+        self._show_stats_table(self.test_parameters)
+
+        for params in self.test_parameters:
+            self.verify(
+                params["measured_mpps"] >= params["expected_mpps"] * (1 - self.delta_tolerance),
+                f"""Packets forwarded is less than {(1 -self.delta_tolerance)*100}%
+                of the expected baseline.
+                Measured MPPS = {params["measured_mpps"]}
+                Expected MPPS = {params["expected_mpps"]}""",
+            )
-- 
2.49.0