Add a simple test for USO. Can be used with netdevsim or real hardware.
Tests both ipv4 and ipv6 with several full segments and a partial
segment.

Suggested-by: Jakub Kicinski <kuba@kernel.org>
Reviewed-by: Pavan Chebbi <pavan.chebbi@broadcom.com>
Signed-off-by: Joe Damato <joe@dama.to>
---
 v5:
   - Added Pavan's Reviewed-by. No functional changes.

 v4:
   - Fix python linter issues (unused imports, docstring, etc).

 rfcv2:
   - new in rfcv2

 tools/testing/selftests/drivers/net/Makefile |  1 +
 tools/testing/selftests/drivers/net/uso.py   | 96 ++++++++++++++++++++
 2 files changed, 97 insertions(+)
 create mode 100755 tools/testing/selftests/drivers/net/uso.py

diff --git a/tools/testing/selftests/drivers/net/Makefile b/tools/testing/selftests/drivers/net/Makefile
index 7c7fa75b80c2..335c2ce4b9ab 100644
--- a/tools/testing/selftests/drivers/net/Makefile
+++ b/tools/testing/selftests/drivers/net/Makefile
@@ -21,6 +21,7 @@ TEST_PROGS := \
 	ring_reconfig.py \
 	shaper.py \
 	stats.py \
+	uso.py \
 	xdp.py \
 # end of TEST_PROGS
 
