brload: fix premature socket closure (scionproto#4591)

Brload borrows the file descriptor from a af_packet.TPacket structure in
order to use it with sendmmsg (not available via TPacket). In the
process it left the original TPacket, only used on the receiving side,
exposed to garbage collection, resulting in the premature closure of the
socket. Fixed it by keeping a reference to TPacket on the sending side.

In passing:
* increase the benchmark packet size to 1500 bytes, to show
bandwidth (~10Gb/s) with normal frames.
* display brload's output to help in diagnosing issues.

Fixes scionproto#4590

acceptance/router_benchmark/benchmarklib.py

@@ -185,6 +185,7 @@ def run_test_case(self, case: str, map_args: list[str]) -> (int, int):
         output = self.exec_br_load(case, map_args, 13)
         end = "0"
         for line in output.splitlines():
+            logger.info("BrLoad output: " + line)
             if line.startswith("metricsBegin"):
                 end = line.split()[3]  # "... metricsEnd: <end>"
 
@@ -322,7 +323,8 @@ def run_bm(self, test_cases: [str]) -> Results:
         self.exec_br_load(test_cases[0], map_args, 5)
 
         # Fetch the core count once. It doesn't change while the router is running.
-        # We can't get it until the router has done some work, but the warmup is enough.
+        # We cannot get this until the router has been up for a few seconds. If you shorten
+        # the warmup for some reason, make sure to add a delay.
         cores = self.core_count()
 
         # At long last, run the tests.

acceptance/router_benchmark/brload/main.go

@@ -224,6 +224,7 @@ func run(cmd *cobra.Command) int {
 				binary.BigEndian.PutUint16(allPkts[j][44:46], uint16(numPkt%int(numStreams)))
 				numPkt++
 			}
+
 			if _, err := sender.sendAll(); err != nil {
 				log.Error("writing input packet", "case", string(caseToRun), "error", err)
 				return 1
@@ -303,7 +304,7 @@ func openDevices(interfaceNames []string) (map[string]*afpacket.TPacket, error)
 	handles := make(map[string]*afpacket.TPacket)
 
 	for _, intf := range interfaceNames {
-		handle, err := afpacket.NewTPacket(afpacket.OptInterface(intf))
+		handle, err := afpacket.NewTPacket(afpacket.OptInterface(intf), afpacket.OptFrameSize(4096))
 		if err != nil {
 			return nil, serrors.WrapStr("creating TPacket", err)
 		}

acceptance/router_benchmark/brload/mmsg.go

@@ -25,12 +25,14 @@ import (
 type mmsgHdr struct {
 	hdr unix.Msghdr
 	len uint32
+	_   [4]byte
 }
 
 // mpktSender is a helper class to add the ability of using the sendmmsg system call
 // with afpacket sockets.
 type mpktSender struct {
 	fd     int
+	tp     *afpacket.TPacket
 	msgs   []mmsgHdr
 	iovecs []unix.Iovec
 }
@@ -42,6 +44,8 @@ func newMpktSender(tp *afpacket.TPacket) *mpktSender {
 	// than this) to the afpacket project and get it merged.
 	fdv := reflect.ValueOf(tp).Elem().FieldByName("fd")
 	sender.fd = int(fdv.Int())
+	// This is to make sure that tp cannot be finalized before we're done abusing its file desc.
+	sender.tp = tp
 	return sender
 }
 
@@ -66,7 +70,7 @@ func (sender *mpktSender) sendAll() (int, error) {
 	// use case.
 	n, _, err := unix.Syscall6(unix.SYS_SENDMMSG,
 		uintptr(sender.fd),
-		(uintptr)(unsafe.Pointer(&sender.msgs[0])),
+		uintptr(unsafe.Pointer(&sender.msgs[0])),
 		uintptr(len(sender.msgs)),
 		0, 0, 0)
 	if err == 0 {

acceptance/router_benchmark/test.py

@@ -34,7 +34,7 @@
 logger = logging.getLogger(__name__)
 
 # Default packet length for CI testing
-BM_PACKET_SIZE = 172
+BM_PACKET_SIZE = 1500
 
 # Router profiling ON or OFF?
 PROFILING = False