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

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

File Facts

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

"""
Tests for RSS hashing on IPv6 Flow Label.
"""

import glob
import os
import socket
from lib.py import CmdExitFailure
from lib.py import ksft_run, ksft_exit, ksft_eq, ksft_ge, ksft_in, \
    ksft_not_in, ksft_raises, KsftSkipEx
from lib.py import bkg, cmd, defer, fd_read_timeout, rand_port
from lib.py import NetDrvEpEnv


def _check_system(cfg):
    if not hasattr(socket, "SO_INCOMING_CPU"):
        raise KsftSkipEx("socket.SO_INCOMING_CPU was added in Python 3.11")

    qcnt = len(glob.glob(f"/sys/class/net/{cfg.ifname}/queues/rx-*"))
    if qcnt < 2:
        raise KsftSkipEx(f"Local has only {qcnt} queues")

    for f in [f"/sys/class/net/{cfg.ifname}/queues/rx-0/rps_flow_cnt",
              f"/sys/class/net/{cfg.ifname}/queues/rx-0/rps_cpus"]:
        try:
            with open(f, 'r') as fp:
                setting = fp.read().strip()
                # CPU mask will be zeros and commas
                if setting.replace("0", "").replace(",", ""):
                    raise KsftSkipEx(f"RPS/RFS is configured: {f}: {setting}")
        except FileNotFoundError:
            pass

    # 1 is the default, if someone changed it we probably shouldn"t mess with it
    af = cmd("cat /proc/sys/net/ipv6/auto_flowlabels", host=cfg.remote).stdout
    if af.strip() != "1":
        raise KsftSkipEx("Remote does not have auto_flowlabels enabled")


def _ethtool_get_cfg(cfg, fl_type):
    descr = cmd(f"ethtool -n {cfg.ifname} rx-flow-hash {fl_type}").stdout

    converter = {
        "IP SA": "s",
        "IP DA": "d",
        "L3 proto": "t",
        "L4 bytes 0 & 1 [TCP/UDP src port]": "f",
        "L4 bytes 2 & 3 [TCP/UDP dst port]": "n",
        "IPv6 Flow Label": "l",
    }

    ret = ""
    for line in descr.split("\n")[1:-2]:
        # if this raises we probably need to add more keys to converter above
        ret += converter[line]
    return ret


def _traffic(cfg, one_sock, one_cpu):
    local_port  = rand_port(socket.SOCK_DGRAM)
    remote_port = rand_port(socket.SOCK_DGRAM)

    sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
    sock.bind(("", local_port))
    sock.connect((cfg.remote_addr_v["6"], 0))
    if one_sock:
        send = f"exec 5<>/dev/udp/{cfg.addr_v['6']}/{local_port}; " \

Annotation

Implementation Notes