Merge pull request #643 from intel-go/gregory/af_xdp

Added AF_XDP input/output. Added ctrl-C handling

Author: gshimansky

Dockerfile

@@ -20,7 +20,11 @@ RUN apt-get -q update && apt-get -q -y install \
     libmnl-dev \
     libibverbs-dev
 
-RUN cd /opt && curl -L -s https://dl.google.com/go/go1.12.5.linux-amd64.tar.gz | tar zx
+RUN cd /opt && curl -L -s https://dl.google.com/go/go1.12.9.linux-amd64.tar.gz | tar zx
+RUN git clone -b v0.0.4 https://github.com/libbpf/libbpf
+RUN make -C libbpf/src all install
+RUN echo "/usr/lib64" > /etc/ld.so.conf.d/usrlib64.conf
+RUN ldconfig
 
 RUN mkdir -p ${NFF_GO}
 COPY . ${NFF_GO}

README.md

@@ -118,6 +118,27 @@ Use Go version 1.11.4 or higher. To check the version of Go, do:
 
         go version
         
+### AF_XDP support
+
+AF_XDP support is enabled by default, and it requires you to install
+`libbpf` package. At the time of writing Ubuntu doesn't have this
+library among its packages, so it is necessary to build `libbpf` from
+sources or disable AF_XDP socket support.
+
+To disable it set variable `NFF_GO_NO_BPF_SUPPORT` to some unempty
+value. When NFF_GO is built with it, AF_XDP support is disaled and
+using it results in errors.
+
+If you want to build `libbpf` from sources you can do it in two
+different ways.
+* If you are using stock Linux kernel from distribution, [download
+  `libbpf` from GitHub](https://github.com/libbpf/libbpf), then
+  execute `cd src; make; sudo make install`. Add /usr/lib64 to your
+  ldconfig path.
+* If you build Linux kernel from sources, you can build `libbpf` from
+  Linux source tree using commands `cd tools/lib/bpf; make; sudo make
+  install install_headers`. Add /usr/local/lib64 to your ldconfig path.
+
 ## Building NFF-GO
 
 When Go compiler runs for the first time it downloads all dependent

common/Makefile

@@ -7,7 +7,7 @@ include $(PATH_TO_MK)/include.mk
 
 .PHONY: testing
 testing: check-pktgen
-	go test
+	go test -tags "${GO_BUILD_TAGS}"
 
 .PHONY: coverage
 coverage:

examples/OSforwarding.go

@@ -10,15 +10,23 @@ import (
 )
 
 func main() {
+	// If you use af-xdp mode you need to configure queues:
+	// e.g. ethtool -N my_device flow-type tcp4 dst-port 4242 action 16
+	afXDP := flag.Bool("af-xdp", false, "use af-xdp. need to use ethtool to setup queues")
 	inport := flag.String("in", "", "device for receiver")
+	inQueue := flag.Int("in-queue", 16, "queue for receiver")
 	outport := flag.String("out", "", "device for sender")
 	flag.Parse()
 
 	flow.CheckFatal(flow.SystemInit(nil))
-
-	inputFlow, err := flow.SetReceiverOS(*inport)
-	flow.CheckFatal(err)
-	flow.CheckFatal(flow.SetSenderOS(inputFlow, *outport))
-
+	if *afXDP {
+		inputFlow, err := flow.SetReceiverXDP(*inport, *inQueue)
+		flow.CheckFatal(err)
+		flow.CheckFatal(flow.SetSenderXDP(inputFlow, *outport))
+	} else {
+		inputFlow, err := flow.SetReceiverOS(*inport)
+		flow.CheckFatal(err)
+		flow.CheckFatal(flow.SetSenderOS(inputFlow, *outport))
+	}
 	flow.CheckFatal(flow.SystemStart())
 }

flow/flow.go

@@ -32,6 +32,7 @@ package flow
 import (
 	"net"
 	"os"
+	"os/signal"
 	"runtime"
 	"sync/atomic"
 	"time"
@@ -48,7 +49,7 @@ var createdPorts []port
 var portPair map[types.IPv4Address](*port)
 var schedState *scheduler
 var vEach [10][vBurstSize]uint8
-var devices map[string]int
+var ioDevices map[string]interface{}
 
 type Timer struct {
 	t        *time.Ticker
@@ -176,6 +177,19 @@ func addOSReceiver(socket int, out low.Rings) {
 	schedState.addFF("OS receiver", nil, recvOS, nil, par, nil, sendReceiveKNI, 0, &par.stats)
 }
 
+type receiveXDPParameters struct {
+	out    low.Rings
+	socket low.XDPSocket
+	stats  common.RXTXStats
+}
+
+func addXDPReceiver(socket low.XDPSocket, out low.Rings) {
+	par := new(receiveXDPParameters)
+	par.socket = socket
+	par.out = out
+	schedState.addFF("AF_XDP receiver", nil, recvXDP, nil, par, nil, sendReceiveKNI, 0, &par.stats)
+}
+
 type KNIParameters struct {
 	in        low.Rings
 	out       low.Rings
@@ -269,6 +283,19 @@ func addSenderOS(socket int, in low.Rings, inIndexNumber int32) {
 	schedState.addFF("sender OS", nil, sendOS, nil, par, nil, sendReceiveKNI, inIndexNumber, &par.stats)
 }
 
+type sendXDPParameters struct {
+	in     low.Rings
+	socket low.XDPSocket
+	stats  common.RXTXStats
+}
+
+func addSenderXDP(socket low.XDPSocket, in low.Rings, inIndexNumber int32) {
+	par := new(sendXDPParameters)
+	par.socket = socket
+	par.in = in
+	schedState.addFF("AF_XDP sender", nil, sendXDP, nil, par, nil, sendReceiveKNI, inIndexNumber, &par.stats)
+}
+
 type copyParameters struct {
 	in      low.Rings
 	out     low.Rings
@@ -439,7 +466,7 @@ const reportMbits = false
 
 var sizeMultiplier uint
 var schedTime uint
-var hwtxchecksum, hwrxpacketstimestamp bool
+var hwtxchecksum, hwrxpacketstimestamp, setSIGINTHandler bool
 var maxRecv int
 
 type port struct {
@@ -538,6 +565,11 @@ type Config struct {
 	// Enables hardware assisted timestamps in packet mbufs. These
 	// timestamps can be accessed with GetPacketTimestamp function.
 	HWRXPacketsTimestamp bool
+	// Disable setting custom handler for SIGINT in
+	// SystemStartScheduler. When handler is enabled
+	// SystemStartScheduler waits for SIGINT notification and calls
+	// SystemStop after it. It is enabled by default.
+	NoSetSIGINTHandler bool
 }
 
 // SystemInit is initialization of system. This function should be always called before graph construction.
@@ -665,7 +697,7 @@ func SystemInit(args *Config) error {
 		}
 	}
 	portPair = make(map[types.IPv4Address](*port))
-	devices = make(map[string]int)
+	ioDevices = make(map[string]interface{})
 	// Init scheduler
 	common.LogTitle(common.Initialization, "------------***------ Initializing scheduler -----***------------")
 	StopRing := low.CreateRings(burstSize*sizeMultiplier, maxInIndex /* Maximum possible rings */)
@@ -690,6 +722,8 @@ func SystemInit(args *Config) error {
 			vEach[i][j] = uint8(i)
 		}
 	}
+
+	setSIGINTHandler = !args.NoSetSIGINTHandler
 	return nil
 }
 
@@ -723,7 +757,19 @@ func SystemStartScheduler() error {
 		return common.WrapWithNFError(err, "scheduler start failed", common.Fail)
 	}
 	common.LogTitle(common.Initialization, "------------***---------- NFF-GO Started ---------***------------")
-	schedState.schedule(schedTime)
+
+	if setSIGINTHandler {
+		signalChan := make(chan os.Signal, 1)
+		signal.Notify(signalChan, os.Interrupt)
+		go func() {
+			schedState.schedule(schedTime)
+		}()
+		<-signalChan
+		common.LogTitle(common.Debug, "Received an interrupt, stopping everything")
+		SystemStop()
+	} else {
+		schedState.schedule(schedTime)
+	}
 	return nil
 }
 
@@ -819,13 +865,14 @@ func SetReceiver(portId uint16) (OUT *Flow, err error) {
 // Gets name of device, will return error if can't initialize socket.
 // Creates RAW socket, returns new opened flow with received packets.
 func SetReceiverOS(device string) (*Flow, error) {
-	socketID, ok := devices[device]
+	v, ok := ioDevices[device]
+	socketID := v.(int)
 	if !ok {
 		socketID = low.InitDevice(device)
 		if socketID == -1 {
 			return nil, common.WrapWithNFError(nil, "Can't initialize socket", common.BadSocket)
 		}
-		devices[device] = socketID
+		ioDevices[device] = socketID
 	}
 	rings := low.CreateRings(burstSize*sizeMultiplier, 1)
 	addOSReceiver(socketID, rings)
@@ -839,18 +886,57 @@ func SetSenderOS(IN *Flow, device string) error {
 	if err := checkFlow(IN); err != nil {
 		return err
 	}
-	socketID, ok := devices[device]
+	v, ok := ioDevices[device]
+	socketID := v.(int)
 	if !ok {
 		socketID = low.InitDevice(device)
 		if socketID == -1 {
 			return common.WrapWithNFError(nil, "Can't initialize socket", common.BadSocket)
 		}
-		devices[device] = socketID
+		ioDevices[device] = socketID
 	}
 	addSenderOS(socketID, finishFlow(IN), IN.inIndexNumber)
 	return nil
 }
 
+// SetReceiverXDP adds function receive from Linux AF_XDP to flow graph.
+// Gets name of device and queue number, will return error if can't initialize socket.
+// Creates AF_XDP socket, returns new opened flow with received packets.
+func SetReceiverXDP(device string, queue int) (*Flow, error) {
+	_, ok := ioDevices[device]
+	if ok {
+		return nil, common.WrapWithNFError(nil, "Device shouldn't have any sockets before AF_XDP. AF_XDP Send and Receive for one device in forbidden now", common.BadSocket)
+	}
+	socketID := low.InitXDP(device, queue)
+	if socketID == nil {
+		return nil, common.WrapWithNFError(nil, "Can't initialize AF_XDP socket", common.BadSocket)
+	}
+	ioDevices[device] = socketID
+	rings := low.CreateRings(burstSize*sizeMultiplier, 1)
+	addXDPReceiver(socketID, rings)
+	return newFlow(rings, 1), nil
+}
+
+// SetSenderXDP adds function send from flow graph to Linux AF_XDP interface.
+// Gets name of device, will return error if can't initialize socket.
+// Creates RAW socket, sends packets, closes input flow.
+func SetSenderXDP(IN *Flow, device string) error {
+	if err := checkFlow(IN); err != nil {
+		return err
+	}
+	_, ok := ioDevices[device]
+	if ok {
+		return common.WrapWithNFError(nil, "Device shouldn't have any sockets before AF_XDP. AF_XDP Send and Receive for one device in forbidden now", common.BadSocket)
+	}
+	socketID := low.InitXDP(device, 0)
+	if socketID == nil {
+		return common.WrapWithNFError(nil, "Can't initialize AF_XDP socket", common.BadSocket)
+	}
+	ioDevices[device] = socketID
+	addSenderXDP(socketID, finishFlow(IN), IN.inIndexNumber)
+	return nil
+}
+
 // SetReceiverKNI adds function receive from KNI to flow graph.
 // Gets KNI device from which packets will be received.
 // Receive queue will be added to port automatically.
@@ -1418,6 +1504,11 @@ func recvOS(parameters interface{}, inIndex []int32, flag *int32, coreID int) {
 	low.ReceiveOS(srp.socket, srp.out[0], flag, coreID, &srp.stats)
 }
 
+func recvXDP(parameters interface{}, inIndex []int32, flag *int32, coreID int) {
+	srp := parameters.(*receiveXDPParameters)
+	low.ReceiveXDP(srp.socket, srp.out[0], flag, coreID, &srp.stats)
+}
+
 func processKNI(parameters interface{}, inIndex []int32, flag *int32, coreID int) {
 	srk := parameters.(*KNIParameters)
 	if srk.linuxCore == true {
@@ -1607,6 +1698,11 @@ func sendOS(parameters interface{}, inIndex []int32, flag *int32, coreID int) {
 	low.SendOS(srp.socket, srp.in, flag, coreID, &srp.stats)
 }
 
+func sendXDP(parameters interface{}, inIndex []int32, flag *int32, coreID int) {
+	srp := parameters.(*sendXDPParameters)
+	low.SendXDP(srp.socket, srp.in, flag, coreID, &srp.stats)
+}
+
 func merge(from low.Rings, to low.Rings) {
 	// We should change out rings in all flow functions which we added before
 	// and change them to one "after merge" ring.
@@ -1620,6 +1716,10 @@ func merge(from low.Rings, to low.Rings) {
 			if parameters.out[0] == from[0] {
 				parameters.out = to
 			}
+		case *receiveOSParameters:
+			if parameters.out[0] == from[0] {
+				parameters.out = to
+			}
 		case *generateParameters:
 			if parameters.out[0] == from[0] {
 				parameters.out = to

internal/low/Makefile

@@ -7,4 +7,4 @@ include $(PATH_TO_MK)/include.mk
 
 .PHONY: testing
 testing: check-pktgen
-	go test -v
+	go test -v -tags "${GO_BUILD_TAGS}"

internal/low/low.go

@@ -760,6 +760,10 @@ func CheckHWRXPacketsTimestamp(port uint16) bool {
 	return bool(C.check_hwrxpackets_timestamp_capability(C.uint16_t(port)))
 }
 
+func InitDevice(device string) int {
+	return int(C.initDevice(C.CString(device)))
+}
+
 func ReceiveOS(socket int, OUT *Ring, flag *int32, coreID int, stats *common.RXTXStats) {
 	m := CreateMempool("receiveOS")
 	C.receiveOS(C.int(socket), OUT.DPDK_ring, (*C.struct_rte_mempool)(unsafe.Pointer(m)),
@@ -772,10 +776,6 @@ func SendOS(socket int, IN Rings, flag *int32, coreID int, stats *common.RXTXSta
 		(*C.RXTXStats)(unsafe.Pointer(stats)))
 }
 
-func InitDevice(device string) int {
-	return int(C.initDevice(C.CString(device)))
-}
-
 func SetCountersEnabledInApplication(enabled bool) {
 	C.counters_enabled_in_application = C.bool(true)
 }
@@ -791,3 +791,21 @@ func GetPacketOffloadFlags(mb *Mbuf) uint64 {
 func GetPacketTimestamp(mb *Mbuf) uint64 {
 	return uint64(mb.timestamp)
 }
+
+type XDPSocket *C.struct_xsk_socket_info
+
+func InitXDP(device string, queue int) XDPSocket {
+	return C.initXDP(C.CString(device), C.int(queue))
+}
+
+func ReceiveXDP(socket XDPSocket, OUT *Ring, flag *int32, coreID int, stats *common.RXTXStats) {
+	m := CreateMempool("receiveXDP")
+	C.receiveXDP(socket, OUT.DPDK_ring, (*C.struct_rte_mempool)(unsafe.Pointer(m)),
+		(*C.int)(unsafe.Pointer(flag)), C.int(coreID), (*C.RXTXStats)(unsafe.Pointer(stats)))
+}
+
+func SendXDP(socket XDPSocket, IN Rings, flag *int32, coreID int, stats *common.RXTXStats) {
+	C.sendXDP(socket, C.extractDPDKRings((**C.struct_nff_go_ring)(unsafe.Pointer(&(IN[0]))),
+		C.int32_t(len(IN))), C.int32_t(len(IN)), (*C.int)(unsafe.Pointer(flag)), C.int(coreID),
+		(*C.RXTXStats)(unsafe.Pointer(stats)))
+}