selftests: net: exercise page pool reporting via netlink

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

Reviewed-by: Petr Machata <petrm@nvidia.com>
Link: https://lore.kernel.org/r/20240412141436.828666-7-kuba@kernel.org
Signed-off-by: Jakub Kicinski <kuba@kernel.org>

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 @@ def ksft_ge(a, b, comment=""):
         _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:

tools/testing/selftests/net/lib/py/nsim.py

@@ -28,6 +28,7 @@ class NetdevSim:
         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)

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 @@ def lo_check(nf) -> None:
     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__":