blob: ebb47efe19a2090fc5d92a1d780b06b292a351ab [file] [log] [blame]
# vim: syntax=py
# -*- mode: python -*-
# SPDX-License-Identifier: Apache-2.0
# Copyright 2020-present Open Networking Foundation
# Copyright(c) 2019 Intel Corporation
# for errnos
import errno
from conf.utils import *
# ====================================================
# Parameters
# ====================================================
conf_file = get_env('CONF_FILE', 'conf/spgwu.json')
conf = get_json_conf(conf_file, False)
# Maximum number of flows to manage ip frags for re-assembly
max_ip_defrag_flows = None
try:
max_ip_defrag_flows = int(conf["max_ip_defrag_flows"])
except ValueError:
print('Invalid value for max_ip_defrag_flows. Not installing IPDefrag module.')
except KeyError:
print('max_ip_defrag_flows value not set. Not installing IPDefrag module.')
# ====================================================
# Port Helpers
# ====================================================
MAX_GATES = 8192
dpdk_ports = {}
def scan_dpdk_ports():
idx = 0
while True:
try:
intf = PMDPort(name="Port {}".format(idx), port_id=idx)
if intf:
# Need to declare mac so that we don't lose key during destroy_port
mac = intf.mac_addr
dpdk_ports[mac] = idx
bess.destroy_port(intf.name)
except bess.Error as e:
if e.code == errno.ENODEV:
break
else:
raise
idx += 1
return True if dpdk_ports else False
class Port:
def __init__(self, name):
self.name = name
self.wid = None
self.fpi = None
self.fpo = None
self.bpf = None
self.bpfgate = 0
self.routes_table = None
def bpf_gate(self):
if self.bpfgate < MAX_GATES - 2:
self.bpfgate += 1
return self.bpfgate
else:
raise Exception('Port {}: Out of BPF gates to allocate'.format(self.name))
def detect_mode(self):
# default case
mode = "unselected"
try:
peer_by_interface(self.name)
mode = "dpdk"
except:
mode = "linux"
return mode
def init_fastpath(self, **kwargs):
# Initialize PMDPort and RX/TX modules
name = self.name
fast = PMDPort(name="{}Fast".format(name), **kwargs)
self.fpi = __bess_module__("{}FastPI".format(name), 'PortInc', port=fast.name)
self.fpo = __bess_module__("{}FastPO".format(name), 'PortOut', port=fast.name)
# Initialize BPF to classify incoming traffic to go to kernel and/or pipeline
self.bpf = __bess_module__("{}FastBPF".format(name), 'BPF')
self.bpf.clear()
# Default drop when no matches
if max_ip_defrag_flows is not None:
ipdefrag = __bess_module__("{}Defrag".format(name), 'IPDefrag', num_flows=max_ip_defrag_flows, numa=0)
self.fpi -> ipdefrag:1 -> self.bpf:0 -> Sink()
ipdefrag:0 -> Sink()
else:
self.fpi -> self.bpf:0 -> Sink()
# Initialize route module
self.rtr = __bess_module__("{}Routes".format(name), 'IPLookup')
# Default route goes to Sink
self.rtr.add(prefix='0.0.0.0', prefix_len=0, gate=MAX_GATES-1)
self.rtr:(MAX_GATES-1) -> Sink()
# Attach fastpath to worker's root TC
self.fpi.attach_task(wid=self.wid)
def setup_port(self, idx, workers):
# Pick the worker handling this port
self.wid = idx % workers
name = self.name
wid = self.wid
print('Setting up port {} on worker {}'.format(name,wid))
# Detect the mode of this interface - DPDK/AF_XDP/AF_PACKET
mode = self.detect_mode()
if mode == 'linux':
try:
# Initialize kernel fastpath.
# AF_XDP requires that num_rx_qs == num_tx_qs
kwargs = {"vdev" : "net_af_xdp{},iface={},start_queue=0,queue_count={}"
.format(idx, name, workers), "num_out_q": workers, "num_inc_q": workers}
self.init_fastpath(**kwargs)
except:
print('Failed to create AF_XDP socket for {}. Creating AF_PACKET socket instead.'.format(name))
kwargs = {"vdev" : "net_af_packet{},iface={},qpairs={}".format(idx, name, workers), "num_out_q": workers}
self.init_fastpath(**kwargs)
elif mode == 'dpdk':
# if port list is empty, scan for dpdk_ports first
if not dpdk_ports and scan_dpdk_ports() == False:
print('Registered dpdk ports do not exist.')
sys.exit()
# Initialize DPDK fastpath
fidx = dpdk_ports.get(mac_by_interface(name))
if fidx is None:
raise Exception('Registered port for {} not detected!'.format(name))
kwargs = {"port_id" : fidx, "num_out_q": workers}
self.init_fastpath( **kwargs)
# Initialize kernel slowpath port and RX/TX modules
try:
peer = peer_by_interface(name)
vdev = "net_af_packet{},iface={}".format(idx, peer)
slow = PMDPort(name="{}Slow".format(name), vdev=vdev)
spi = __bess_module__("{}SlowPI".format(name), 'PortInc', port=slow.name)
spo = __bess_module__("{}SlowPO".format(name), 'PortOut', port=slow.name)
qspo = __bess_module__("{}QSlowPO".format(name), 'Queue')
# host_ip_filter: tcpdump -i foo 'dst host 198.19.0.1 or 198.18.0.1' -d
# Should always be set to lowest priority
HostGate = MAX_GATES - 1
ips = ips_by_interface(name)
host_ip_filter = {"priority": -HostGate, "filter": "dst host "
+ " or ".join(str(x) for x in ips), "gate": HostGate}
self.bpf.add(filters=[host_ip_filter])
# Direct control traffic from DPDK to kernel
self.bpf:HostGate -> qspo -> spo
# Direct control traffic from kernel to DPDK
spi -> self.fpo
tc = 'slow{}'.format(wid)
try:
bess.add_tc(tc, policy='round_robin', wid=wid)
except Exception as e:
if e.errmsg == "Name '{}' already exists".format(tc):
pass
else:
raise e
# Limit scheduling slow path RX/TX to 1000 times/second each
for mod in spi, qspo:
bess.add_tc(mod.name,
parent=tc,
policy='rate_limit',
resource='count',
limit={'count': 1000})
mod.attach_task(mod.name)
except Exception as e:
print('Mirror veth interface: {} misconfigured: {}'.format(name, e))
else:
raise Exception('Invalid mode selected.')
# ====================================================
# Validate
# ====================================================
# CIDRs
enb_cidr = conf["enb_cidr"]
ue_cidr = conf["ue_cidr"]
cidrs = [enb_cidr, ue_cidr]
for cidr in cidrs:
if validate_cidr(cidr) is False:
exit(1, 'Invalid {}'.format(cidr))
# ====================================================
# Core Setup
# ====================================================
# Initialize workers
workers = 1
try:
workers = int(conf["workers"])
except ValueError:
print('Invalid workers value! Re-setting # of workers to 1.')
cores = get_process_affinity()
for wid in xrange(workers):
bess.add_worker(wid=wid, core=int(cores[wid % len(cores)]))
# ====================================================
# Port Setup
# ====================================================
interfaces = ["s1u", "sgi"]
ports = {}
for idx, interface in enumerate(interfaces):
port = Port(conf[interface]["ifname"])
if port.name in ports:
continue
port.setup_port(idx, workers)
ports[port.name] = port
s1u_ifname = conf["s1u"]["ifname"]
sgi_ifname = conf["sgi"]["ifname"]
# ====================================================
# Downlink Pipeline
# ====================================================
s1uRoutes = ports[s1u_ifname].rtr
# Maximum number of sessions to manage
try:
max_sessions = int(conf["max_sessions"])
except ValueError:
print('Invalid max_sessions value!')
# Setting filter to detect UE subnet
sgiFastBPF = ports[sgi_ifname].bpf
UEGate = ports[sgi_ifname].bpf_gate()
ue_filter = {"priority": -UEGate,
"filter": "ip dst net {}".format(ue_cidr), "gate": UEGate}
sgiFastBPF.add(filters=[ue_filter])
sgiFastBPF:UEGate \
-> EtherTrim::GenericDecap(bytes=14) \
-> GTPUEncap::GtpuEncap(s1u_sgw_ip=ip2long(ips_by_interface(s1u_ifname)[0]), num_subscribers=max_sessions):1 \
-> S1UEtherAdd::GenericEncap(fields=[
{'size': 6, 'value': {'value_int': 0x0}},
{'size': 6, 'value': {'value_int': mac2hex(mac_by_interface(s1u_ifname))}},
{'size': 2, 'value': {'value_int': 0x0800}}]) \
-> OuterUDPCsum::L4Checksum() \
-> OuterIPCsum::IPChecksum() \
-> s1uRoutes
# Drop unknown packets
GTPUEncap:0 -> Sink()
# ====================================================
# Uplink Pipeline
# ====================================================
# Setting filter to detect gtpu traffic
# src net 11.1.1.0 mask 255.255.255.0 # check eNB subnet
# and dst host 11.1.1.1 # check S/PGWU IP
# and udp dst port 2152 # check GTPU port
# and (udp[28:4] & 0xffffff00) = 0x10000000 # check UE subnet
s1uFastBPF = ports[s1u_ifname].bpf
check_enb_subnet = "src net {} ".format(enb_cidr)
check_spgwu_ip = " and dst host " + \
" or ".join(str(x) for x in ips_by_interface(s1u_ifname))
check_gtpu_port = " and udp dst port 2152"
check_ue_subnet = " and (udp[28:4] & 0x{}) = 0x{}".format(
cidr2mask(ue_cidr), ip2hex(cidr2block(ue_cidr)[0]))
check_gtpu_msg_echo = " and udp[9] = 0x1"
GTPUEchoGate = ports[s1u_ifname].bpf_gate()
uplink_echo_filter = {"priority": -GTPUEchoGate, "filter": check_enb_subnet +
check_spgwu_ip + check_gtpu_port +
check_gtpu_msg_echo, "gate": GTPUEchoGate}
s1uFastBPF.add(filters=[uplink_echo_filter])
GTPUGate = ports[s1u_ifname].bpf_gate()
uplink_filter = {"priority": -GTPUGate, "filter": check_enb_subnet +
check_spgwu_ip + check_gtpu_port, "gate": GTPUGate}
s1uFastBPF.add(filters=[uplink_filter])
sgiRoutes = ports[sgi_ifname].rtr
s1uFastBPF:GTPUGate -> EtherDecapTrim::GenericDecap(bytes=14) -> GTPUDecap::GtpuDecap(ename="GTPUEncap"):1 \
-> SGIEtherAdd::GenericEncap(fields=[
{'size': 6, 'value': {'value_int': 0x0}},
{'size': 6, 'value': {'value_int': mac2hex(mac_by_interface(sgi_ifname))}},
{'size': 2, 'value': {'value_int': 0x0800}}]) \
-> sgiRoutes
s1uFastBPF:GTPUEchoGate \
-> GTPUEcho::GtpuEcho(s1u_sgw_ip=ip2long(ips_by_interface(s1u_ifname)[0])):1 \
-> EthSwap::MACSwap() \
-> 1:OuterUDPCsum
# Drop unknown packets
GTPUEcho:0 -> Sink()
GTPUDecap:0 -> Sink()
# ====================================================
# SIM_TEST
# ====================================================
import time
def sim_start_test():
start_teid = 0xf0000000
start_ue_ip = 0x10000001
start_enb_ip = 0x0b010181
NG4T_MAX_UE_RAN = 500000
NG4T_MAX_ENB_RAN = 80
for i in range(max_sessions):
# NG4T-based formula to calculate enodeB IP address against a given UE IP address
# il_trafficgen also uses the same scheme
# See SimuCPEnbv4Teid(...) in ngic code for more details
ue_of_ran = i % NG4T_MAX_UE_RAN
ran = i / NG4T_MAX_UE_RAN
enb_of_ran = ue_of_ran % NG4T_MAX_ENB_RAN
enb_idx = ran * NG4T_MAX_ENB_RAN + enb_of_ran
GTPUEncap.add(teid=start_teid+i, eteid=i+1, ueaddr=start_ue_ip+i, enodeb_ip=start_enb_ip+enb_idx)
def sim_end_test():
start_ue_ip = 0x10000001
for i in range(max_sessions):
GTPUEncap.remove(ueaddr=start_ue_ip+i)
GTPUEncap.show_records()
### Uncomment the following lines to test with il_trafficgen ###
# Start the test
#sim_start_test()
# Wait for 30 seconds before deleting the session info entries (optional)
#time.sleep(30)
#sim_end_test()
# Finally send SIGHUP to route_control daemon on reload
# TODO: behavior is unspecified if route_control.py pid is not found
route_control_pid = getpythonpid('route_control.py')
if route_control_pid:
os.kill(route_control_pid, signal.SIGHUP)