557 lines
18 KiB
Python
Executable File
557 lines
18 KiB
Python
Executable File
#! /usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-2.0
|
|
"""
|
|
This module provides functional testing for the net/rds component.
|
|
"""
|
|
|
|
import argparse
|
|
import atexit
|
|
import ctypes
|
|
import errno
|
|
import hashlib
|
|
import os
|
|
import select
|
|
import re
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
# Allow utils module to be imported from different directory
|
|
this_dir = os.path.dirname(os.path.realpath(__file__))
|
|
sys.path.append(os.path.join(this_dir, "../"))
|
|
# pylint: disable-next=wrong-import-position,import-error,no-name-in-module
|
|
from lib.py.utils import ip, cmd # noqa: E402
|
|
# pylint: disable-next=wrong-import-position,import-error,no-name-in-module
|
|
from lib.py.ksft import ksft_pr # noqa: E402
|
|
|
|
libc = ctypes.cdll.LoadLibrary('libc.so.6')
|
|
setns = libc.setns
|
|
|
|
NET0 = 'net0'
|
|
NET1 = 'net1'
|
|
|
|
VETH0 = 'veth0'
|
|
VETH1 = 'veth1'
|
|
|
|
tcpdump_procs = []
|
|
tcp_addrs = [
|
|
# we technically don't need different port numbers, but this will
|
|
# help identify traffic in the network analyzer
|
|
('10.0.0.1', 10000),
|
|
('10.0.0.2', 20000),
|
|
]
|
|
|
|
# RDMA network configs
|
|
RXE_DEV0 = 'rxe0'
|
|
RXE_DEV1 = 'rxe1'
|
|
|
|
VETH_RDMA0 = 'veth_rdma0'
|
|
VETH_RDMA1 = 'veth_rdma1'
|
|
|
|
rdma_addrs = [
|
|
('10.0.0.3', 30000),
|
|
('10.0.0.4', 30000),
|
|
]
|
|
|
|
# send_packets flag space
|
|
OP_FLAG_TCP = 0x1
|
|
OP_FLAG_RDMA = 0x2
|
|
|
|
# from include/uapi/linux/rds.h: SO_RDS_TRANSPORT pins a socket to a
|
|
# specific RDS transport so connection setup cannot silently fall back
|
|
# to another (e.g. loopback) transport.
|
|
SOL_RDS = 276
|
|
SO_RDS_TRANSPORT = 8
|
|
RDS_TRANS_TCP = 2
|
|
RDS_TRANS_IB = 0
|
|
|
|
signal_handler_label = ""
|
|
|
|
tap_idx = 0
|
|
nr_pass = 0
|
|
nr_fail = 0
|
|
|
|
# Helper function for creating a socket inside a network namespace.
|
|
# We need this because otherwise RDS will detect that the two TCP
|
|
# sockets are on the same interface and use the loop transport instead
|
|
# of the TCP transport.
|
|
def netns_socket(netns, *sock_args):
|
|
"""
|
|
Creates sockets inside of network namespace
|
|
|
|
:param netns: the name of the network namespace
|
|
:param sock_args: socket family and type
|
|
"""
|
|
u0, u1 = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
|
|
child = os.fork()
|
|
if child == 0:
|
|
try:
|
|
# change network namespace
|
|
with open(f'/var/run/netns/{netns}', encoding='utf-8') as f:
|
|
setns(f.fileno(), 0)
|
|
# create socket in target namespace
|
|
sock = socket.socket(*sock_args)
|
|
|
|
# send resulting socket to parent
|
|
socket.send_fds(u0, [], [sock.fileno()])
|
|
|
|
os._exit(0)
|
|
except BaseException:
|
|
os._exit(1)
|
|
|
|
# receive socket from child
|
|
_, fds, _, _ = socket.recv_fds(u1, 0, 1)
|
|
_, status = os.waitpid(child, 0)
|
|
u0.close()
|
|
u1.close()
|
|
if not os.WIFEXITED(status) or os.WEXITSTATUS(status) != 0:
|
|
raise RuntimeError(
|
|
f"netns_socket child failed in netns {netns} (status={status})")
|
|
return socket.fromfd(fds[0], *sock_args)
|
|
|
|
def send_burst(socks, ip_addrs, snd_hashes, nr_sent, nr_total):
|
|
"""Send until blocked or nr_total reached. Return updated nr_sent."""
|
|
|
|
while nr_sent < nr_total:
|
|
data = hashlib.sha256(
|
|
f'packet {nr_sent}'.encode('utf-8')).hexdigest().encode('utf-8')
|
|
# pseudo-random send/receive pattern
|
|
snd_idx = nr_sent % 2
|
|
rcv_idx = 1 - (nr_sent % 3) % 2
|
|
|
|
snd = socks[snd_idx]
|
|
rcv = socks[rcv_idx]
|
|
try:
|
|
snd.sendto(data, ip_addrs[rcv_idx])
|
|
except BlockingIOError:
|
|
return nr_sent
|
|
except OSError as e:
|
|
if e.errno in (errno.ENOBUFS, errno.ECONNRESET, errno.EPIPE):
|
|
return nr_sent
|
|
raise
|
|
snd_hashes.setdefault((snd.fileno(), rcv.fileno()),
|
|
hashlib.sha256()).update(f'<{data}>'.encode('utf-8'))
|
|
nr_sent += 1
|
|
return nr_sent
|
|
|
|
def recv_burst(epoll, socks, ip_addrs, rcv_hashes, nr_rcv):
|
|
"""Drain whatever's readable from epoll. Return updated nr_recv."""
|
|
for filen, evntmask in epoll.poll():
|
|
if not evntmask & select.EPOLLRDNORM:
|
|
continue
|
|
rcv = next(s for s in socks if s.fileno() == filen)
|
|
while True:
|
|
try:
|
|
data, adr = rcv.recvfrom(1024)
|
|
except BlockingIOError:
|
|
break
|
|
snd_idx = ip_addrs.index(adr)
|
|
snd = socks[snd_idx]
|
|
rcv_hashes.setdefault((snd.fileno(), rcv.fileno()),
|
|
hashlib.sha256()).update(f'<{data}>'.encode('utf-8'))
|
|
nr_rcv += 1
|
|
return nr_rcv
|
|
|
|
def check_info(socks):
|
|
"""
|
|
Check all rds info pages for errors
|
|
|
|
:param socks: list of sockets to check
|
|
"""
|
|
|
|
# the Python socket module doesn't know these
|
|
rds_info_first = 10000
|
|
rds_info_last = 10017
|
|
|
|
nr_success = 0
|
|
nr_error = 0
|
|
|
|
for sock in socks:
|
|
for optname in range(rds_info_first, rds_info_last + 1):
|
|
# Sigh, the Python socket module doesn't allow us to pass
|
|
# buffer lengths greater than 1024 for some reason. RDS
|
|
# wants multiple pages.
|
|
try:
|
|
sock.getsockopt(socket.SOL_RDS, optname, 1024)
|
|
nr_success = nr_success + 1
|
|
except OSError as e:
|
|
nr_error = nr_error + 1
|
|
if e.errno == errno.ENOSPC:
|
|
# ignore
|
|
pass
|
|
|
|
ksft_pr(f"getsockopt(): {nr_success}/{nr_error}")
|
|
|
|
def verify_hashes(snd_hashes, rcv_hashes):
|
|
"""Compare send/recv hashes per (sender, receiver) pair."""
|
|
for key, snd_hash in snd_hashes.items():
|
|
rcv_hash = rcv_hashes.get(key)
|
|
if rcv_hash is None:
|
|
ksft_pr("FAIL: No data received")
|
|
return 1
|
|
if snd_hash.hexdigest() != rcv_hash.hexdigest():
|
|
ksft_pr("FAIL: Send/recv mismatch")
|
|
ksft_pr("hash expected:", snd_hash.hexdigest())
|
|
ksft_pr("hash received:", rcv_hash.hexdigest())
|
|
return 1
|
|
ksft_pr(f"{key[0]}/{key[1]}: ok")
|
|
return 0
|
|
|
|
def snd_rcv_packets(env):
|
|
"""
|
|
Send packets on the given network interfaces
|
|
|
|
:param env: transport-environment dict for setup_tcp() / setup_rdma().
|
|
"addrs": list of (ip, port) tuples matching the sockets
|
|
"netns": list of netns names for TCP or None for RDMA
|
|
"flags": OP_FLAG_TCP or OP_FLAG_RDMA, selects sockets
|
|
"""
|
|
|
|
addrs = env["addrs"]
|
|
netns_list = env["netns"]
|
|
flags = env.get("flags", 0)
|
|
|
|
if (flags & OP_FLAG_TCP) and (flags & OP_FLAG_RDMA):
|
|
raise RuntimeError(f"Invalid transport flag sets multiple transports: {flags}")
|
|
|
|
if flags & OP_FLAG_TCP:
|
|
sockets = [
|
|
netns_socket(netns_list[0], socket.AF_RDS, socket.SOCK_SEQPACKET),
|
|
netns_socket(netns_list[1], socket.AF_RDS, socket.SOCK_SEQPACKET),
|
|
]
|
|
|
|
# Pin the sockets to the TCP transport so it doesn't fail over to a
|
|
# different transport during this test
|
|
for s in sockets:
|
|
s.setsockopt(SOL_RDS, SO_RDS_TRANSPORT, RDS_TRANS_TCP)
|
|
elif flags & OP_FLAG_RDMA:
|
|
sockets = [
|
|
socket.socket(socket.AF_RDS, socket.SOCK_SEQPACKET),
|
|
socket.socket(socket.AF_RDS, socket.SOCK_SEQPACKET),
|
|
]
|
|
|
|
# Pin the sockets to the RDMA transport so it doesn't fail over to a
|
|
# different transport during this test
|
|
for s in sockets:
|
|
s.setsockopt(SOL_RDS, SO_RDS_TRANSPORT, RDS_TRANS_IB)
|
|
else:
|
|
raise RuntimeError(f"Invalid transport flag sets no transports: {flags}")
|
|
|
|
for s, addr in zip(sockets, addrs):
|
|
s.bind(addr)
|
|
s.setblocking(0)
|
|
|
|
send_hashes = {}
|
|
recv_hashes = {}
|
|
|
|
ep = select.epoll()
|
|
|
|
for s in sockets:
|
|
ep.register(s, select.EPOLLRDNORM)
|
|
|
|
num_packets = 50000
|
|
nr_send = 0
|
|
nr_recv = 0
|
|
|
|
while nr_send < num_packets:
|
|
|
|
# Send as much as we can without blocking
|
|
ksft_pr("sending...", nr_send, nr_recv)
|
|
nr_send = send_burst(sockets, addrs, send_hashes, nr_send, num_packets)
|
|
|
|
# Receive as much as we can without blocking
|
|
ksft_pr("receiving...", nr_send, nr_recv)
|
|
while nr_recv < nr_send:
|
|
nr_recv = recv_burst(ep, sockets, addrs, recv_hashes, nr_recv)
|
|
|
|
# exercise net/rds/tcp.c:rds_tcp_sysctl_reset()
|
|
if netns_list:
|
|
for net in netns_list:
|
|
ip(f"netns exec {net} /usr/sbin/sysctl net.rds.tcp.rds_tcp_rcvbuf=10000")
|
|
ip(f"netns exec {net} /usr/sbin/sysctl net.rds.tcp.rds_tcp_sndbuf=10000")
|
|
|
|
ksft_pr("done", nr_send, nr_recv)
|
|
|
|
check_info(sockets)
|
|
|
|
# We're done sending and receiving stuff, now let's check if what
|
|
# we received is what we sent.
|
|
rc = verify_hashes(send_hashes, recv_hashes)
|
|
|
|
ep.close()
|
|
for s in sockets:
|
|
s.close()
|
|
|
|
return rc
|
|
|
|
def stop_pcaps():
|
|
"""Stop tcpdump processes.
|
|
|
|
We use pop() here to drain the list in the event that the test
|
|
completes after the signal handler is fired. List will be empty
|
|
if logdir is not set
|
|
"""
|
|
|
|
if not tcpdump_procs:
|
|
return
|
|
|
|
ksft_pr("Stopping network packet captures")
|
|
while tcpdump_procs:
|
|
proc = tcpdump_procs.pop()
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
proc.wait()
|
|
|
|
def signal_handler(_sig, _frame):
|
|
"""
|
|
Test timed out signal handler
|
|
"""
|
|
ksft_pr(f"Test timed out: {signal_handler_label}")
|
|
print(f"not ok {tap_idx} rds selftest {signal_handler_label}")
|
|
sys.exit(1)
|
|
|
|
def setup_tcp():
|
|
"""
|
|
Configure tcp network
|
|
"""
|
|
|
|
# clean up any leftovers from a previously interrupted run
|
|
teardown_tcp()
|
|
|
|
ip(f"netns add {NET0}")
|
|
ip(f"netns add {NET1}")
|
|
ip("link add type veth")
|
|
|
|
# Move TCP interfaces into separate namespaces so they can no longer be
|
|
# bound directly; this prevents rds from switching over from the tcp
|
|
# transport to the loop transport.
|
|
ip(f"link set {VETH0} netns {NET0} up")
|
|
ip(f"link set {VETH1} netns {NET1} up")
|
|
|
|
# add addresses
|
|
ip(f"-n {NET0} addr add {tcp_addrs[0][0]}/32 dev {VETH0}")
|
|
ip(f"-n {NET1} addr add {tcp_addrs[1][0]}/32 dev {VETH1}")
|
|
|
|
# add routes
|
|
ip(f"-n {NET0} route add {tcp_addrs[1][0]}/32 dev {VETH0}")
|
|
ip(f"-n {NET1} route add {tcp_addrs[0][0]}/32 dev {VETH1}")
|
|
|
|
# sanity check that our two interfaces/addresses are correctly set up
|
|
# and communicating by doing a single ping
|
|
ip(f"netns exec {NET0} ping -c 1 {tcp_addrs[1][0]}")
|
|
|
|
# Start a packet capture on each network
|
|
if logdir is not None:
|
|
for netn in [NET0, NET1]:
|
|
pcap = logdir+'/rds-'+netn+'.pcap'
|
|
|
|
tcpdump_cmd = ['ip', 'netns', 'exec', netn, '/usr/sbin/tcpdump']
|
|
sudo_user = os.environ.get('SUDO_USER')
|
|
if sudo_user:
|
|
tcpdump_cmd.extend(['-Z', sudo_user])
|
|
tcpdump_cmd.extend(['-i', 'any', '-w', pcap])
|
|
|
|
# pylint: disable-next=consider-using-with
|
|
p = subprocess.Popen(tcpdump_cmd,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
tcpdump_procs.append(p)
|
|
|
|
# simulate packet loss, duplication and corruption
|
|
for netn, iface in [(NET0, VETH0), (NET1, VETH1)]:
|
|
ip(f"netns exec {netn} /usr/sbin/tc qdisc add dev {iface} root netem \
|
|
corrupt {PACKET_CORRUPTION} loss {PACKET_LOSS} duplicate \
|
|
{PACKET_DUPLICATE}")
|
|
|
|
def teardown_tcp():
|
|
"""
|
|
Tear down the tcp network configured by setup_tcp().
|
|
|
|
Removing the namespaces also removes the veth pair, addresses,
|
|
routes, and netem qdisc that live inside them. fail=False so
|
|
this is safe to call in error paths after a partial or complete setup.
|
|
"""
|
|
cmd(f"ip netns del {NET0}", fail=False)
|
|
cmd(f"ip netns del {NET1}", fail=False)
|
|
|
|
def get_iface_mac(iface):
|
|
"""Return the MAC address of a local network interface."""
|
|
out = subprocess.check_output(['ip', 'link', 'show', iface], text=True)
|
|
mac = re.search(r'link/ether\s+([0-9a-f:]+)', out)
|
|
if not mac:
|
|
raise RuntimeError(f"Cannot determine MAC address of {iface}")
|
|
return mac.group(1)
|
|
|
|
def setup_rdma():
|
|
"""
|
|
Configure rdma network
|
|
"""
|
|
|
|
# remove links left over by previously interrupted run.
|
|
teardown_rdma()
|
|
|
|
# use call here since modprobe may fail if the rdma_rxe
|
|
# module is built-in
|
|
subprocess.call(['modprobe', 'rdma_rxe'],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
ip(f"link add {VETH_RDMA0} type veth peer name {VETH_RDMA1}")
|
|
|
|
ip(f"link set {VETH_RDMA0} up")
|
|
ip(f"link set {VETH_RDMA1} up")
|
|
|
|
# Since both addresses are in the same namespace, the source address
|
|
# is always local, so enable accept_local
|
|
cmd(f"/usr/sbin/sysctl -q net.ipv4.conf.{VETH_RDMA0}.accept_local=1")
|
|
cmd(f"/usr/sbin/sysctl -q net.ipv4.conf.{VETH_RDMA1}.accept_local=1")
|
|
|
|
# Reverse path filters must be disabled so that the local routes don't
|
|
# cause RPF failures.
|
|
cmd(f"/usr/sbin/sysctl -q net.ipv4.conf.{VETH_RDMA0}.rp_filter=0")
|
|
cmd(f"/usr/sbin/sysctl -q net.ipv4.conf.{VETH_RDMA1}.rp_filter=0")
|
|
|
|
# add addresses
|
|
ip(f"addr add {rdma_addrs[0][0]}/32 dev {VETH_RDMA0}")
|
|
ip(f"addr add {rdma_addrs[1][0]}/32 dev {VETH_RDMA1}")
|
|
|
|
# add routes
|
|
ip(f"route add {rdma_addrs[1][0]}/32 dev {VETH_RDMA0}")
|
|
ip(f"route add {rdma_addrs[0][0]}/32 dev {VETH_RDMA1}")
|
|
|
|
# ARP will not resolve neighbor IPs on /32 routes without a subnet.
|
|
# Avoid this by adding neighbors directly so RDMA CM can populate path
|
|
# records with correct mac addrs without waiting for the ARP.
|
|
mac0 = get_iface_mac(VETH_RDMA0)
|
|
mac1 = get_iface_mac(VETH_RDMA1)
|
|
ip(f"neigh add {rdma_addrs[1][0]} lladdr {mac1} dev {VETH_RDMA0} nud permanent")
|
|
ip(f"neigh add {rdma_addrs[0][0]} lladdr {mac0} dev {VETH_RDMA1} nud permanent")
|
|
|
|
cmd(f'rdma link add {RXE_DEV0} type rxe netdev {VETH_RDMA0}')
|
|
cmd(f'rdma link add {RXE_DEV1} type rxe netdev {VETH_RDMA1}')
|
|
|
|
time.sleep(1) # allow RXE devices to initialise
|
|
|
|
# Start a packet capture on each network
|
|
if logdir is not None:
|
|
for iface in [VETH_RDMA0, VETH_RDMA1]:
|
|
pcap = logdir+'/rds-roce-'+iface+'.pcap'
|
|
|
|
tcpdump_cmd = ['/usr/sbin/tcpdump']
|
|
sudo_user = os.environ.get('SUDO_USER')
|
|
if sudo_user:
|
|
tcpdump_cmd.extend(['-Z', sudo_user])
|
|
tcpdump_cmd.extend(['-i', iface, '-w', pcap])
|
|
|
|
# pylint: disable-next=consider-using-with
|
|
p = subprocess.Popen(tcpdump_cmd,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
tcpdump_procs.append(p)
|
|
|
|
# simulate packet loss, duplication and corruption
|
|
for iface in [VETH_RDMA0, VETH_RDMA1]:
|
|
cmd(f"/usr/sbin/tc qdisc add dev {iface} root netem \
|
|
corrupt {PACKET_CORRUPTION} loss {PACKET_LOSS} duplicate \
|
|
{PACKET_DUPLICATE}")
|
|
|
|
def teardown_rdma():
|
|
"""
|
|
Tear down the rdma network configured by setup_rdma().
|
|
"""
|
|
|
|
# remove links left over by previously interrupted run.
|
|
cmd(f'rdma link del {RXE_DEV0}', fail=False)
|
|
cmd(f'rdma link del {RXE_DEV1}', fail=False)
|
|
cmd(f'ip link del {VETH_RDMA0}', fail=False)
|
|
|
|
|
|
#Parse out command line arguments. We take an optional
|
|
# timeout parameter and an optional log output folder
|
|
parser = argparse.ArgumentParser(description="init script args",
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|
parser.add_argument("-d", "--logdir", action="store",
|
|
help="directory to store logs", default=None)
|
|
parser.add_argument("-T", "--transport", default="tcp",
|
|
help="Comma-separated list of transports to test: "
|
|
"tcp, rdma, or tcp,rdma. Each matching test "
|
|
"is run once per transport. "
|
|
"'rdma' requires CONFIG_RDS_RDMA and rdma_rxe.")
|
|
parser.add_argument('-t', '--timeout', help="timeout to terminate hung test",
|
|
type=int, default=0)
|
|
parser.add_argument('-l', '--loss', help="Simulate tcp packet loss",
|
|
type=int, default=0)
|
|
parser.add_argument('-c', '--corruption', help="Simulate tcp packet corruption",
|
|
type=int, default=0)
|
|
parser.add_argument('-u', '--duplicate', help="Simulate tcp packet duplication",
|
|
type=int, default=0)
|
|
args = parser.parse_args()
|
|
logdir=args.logdir
|
|
PACKET_LOSS=str(args.loss)+'%'
|
|
PACKET_CORRUPTION=str(args.corruption)+'%'
|
|
PACKET_DUPLICATE=str(args.duplicate)+'%'
|
|
|
|
# check transport is either tcp or rdma
|
|
transports = [t.strip() for t in args.transport.split(',')]
|
|
for t in transports:
|
|
if t not in ('tcp', 'rdma'):
|
|
raise SystemExit(f"test.py: unknown transport: {t!r}")
|
|
|
|
# Register stop_pcaps before any network setups so that any partially setup
|
|
# tcpdumps are still cleaned up on error
|
|
atexit.register(stop_pcaps)
|
|
|
|
# Set up all requested transports upfront so network plumbing is
|
|
# ready before any test runs.
|
|
transport_envs = {}
|
|
FLAGS = 0
|
|
if 'tcp' in transports:
|
|
# Register cleanups before setups to handle partial setups that error'd out
|
|
atexit.register(teardown_tcp)
|
|
setup_tcp()
|
|
transport_envs['tcp'] = {
|
|
'addrs': tcp_addrs,
|
|
'netns': [NET0, NET1],
|
|
'flags': FLAGS | OP_FLAG_TCP,
|
|
}
|
|
|
|
if 'rdma' in transports:
|
|
atexit.register(teardown_rdma)
|
|
setup_rdma()
|
|
transport_envs['rdma'] = {
|
|
'addrs': rdma_addrs,
|
|
'netns': None,
|
|
'flags': FLAGS | OP_FLAG_RDMA,
|
|
}
|
|
|
|
print("TAP version 13")
|
|
print(f"1..{len(transport_envs)}")
|
|
|
|
for transport, tenv in transport_envs.items():
|
|
tap_idx += 1
|
|
|
|
# add a timeout
|
|
if args.timeout > 0:
|
|
signal_handler_label = transport
|
|
signal.alarm(args.timeout)
|
|
signal.signal(signal.SIGALRM, signal_handler)
|
|
|
|
ret = snd_rcv_packets(tenv)
|
|
|
|
# cancel timeout
|
|
signal.alarm(0)
|
|
|
|
if ret == 0:
|
|
ksft_pr("Success")
|
|
print(f"ok {tap_idx} rds selftest {transport}")
|
|
nr_pass += 1
|
|
else:
|
|
print(f"not ok {tap_idx} rds selftest {transport}")
|
|
nr_fail += 1
|
|
|
|
ksft_pr(f"Totals: pass:{nr_pass} fail:{nr_fail} skip:0")
|
|
sys.exit(1 if nr_fail else 0)
|