Add a Python test for the basic ops.
# ./net/nl_netdev.py
KTAP version 1
1..3
ok 1 nl_netdev.empty_check
ok 2 nl_netdev.lo_check
ok 3 nl_netdev.page_pool_check
# Totals: pass:3 fail:0 xfail:0 xpass:0 skip:0 error:0
Signed-off-by: Jakub Kicinski <kuba@xxxxxxxxxx>
---
v2:
- move up / down to the test
- add helper for getting pp
- add busy_wait helper
- rename undeteched
---
tools/testing/selftests/net/lib/py/ksft.py | 12 ++++
tools/testing/selftests/net/lib/py/nsim.py | 1 +
tools/testing/selftests/net/nl_netdev.py | 76 +++++++++++++++++++++-
3 files changed, 87 insertions(+), 2 deletions(-)
diff --git a/tools/testing/selftests/net/lib/py/ksft.py b/tools/testing/selftests/net/lib/py/ksft.py
index 793e4761645e..3769b9197213 100644
--- a/tools/testing/selftests/net/lib/py/ksft.py
+++ b/tools/testing/selftests/net/lib/py/ksft.py
@@ -2,6 +2,7 @@
import builtins
import inspect
+import time
import traceback
from .consts import KSFT_MAIN_NAME
@@ -50,6 +51,17 @@ KSFT_RESULT = None
_fail("Check failed", a, "<", b, comment)
+def ksft_busy_wait(cond, sleep=0.005, deadline=1, comment=""):
+ end = time.monotonic() + deadline
+ while True:
+ if cond():
+ return
+ if time.monotonic() > end:
+ _fail("Waiting for condition timed out", comment)
+ return
+ time.sleep(sleep)
+
+
def ktap_result(ok, cnt=1, case="", comment=""):
res = ""
if not ok:
diff --git a/tools/testing/selftests/net/lib/py/nsim.py b/tools/testing/selftests/net/lib/py/nsim.py
index 94aa32f59fdb..06896cdf7c18 100644
--- a/tools/testing/selftests/net/lib/py/nsim.py
+++ b/tools/testing/selftests/net/lib/py/nsim.py
@@ -28,6 +28,7 @@ from .utils import cmd, ip
self.dfs_dir = "%s/ports/%u/" % (nsimdev.dfs_dir, port_index)
ret = ip("-j link show dev %s" % ifname, ns=ns)
self.dev = json.loads(ret.stdout)[0]
+ self.ifindex = self.dev["ifindex"]
def dfs_write(self, path, val):
self.nsimdev.dfs_write(f'ports/{self.port_index}/' + path, val)
diff --git a/tools/testing/selftests/net/nl_netdev.py b/tools/testing/selftests/net/nl_netdev.py
index 2b8b488fb419..6909b1760739 100755
--- a/tools/testing/selftests/net/nl_netdev.py
+++ b/tools/testing/selftests/net/nl_netdev.py
@@ -1,7 +1,9 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
-from lib.py import ksft_run, ksft_pr, ksft_eq, ksft_ge, NetdevFamily
+import time
+from lib.py import ksft_run, ksft_pr, ksft_eq, ksft_ge, ksft_busy_wait
+from lib.py import NetdevFamily, NetdevSimDev, ip
def empty_check(nf) -> None:
@@ -15,9 +17,79 @@ from lib.py import ksft_run, ksft_pr, ksft_eq, ksft_ge, NetdevFamily
ksft_eq(len(lo_info['xdp-rx-metadata-features']), 0)
+def page_pool_check(nf) -> None:
+ with NetdevSimDev() as nsimdev:
+ nsim = nsimdev.nsims[0]
+
+ def up():
+ ip(f"link set dev {nsim.ifname} up")
+
+ def down():
+ ip(f"link set dev {nsim.ifname} down")
+
+ def get_pp():
+ pp_list = nf.page_pool_get({}, dump=True)
+ return [pp for pp in pp_list if pp.get("ifindex") == nsim.ifindex]
+
+ # No page pools when down
+ down()
+ ksft_eq(len(get_pp()), 0)
+
+ # Up, empty page pool appears
+ up()
+ pp_list = get_pp()
+ ksft_ge(len(pp_list), 0)
+ refs = sum([pp["inflight"] for pp in pp_list])
+ ksft_eq(refs, 0)
+
+ # Down, it disappears, again
+ down()
+ pp_list = get_pp()
+ ksft_eq(len(pp_list), 0)
+
+ # Up, allocate a page
+ up()
+ nsim.dfs_write("pp_hold", "y")
+ pp_list = nf.page_pool_get({}, dump=True)
+ refs = sum([pp["inflight"] for pp in pp_list if pp.get("ifindex") == nsim.ifindex])
+ ksft_ge(refs, 1)
+
+ # Now let's leak a page
+ down()
+ pp_list = get_pp()
+ ksft_eq(len(pp_list), 1)
+ refs = sum([pp["inflight"] for pp in pp_list])
+ ksft_eq(refs, 1)
+ attached = [pp for pp in pp_list if "detach-time" not in pp]
+ ksft_eq(len(attached), 0)
+
+ # New pp can get created, and we'll have two
+ up()
+ pp_list = get_pp()
+ attached = [pp for pp in pp_list if "detach-time" not in pp]
+ detached = [pp for pp in pp_list if "detach-time" in pp]
+ ksft_eq(len(attached), 1)
+ ksft_eq(len(detached), 1)
+
+ # Free the old page and the old pp is gone
+ nsim.dfs_write("pp_hold", "n")
+ # Freeing check is once a second so we may need to retry
+ ksft_busy_wait(lambda: len(get_pp()) == 1, deadline=2)
+
+ # And down...
+ down()
+ ksft_eq(len(get_pp()), 0)
+
+ # Last, leave the page hanging for destroy, nothing to check
+ # we're trying to exercise the orphaning path in the kernel
+ up()
+ nsim.dfs_write("pp_hold", "y")
+
+
def main() -> None:
nf = NetdevFamily()
- ksft_run([empty_check, lo_check], args=(nf, ))
+ ksft_run([empty_check, lo_check, page_pool_check],
+ args=(nf, ))
if __name__ == "__main__":
--
2.44.0
