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

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

File Facts

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

"""
Toeplitz Rx hashing test:
 - rxhash (the hash value calculation itself);
 - RSS mapping from rxhash to rx queue;
 - RPS mapping from rxhash to cpu.
"""

import glob
import os
import socket
from lib.py import ksft_run, ksft_exit, ksft_pr
from lib.py import NetDrvEpEnv, EthtoolFamily, NetdevFamily
from lib.py import cmd, bkg, rand_port, defer
from lib.py import ksft_in
from lib.py import ksft_variants, KsftNamedVariant, KsftSkipEx, KsftFailEx

# "define" for the ID of the Toeplitz hash function
ETH_RSS_HASH_TOP = 1
# Must match RPS_MAX_CPUS in toeplitz.c
RPS_MAX_CPUS = 16


def _check_rps_and_rfs_not_configured(cfg):
    """Verify that RPS is not already configured."""

    for rps_file in glob.glob(f"/sys/class/net/{cfg.ifname}/queues/rx-*/rps_cpus"):
        with open(rps_file, "r", encoding="utf-8") as fp:
            val = fp.read().strip()
            if set(val) - {"0", ","}:
                raise KsftSkipEx(f"RPS already configured on {rps_file}: {val}")

    rfs_file = "/proc/sys/net/core/rps_sock_flow_entries"
    with open(rfs_file, "r", encoding="utf-8") as fp:
        val = fp.read().strip()
        if val != "0":
            raise KsftSkipEx(f"RFS already configured {rfs_file}: {val}")


def _get_cpu_for_irq(irq):
    with open(f"/proc/irq/{irq}/smp_affinity_list", "r",
              encoding="utf-8") as fp:
        data = fp.read().strip()
        if "," in data or "-" in data:
            raise KsftFailEx(f"IRQ{irq} is not mapped to a single core: {data}")
        return int(data)


def _get_irq_cpus(cfg):
    """
    Read the list of IRQs for the device Rx queues.
    """
    queues = cfg.netnl.queue_get({"ifindex": cfg.ifindex}, dump=True)
    napis = cfg.netnl.napi_get({"ifindex": cfg.ifindex}, dump=True)

    # Remap into ID-based dicts
    napis = {n["id"]: n for n in napis}
    queues = {f"{q['type']}{q['id']}": q for q in queues}

    cpus = []
    for rx in range(9999):
        name = f"rx{rx}"
        if name not in queues:
            break
        cpus.append(_get_cpu_for_irq(napis[queues[name]["napi-id"]]["irq"]))

    return cpus

Annotation

Implementation Notes