tools/testing/selftests/drivers/net/hw/uso.py

Source file repositories/reference/linux-study-clean/tools/testing/selftests/drivers/net/hw/uso.py

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/hw/uso.py
Extension
.py
Size
3192 bytes
Lines
104
Domain
Support Tooling And Documentation
Bucket
tools
Inferred role
Support Tooling And Documentation: tools
Status
atlas-only

Why This File Exists

Repository support layer: documentation, build tooling, samples, user-space helper tools, generated initramfs support, licenses, and validation utilities.

Dependency Surface

Detected Declarations

Annotated Snippet

#!/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 expected total payload and that the NIC transmitted at least
the expected number of segments.
"""
import random
import socket
import string

from lib.py import ksft_run, ksft_exit, KsftSkipEx
from lib.py import ksft_eq, ksft_ge, ksft_variants, KsftNamedVariant
from lib.py import NetDrvEpEnv
from lib.py import bkg, defer, ethtool, ip, rand_port, wait_port_listen

# 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 = ''.join(random.choice(string.ascii_lowercase)
                      for _ in range(total_payload))
    sock.sendto(payload.encode(), dst)
    sock.close()


def _get_tx_packets(cfg):
    stats = ip(f"-s link show dev {cfg.ifname}", json=True)[0]
    return stats['stats64']['tx']['packets']


def _test_uso(cfg, ipver, mss, total_payload):
    cfg.require_ipver(ipver)
    cfg.require_cmd("socat", remote=True)

    features = ethtool(f"-k {cfg.ifname}", json=True)
    uso_was_on = features[0]["tx-udp-segmentation"]["active"]

    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
    if not uso_was_on:
        defer(ethtool, f"-K {cfg.ifname} tx-udp-segmentation off")

    expected_segs = (total_payload + mss - 1) // mss

    port = rand_port(stype=socket.SOCK_DGRAM)
    rx_cmd = f"socat -{ipver} -T 2 -u UDP-LISTEN:{port},reuseport STDOUT"

    tx_before = _get_tx_packets(cfg)

    with bkg(rx_cmd, host=cfg.remote, exit_wait=True) as rx:
        wait_port_listen(port, proto="udp", host=cfg.remote)
        _send_uso(cfg, ipver, mss, total_payload, port)

    ksft_eq(len(rx.stdout), total_payload,

Annotation

Implementation Notes