tools/testing/selftests/drivers/net/lib/py/load.py

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/drivers/net/lib/py/load.py
Extension
.py
Size
5060 bytes
Lines
143
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

import re
import time
import json

from lib.py import ksft_pr, cmd, ip, rand_port, wait_port_listen


class Iperf3Runner:
    """
    Sets up and runs iperf3 traffic.
    """
    def __init__(self, env, port=None, server_ip=None, client_ip=None):
        env.require_cmd("iperf3", local=True, remote=True)
        self.env = env
        self.port = rand_port() if port is None else port
        self.server_ip = server_ip
        self.client_ip = client_ip

    def _build_server(self):
        cmdline = f"iperf3 -s -1 -p {self.port}"
        if self.server_ip:
            cmdline += f" -B {self.server_ip}"
        return cmdline

    def _build_client(self, streams, duration, reverse):
        host = self.env.addr if self.server_ip is None else self.server_ip
        cmdline = f"iperf3 -c {host} -p {self.port} -P {streams} -t {duration} -J"
        if self.client_ip:
            cmdline += f" -B {self.client_ip}"
        if reverse:
            cmdline += " --reverse"
        return cmdline

    def start_server(self):
        """
        Starts an iperf3 server with optional bind IP.
        """
        cmdline = self._build_server()
        proc = cmd(cmdline, background=True)
        wait_port_listen(self.port)
        time.sleep(0.1)
        return proc

    def start_client(self, background=False, streams=1, duration=10, reverse=False):
        """
        Starts the iperf3 client with the configured options.
        """
        cmdline = self._build_client(streams, duration, reverse)
        kwargs = {"background": background, "host": self.env.remote}
        if not background:
            kwargs["timeout"] = duration + 5
        return cmd(cmdline, **kwargs)

    def measure_bandwidth(self, reverse=False):
        """
        Runs an iperf3 measurement and returns the average bandwidth (Gbps).
        Discards the first and last few reporting intervals and uses only the
        middle part of the run where throughput is typically stable.
        """
        self.start_server()
        result = self.start_client(duration=10, reverse=reverse)

        if result.ret != 0:
            raise RuntimeError("iperf3 failed to run successfully")
        try:
            out = json.loads(result.stdout)
        except json.JSONDecodeError as exc:
            raise ValueError("Failed to parse iperf3 JSON output") from exc

Annotation

Implementation Notes