1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""Regression tests for the SO_TXTIME interface.
Test delivery time in FQ and ETF qdiscs.
"""
import os
import time
from lib.py import ksft_exit, ksft_run, ksft_variants
from lib.py import KsftNamedVariant, KsftSkipEx
from lib.py import NetDrvEpEnv, bkg, cmd, defer, tc
from lib.py import CmdExitFailure
def test_so_txtime(cfg, clockid, ipver, args_tx, args_rx, expect_success):
"""Main function. Run so_txtime as sender and receiver."""
slow_machine = os.environ.get('KSFT_MACHINE_SLOW')
if not hasattr(cfg, "bin_remote"):
cfg.bin_local = cfg.test_dir / "so_txtime"
cfg.bin_remote = cfg.remote.deploy(cfg.bin_local)
tstart = time.time_ns() + (2000_000_000 if slow_machine else 200_000_000)
cmd_addr = f"-S {cfg.addr_v[ipver]} -D {cfg.remote_addr_v[ipver]}"
cmd_args = f"-{ipver} -c {clockid} -t {tstart} {cmd_addr}"
cmd_rx = f"{cfg.bin_remote} {cmd_args} {args_rx} -r"
cmd_tx = f"{cfg.bin_local} -m 100 {cmd_args} {args_tx}"
expect_fail = not expect_success
if slow_machine:
expect_success = False
with bkg(cmd_rx, host=cfg.remote, fail=expect_success,
expect_fail=expect_fail, exit_wait=True):
cmd(cmd_tx)
def _qdisc_setup(ifname, qdisc, optargs=""):
"""Replace root qdisc. Restore the original after the test.
If the original is mq, children will be of type default_qdisc.
"""
orig = tc(f"qdisc show dev {ifname} root", json=True)[0].get("kind", None)
defer(tc, f"qdisc replace dev {ifname} root {orig}")
try:
tc(f"qdisc del dev {ifname} root")
except CmdExitFailure:
pass
tc(f"qdisc replace dev {ifname} root handle 1: {qdisc} {optargs}")
def _test_variants_fq():
for ipver in ["4", "6"]:
for testcase in [
["no_delay", "a,-1", "a,-1"],
["zero_delay", "a,0", "a,0"],
["one_pkt", "a,10", "a,10"],
["in_order", "a,10,b,20", "a,10,b,20"],
["reverse_order", "a,20,b,10", "b,10,a,20"],
]:
name = f"v{ipver}_{testcase[0]}"
yield KsftNamedVariant(name, ipver, testcase[1], testcase[2])
@ksft_variants(_test_variants_fq())
def test_so_txtime_fq_mono(cfg, ipver, args_tx, args_rx):
"""Run all variants of monotonic (fq) tests."""
cfg.require_ipver(ipver)
_qdisc_setup(cfg.ifname, "fq")
test_so_txtime(cfg, "mono", ipver, args_tx, args_rx, True)
@ksft_variants(_test_variants_fq())
def test_so_txtime_fq_tai(cfg, ipver, args_tx, args_rx):
"""Run all variants of fq tests, but pass CLOCK_TAI to test conversion."""
cfg.require_ipver(ipver)
_qdisc_setup(cfg.ifname, "fq")
test_so_txtime(cfg, "tai", ipver, args_tx, args_rx, True)
def _test_variants_etf():
for ipver in ["4", "6"]:
for testcase in [
["no_delay", "a,-1", "a,-1", False],
["zero_delay", "a,0", "a,0", False],
["one_pkt", "a,10", "a,10", True],
["in_order", "a,10,b,20", "a,10,b,20", True],
["reverse_order", "a,20,b,10", "b,10,a,20", True],
]:
name = f"v{ipver}_{testcase[0]}"
yield KsftNamedVariant(
name, ipver, testcase[1], testcase[2], testcase[3]
)
@ksft_variants(_test_variants_etf())
def test_so_txtime_etf(cfg, ipver, args_tx, args_rx, expect_fail):
"""Run all variants of etf tests."""
cfg.require_ipver(ipver)
# root qdisc for background traffic (e.g., bkg())
_qdisc_setup(cfg.ifname, "prio")
# leaf ETF qdisc only for intended packets
try:
etf_args = "clockid CLOCK_TAI delta 400000"
tc(f"qdisc add dev {cfg.ifname} parent 1:1 handle 10: etf {etf_args}")
except Exception as e:
raise KsftSkipEx("tc does not support qdisc etf. skipping") from e
# redirect mark 100 to leaf
filter_args = "protocol all handle 100 fw flowid 1:1"
tc(f"filter add dev {cfg.ifname} parent 1: {filter_args}")
test_so_txtime(cfg, "tai", ipver, args_tx, args_rx, expect_fail)
def main() -> None:
"""Boilerplate ksft main."""
with NetDrvEpEnv(__file__) as cfg:
ksft_run(
[test_so_txtime_fq_mono, test_so_txtime_fq_tai, test_so_txtime_etf],
args=(cfg,),
)
ksft_exit()
if __name__ == "__main__":
main()
|
