🧵 service: join multiple wg messages into coalesced segments

As of Linux 6.11, the in-kernel wg implementation does not send messages with GSO. A proxy client or server receiving from lo on the same system won't get any coalesced messages.

This commit updates the mmsg relay routines to attempt to join as many received uncoalesced messages as possible after encrypting, so sending them can take advantage of GSO.

service/client.go

@@ -544,16 +544,14 @@ func (c *client) relayWgToProxyGeneric(uplink clientNatUplinkGeneric) {
 	)
 
 	for rqp := range uplink.proxyConnSendCh {
-		wgPacketBuf := rqp.buf
-
 		var (
 			isHandshake     bool
 			sqpLength       uint32
 			sqpSegmentSize  uint32
 			sqpSegmentCount uint32
 		)
 
-		for len(wgPacketBuf) > 0 {
+		for wgPacketBuf := rqp.buf; len(wgPacketBuf) > 0; {
 			wgPacketLength := min(len(wgPacketBuf), int(rqp.segmentSize))
 			wgPacket := wgPacketBuf[:wgPacketLength]
 			wgPacketBuf = wgPacketBuf[wgPacketLength:]

service/client_mmsg.go

@@ -422,25 +422,22 @@ func (c *client) relayWgToProxySendmmsg(uplink clientNatUplinkMmsg) {
 
 main:
 	for {
-		var isHandshake bool
-
 		// Block on first dequeue op.
 		rqp, ok := <-uplink.proxyConnSendCh
 		if !ok {
 			break
 		}
 
+		var (
+			isHandshake     bool
+			sqpLength       uint32
+			sqpSegmentSize  uint32
+			sqpSegmentCount uint32
+		)
+
 	dequeue:
 		for {
-			wgPacketBuf := rqp.buf
-
-			var (
-				sqpLength       uint32
-				sqpSegmentSize  uint32
-				sqpSegmentCount uint32
-			)
-
-			for len(wgPacketBuf) > 0 {
+			for wgPacketBuf := rqp.buf; len(wgPacketBuf) > 0; {
 				wgPacketLength := min(len(wgPacketBuf), int(rqp.segmentSize))
 				wgPacket := wgPacketBuf[:wgPacketLength]
 				wgPacketBuf = wgPacketBuf[wgPacketLength:]
@@ -498,17 +495,9 @@ main:
 				packetBuf = dst
 			}
 
-			if sqpLength > 0 {
-				sendQueuedPackets = append(sendQueuedPackets, queuedPacket{
-					buf:          packetBuf[len(packetBuf)-int(sqpLength):],
-					segmentSize:  sqpSegmentSize,
-					segmentCount: sqpSegmentCount,
-				})
-			}
-
 			c.putPacketBuf(rqp.buf)
 
-			if len(sendQueuedPackets) == 0 {
+			if len(sendQueuedPackets) == 0 && sqpLength == 0 {
 				continue main
 			}
 
@@ -526,6 +515,14 @@ main:
 			}
 		}
 
+		if sqpLength > 0 {
+			sendQueuedPackets = append(sendQueuedPackets, queuedPacket{
+				buf:          packetBuf[len(packetBuf)-int(sqpLength):],
+				segmentSize:  sqpSegmentSize,
+				segmentCount: sqpSegmentCount,
+			})
+		}
+
 		for _, sqp := range sendQueuedPackets {
 			b := sqp.buf
 			segmentsRemaining := sqp.segmentCount

service/server_mmsg.go

@@ -693,6 +693,12 @@ func (s *server) relayWgToProxySendmmsg(downlink serverNatDownlinkMmsg) {
 		msgsReceived += uint64(nr)
 		burstRecvBatchSize = max(burstRecvBatchSize, nr)
 
+		var (
+			qpLength       uint32
+			qpSegmentSize  uint32
+			qpSegmentCount uint32
+		)
+
 		rmsgvecn := rmsgvec[:nr]
 
 		for i := range rmsgvecn {
@@ -760,12 +766,7 @@ func (s *server) relayWgToProxySendmmsg(downlink serverNatDownlinkMmsg) {
 				recvSegmentSize = len(wgPacketBuf)
 			}
 
-			var (
-				recvSegmentCount uint32
-				qpLength         uint32
-				qpSegmentSize    uint32
-				qpSegmentCount   uint32
-			)
+			var recvSegmentCount uint32
 
 			for len(wgPacketBuf) > 0 {
 				wgPacketLength := min(len(wgPacketBuf), recvSegmentSize)
@@ -824,14 +825,14 @@ func (s *server) relayWgToProxySendmmsg(downlink serverNatDownlinkMmsg) {
 
 			packetsReceived += uint64(recvSegmentCount)
 			burstRecvSegmentCount = max(burstRecvSegmentCount, recvSegmentCount)
+		}
 
-			if qpLength > 0 {
-				queuedPackets = append(queuedPackets, queuedPacket{
-					buf:          sendPacketBuf[len(sendPacketBuf)-int(qpLength):],
-					segmentSize:  qpSegmentSize,
-					segmentCount: qpSegmentCount,
-				})
-			}
+		if qpLength > 0 {
+			queuedPackets = append(queuedPackets, queuedPacket{
+				buf:          sendPacketBuf[len(sendPacketBuf)-int(qpLength):],
+				segmentSize:  qpSegmentSize,
+				segmentCount: qpSegmentCount,
+			})
 		}
 
 		if len(queuedPackets) == 0 {