tools/testing/selftests/net/nl_netdev.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: GPL-2.0
   3
   4 import time
   5 from lib.py import ksft_run, ksft_exit, ksft_pr
   6 from lib.py import ksft_eq, ksft_ge, ksft_busy_wait
   7 from lib.py import NetdevFamily, NetdevSimDev, ip
   8
   9
  10 def empty_check(nf) -> None:
  11     devs = nf.dev_get({}, dump=True)
  12     ksft_ge(len(devs), 1)
  13
  14
  15 def 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
  21 def page_pool_check(nf) -> None:
  22     with NetdevSimDev() as nsimdev:
  23         nsim = nsimdev.nsims[0]
  24
  25         def up():
  26             ip(f"link set dev {nsim.ifname} up")
  27
  28         def down():
  29             ip(f"link set dev {nsim.ifname} down")
  30
  31         def get_pp():
  32             pp_list = nf.page_pool_get({}, dump=True)
  33             return [pp for pp in pp_list if pp.get("ifindex") == nsim.ifindex]
  34
  35         # No page pools when down
  36         down()
  37         ksft_eq(len(get_pp()), 0)
  38
  39         # Up, empty page pool appears
  40         up()
  41         pp_list = get_pp()
  42         ksft_ge(len(pp_list), 0)
  43         refs = sum([pp["inflight"] for pp in pp_list])
  44         ksft_eq(refs, 0)
  45
  46         # Down, it disappears, again
  47         down()
  48         pp_list = get_pp()
  49         ksft_eq(len(pp_list), 0)
  50
  51         # Up, allocate a page
  52         up()
  53         nsim.dfs_write("pp_hold", "y")
  54         pp_list = nf.page_pool_get({}, dump=True)
  55         refs = sum([pp["inflight"] for pp in pp_list if pp.get("ifindex") == nsim.ifindex])
  56         ksft_ge(refs, 1)
  57
  58         # Now let's leak a page
  59         down()
  60         pp_list = get_pp()
  61         ksft_eq(len(pp_list), 1)
  62         refs = sum([pp["inflight"] for pp in pp_list])
  63         ksft_eq(refs, 1)
  64         attached = [pp for pp in pp_list if "detach-time" not in pp]
  65         ksft_eq(len(attached), 0)
  66
  67         # New pp can get created, and we'll have two
  68         up()
  69         pp_list = get_pp()
  70         attached = [pp for pp in pp_list if "detach-time" not in pp]
  71         detached = [pp for pp in pp_list if "detach-time" in pp]
  72         ksft_eq(len(attached), 1)
  73         ksft_eq(len(detached), 1)
  74
  75         # Free the old page and the old pp is gone
  76         nsim.dfs_write("pp_hold", "n")
  77         # Freeing check is once a second so we may need to retry
  78         ksft_busy_wait(lambda: len(get_pp()) == 1, deadline=2)
  79
  80         # And down...
  81         down()
  82         ksft_eq(len(get_pp()), 0)
  83
  84         # Last, leave the page hanging for destroy, nothing to check
  85         # we're trying to exercise the orphaning path in the kernel
  86         up()
  87         nsim.dfs_write("pp_hold", "y")
  88
  89
  90 def main() -> None:
  91     nf = NetdevFamily()
  92     ksft_run([empty_check, lo_check, page_pool_check],
  93              args=(nf, ))
  94     ksft_exit()
  95
  96
  97 if __name__ == "__main__":
  98     main()