tools/testing/selftests/drivers/net/netpoll_basic.py

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/netpoll_basic.py
Extension
.py
Size
12984 bytes
Lines
397
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
# Author: Breno Leitao <leitao@debian.org>
"""
 This test aims to evaluate the netpoll polling mechanism (as in
 netpoll_poll_dev()). It presents a complex scenario where the network
 attempts to send a packet but fails, prompting it to poll the NIC from within
 the netpoll TX side.

 This has been a crucial path in netpoll that was previously untested. Jakub
 suggested using a single RX/TX queue, pushing traffic to the NIC, and then
 sending netpoll messages (via netconsole) to trigger the poll.

 In parallel, bpftrace is used to detect if netpoll_poll_dev() was called. If
 so, the test passes, otherwise it will be skipped. This test is very dependent on
 the driver and environment, given we are trying to trigger a tricky scenario.
"""

import errno
import logging
import os
import random
import string
import threading
import time
from typing import Optional

from lib.py import (
    bpftrace,
    CmdExitFailure,
    defer,
    ethtool,
    GenerateTraffic,
    ksft_exit,
    ksft_pr,
    ksft_run,
    KsftFailEx,
    KsftSkipEx,
    NetDrvEpEnv,
    KsftXfailEx,
)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)

NETCONSOLE_CONFIGFS_PATH: str = "/sys/kernel/config/netconsole"
NETCONS_REMOTE_PORT: int = 6666
NETCONS_LOCAL_PORT: int = 1514

# Max number of netcons messages to send. Each iteration will setup
# netconsole and send MAX_WRITES messages
ITERATIONS: int = 20
# Number of writes to /dev/kmsg per iteration
MAX_WRITES: int = 40
# MAPS contains the information coming from bpftrace it will have only one
# key: "hits", which tells the number of times netpoll_poll_dev() was called
MAPS: dict[str, int] = {}
# Thread to run bpftrace in parallel
BPF_THREAD: Optional[threading.Thread] = None
# Time bpftrace will be running in parallel.
BPFTRACE_TIMEOUT: int = 10


def ethtool_get_ringsize(interface_name: str) -> tuple[int, int]:
    """
    Read the ringsize using ethtool. This will be used to restore it after the test
    """

Annotation

Implementation Notes