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