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

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/hw/nk_qlease.py
Extension
.py
Size
13877 bytes
Lines
466
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 time
import threading
from os import path
from lib.py import (
    ksft_run,
    ksft_exit,
    ksft_eq,
    ksft_in,
    ksft_not_in,
    ksft_raises,
)
from lib.py import (
    NetDrvContEnv,
    NetNSEnter,
    EthtoolFamily,
    NetdevFamily,
    RtnlFamily,
)
from lib.py import (
    Netlink,
    bkg,
    cmd,
    defer,
    ethtool,
    ip,
    rand_port,
    wait_port_listen,
)
from lib.py import KsftSkipEx, CmdExitFailure

# iou-zcrx exits with 42 from setup_zcrx() when the NIC does not advertise
# QCFG_RX_PAGE_SIZE (or otherwise rejects the requested rx_buf_len).
SKIP_CODE = 42


def _restore_hugepages(count):
    with open("/proc/sys/vm/nr_hugepages", "w", encoding="utf-8") as f:
        f.write(str(count))


def _mp_clear_wait(cfg, src_queue):
    """Wait for the io_uring memory provider to clear from the leased
    physical queue; io_uring tears it down asynchronously after the
    process holding the ifq exits."""
    netdevnl = NetdevFamily()
    deadline = time.time() + 5
    while time.time() < deadline:
        queue_info = netdevnl.queue_get(
            {"ifindex": cfg.ifindex, "id": src_queue, "type": "rx"}
        )
        if "io-uring" not in queue_info:
            return
        time.sleep(0.1)
    raise TimeoutError("Timed out waiting for memory provider to clear")


def _create_netkit_pair(cfg, rxqueues=2):
    if cfg.nk_host_ifname:
        cmd(f"ip link del dev {cfg.nk_host_ifname}", fail=False)
        cfg.nk_host_ifname = None
        cfg.nk_guest_ifname = None
    cfg.detach_bpf()

    all_links = ip("-d link show", json=True)
    old_idxs = {
        link["ifindex"]

Annotation

Implementation Notes