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

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/hw/rss_input_xfrm.py
Extension
.py
Size
3912 bytes
Lines
125
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

import multiprocessing
import socket
from lib.py import ksft_run, ksft_exit, ksft_eq, ksft_ge, cmd, fd_read_timeout
from lib.py import NetDrvEpEnv
from lib.py import EthtoolFamily, NetdevFamily, NlError
from lib.py import KsftSkipEx, KsftFailEx
from lib.py import defer, ksft_pr, rand_port


def traffic(cfg, local_port, remote_port, ipver):
    af_inet = socket.AF_INET if ipver == "4" else socket.AF_INET6
    sock = socket.socket(af_inet, socket.SOCK_DGRAM)
    sock.bind(("", local_port))
    sock.connect((cfg.remote_addr_v[ipver], remote_port))
    tgt = f"{ipver}:[{cfg.addr_v[ipver]}]:{local_port},sourceport={remote_port}"
    cmd("echo a | socat - UDP" + tgt, host=cfg.remote)
    fd_read_timeout(sock.fileno(), 5)
    return sock.getsockopt(socket.SOL_SOCKET, socket.SO_INCOMING_CPU)


def _rss_input_xfrm_try_enable(cfg):
    """
    Check if symmetric input-xfrm is already enabled, if not try to enable it
    and register a cleanup.
    """
    rss = cfg.ethnl.rss_get({'header': {'dev-name': cfg.ifname}})
    orig_xfrm = rss.get('input-xfrm', set())
    sym_xfrm = set(filter(lambda x: 'sym' in x, orig_xfrm))

    if sym_xfrm:
        ksft_pr("Sym input xfrm already enabled:", sym_xfrm)
        return sym_xfrm

    for xfrm in cfg.ethnl.consts["input-xfrm"].entries:
        # Skip non-symmetric transforms
        if "sym" not in xfrm:
            continue

        try_xfrm = {xfrm} | orig_xfrm
        try:
            cfg.ethnl.rss_set({"header": {"dev-index": cfg.ifindex},
                               "input-xfrm": try_xfrm})
        except NlError:
            continue

        ksft_pr("Sym input xfrm configured:", try_xfrm)
        defer(cfg.ethnl.rss_set,
              {"header": {"dev-index": cfg.ifindex},
               "input-xfrm": orig_xfrm})
        return {xfrm}

    return set()


def test_rss_input_xfrm(cfg, ipver):
    """
    Test symmetric input_xfrm.
    If symmetric RSS hash is configured, send traffic twice, swapping the
    src/dst UDP ports, and verify that the same queue is receiving the traffic
    in both cases (IPs are constant).
    """

    if multiprocessing.cpu_count() < 2:
        raise KsftSkipEx("Need at least two CPUs to test symmetric RSS hash")

    cfg.require_cmd("socat", local=False, remote=True)

Annotation

Implementation Notes