diff --git a/tools/testing/selftests/drivers/net/uso.py b/tools/testing/selftests/drivers/net/uso.py
new file mode 100755
index 000000000000..2ddeae99b4d6
--- /dev/null
+++ b/tools/testing/selftests/drivers/net/uso.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+
+"""Test USO
+
+Sends large UDP datagrams with UDP_SEGMENT and verifies that the peer
+receives the correct number of individual segments with correct sizes.
+"""
+import socket
+import time
+
+from lib.py import ksft_run, ksft_exit, KsftSkipEx
+from lib.py import ksft_ge
+from lib.py import NetDrvEpEnv
+from lib.py import defer, ethtool, ip, rand_port
+
+# python doesn't expose this constant, so we need to hardcode it to enable UDP
+# segmentation for large payloads
+UDP_SEGMENT = 103
+
+
+def _send_uso(cfg, ipver, mss, total_payload, port):
+    if ipver == "4":
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        dst = (cfg.remote_addr_v["4"], port)
+    else:
+        sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
+        dst = (cfg.remote_addr_v["6"], port)
+
+    sock.setsockopt(socket.IPPROTO_UDP, UDP_SEGMENT, mss)
+    payload = bytes(range(256)) * ((total_payload // 256) + 1)
+    payload = payload[:total_payload]
+    sock.sendto(payload, dst)
+    sock.close()
+    return payload
+
+
+def _get_rx_packets(cfg):
+    stats = ip(f"-s link show dev {cfg.remote_ifname}",
+               json=True, host=cfg.remote)[0]
+    return stats['stats64']['rx']['packets']
+
+
+def _test_uso(cfg, ipver, mss, total_payload):
+    cfg.require_ipver(ipver)
+
+    try:
+        ethtool(f"-K {cfg.ifname} tx-udp-segmentation on")
+    except Exception as exc:
+        raise KsftSkipEx(
+            "Device does not support tx-udp-segmentation") from exc
+    defer(ethtool, f"-K {cfg.ifname} tx-udp-segmentation off")
+
+    expected_segs = (total_payload + mss - 1) // mss
+
+    rx_before = _get_rx_packets(cfg)
+
+    port = rand_port(stype=socket.SOCK_DGRAM)
+    _send_uso(cfg, ipver, mss, total_payload, port)
+
+    time.sleep(0.5)
+
+    rx_after = _get_rx_packets(cfg)
+    rx_delta = rx_after - rx_before
+
+    ksft_ge(rx_delta, expected_segs,
+            comment=f"Expected >= {expected_segs} rx packets, got {rx_delta}")
+
+
+def test_uso_v4(cfg):
+    """USO IPv4: 11 segments (10 full + 1 partial)."""
+    _test_uso(cfg, "4", 1400, 1400 * 10 + 500)
+
+
+def test_uso_v6(cfg):
+    """USO IPv6: 11 segments (10 full + 1 partial)."""
+    _test_uso(cfg, "6", 1400, 1400 * 10 + 500)
+
+
+def test_uso_v4_exact(cfg):
+    """USO IPv4: exact multiple of MSS (5 full segments)."""
+    _test_uso(cfg, "4", 1400, 1400 * 5)
+
+
+def main() -> None:
+    """Run USO tests."""
+    with NetDrvEpEnv(__file__) as cfg:
+        ksft_run([test_uso_v4,
+                  test_uso_v6,
+                  test_uso_v4_exact],
+                 args=(cfg, ))
+    ksft_exit()
+
+
+if __name__ == "__main__":
+    main()
-- 
2.52.0

On Thu, 26 Mar 2026 16:52:31 -0700 Joe Damato wrote:
> Add a simple test for USO. Can be used with netdevsim or real hardware.
> Tests both ipv4 and ipv6 with several full segments and a partial
> segment.


> +def _test_uso(cfg, ipver, mss, total_payload):
> +    cfg.require_ipver(ipver)
> +
> +    try:
> +        ethtool(f"-K {cfg.ifname} tx-udp-segmentation on")
> +    except Exception as exc:
> +        raise KsftSkipEx(
> +            "Device does not support tx-udp-segmentation") from exc
> +    defer(ethtool, f"-K {cfg.ifname} tx-udp-segmentation off")

If may have been on already when we started, no?

> +    expected_segs = (total_payload + mss - 1) // mss
> +
> +    rx_before = _get_rx_packets(cfg)

Let's run a little program on the remote that receives the packets
and validates they were correct? Maybe socat can do?

> +    port = rand_port(stype=socket.SOCK_DGRAM)
> +    _send_uso(cfg, ipver, mss, total_payload, port)
> +
> +    time.sleep(0.5)

What's this? Did you mean to wait for stats to settle?
cfg.wait_hw_stats_settle()

> +    rx_after = _get_rx_packets(cfg)
> +    rx_delta = rx_after - rx_before

Instead of checking Rx (which is probably fine) we may want to check Tx
has the right number of frames. Sender may have mis-counted the whole
USO as one packet if it's buggy?

> +    ksft_ge(rx_delta, expected_segs,
> +            comment=f"Expected >= {expected_segs} rx packets, got {rx_delta}")
> +
> +
> +def test_uso_v4(cfg):
> +    """USO IPv4: 11 segments (10 full + 1 partial)."""
> +    _test_uso(cfg, "4", 1400, 1400 * 10 + 500)
> +
> +
> +def test_uso_v6(cfg):
> +    """USO IPv6: 11 segments (10 full + 1 partial)."""
> +    _test_uso(cfg, "6", 1400, 1400 * 10 + 500)
> +
> +
> +def test_uso_v4_exact(cfg):
> +    """USO IPv4: exact multiple of MSS (5 full segments)."""
> +    _test_uso(cfg, "4", 1400, 1400 * 5)

Variants are probably a good fit here.

On Sun, Mar 29, 2026 at 03:31:30PM -0700, Jakub Kicinski wrote:
> On Thu, 26 Mar 2026 16:52:31 -0700 Joe Damato wrote:
> > Add a simple test for USO. Can be used with netdevsim or real hardware.
> > Tests both ipv4 and ipv6 with several full segments and a partial
> > segment.
> 
> 
> > +def _test_uso(cfg, ipver, mss, total_payload):
> > +    cfg.require_ipver(ipver)
> > +
> > +    try:
> > +        ethtool(f"-K {cfg.ifname} tx-udp-segmentation on")
> > +    except Exception as exc:
> > +        raise KsftSkipEx(
> > +            "Device does not support tx-udp-segmentation") from exc
> > +    defer(ethtool, f"-K {cfg.ifname} tx-udp-segmentation off")
> 
> If may have been on already when we started, no?

Right, yea. I'll fix it. Is there a helper that does this generically for any
features, by any chance?

> > +    expected_segs = (total_payload + mss - 1) // mss
> > +
> > +    rx_before = _get_rx_packets(cfg)
> 
> Let's run a little program on the remote that receives the packets
> and validates they were correct? Maybe socat can do?

Sure, I can do that in the v7.

> > +    port = rand_port(stype=socket.SOCK_DGRAM)
> > +    _send_uso(cfg, ipver, mss, total_payload, port)
> > +
> > +    time.sleep(0.5)
> 
> What's this? Did you mean to wait for stats to settle?
> cfg.wait_hw_stats_settle()

Yea. cfg.wait_hw_stats_settle is clearly better, thanks.
 
> > +    rx_after = _get_rx_packets(cfg)
> > +    rx_delta = rx_after - rx_before
> 
> Instead of checking Rx (which is probably fine) we may want to check Tx
> has the right number of frames. Sender may have mis-counted the whole
> USO as one packet if it's buggy?

Sure.

> > +    ksft_ge(rx_delta, expected_segs,
> > +            comment=f"Expected >= {expected_segs} rx packets, got {rx_delta}")
> > +
> > +
> > +def test_uso_v4(cfg):
> > +    """USO IPv4: 11 segments (10 full + 1 partial)."""
> > +    _test_uso(cfg, "4", 1400, 1400 * 10 + 500)
> > +
> > +
> > +def test_uso_v6(cfg):
> > +    """USO IPv6: 11 segments (10 full + 1 partial)."""
> > +    _test_uso(cfg, "6", 1400, 1400 * 10 + 500)
> > +
> > +
> > +def test_uso_v4_exact(cfg):
> > +    """USO IPv4: exact multiple of MSS (5 full segments)."""
> > +    _test_uso(cfg, "4", 1400, 1400 * 5)
> 
> Variants are probably a good fit here.

Is the right way to do this by using test_builder?

On Mon, 30 Mar 2026 10:25:28 -0700 Joe Damato wrote:
> > > +def test_uso_v4_exact(cfg):
> > > +    """USO IPv4: exact multiple of MSS (5 full segments)."""
> > > +    _test_uso(cfg, "4", 1400, 1400 * 5)  
> > 
> > Variants are probably a good fit here.  
> 
> Is the right way to do this by using test_builder?

grep for ksft_variants, I'll send a patch to mention it in the README
shortly cause I think it's missing there