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

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

File Facts

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

# SPDX-License-Identifier: GPL-2.0
"""Shared helpers for devmem TCP selftests."""

import re

from lib.py import (bkg, cmd, defer, ethtool, rand_port, wait_port_listen,
                    ksft_eq, KsftSkipEx, NetNSEnter, EthtoolFamily,
                    NetdevFamily)


def require_devmem(cfg):
    """Probe ncdevmem on cfg.ifname and SKIP the test if devmem isn't supported."""
    if not hasattr(cfg, "devmem_probed"):
        probe_command = f"{cfg.bin_local} -f {cfg.ifname}"
        cfg.devmem_supported = cmd(probe_command, fail=False, shell=True).ret == 0
        cfg.devmem_probed = True

    if not cfg.devmem_supported:
        raise KsftSkipEx("Test requires devmem support")


def configure_nic(cfg):
    """Channels, rings, RSS, queue lease for netkit devmem."""
    if not hasattr(cfg, 'netns'):
        return

    cfg.require_ipver('6')
    ethnl = EthtoolFamily()

    channels = 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 = ethnl.rings_get({'header': {'dev-index': cfg.ifindex}})
    orig_rx_rings = rings['rx']
    orig_hds_thresh = rings.get('hds-thresh', 0)
    orig_data_split = rings.get('tcp-data-split', 'unknown')

    ethnl.rings_set({'header': {'dev-index': cfg.ifindex},
                     'tcp-data-split': 'enabled',
                     'hds-thresh': 0,
                     'rx': min(64, orig_rx_rings)})
    defer(ethnl.rings_set, {'header': {'dev-index': cfg.ifindex},
                            'tcp-data-split': orig_data_split,
                            'hds-thresh': orig_hds_thresh,
                            'rx': orig_rx_rings})

    cfg.src_queue = channels - 1
    ethtool(f"-X {cfg.ifname} equal {cfg.src_queue}")
    defer(ethtool, f"-X {cfg.ifname} default")

    if not hasattr(cfg, 'nk_queue'):
        with NetNSEnter(str(cfg.netns)):
            netdevnl = NetdevFamily()
            lease_result = netdevnl.queue_create({
                "ifindex": cfg.nk_guest_ifindex,
                "type": "rx",
                "lease": {
                    "ifindex": cfg.ifindex,
                    "queue": {"id": cfg.src_queue, "type": "rx"},
                    "netns-id": 0,
                },
            })
            cfg.nk_queue = lease_result['id']


def set_flow_rule(cfg, port):

Annotation

Implementation Notes