* [RFC PATCH 1/2] dts: update parsing for cryptodev latency
@ 2026-04-28 18:15 Andrew Bailey
2026-04-28 18:15 ` [RFC PATCH 2/2] dts: add latency coverage for cryptodev testing Andrew Bailey
2026-05-13 15:27 ` [PATCH v1 1/2] dts: update parsing for cryptodev latency Andrew Bailey
0 siblings, 2 replies; 4+ messages in thread
From: Andrew Bailey @ 2026-04-28 18:15 UTC (permalink / raw)
To: dev, luca.vizzarro, patrickrobb1997
Cc: lylavoie, ahassick, knimoji, dmarx, Andrew Bailey
Previously, the parsing for cryptodev latency output would fail to parse
due to an incorrect regex. This commit updates this regex to properly
capture the burst size attribute for latency results.
Bugzilla ID: 1938
Fixes: 8ee2df9da125 ("dts: add cryptodev package")
Signed-off-by: Andrew Bailey <abailey@iol.unh.edu>
---
dts/api/cryptodev/types.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/dts/api/cryptodev/types.py b/dts/api/cryptodev/types.py
index df73a86fa4..0942d08371 100644
--- a/dts/api/cryptodev/types.py
+++ b/dts/api/cryptodev/types.py
@@ -69,7 +69,7 @@ class LatencyResults(CryptodevResults):
)
#: burst size ran with app
burst_size: int = field(
- metadata=TextParser.find_int(rf"Burst(?:.*\n\s+\d+\s+){2}?(?: size:\s+)?(\d+)"),
+ metadata=TextParser.find_int(r"Burst size:\s+(\d+)"),
)
#: total operations ran
total_ops: int = field(metadata=TextParser.find_int(r"total operations:\s+(\d+)"))
--
2.50.1
^ permalink raw reply related [flat|nested] 4+ messages in thread
* [RFC PATCH 2/2] dts: add latency coverage for cryptodev testing
2026-04-28 18:15 [RFC PATCH 1/2] dts: update parsing for cryptodev latency Andrew Bailey
@ 2026-04-28 18:15 ` Andrew Bailey
2026-05-13 15:27 ` [PATCH v1 1/2] dts: update parsing for cryptodev latency Andrew Bailey
1 sibling, 0 replies; 4+ messages in thread
From: Andrew Bailey @ 2026-04-28 18:15 UTC (permalink / raw)
To: dev, luca.vizzarro, patrickrobb1997
Cc: lylavoie, ahassick, knimoji, dmarx, Andrew Bailey
Currently, next DTS only has cryptodev testing coverage for throughput
metrics. This patch adds a test suite to include latency testing for
crypto devices.
Signed-off-by: Andrew Bailey <abailey@iol.unh.edu>
---
.../dts/tests.TestSuite_cryptodev_latency.rst | 8 +
dts/tests/TestSuite_cryptodev_latency.py | 356 ++++++++++++++++++
2 files changed, 364 insertions(+)
create mode 100644 doc/api/dts/tests.TestSuite_cryptodev_latency.rst
create mode 100644 dts/tests/TestSuite_cryptodev_latency.py
diff --git a/doc/api/dts/tests.TestSuite_cryptodev_latency.rst b/doc/api/dts/tests.TestSuite_cryptodev_latency.rst
new file mode 100644
index 0000000000..1bf4a0f4b0
--- /dev/null
+++ b/doc/api/dts/tests.TestSuite_cryptodev_latency.rst
@@ -0,0 +1,8 @@
+.. SPDX-License-Identifier: BSD-3-Clause
+
+cryptodev_latency Test Suite
+============================
+
+.. automodule:: tests.TestSuite_cryptodev_latency
+ :members:
+ :show-inheritance:
diff --git a/dts/tests/TestSuite_cryptodev_latency.py b/dts/tests/TestSuite_cryptodev_latency.py
new file mode 100644
index 0000000000..39c9ee02c4
--- /dev/null
+++ b/dts/tests/TestSuite_cryptodev_latency.py
@@ -0,0 +1,356 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2026 University of New Hampshire
+
+"""DPDK cryptodev performance test suite.
+
+The main goal of this test suite is to utilize the dpdk-test-cryptodev application to gather
+performance metrics for various cryptographic operations supported by DPDK cryptodev-pmd.
+It will then compare the results against a predefined baseline given in the test_config file to
+ensure performance standards are met.
+"""
+
+from api.capabilities import (
+ LinkTopology,
+ requires_link_topology,
+)
+from api.cryptodev import Cryptodev
+from api.cryptodev.config import (
+ AeadAlgName,
+ AuthenticationAlgorithm,
+ AuthenticationOpMode,
+ CipherAlgorithm,
+ DeviceType,
+ EncryptDecryptSwitch,
+ ListWrapper,
+ OperationType,
+ TestType,
+ get_device_from_str,
+)
+from api.cryptodev.types import (
+ CryptodevResults,
+)
+from api.test import skip, verify
+from framework.context import get_ctx
+from framework.test_suite import BaseConfig, TestSuite, crypto_test
+from framework.testbed_model.virtual_device import VirtualDevice
+
+config_list: list[dict[str, int | float | str]] = [
+ {"buff_size": 64, "avg_cycles": 9999.00, "avg_time_us": 9999.0},
+ {"buff_size": 512, "avg_cycles": 9999.00, "avg_time_us": 9999.0},
+ {"buff_size": 2048, "avg_cycles": 9999.00, "avg_time_us": 9999.0},
+]
+
+TOTAL_OPS = 10_000_000
+
+
+class Config(BaseConfig):
+ """Performance test metrics.
+
+ Attributes:
+ delta_tolerance: The allowed tolerance below a given baseline.
+ latency_test_parameters: The test parameters to use in the test suite.
+ """
+
+ delta_tolerance: float = 0.05
+
+ latency_test_parameters: dict[str, list[dict[str, int | float | str]]] = {
+ "aes_gcm": config_list,
+ "aes_cbc_sha1_hmac": config_list,
+ "aes_cbc": config_list,
+ "vdev_aesni_gcm": config_list,
+ "vdev_aesni_mb": config_list,
+ "vdev_aesni_mb_cipher_then_auth": config_list,
+ }
+
+
+@requires_link_topology(LinkTopology.NO_LINK)
+class TestCryptodevLatency(TestSuite):
+ """DPDK Crypto Device Testing Suite."""
+
+ config: Config
+
+ def set_up_suite(self) -> None:
+ """Set up the test suite."""
+ self.latency_test_parameters: dict[str, list[dict[str, int | float | str]]] = (
+ self.config.latency_test_parameters
+ )
+ self.delta_tolerance: float = self.config.delta_tolerance
+ self.device_type: DeviceType | None = get_device_from_str(
+ str(get_ctx().sut_node.crypto_device_type)
+ )
+ self.buffer_sizes = {}
+
+ for k, v in self.latency_test_parameters.items():
+ self.buffer_sizes[k] = ListWrapper([int(run["buff_size"]) for run in v])
+
+ def _print_stats(self, test_vals: list[dict[str, int | float | str]]) -> None:
+ element_len = len("Avg Time us Target")
+ border_len = (element_len + 1) * (len(test_vals[0]))
+
+ print(f"{'Latency Results'.center(border_len)}\n{'=' * border_len}")
+ for k, v in test_vals[0].items():
+ print(f"|{k.title():<{element_len}}", end="")
+ print(f"|\n{'='*border_len}")
+
+ for test_val in test_vals:
+ for k, v in test_val.items():
+ print(f"|{v:<{element_len}}", end="")
+ print(f"|\n{'='*border_len}")
+
+ def _verify_latency(
+ self,
+ results: list[CryptodevResults],
+ key: str,
+ ) -> list[dict[str, int | float | str]]:
+ result_list: list[dict[str, int | float | str]] = []
+
+ for result in results:
+ # get the corresponding baseline for the current buffer size
+ parameters: dict[str, int | float | str] = list(
+ filter(
+ lambda x: x["buff_size"] == result.buffer_size,
+ self.latency_test_parameters[key],
+ )
+ )[0]
+ test_result = True
+ expected_cycles = parameters["avg_cycles"]
+ expected_time_us = parameters["avg_time_us"]
+ measured_time_delta = abs(
+ (getattr(result, "avg_time_us") - expected_time_us) / expected_time_us
+ )
+ measured_cycles_delta = abs(
+ (getattr(result, "avg_cycles") - expected_cycles) / expected_cycles
+ )
+
+ # result did not meet the given cycles parameter, check if within delta.
+ if getattr(result, "avg_cycles") > expected_cycles:
+ if self.delta_tolerance < measured_cycles_delta:
+ test_result = False
+ # result did not meet the given time parameter, check if within delta.
+ if getattr(result, "avg_time_us") > expected_time_us:
+ if self.delta_tolerance < measured_time_delta:
+ test_result = False
+ result_list.append(
+ {
+ "Buffer Size": parameters["buff_size"],
+ "delta tolerance": self.delta_tolerance,
+ "cycles delta": round(measured_cycles_delta, 5),
+ "Avg cycles": round(getattr(result, "avg_cycles"), 5),
+ "Avg cycles target": expected_cycles,
+ "time delta": round(measured_time_delta, 5),
+ "Avg time us": round(getattr(result, "avg_time_us"), 5),
+ "Avg time us target": expected_time_us,
+ "passed": "PASS" if test_result else "FAIL",
+ }
+ )
+ return result_list
+
+ @crypto_test
+ def aes_gcm(self) -> None:
+ """aes_gcm latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_gcm" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ burst_sz=32,
+ buffer_sz=ListWrapper(self.buffer_sizes["aes_gcm"]),
+ devtype=self.device_type,
+ optype=OperationType.aead,
+ aead_algo=AeadAlgName.aes_gcm,
+ aead_op=EncryptDecryptSwitch.encrypt,
+ aead_key_sz=16,
+ aead_iv_sz=12,
+ aead_aad_sz=16,
+ digest_sz=16,
+ total_ops=TOTAL_OPS,
+ )
+ results = self._verify_latency(app.run_app(), "aes_gcm")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aes_cbc_sha1_hmac(self) -> None:
+ """aes_cbc_sha1_hmac latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_cbc_sha1_hmac" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ burst_sz=32,
+ buffer_sz=ListWrapper(self.buffer_sizes["aes_cbc_sha1_hmac"]),
+ devtype=self.device_type,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ auth_algo=AuthenticationAlgorithm.sha1_hmac,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=64,
+ digest_sz=12,
+ total_ops=TOTAL_OPS,
+ )
+ results = self._verify_latency(app.run_app(), "aes_cbc_sha1_hmac")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aes_cbc(self) -> None:
+ """aes_cbc latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_cbc" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ burst_sz=32,
+ buffer_sz=ListWrapper(self.buffer_sizes["aes_cbc"]),
+ devtype=self.device_type,
+ optype=OperationType.cipher_only,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ total_ops=TOTAL_OPS,
+ )
+ results = self._verify_latency(app.run_app(), "aes_cbc")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def vdev_aesni_gcm(self) -> None:
+ """aesni_gcm virtual device latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "vdev_aesni_gcm" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_aesni_gcm0")],
+ burst_sz=32,
+ buffer_sz=self.buffer_sizes["vdev_aesni_gcm"],
+ devtype=DeviceType.crypto_aesni_gcm,
+ optype=OperationType.aead,
+ aead_op=EncryptDecryptSwitch.encrypt,
+ aead_key_sz=16,
+ aead_iv_sz=12,
+ aead_aad_sz=16,
+ digest_sz=16,
+ total_ops=TOTAL_OPS,
+ )
+ results = self._verify_latency(app.run_app(), "vdev_aesni_gcm")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "latency fell more than the delta tolerance")
+
+ @crypto_test
+ def vdev_aesni_mb(self) -> None:
+ """aesni_mb vdev latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "vdev_aesni_mb" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_aesni_mb0")],
+ burst_sz=32,
+ buffer_sz=self.buffer_sizes["vdev_aesni_mb"],
+ devtype=DeviceType.crypto_aesni_mb,
+ optype=OperationType.cipher_only,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ total_ops=TOTAL_OPS,
+ )
+ results = self._verify_latency(app.run_app(), "vdev_aesni_mb")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "Gbps fell below delta tolerance")
+
+ @crypto_test
+ def vdev_aesni_mb_cipher_then_auth(self) -> None:
+ """aesni_mb vdev cipher and auth latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "vdev_aesni_mb" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_aesni_mb0")],
+ burst_sz=32,
+ buffer_sz=self.buffer_sizes["vdev_aesni_mb"],
+ devtype=DeviceType.crypto_aesni_mb,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ auth_algo=AuthenticationAlgorithm.sha1_hmac,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=64,
+ digest_sz=12,
+ total_ops=TOTAL_OPS,
+ )
+ results = self._verify_latency(app.run_app(), "vdev_aesni_mb_cipher_then_auth")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
--
2.50.1
^ permalink raw reply related [flat|nested] 4+ messages in thread
* [PATCH v1 1/2] dts: update parsing for cryptodev latency
2026-04-28 18:15 [RFC PATCH 1/2] dts: update parsing for cryptodev latency Andrew Bailey
2026-04-28 18:15 ` [RFC PATCH 2/2] dts: add latency coverage for cryptodev testing Andrew Bailey
@ 2026-05-13 15:27 ` Andrew Bailey
2026-05-13 15:27 ` [PATCH v1 2/2] dts: add latency coverage for cryptodev testing Andrew Bailey
1 sibling, 1 reply; 4+ messages in thread
From: Andrew Bailey @ 2026-05-13 15:27 UTC (permalink / raw)
To: luca.vizzarro, patrickrobb1997
Cc: dev, lylavoie, ahassick, knimoji, Andrew Bailey
Previously, the parsing for cryptodev latency output would fail to parse
due to an incorrect regex. This commit updates this regex to properly
capture the burst size attribute for latency results.
Bugzilla ID: 1938
Fixes: 8ee2df9da125 ("dts: add cryptodev package")
Signed-off-by: Andrew Bailey <abailey@iol.unh.edu>
---
dts/api/cryptodev/__init__.py | 3 +--
dts/api/cryptodev/types.py | 4 ++--
2 files changed, 3 insertions(+), 4 deletions(-)
diff --git a/dts/api/cryptodev/__init__.py b/dts/api/cryptodev/__init__.py
index a4fafc3713..15cffcb409 100644
--- a/dts/api/cryptodev/__init__.py
+++ b/dts/api/cryptodev/__init__.py
@@ -117,7 +117,6 @@ def run_app(self, num_vfs: int = 1) -> list[CryptodevResults]:
f"dependencies missing for virtual device {self._app_params['vdevs'][0].name}"
)
raise e
-
regex = r"^\s+\d+.*$"
parser_options = re.MULTILINE
parser: type[CryptodevResults]
@@ -126,7 +125,7 @@ def run_app(self, num_vfs: int = 1) -> list[CryptodevResults]:
case TestType.throughput:
parser = ThroughputResults
case TestType.latency:
- regex = r"total operations:.*time[^\n]*"
+ regex = r"total operations:.*?time[^\n]*"
parser_options |= re.DOTALL
parser = LatencyResults
case TestType.pmd_cyclecount:
diff --git a/dts/api/cryptodev/types.py b/dts/api/cryptodev/types.py
index df73a86fa4..7b4fd17cb6 100644
--- a/dts/api/cryptodev/types.py
+++ b/dts/api/cryptodev/types.py
@@ -65,11 +65,11 @@ class LatencyResults(CryptodevResults):
#: buffer size ran with app
buffer_size: int = field(
- metadata=TextParser.find_int(r"Buf(?:.*\n\s+\d+\s+)?(?:fer size:\s+)?(\d+)"),
+ metadata=TextParser.find_int(r"Buffer size:\s+(\d+)"),
)
#: burst size ran with app
burst_size: int = field(
- metadata=TextParser.find_int(rf"Burst(?:.*\n\s+\d+\s+){2}?(?: size:\s+)?(\d+)"),
+ metadata=TextParser.find_int(r"Burst size:\s+(\d+)"),
)
#: total operations ran
total_ops: int = field(metadata=TextParser.find_int(r"total operations:\s+(\d+)"))
--
2.50.1
^ permalink raw reply related [flat|nested] 4+ messages in thread
* [PATCH v1 2/2] dts: add latency coverage for cryptodev testing
2026-05-13 15:27 ` [PATCH v1 1/2] dts: update parsing for cryptodev latency Andrew Bailey
@ 2026-05-13 15:27 ` Andrew Bailey
0 siblings, 0 replies; 4+ messages in thread
From: Andrew Bailey @ 2026-05-13 15:27 UTC (permalink / raw)
To: luca.vizzarro, patrickrobb1997
Cc: dev, lylavoie, ahassick, knimoji, Andrew Bailey
Currently, next DTS only has cryptodev testing coverage for throughput
metrics. This patch adds a test suite to include latency testing for
crypto devices.
Signed-off-by: Andrew Bailey <abailey@iol.unh.edu>
---
.../dts/tests.TestSuite_cryptodev_latency.rst | 8 +
dts/tests/TestSuite_cryptodev_latency.py | 695 ++++++++++++++++++
2 files changed, 703 insertions(+)
create mode 100644 doc/api/dts/tests.TestSuite_cryptodev_latency.rst
create mode 100644 dts/tests/TestSuite_cryptodev_latency.py
diff --git a/doc/api/dts/tests.TestSuite_cryptodev_latency.rst b/doc/api/dts/tests.TestSuite_cryptodev_latency.rst
new file mode 100644
index 0000000000..1bf4a0f4b0
--- /dev/null
+++ b/doc/api/dts/tests.TestSuite_cryptodev_latency.rst
@@ -0,0 +1,8 @@
+.. SPDX-License-Identifier: BSD-3-Clause
+
+cryptodev_latency Test Suite
+============================
+
+.. automodule:: tests.TestSuite_cryptodev_latency
+ :members:
+ :show-inheritance:
diff --git a/dts/tests/TestSuite_cryptodev_latency.py b/dts/tests/TestSuite_cryptodev_latency.py
new file mode 100644
index 0000000000..54edd4e0a6
--- /dev/null
+++ b/dts/tests/TestSuite_cryptodev_latency.py
@@ -0,0 +1,695 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2026 University of New Hampshire
+
+"""DPDK cryptodev performance test suite.
+
+The main goal of this test suite is to utilize the dpdk-test-cryptodev application to gather
+performance metrics for various cryptographic operations supported by DPDK cryptodev-pmd.
+It will then compare the results against a predefined baseline given in the test_config file to
+ensure performance standards are met.
+"""
+
+from api.capabilities import (
+ LinkTopology,
+ requires_link_topology,
+)
+from api.cryptodev import Cryptodev
+from api.cryptodev.config import (
+ AeadAlgName,
+ AuthenticationAlgorithm,
+ AuthenticationOpMode,
+ CipherAlgorithm,
+ DeviceType,
+ EncryptDecryptSwitch,
+ ListWrapper,
+ OperationType,
+ TestType,
+ get_device_from_str,
+)
+from api.cryptodev.types import (
+ CryptodevResults,
+)
+from api.test import skip, verify
+from framework.context import get_ctx
+from framework.test_suite import BaseConfig, TestSuite, crypto_test
+from framework.testbed_model.virtual_device import VirtualDevice
+
+config_list: list[dict[str, int | float | str]] = [
+ {"buff_size": 64, "avg_cycles": 99_999.00, "avg_time_us": 9999.0},
+ {"buff_size": 512, "avg_cycles": 99_999.00, "avg_time_us": 9999.0},
+ {"buff_size": 2048, "avg_cycles": 99_999.00, "avg_time_us": 9999.0},
+]
+
+TOTAL_OPS = 10_000_000
+
+
+class Config(BaseConfig):
+ """Performance test metrics.
+
+ Attributes:
+ delta_tolerance: The allowed tolerance below a given baseline.
+ latency_test_parameters: The test parameters to use in the test suite.
+ """
+
+ delta_tolerance: float = 0.05
+
+ latency_test_parameters: dict[str, list[dict[str, int | float | str]]] = {
+ "aes_cbc": config_list,
+ "aes_cbc_sha1_hmac": config_list,
+ "aes_cbc_sha2_hmac": config_list,
+ "aes_docsisbpi": config_list,
+ "aes_gcm": config_list,
+ "kasumi_f8_kasumi_f9": config_list,
+ "snow3g_uea2_snow3g_uia2": config_list,
+ "zuc_eea3_zuc_eia3": config_list,
+ "aesni_gcm_vdev": config_list,
+ "aesni_mb_cipher_then_auth_vdev": config_list,
+ "aesni_mb_vdev": config_list,
+ "kasumi_vdev": config_list,
+ "open_ssl_vdev": config_list,
+ "snow3g_vdev": config_list,
+ "zuc_vdev": config_list,
+ }
+
+
+@requires_link_topology(LinkTopology.NO_LINK)
+class TestCryptodevLatency(TestSuite):
+ """DPDK Crypto Device Testing Suite."""
+
+ config: Config
+
+ def set_up_suite(self) -> None:
+ """Set up the test suite."""
+ self.latency_test_parameters: dict[str, list[dict[str, int | float | str]]] = (
+ self.config.latency_test_parameters
+ )
+ self.delta_tolerance: float = self.config.delta_tolerance
+ self.device_type: DeviceType | None = get_device_from_str(
+ str(get_ctx().sut_node.crypto_device_type)
+ )
+ self.buffer_sizes = {}
+
+ for k, v in self.latency_test_parameters.items():
+ self.buffer_sizes[k] = ListWrapper([int(run["buff_size"]) for run in v])
+
+ def _print_stats(self, test_vals: list[dict[str, int | float | str]]) -> None:
+ element_len = len("Avg Time us Target")
+ border_len = (element_len + 1) * (len(test_vals[0]))
+
+ print(f"{'Latency Results'.center(border_len)}\n{'=' * border_len}")
+ for k, v in test_vals[0].items():
+ print(f"|{k.title():<{element_len}}", end="")
+ print(f"|\n{'='*border_len}")
+
+ for test_val in test_vals:
+ for k, v in test_val.items():
+ print(f"|{v:<{element_len}}", end="")
+ print(f"|\n{'='*border_len}")
+
+ def _verify_latency(
+ self,
+ results: list[CryptodevResults],
+ key: str,
+ ) -> list[dict[str, int | float | str]]:
+ result_list: list[dict[str, int | float | str]] = []
+
+ for result in results:
+ # get the corresponding baseline for the current buffer size
+ parameters: dict[str, int | float | str] = list(
+ filter(
+ lambda x: x["buff_size"] == result.buffer_size,
+ self.latency_test_parameters[key],
+ )
+ )[0]
+ test_result = True
+ expected_cycles = parameters["avg_cycles"]
+ expected_time_us = parameters["avg_time_us"]
+ measured_time_delta = abs(
+ (getattr(result, "avg_time_us") - expected_time_us) / expected_time_us
+ )
+ measured_cycles_delta = abs(
+ (getattr(result, "avg_cycles") - expected_cycles) / expected_cycles
+ )
+
+ # result did not meet the given cycles parameter, check if within delta.
+ if getattr(result, "avg_cycles") > expected_cycles:
+ if self.delta_tolerance < measured_cycles_delta:
+ test_result = False
+ # result did not meet the given time parameter, check if within delta.
+ if getattr(result, "avg_time_us") > expected_time_us:
+ if self.delta_tolerance < measured_time_delta:
+ test_result = False
+ result_list.append(
+ {
+ "Buffer Size": parameters["buff_size"],
+ "delta tolerance": self.delta_tolerance,
+ "cycles delta": round(measured_cycles_delta, 5),
+ "Avg cycles": round(getattr(result, "avg_cycles"), 5),
+ "Avg cycles target": expected_cycles,
+ "time delta": round(measured_time_delta, 5),
+ "Avg time us": round(getattr(result, "avg_time_us"), 5),
+ "Avg time us target": expected_time_us,
+ "passed": "PASS" if test_result else "FAIL",
+ }
+ )
+ return result_list
+
+ @crypto_test
+ def aes_cbc(self) -> None:
+ """aes_cbc latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_cbc" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_only,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aes_cbc"],
+ )
+ results = self._verify_latency(app.run_app(), "aes_cbc")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aes_cbc_sha1_hmac(self) -> None:
+ """aes_cbc_sha1_hmac latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_cbc_sha1_hmac" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ auth_algo=AuthenticationAlgorithm.sha1_hmac,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=64,
+ digest_sz=12,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aes_cbc_sha1_hmac"],
+ )
+ results = self._verify_latency(app.run_app(), "aes_cbc_sha1_hmac")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aes_cbc_sha2_hmac(self) -> None:
+ """aes_cbc_sha2_hmac latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_cbc_sha2_hmac" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ auth_algo=AuthenticationAlgorithm.sha2_256_hmac,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=64,
+ digest_sz=32,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aes_cbc_sha2_hmac"],
+ )
+ results = self._verify_latency(app.run_app(), "aes_cbc_sha2_hmac")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aes_docsisbpi(self) -> None:
+ """aes_docsisbpi latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_docsisbpi" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_only,
+ cipher_algo=CipherAlgorithm.aes_docsisbpi,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=32,
+ cipher_iv_sz=16,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aes_docsisbpi"],
+ )
+ results = self._verify_latency(app.run_app(), "aes_docsisbpi")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aes_gcm(self) -> None:
+ """aes_gcm latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aes_gcm" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.aead,
+ aead_algo=AeadAlgName.aes_gcm,
+ aead_op=EncryptDecryptSwitch.encrypt,
+ aead_key_sz=16,
+ aead_iv_sz=12,
+ aead_aad_sz=16,
+ digest_sz=16,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aes_gcm"],
+ )
+ results = self._verify_latency(app.run_app(), "aes_gcm")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def kasumi_f8_kasumi_f9(self) -> None:
+ """kasumi_f8_kasumi_f9 latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "kasumi_f8_kasumi_f9" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.kasumi_f8,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=8,
+ auth_algo=AuthenticationAlgorithm.kasumi_f9,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=16,
+ digest_sz=4,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["kasumi_f8_kasumi_f9"],
+ )
+ results = self._verify_latency(app.run_app(), "kasumi_f8_kasumi_f9")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def snow3g_uea2_snow3g_uia2(self) -> None:
+ """snow3g_uea2_snow3g_uia2 latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "snow3g_uea2_snow3g_uia2" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.snow3g_uea2,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ auth_algo=AuthenticationAlgorithm.snow3g_uia2,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=16,
+ auth_iv_sz=16,
+ digest_sz=4,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["snow3g_uea2_snow3g_uia2"],
+ )
+ results = self._verify_latency(app.run_app(), "snow3g_uea2_snow3g_uia2")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def zuc_eea3_zuc_eia3(self) -> None:
+ """zuc_eea3_zuc_eia3 vdev cipher and auth latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "zuc_eea3_zuc_eia3" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ devtype=self.device_type,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.zuc_eea3,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ auth_algo=AuthenticationAlgorithm.zuc_eia3,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=16,
+ auth_iv_sz=16,
+ digest_sz=4,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["zuc_eea3_zuc_eia3"],
+ )
+ results = self._verify_latency(app.run_app(), "zuc_eea3_zuc_eia3")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ # BEGIN VDEV TESTS
+
+ @crypto_test
+ def aesni_gcm_vdev(self) -> None:
+ """aesni_gcm virtual device latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aesni_gcm_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_aesni_gcm0")],
+ devtype=DeviceType.crypto_aesni_gcm,
+ optype=OperationType.aead,
+ aead_op=EncryptDecryptSwitch.encrypt,
+ aead_key_sz=16,
+ aead_iv_sz=12,
+ aead_aad_sz=16,
+ digest_sz=16,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aesni_gcm_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "aesni_gcm_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "latency fell more than the delta tolerance")
+
+ @crypto_test
+ def aesni_mb_cipher_then_auth_vdev(self) -> None:
+ """aesni_mb vdev cipher and auth latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aesni_mb_cipher_then_auth_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_aesni_mb0")],
+ devtype=DeviceType.crypto_aesni_mb,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ auth_algo=AuthenticationAlgorithm.sha1_hmac,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=64,
+ digest_sz=12,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aesni_mb_cipher_then_auth_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "aesni_mb_cipher_then_auth_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(
+ result["passed"] == "PASS",
+ "latency fell more than the delta tolerance below baseline",
+ )
+
+ @crypto_test
+ def aesni_mb_vdev(self) -> None:
+ """aesni_mb vdev latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "aesni_mb_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_aesni_mb0")],
+ devtype=DeviceType.crypto_aesni_mb,
+ optype=OperationType.cipher_only,
+ cipher_algo=CipherAlgorithm.aes_cbc,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["aesni_mb_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "aesni_mb_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "Gbps fell below delta tolerance")
+
+ @crypto_test
+ def kasumi_vdev(self) -> None:
+ """Kasumi vdev latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "kasumi_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_kasumi0")],
+ devtype=DeviceType.crypto_kasumi,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.kasumi_f8,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=8,
+ auth_algo=AuthenticationAlgorithm.kasumi_f9,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=16,
+ digest_sz=4,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["kasumi_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "kasmui_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "Gbps fell below delta tolerance")
+
+ @crypto_test
+ def open_ssl_vdev(self) -> None:
+ """open_ssl vdev latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "open_ssl_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_openssl0")],
+ devtype=DeviceType.crypto_openssl,
+ optype=OperationType.aead,
+ aead_algo=AeadAlgName.aes_gcm,
+ aead_op=EncryptDecryptSwitch.encrypt,
+ aead_key_sz=16,
+ aead_iv_sz=16,
+ aead_aad_sz=16,
+ digest_sz=16,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["open_ssl_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "open_ssl_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "Gbps fell below delta tolerance")
+
+ @crypto_test
+ def snow3g_vdev(self) -> None:
+ """snow3g vdev latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "snow3g_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_snow3g0")],
+ devtype=DeviceType.crypto_snow3g,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.snow3g_uea2,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ auth_algo=AuthenticationAlgorithm.snow3g_uia2,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=16,
+ auth_iv_sz=16,
+ digest_sz=16,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["open_ssl_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "open_ssl_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "Gbps fell below delta tolerance")
+
+ @crypto_test
+ def zuc_vdev(self) -> None:
+ """Zuc vdev latency test.
+
+ Steps:
+ * Create a cryptodev instance with provided device type and buffer sizes.
+ Verify:
+ * The latency is below or within delta of provided baseline.
+
+ Raises:
+ SkippedTestException: When configuration is not provided.
+ """
+ if "zuc_vdev" not in self.latency_test_parameters:
+ skip("test not configured")
+ app = Cryptodev(
+ ptest=TestType.latency,
+ vdevs=[VirtualDevice("crypto_zuc0")],
+ devtype=DeviceType.crypto_zuc,
+ optype=OperationType.cipher_then_auth,
+ cipher_algo=CipherAlgorithm.zuc_eea3,
+ cipher_op=EncryptDecryptSwitch.encrypt,
+ cipher_key_sz=16,
+ cipher_iv_sz=16,
+ auth_algo=AuthenticationAlgorithm.zuc_eia3,
+ auth_op=AuthenticationOpMode.generate,
+ auth_key_sz=16,
+ auth_iv_sz=16,
+ digest_sz=4,
+ burst_sz=32,
+ total_ops=TOTAL_OPS,
+ buffer_sz=self.buffer_sizes["zuc_vdev"],
+ )
+ results = self._verify_latency(app.run_app(), "zuc_vdev")
+ self._print_stats(results)
+ for result in results:
+ verify(result["passed"] == "PASS", "Gbps fell below delta tolerance")
--
2.50.1
^ permalink raw reply related [flat|nested] 4+ messages in thread
end of thread, other threads:[~2026-05-13 15:27 UTC | newest]
Thread overview: 4+ messages (download: mbox.gz follow: Atom feed
-- links below jump to the message on this page --
2026-04-28 18:15 [RFC PATCH 1/2] dts: update parsing for cryptodev latency Andrew Bailey
2026-04-28 18:15 ` [RFC PATCH 2/2] dts: add latency coverage for cryptodev testing Andrew Bailey
2026-05-13 15:27 ` [PATCH v1 1/2] dts: update parsing for cryptodev latency Andrew Bailey
2026-05-13 15:27 ` [PATCH v1 2/2] dts: add latency coverage for cryptodev testing Andrew Bailey
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox