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

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

File Facts

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

"""
API level tests for RSS (mostly Netlink vs IOCTL).
"""

import errno
import glob
import random
from lib.py import ksft_run, ksft_exit, ksft_eq, ksft_is, ksft_ne, ksft_raises
from lib.py import KsftSkipEx, KsftFailEx
from lib.py import defer, ethtool, CmdExitFailure
from lib.py import EthtoolFamily, NlError
from lib.py import NetDrvEnv


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


def _ethtool_create(cfg, act, opts):
    output = ethtool(f"{act} {cfg.ifname} {opts}").stdout
    # Output will be something like: "New RSS context is 1" or
    # "Added rule with ID 7", we want the integer from the end
    return int(output.split()[-1])


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

    if to_nl:
        converter = {
            "IP SA": "ip-src",
            "IP DA": "ip-dst",
            "L4 bytes 0 & 1 [TCP/UDP src port]": "l4-b-0-1",
            "L4 bytes 2 & 3 [TCP/UDP dst port]": "l4-b-2-3",
        }

        ret = set()
    else:
        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",
        }

        ret = ""

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


def test_rxfh_nl_set_fail(cfg):
    """
    Test error path of Netlink SET.
    """
    _require_2qs(cfg)

    ethnl = EthtoolFamily()

Annotation

Implementation Notes