| #! /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) |