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

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/hw/rss_ctx.py
Extension
.py
Size
38085 bytes
Lines
1036
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 datetime
import random
import re
import time
from lib.py import ksft_disruptive
from lib.py import ksft_run, ksft_pr, ksft_exit
from lib.py import ksft_eq, ksft_ne, ksft_ge, ksft_in, ksft_lt, ksft_true, ksft_raises
from lib.py import NetDrvEpEnv
from lib.py import EthtoolFamily, NetdevFamily, NlError
from lib.py import KsftSkipEx, KsftFailEx
from lib.py import rand_port, rand_ports
from lib.py import cmd, ethtool, ip, defer, CmdExitFailure, wait_file
from lib.py import GenerateTraffic


def _rss_key_str(key):
    return ":".join(["{:02x}".format(x) for x in key])


def _rss_key_rand(length):
    return [random.randint(0, 255) for _ in range(length)]


def _rss_key_check(cfg, data=None, context=0):
    if data is None:
        data = get_rss(cfg, context=context)
    if 'rss-hash-key' not in data:
        return
    non_zero = [x for x in data['rss-hash-key'] if x != 0]
    ksft_eq(bool(non_zero), True, comment=f"RSS key is all zero {data['rss-hash-key']}")


def get_rss(cfg, context=0):
    return ethtool(f"-x {cfg.ifname} context {context}", json=True)[0]


def get_drop_err_sum(cfg):
    stats = ip("-s -s link show dev " + cfg.ifname, json=True)[0]
    cnt = 0
    for key in ['errors', 'dropped', 'over_errors', 'fifo_errors',
                'length_errors', 'crc_errors', 'missed_errors',
                'frame_errors']:
        cnt += stats["stats64"]["rx"][key]
    return cnt, stats["stats64"]["tx"]["carrier_changes"]


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 require_ntuple(cfg):
    features = ethtool(f"-k {cfg.ifname}", json=True)[0]
    if not features["ntuple-filters"]["active"]:
        if features["ntuple-filters"]["fixed"]:
            raise KsftSkipEx("Device does not support ntuple-filters")
        ethtool(f"-K {cfg.ifname} ntuple-filters on")
        defer(ethtool, f"-K {cfg.ifname} ntuple-filters off")


def require_context_cnt(cfg, need_cnt):
    # There's no good API to get the context count, so the tests
    # which try to add a lot opportunisitically set the count they
    # discovered. Careful with test ordering!
    if need_cnt and cfg.context_cnt and cfg.context_cnt < need_cnt:

Annotation

Implementation Notes