AuxXxilium/linux_dsm_epyc7002
1
0
Fork 0
mirror of https://github.com/AuxXxilium/linux_dsm_epyc7002.git synced 2024-12-28 11:18:45 +07:00
Code Issues Actions Packages Projects Releases Wiki Activity
752d7b4501
linux_dsm_epyc7002/tools/testing/selftests/bpf/test_offload.py
Jakub Kicinski 752d7b4501 selftests/bpf: test device info reporting for bound progs 
Check if bound programs report correct device info.  Test
in local namespace, in remote one, back to the local ns,
remove the device and check that information is cleared.

Signed-off-by: Jakub Kicinski <jakub.kicinski@netronome.com>
Reviewed-by: Quentin Monnet <quentin.monnet@netronome.com>
Signed-off-by: Daniel Borkmann <daniel@iogearbox.net>
2017-12-31 16:12:23 +01:00

772 lines
25 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/python3
# Copyright (C) 2017 Netronome Systems, Inc.
#
# This software is licensed under the GNU General License Version 2,
# June 1991 as shown in the file COPYING in the top-level directory of this
# source tree.
#
# THE COPYRIGHT HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS"
# WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING,
# BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE
# OF THE PROGRAM IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME
# THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION.
from datetime import datetime
import argparse
import json
import os
import pprint
import random
import string
import subprocess
import time
logfile = None
log_level = 1
bpf_test_dir = os.path.dirname(os.path.realpath(__file__))
pp = pprint.PrettyPrinter()
devs = [] # devices we created for clean up
files = [] # files to be removed
netns = [] # net namespaces to be removed
def log_get_sec(level=0):
return "*" * (log_level + level)
def log_level_inc(add=1):
global log_level
log_level += add
def log_level_dec(sub=1):
global log_level
log_level -= sub
def log_level_set(level):
global log_level
log_level = level
def log(header, data, level=None):
"""
Output to an optional log.
"""
if logfile is None:
return
if level is not None:
log_level_set(level)
if not isinstance(data, str):
data = pp.pformat(data)
if len(header):
logfile.write("\n" + log_get_sec() + " ")
logfile.write(header)
if len(header) and len(data.strip()):
logfile.write("\n")
logfile.write(data)
def skip(cond, msg):
if not cond:
return
print("SKIP: " + msg)
log("SKIP: " + msg, "", level=1)
os.sys.exit(0)
def fail(cond, msg):
if not cond:
return
print("FAIL: " + msg)
log("FAIL: " + msg, "", level=1)
os.sys.exit(1)
def start_test(msg):
log(msg, "", level=1)
log_level_inc()
print(msg)
def cmd(cmd, shell=True, include_stderr=False, background=False, fail=True):
"""
Run a command in subprocess and return tuple of (retval, stdout);
optionally return stderr as well as third value.
"""
proc = subprocess.Popen(cmd, shell=shell, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
if background:
msg = "%s START: %s" % (log_get_sec(1),
datetime.now().strftime("%H:%M:%S.%f"))
log("BKG " + proc.args, msg)
return proc
return cmd_result(proc, include_stderr=include_stderr, fail=fail)
def cmd_result(proc, include_stderr=False, fail=False):
stdout, stderr = proc.communicate()
stdout = stdout.decode("utf-8")
stderr = stderr.decode("utf-8")
proc.stdout.close()
proc.stderr.close()
stderr = "\n" + stderr
if stderr[-1] == "\n":
stderr = stderr[:-1]
sec = log_get_sec(1)
log("CMD " + proc.args,
"RETCODE: %d\n%s STDOUT:\n%s%s STDERR:%s\n%s END: %s" %
(proc.returncode, sec, stdout, sec, stderr,
sec, datetime.now().strftime("%H:%M:%S.%f")))
if proc.returncode != 0 and fail:
if len(stderr) > 0 and stderr[-1] == "\n":
stderr = stderr[:-1]
raise Exception("Command failed: %s\n%s" % (proc.args, stderr))
if include_stderr:
return proc.returncode, stdout, stderr
else:
return proc.returncode, stdout
def rm(f):
cmd("rm -f %s" % (f))
if f in files:
files.remove(f)
def tool(name, args, flags, JSON=True, ns="", fail=True):
params = ""
if JSON:
params += "%s " % (flags["json"])
if ns != "":
ns = "ip netns exec %s " % (ns)
ret, out = cmd(ns + name + " " + params + args, fail=fail)
if JSON and len(out.strip()) != 0:
return ret, json.loads(out)
else:
return ret, out
def bpftool(args, JSON=True, ns="", fail=True):
return tool("bpftool", args, {"json":"-p"}, JSON=JSON, ns=ns, fail=fail)
def bpftool_prog_list(expected=None, ns=""):
_, progs = bpftool("prog show", JSON=True, ns=ns, fail=True)
if expected is not None:
if len(progs) != expected:
fail(True, "%d BPF programs loaded, expected %d" %
(len(progs), expected))
return progs
def bpftool_prog_list_wait(expected=0, n_retry=20):
for i in range(n_retry):
nprogs = len(bpftool_prog_list())
if nprogs == expected:
return
time.sleep(0.05)
raise Exception("Time out waiting for program counts to stabilize want %d, have %d" % (expected, nprogs))
def ip(args, force=False, JSON=True, ns="", fail=True):
if force:
args = "-force " + args
return tool("ip", args, {"json":"-j"}, JSON=JSON, ns=ns, fail=fail)
def tc(args, JSON=True, ns="", fail=True):
return tool("tc", args, {"json":"-p"}, JSON=JSON, ns=ns, fail=fail)
def ethtool(dev, opt, args, fail=True):
return cmd("ethtool %s %s %s" % (opt, dev["ifname"], args), fail=fail)
def bpf_obj(name, sec=".text", path=bpf_test_dir,):
return "obj %s sec %s" % (os.path.join(path, name), sec)
def bpf_pinned(name):
return "pinned %s" % (name)
def bpf_bytecode(bytecode):
return "bytecode \"%s\"" % (bytecode)
def mknetns(n_retry=10):
for i in range(n_retry):
name = ''.join([random.choice(string.ascii_letters) for i in range(8)])
ret, _ = ip("netns add %s" % (name), fail=False)
if ret == 0:
netns.append(name)
return name
return None
class DebugfsDir:
"""
Class for accessing DebugFS directories as a dictionary.
"""
def __init__(self, path):
self.path = path
self._dict = self._debugfs_dir_read(path)
def __len__(self):
return len(self._dict.keys())
def __getitem__(self, key):
if type(key) is int:
key = list(self._dict.keys())[key]
return self._dict[key]
def __setitem__(self, key, value):
log("DebugFS set %s = %s" % (key, value), "")
log_level_inc()
cmd("echo '%s' > %s/%s" % (value, self.path, key))
log_level_dec()
_, out = cmd('cat %s/%s' % (self.path, key))
self._dict[key] = out.strip()
def _debugfs_dir_read(self, path):
dfs = {}
log("DebugFS state for %s" % (path), "")
log_level_inc(add=2)
_, out = cmd('ls ' + path)
for f in out.split():
p = os.path.join(path, f)
if os.path.isfile(p):
_, out = cmd('cat %s/%s' % (path, f))
dfs[f] = out.strip()
elif os.path.isdir(p):
dfs[f] = DebugfsDir(p)
else:
raise Exception("%s is neither file nor directory" % (p))
log_level_dec()
log("DebugFS state", dfs)
log_level_dec()
return dfs
class NetdevSim:
"""
Class for netdevsim netdevice and its attributes.
"""
def __init__(self):
self.dev = self._netdevsim_create()
devs.append(self)
self.ns = ""
self.dfs_dir = '/sys/kernel/debug/netdevsim/%s' % (self.dev['ifname'])
self.dfs_refresh()
def __getitem__(self, key):
return self.dev[key]
def _netdevsim_create(self):
_, old = ip("link show")
ip("link add sim%d type netdevsim")
_, new = ip("link show")
for dev in new:
f = filter(lambda x: x["ifname"] == dev["ifname"], old)
if len(list(f)) == 0:
return dev
raise Exception("failed to create netdevsim device")
def remove(self):
devs.remove(self)
ip("link del dev %s" % (self.dev["ifname"]), ns=self.ns)
def dfs_refresh(self):
self.dfs = DebugfsDir(self.dfs_dir)
return self.dfs
def dfs_num_bound_progs(self):
path = os.path.join(self.dfs_dir, "bpf_bound_progs")
_, progs = cmd('ls %s' % (path))
return len(progs.split())
def dfs_get_bound_progs(self, expected):
progs = DebugfsDir(os.path.join(self.dfs_dir, "bpf_bound_progs"))
if expected is not None:
if len(progs) != expected:
fail(True, "%d BPF programs bound, expected %d" %
(len(progs), expected))
return progs
def wait_for_flush(self, bound=0, total=0, n_retry=20):
for i in range(n_retry):
nbound = self.dfs_num_bound_progs()
nprogs = len(bpftool_prog_list())
if nbound == bound and nprogs == total:
return
time.sleep(0.05)
raise Exception("Time out waiting for program counts to stabilize want %d/%d, have %d bound, %d loaded" % (bound, total, nbound, nprogs))
def set_ns(self, ns):
name = "1" if ns == "" else ns
ip("link set dev %s netns %s" % (self.dev["ifname"], name), ns=self.ns)
self.ns = ns
def set_mtu(self, mtu, fail=True):
return ip("link set dev %s mtu %d" % (self.dev["ifname"], mtu),
fail=fail)
def set_xdp(self, bpf, mode, force=False, fail=True):
return ip("link set dev %s xdp%s %s" % (self.dev["ifname"], mode, bpf),
force=force, fail=fail)
def unset_xdp(self, mode, force=False, fail=True):
return ip("link set dev %s xdp%s off" % (self.dev["ifname"], mode),
force=force, fail=fail)
def ip_link_show(self, xdp):
_, link = ip("link show dev %s" % (self['ifname']))
if len(link) > 1:
raise Exception("Multiple objects on ip link show")
if len(link) < 1:
return {}
fail(xdp != "xdp" in link,
"XDP program not reporting in iplink (reported %s, expected %s)" %
("xdp" in link, xdp))
return link[0]
def tc_add_ingress(self):
tc("qdisc add dev %s ingress" % (self['ifname']))
def tc_del_ingress(self):
tc("qdisc del dev %s ingress" % (self['ifname']))
def tc_flush_filters(self, bound=0, total=0):
self.tc_del_ingress()
self.tc_add_ingress()
self.wait_for_flush(bound=bound, total=total)
def tc_show_ingress(self, expected=None):
# No JSON support, oh well...
flags = ["skip_sw", "skip_hw", "in_hw"]
named = ["protocol", "pref", "chain", "handle", "id", "tag"]
args = "-s filter show dev %s ingress" % (self['ifname'])
_, out = tc(args, JSON=False)
filters = []
lines = out.split('\n')
for line in lines:
words = line.split()
if "handle" not in words:
continue
fltr = {}
for flag in flags:
fltr[flag] = flag in words
for name in named:
try:
idx = words.index(name)
fltr[name] = words[idx + 1]
except ValueError:
pass
filters.append(fltr)
if expected is not None:
fail(len(filters) != expected,
"%d ingress filters loaded, expected %d" %
(len(filters), expected))
return filters
def cls_bpf_add_filter(self, bpf, da=False, skip_sw=False, skip_hw=False,
fail=True):
params = ""
if da:
params += " da"
if skip_sw:
params += " skip_sw"
if skip_hw:
params += " skip_hw"
return tc("filter add dev %s ingress bpf %s %s" %
(self['ifname'], bpf, params), fail=fail)
def set_ethtool_tc_offloads(self, enable, fail=True):
args = "hw-tc-offload %s" % ("on" if enable else "off")
return ethtool(self, "-K", args, fail=fail)
################################################################################
def clean_up():
for dev in devs:
dev.remove()
for f in files:
cmd("rm -f %s" % (f))
for ns in netns:
cmd("ip netns delete %s" % (ns))
def pin_prog(file_name, idx=0):
progs = bpftool_prog_list(expected=(idx + 1))
prog = progs[idx]
bpftool("prog pin id %d %s" % (prog["id"], file_name))
files.append(file_name)
return file_name, bpf_pinned(file_name)
def check_dev_info(other_ns, ns, pin_file=None, removed=False):
if removed:
bpftool_prog_list(expected=0)
ret, err = bpftool("prog show pin %s" % (pin_file), fail=False)
fail(ret == 0, "Showing prog with removed device did not fail")
fail(err["error"].find("No such device") == -1,
"Showing prog with removed device expected ENODEV, error is %s" %
(err["error"]))
return
progs = bpftool_prog_list(expected=int(not removed), ns=ns)
prog = progs[0]
fail("dev" not in prog.keys(), "Device parameters not reported")
dev = prog["dev"]
fail("ifindex" not in dev.keys(), "Device parameters not reported")
fail("ns_dev" not in dev.keys(), "Device parameters not reported")
fail("ns_inode" not in dev.keys(), "Device parameters not reported")
if not removed and not other_ns:
fail("ifname" not in dev.keys(), "Ifname not reported")
fail(dev["ifname"] != sim["ifname"],
"Ifname incorrect %s vs %s" % (dev["ifname"], sim["ifname"]))
else:
fail("ifname" in dev.keys(), "Ifname is reported for other ns")
if removed:
fail(dev["ifindex"] != 0, "Device perameters not zero on removed")
fail(dev["ns_dev"] != 0, "Device perameters not zero on removed")
fail(dev["ns_inode"] != 0, "Device perameters not zero on removed")
# Parse command line
parser = argparse.ArgumentParser()
parser.add_argument("--log", help="output verbose log to given file")
args = parser.parse_args()
if args.log:
logfile = open(args.log, 'w+')
logfile.write("# -*-Org-*-")
log("Prepare...", "", level=1)
log_level_inc()
# Check permissions
skip(os.getuid() != 0, "test must be run as root")
# Check tools
ret, progs = bpftool("prog", fail=False)
skip(ret != 0, "bpftool not installed")
# Check no BPF programs are loaded
skip(len(progs) != 0, "BPF programs already loaded on the system")
# Check netdevsim
ret, out = cmd("modprobe netdevsim", fail=False)
skip(ret != 0, "netdevsim module could not be loaded")
# Check debugfs
_, out = cmd("mount")
if out.find("/sys/kernel/debug type debugfs") == -1:
cmd("mount -t debugfs none /sys/kernel/debug")
# Check samples are compiled
samples = ["sample_ret0.o"]
for s in samples:
ret, out = cmd("ls %s/%s" % (bpf_test_dir, s), fail=False)
skip(ret != 0, "sample %s/%s not found, please compile it" %
(bpf_test_dir, s))
# Check if net namespaces seem to work
ns = mknetns()
skip(ns is None, "Could not create a net namespace")
cmd("ip netns delete %s" % (ns))
netns = []
try:
obj = bpf_obj("sample_ret0.o")
bytecode = bpf_bytecode("1,6 0 0 4294967295,")
start_test("Test destruction of generic XDP...")
sim = NetdevSim()
sim.set_xdp(obj, "generic")
sim.remove()
bpftool_prog_list_wait(expected=0)
sim = NetdevSim()
sim.tc_add_ingress()
start_test("Test TC non-offloaded...")
ret, _ = sim.cls_bpf_add_filter(obj, skip_hw=True, fail=False)
fail(ret != 0, "Software TC filter did not load")
start_test("Test TC non-offloaded isn't getting bound...")
ret, _ = sim.cls_bpf_add_filter(obj, fail=False)
fail(ret != 0, "Software TC filter did not load")
sim.dfs_get_bound_progs(expected=0)
sim.tc_flush_filters()
start_test("Test TC offloads are off by default...")
ret, _ = sim.cls_bpf_add_filter(obj, skip_sw=True, fail=False)
fail(ret == 0, "TC filter loaded without enabling TC offloads")
sim.wait_for_flush()
sim.set_ethtool_tc_offloads(True)
sim.dfs["bpf_tc_non_bound_accept"] = "Y"
start_test("Test TC offload by default...")
ret, _ = sim.cls_bpf_add_filter(obj, fail=False)
fail(ret != 0, "Software TC filter did not load")
sim.dfs_get_bound_progs(expected=0)
ingress = sim.tc_show_ingress(expected=1)
fltr = ingress[0]
fail(not fltr["in_hw"], "Filter not offloaded by default")
sim.tc_flush_filters()
start_test("Test TC cBPF bytcode tries offload by default...")
ret, _ = sim.cls_bpf_add_filter(bytecode, fail=False)
fail(ret != 0, "Software TC filter did not load")
sim.dfs_get_bound_progs(expected=0)
ingress = sim.tc_show_ingress(expected=1)
fltr = ingress[0]
fail(not fltr["in_hw"], "Bytecode not offloaded by default")
sim.tc_flush_filters()
sim.dfs["bpf_tc_non_bound_accept"] = "N"
start_test("Test TC cBPF unbound bytecode doesn't offload...")
ret, _ = sim.cls_bpf_add_filter(bytecode, skip_sw=True, fail=False)
fail(ret == 0, "TC bytecode loaded for offload")
sim.wait_for_flush()
start_test("Test TC offloads work...")
ret, _ = sim.cls_bpf_add_filter(obj, skip_sw=True, fail=False)
fail(ret != 0, "TC filter did not load with TC offloads enabled")
start_test("Test TC offload basics...")
dfs = sim.dfs_get_bound_progs(expected=1)
progs = bpftool_prog_list(expected=1)
ingress = sim.tc_show_ingress(expected=1)
dprog = dfs[0]
prog = progs[0]
fltr = ingress[0]
fail(fltr["skip_hw"], "TC does reports 'skip_hw' on offloaded filter")
fail(not fltr["in_hw"], "TC does not report 'in_hw' for offloaded filter")
fail(not fltr["skip_sw"], "TC does not report 'skip_sw' back")
start_test("Test TC offload is device-bound...")
fail(str(prog["id"]) != fltr["id"], "Program IDs don't match")
fail(prog["tag"] != fltr["tag"], "Program tags don't match")
fail(fltr["id"] != dprog["id"], "Program IDs don't match")
fail(dprog["state"] != "xlated", "Offloaded program state not translated")
fail(dprog["loaded"] != "Y", "Offloaded program is not loaded")
start_test("Test disabling TC offloads is rejected while filters installed...")
ret, _ = sim.set_ethtool_tc_offloads(False, fail=False)
fail(ret == 0, "Driver should refuse to disable TC offloads with filters installed...")
start_test("Test qdisc removal frees things...")
sim.tc_flush_filters()
sim.tc_show_ingress(expected=0)
start_test("Test disabling TC offloads is OK without filters...")
ret, _ = sim.set_ethtool_tc_offloads(False, fail=False)
fail(ret != 0,
"Driver refused to disable TC offloads without filters installed...")
sim.set_ethtool_tc_offloads(True)
start_test("Test destroying device gets rid of TC filters...")
sim.cls_bpf_add_filter(obj, skip_sw=True)
sim.remove()
bpftool_prog_list_wait(expected=0)
sim = NetdevSim()
sim.set_ethtool_tc_offloads(True)
start_test("Test destroying device gets rid of XDP...")
sim.set_xdp(obj, "offload")
sim.remove()
bpftool_prog_list_wait(expected=0)
sim = NetdevSim()
sim.set_ethtool_tc_offloads(True)
start_test("Test XDP prog reporting...")
sim.set_xdp(obj, "drv")
ipl = sim.ip_link_show(xdp=True)
progs = bpftool_prog_list(expected=1)
fail(ipl["xdp"]["prog"]["id"] != progs[0]["id"],
"Loaded program has wrong ID")
start_test("Test XDP prog replace without force...")
ret, _ = sim.set_xdp(obj, "drv", fail=False)
fail(ret == 0, "Replaced XDP program without -force")
sim.wait_for_flush(total=1)
start_test("Test XDP prog replace with force...")
ret, _ = sim.set_xdp(obj, "drv", force=True, fail=False)
fail(ret != 0, "Could not replace XDP program with -force")
bpftool_prog_list_wait(expected=1)
ipl = sim.ip_link_show(xdp=True)
progs = bpftool_prog_list(expected=1)
fail(ipl["xdp"]["prog"]["id"] != progs[0]["id"],
"Loaded program has wrong ID")
fail("dev" in progs[0].keys(),
"Device parameters reported for non-offloaded program")
start_test("Test XDP prog replace with bad flags...")
ret, _ = sim.set_xdp(obj, "offload", force=True, fail=False)
fail(ret == 0, "Replaced XDP program with a program in different mode")
ret, _ = sim.set_xdp(obj, "", force=True, fail=False)
fail(ret == 0, "Replaced XDP program with a program in different mode")
start_test("Test XDP prog remove with bad flags...")
ret, _ = sim.unset_xdp("offload", force=True, fail=False)
fail(ret == 0, "Removed program with a bad mode mode")
ret, _ = sim.unset_xdp("", force=True, fail=False)
fail(ret == 0, "Removed program with a bad mode mode")
start_test("Test MTU restrictions...")
ret, _ = sim.set_mtu(9000, fail=False)
fail(ret == 0,
"Driver should refuse to increase MTU to 9000 with XDP loaded...")
sim.unset_xdp("drv")
bpftool_prog_list_wait(expected=0)
sim.set_mtu(9000)
ret, _ = sim.set_xdp(obj, "drv", fail=False)
fail(ret == 0, "Driver should refuse to load program with MTU of 9000...")
sim.set_mtu(1500)
sim.wait_for_flush()
start_test("Test XDP offload...")
sim.set_xdp(obj, "offload")
ipl = sim.ip_link_show(xdp=True)
link_xdp = ipl["xdp"]["prog"]
progs = bpftool_prog_list(expected=1)
prog = progs[0]
fail(link_xdp["id"] != prog["id"], "Loaded program has wrong ID")
start_test("Test XDP offload is device bound...")
dfs = sim.dfs_get_bound_progs(expected=1)
dprog = dfs[0]
fail(prog["id"] != link_xdp["id"], "Program IDs don't match")
fail(prog["tag"] != link_xdp["tag"], "Program tags don't match")
fail(str(link_xdp["id"]) != dprog["id"], "Program IDs don't match")
fail(dprog["state"] != "xlated", "Offloaded program state not translated")
fail(dprog["loaded"] != "Y", "Offloaded program is not loaded")
start_test("Test removing XDP program many times...")
sim.unset_xdp("offload")
sim.unset_xdp("offload")
sim.unset_xdp("drv")
sim.unset_xdp("drv")
sim.unset_xdp("")
sim.unset_xdp("")
bpftool_prog_list_wait(expected=0)
start_test("Test attempt to use a program for a wrong device...")
sim2 = NetdevSim()
sim2.set_xdp(obj, "offload")
pin_file, pinned = pin_prog("/sys/fs/bpf/tmp")
ret, _ = sim.set_xdp(pinned, "offload", fail=False)
fail(ret == 0, "Pinned program loaded for a different device accepted")
sim2.remove()
ret, _ = sim.set_xdp(pinned, "offload", fail=False)
fail(ret == 0, "Pinned program loaded for a removed device accepted")
rm(pin_file)
bpftool_prog_list_wait(expected=0)
start_test("Test mixing of TC and XDP...")
sim.tc_add_ingress()
sim.set_xdp(obj, "offload")
ret, _ = sim.cls_bpf_add_filter(obj, skip_sw=True, fail=False)
fail(ret == 0, "Loading TC when XDP active should fail")
sim.unset_xdp("offload")
sim.wait_for_flush()
sim.cls_bpf_add_filter(obj, skip_sw=True)
ret, _ = sim.set_xdp(obj, "offload", fail=False)
fail(ret == 0, "Loading XDP when TC active should fail")
start_test("Test binding TC from pinned...")
pin_file, pinned = pin_prog("/sys/fs/bpf/tmp")
sim.tc_flush_filters(bound=1, total=1)
sim.cls_bpf_add_filter(pinned, da=True, skip_sw=True)
sim.tc_flush_filters(bound=1, total=1)
start_test("Test binding XDP from pinned...")
sim.set_xdp(obj, "offload")
pin_file, pinned = pin_prog("/sys/fs/bpf/tmp2", idx=1)
sim.set_xdp(pinned, "offload", force=True)
sim.unset_xdp("offload")
sim.set_xdp(pinned, "offload", force=True)
sim.unset_xdp("offload")
start_test("Test offload of wrong type fails...")
ret, _ = sim.cls_bpf_add_filter(pinned, da=True, skip_sw=True, fail=False)
fail(ret == 0, "Managed to attach XDP program to TC")
start_test("Test asking for TC offload of two filters...")
sim.cls_bpf_add_filter(obj, da=True, skip_sw=True)
ret, _ = sim.cls_bpf_add_filter(obj, da=True, skip_sw=True, fail=False)
fail(ret == 0, "Managed to offload two TC filters at the same time")
sim.tc_flush_filters(bound=2, total=2)
start_test("Test if netdev removal waits for translation...")
delay_msec = 500
sim.dfs["bpf_bind_verifier_delay"] = delay_msec
start = time.time()
cmd_line = "tc filter add dev %s ingress bpf %s da skip_sw" % \
(sim['ifname'], obj)
tc_proc = cmd(cmd_line, background=True, fail=False)
# Wait for the verifier to start
while sim.dfs_num_bound_progs() <= 2:
pass
sim.remove()
end = time.time()
ret, _ = cmd_result(tc_proc, fail=False)
time_diff = end - start
log("Time", "start:\t%s\nend:\t%s\ndiff:\t%s" % (start, end, time_diff))
fail(ret == 0, "Managed to load TC filter on a unregistering device")
delay_sec = delay_msec * 0.001
fail(time_diff < delay_sec, "Removal process took %s, expected %s" %
(time_diff, delay_sec))
# Remove all pinned files and reinstantiate the netdev
clean_up()
bpftool_prog_list_wait(expected=0)
sim = NetdevSim()
sim.set_ethtool_tc_offloads(True)
sim.set_xdp(obj, "offload")
start_test("Test bpftool bound info reporting (own ns)...")
check_dev_info(False, "")
start_test("Test bpftool bound info reporting (other ns)...")
ns = mknetns()
sim.set_ns(ns)
check_dev_info(True, "")
start_test("Test bpftool bound info reporting (remote ns)...")
check_dev_info(False, ns)
start_test("Test bpftool bound info reporting (back to own ns)...")
sim.set_ns("")
check_dev_info(False, "")
pin_file, _ = pin_prog("/sys/fs/bpf/tmp")
sim.remove()
start_test("Test bpftool bound info reporting (removed dev)...")
check_dev_info(True, "", pin_file=pin_file, removed=True)
print("%s: OK" % (os.path.basename(__file__)))
finally:
log("Clean up...", "", level=1)
log_level_inc()
clean_up()
Reference in New Issue View Git Blame Copy Permalink