tools/testing/selftests/drivers/net/hw/iou-zcrx.py

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/hw/iou-zcrx.py
Extension
.py
Size
7443 bytes
Lines
211
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 re
import resource
import time
from os import path
from lib.py import ksft_run, ksft_exit, KsftSkipEx, ksft_variants, KsftNamedVariant
from lib.py import NetDrvEpEnv
from lib.py import bkg, cmd, defer, ethtool, rand_port, wait_port_listen
from lib.py import EthtoolFamily, NetdevFamily

SKIP_CODE = 42


def mp_clear_wait(cfg):
    """Wait for io_uring memory providers to clear from all device queues."""
    deadline = time.time() + 5
    while time.time() < deadline:
        queues = cfg.netnl.queue_get({'ifindex': cfg.ifindex}, dump=True)
        if not any('io-uring' in q for q in queues):
            return
        time.sleep(0.1)
    raise TimeoutError("Timed out waiting for memory provider to clear")


def create_rss_ctx(cfg):
    output = ethtool(f"-X {cfg.ifname} context new start {cfg.target} equal 1").stdout
    values = re.search(r'New RSS context is (\d+)', output).group(1)
    return int(values)


def set_flow_rule(cfg):
    output = ethtool(f"-N {cfg.ifname} flow-type tcp6 dst-port {cfg.port} action {cfg.target}").stdout
    values = re.search(r'ID (\d+)', output).group(1)
    return int(values)


def set_flow_rule_rss(cfg, rss_ctx_id):
    output = ethtool(f"-N {cfg.ifname} flow-type tcp6 dst-port {cfg.port} context {rss_ctx_id}").stdout
    values = re.search(r'ID (\d+)', output).group(1)
    return int(values)


def check_iou_rx_buf_len(cfg, expected_rx_buf_len):
    """Check the io-uring memory provider exposes the expected rx_buf_len."""
    q = cfg.netnl.queue_get({'ifindex': cfg.ifindex, 'type': 'rx', 'id': cfg.target})
    napi_id = q['napi-id']
    pools = cfg.netnl.page_pool_get({}, dump=True)
    pools = [p for p in pools if p.get('napi-id') == napi_id
             and 'io-uring' in p]
    if len(pools) != 1:
        raise Exception(f"Expected 1 io-uring page pool, found {len(pools)}")
    rx_buf_len = pools[0]['io-uring'].get('rx-buf-len')
    if rx_buf_len is None:
        raise KsftSkipEx("io-uring 'rx-buf-len' attribute not supported")
    if rx_buf_len != expected_rx_buf_len:
        raise Exception(f'Expected io-uring rx-buf-len {expected_rx_buf_len}, '
                        f'got {rx_buf_len}')


def single(cfg):
    channels = cfg.ethnl.channels_get({'header': {'dev-index': cfg.ifindex}})
    channels = channels['combined-count']
    if channels < 2:
        raise KsftSkipEx('Test requires NETIF with at least 2 combined channels')

    rings = cfg.ethnl.rings_get({'header': {'dev-index': cfg.ifindex}})
    rx_rings = rings['rx']
    hds_thresh = rings.get('hds-thresh', 0)

Annotation

Implementation Notes