Linux kernel mirror (for testing)
git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
kernel
os
linux
1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-2.0
3
4from os import path
5from lib.py import ksft_run, ksft_exit
6from lib.py import ksft_eq, KsftSkipEx
7from lib.py import NetDrvEpEnv
8from lib.py import bkg, cmd, rand_port, wait_port_listen
9from lib.py import ksft_disruptive
10
11
12def require_devmem(cfg):
13 if not hasattr(cfg, "_devmem_probed"):
14 probe_command = f"{cfg.bin_local} -f {cfg.ifname}"
15 cfg._devmem_supported = cmd(probe_command, fail=False, shell=True).ret == 0
16 cfg._devmem_probed = True
17
18 if not cfg._devmem_supported:
19 raise KsftSkipEx("Test requires devmem support")
20
21
22@ksft_disruptive
23def check_rx(cfg) -> None:
24 require_devmem(cfg)
25
26 port = rand_port()
27 socat = f"socat -u - TCP{cfg.addr_ipver}:{cfg.baddr}:{port},bind={cfg.remote_baddr}:{port}"
28 listen_cmd = f"{cfg.bin_local} -l -f {cfg.ifname} -s {cfg.addr} -p {port} -c {cfg.remote_addr} -v 7"
29
30 with bkg(listen_cmd, exit_wait=True) as ncdevmem:
31 wait_port_listen(port)
32 cmd(f"yes $(echo -e \x01\x02\x03\x04\x05\x06) | \
33 head -c 1K | {socat}", host=cfg.remote, shell=True)
34
35 ksft_eq(ncdevmem.ret, 0)
36
37
38@ksft_disruptive
39def check_tx(cfg) -> None:
40 require_devmem(cfg)
41
42 port = rand_port()
43 listen_cmd = f"socat -U - TCP{cfg.addr_ipver}-LISTEN:{port}"
44
45 with bkg(listen_cmd, host=cfg.remote, exit_wait=True) as socat:
46 wait_port_listen(port, host=cfg.remote)
47 cmd(f"echo -e \"hello\\nworld\"| {cfg.bin_local} -f {cfg.ifname} -s {cfg.remote_addr} -p {port}", shell=True)
48
49 ksft_eq(socat.stdout.strip(), "hello\nworld")
50
51
52@ksft_disruptive
53def check_tx_chunks(cfg) -> None:
54 require_devmem(cfg)
55
56 port = rand_port()
57 listen_cmd = f"socat -U - TCP{cfg.addr_ipver}-LISTEN:{port}"
58
59 with bkg(listen_cmd, host=cfg.remote, exit_wait=True) as socat:
60 wait_port_listen(port, host=cfg.remote)
61 cmd(f"echo -e \"hello\\nworld\"| {cfg.bin_local} -f {cfg.ifname} -s {cfg.remote_addr} -p {port} -z 3", shell=True)
62
63 ksft_eq(socat.stdout.strip(), "hello\nworld")
64
65
66def check_rx_hds(cfg) -> None:
67 """Test HDS splitting across payload sizes."""
68 require_devmem(cfg)
69
70 for size in [1, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192]:
71 port = rand_port()
72 listen_cmd = f"{cfg.bin_local} -L -l -f {cfg.ifname} -s {cfg.addr} -p {port}"
73
74 with bkg(listen_cmd, exit_wait=True) as ncdevmem:
75 wait_port_listen(port)
76 cmd(f"dd if=/dev/zero bs={size} count=1 2>/dev/null | " +
77 f"socat -b {size} -u - TCP{cfg.addr_ipver}:{cfg.baddr}:{port},nodelay",
78 host=cfg.remote, shell=True)
79
80 ksft_eq(ncdevmem.ret, 0, f"HDS failed for payload size {size}")
81
82
83def main() -> None:
84 with NetDrvEpEnv(__file__) as cfg:
85 cfg.bin_local = path.abspath(path.dirname(__file__) + "/ncdevmem")
86 cfg.bin_remote = cfg.remote.deploy(cfg.bin_local)
87
88 ksft_run([check_rx, check_tx, check_tx_chunks, check_rx_hds],
89 args=(cfg, ))
90 ksft_exit()
91
92
93if __name__ == "__main__":
94 main()