Linux kernel mirror (for testing)
git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
tjh.dev
kernel
os
linux
1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-2.0
3
4import time
5from lib.py import ksft_run, ksft_exit, ksft_pr
6from lib.py import ksft_eq, ksft_ge, ksft_busy_wait
7from lib.py import NetdevFamily, NetdevSimDev, ip
8
9
10def empty_check(nf) -> None:
11 devs = nf.dev_get({}, dump=True)
12 ksft_ge(len(devs), 1)
13
14
15def lo_check(nf) -> None:
16 lo_info = nf.dev_get({"ifindex": 1})
17 ksft_eq(len(lo_info['xdp-features']), 0)
18 ksft_eq(len(lo_info['xdp-rx-metadata-features']), 0)
19
20
21def napi_list_check(nf) -> None:
22 with NetdevSimDev(queue_count=100) as nsimdev:
23 nsim = nsimdev.nsims[0]
24
25 ip(f"link set dev {nsim.ifname} up")
26
27 napis = nf.napi_get({'ifindex': nsim.ifindex}, dump=True)
28 ksft_eq(len(napis), 100)
29
30 for q in [50, 0, 99]:
31 for i in range(4):
32 nsim.dfs_write("queue_reset", f"{q} {i}")
33 napis = nf.napi_get({'ifindex': nsim.ifindex}, dump=True)
34 ksft_eq(len(napis), 100,
35 comment=f"queue count after reset queue {q} mode {i}")
36
37
38def nsim_rxq_reset_down(nf) -> None:
39 """
40 Test that the queue API supports resetting a queue
41 while the interface is down. We should convert this
42 test to testing real HW once more devices support
43 queue API.
44 """
45 with NetdevSimDev(queue_count=4) as nsimdev:
46 nsim = nsimdev.nsims[0]
47
48 ip(f"link set dev {nsim.ifname} down")
49 for i in [0, 2, 3]:
50 nsim.dfs_write("queue_reset", f"1 {i}")
51
52
53def page_pool_check(nf) -> None:
54 with NetdevSimDev() as nsimdev:
55 nsim = nsimdev.nsims[0]
56
57 def up():
58 ip(f"link set dev {nsim.ifname} up")
59
60 def down():
61 ip(f"link set dev {nsim.ifname} down")
62
63 def get_pp():
64 pp_list = nf.page_pool_get({}, dump=True)
65 return [pp for pp in pp_list if pp.get("ifindex") == nsim.ifindex]
66
67 # No page pools when down
68 down()
69 ksft_eq(len(get_pp()), 0)
70
71 # Up, empty page pool appears
72 up()
73 pp_list = get_pp()
74 ksft_ge(len(pp_list), 0)
75 refs = sum([pp["inflight"] for pp in pp_list])
76 ksft_eq(refs, 0)
77
78 # Down, it disappears, again
79 down()
80 pp_list = get_pp()
81 ksft_eq(len(pp_list), 0)
82
83 # Up, allocate a page
84 up()
85 nsim.dfs_write("pp_hold", "y")
86 pp_list = nf.page_pool_get({}, dump=True)
87 refs = sum([pp["inflight"] for pp in pp_list if pp.get("ifindex") == nsim.ifindex])
88 ksft_ge(refs, 1)
89
90 # Now let's leak a page
91 down()
92 pp_list = get_pp()
93 ksft_eq(len(pp_list), 1)
94 refs = sum([pp["inflight"] for pp in pp_list])
95 ksft_eq(refs, 1)
96 attached = [pp for pp in pp_list if "detach-time" not in pp]
97 ksft_eq(len(attached), 0)
98
99 # New pp can get created, and we'll have two
100 up()
101 pp_list = get_pp()
102 attached = [pp for pp in pp_list if "detach-time" not in pp]
103 detached = [pp for pp in pp_list if "detach-time" in pp]
104 ksft_eq(len(attached), 1)
105 ksft_eq(len(detached), 1)
106
107 # Free the old page and the old pp is gone
108 nsim.dfs_write("pp_hold", "n")
109 # Freeing check is once a second so we may need to retry
110 ksft_busy_wait(lambda: len(get_pp()) == 1, deadline=2)
111
112 # And down...
113 down()
114 ksft_eq(len(get_pp()), 0)
115
116 # Last, leave the page hanging for destroy, nothing to check
117 # we're trying to exercise the orphaning path in the kernel
118 up()
119 nsim.dfs_write("pp_hold", "y")
120
121
122def main() -> None:
123 nf = NetdevFamily()
124 ksft_run([empty_check, lo_check, page_pool_check, napi_list_check,
125 nsim_rxq_reset_down],
126 args=(nf, ))
127 ksft_exit()
128
129
130if __name__ == "__main__":
131 main()