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

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/hw/ntuple.py
Extension
.py
Size
5766 bytes
Lines
166
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
"""Test ethtool NFC (ntuple) flow steering rules."""

import random
from enum import Enum, auto
from lib.py import ksft_run, ksft_exit
from lib.py import ksft_eq, ksft_ge
from lib.py import ksft_variants, KsftNamedVariant
from lib.py import EthtoolFamily, NetDrvEpEnv, NetdevFamily
from lib.py import KsftSkipEx
from lib.py import cmd, ethtool, defer, rand_ports, bkg, wait_port_listen


class NtupleField(Enum):
    SRC_IP = auto()
    DST_IP = auto()
    SRC_PORT = auto()
    DST_PORT = auto()


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 _get_rx_cnts(cfg, prev=None):
    """Get Rx packet counts for all queues, as a simple list of integers
       if @prev is specified the prev counts will be subtracted"""
    cfg.wait_hw_stats_settle()
    data = cfg.netdevnl.qstats_get({"ifindex": cfg.ifindex, "scope": ["queue"]}, dump=True)
    data = [x for x in data if x['queue-type'] == "rx"]
    max_q = max([x["queue-id"] for x in data])
    queue_stats = [0] * (max_q + 1)
    for q in data:
        queue_stats[q["queue-id"]] = q["rx-packets"]
        if prev and q["queue-id"] < len(prev):
            queue_stats[q["queue-id"]] -= prev[q["queue-id"]]
    return queue_stats


def _ntuple_rule_add(cfg, flow_spec):
    """Install an NFC rule via ethtool."""

    output = ethtool(f"-N {cfg.ifname} {flow_spec}").stdout
    rule_id = int(output.split()[-1])
    defer(ethtool, f"-N {cfg.ifname} delete {rule_id}")


def _setup_isolated_queue(cfg):
    """Default all traffic to queue 0, and pick a random queue to
       steer NFC traffic to."""

    channels = cfg.ethnl.channels_get({'header': {'dev-index': cfg.ifindex}})
    ch_max = channels['combined-max']
    qcnt = channels['combined-count']

    if ch_max < 2:
        raise KsftSkipEx(f"Need at least 2 combined channels, max is {ch_max}")

    desired_queues = min(ch_max, 4)
    if qcnt >= desired_queues:
        desired_queues = qcnt
    else:
        ethtool(f"-L {cfg.ifname} combined {desired_queues}")
        defer(ethtool, f"-L {cfg.ifname} combined {qcnt}")

Annotation

Implementation Notes