perf: reduce heap allocations

.dockerignore

@@ -0,0 +1,42 @@
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work
+go.work.sum
+.idea
+
+# env file
+.env
+
+.DS_Store
+nylon
+
+central.key
+central.yaml
+node.yaml
+central.nybundle
+
+distribute.sh
+distribute-conf.sh
+log
+
+log.csv
+
+vendor
+
+workdir
+Dockerfile

.gitignore

@@ -36,4 +36,6 @@ log
 
 log.csv
 
-vendor
+vendor
+
+workdir

polyamide/device/peer.go

@@ -135,31 +135,41 @@ func (peer *Peer) SendBuffers(buffers [][]byte, eps []conn.Endpoint) error {
 		for _, ep := range endpoints {
 			ep.ClearSrc()
 		}
+		for _, ep := range eps {
+			if ep != nil {
+				ep.ClearSrc()
+			}
+		}
 		peer.endpoints.clearSrcOnTx = false
 	}
 	peer.endpoints.Unlock()
 
-	// group by endpoints
-	bufByEp := make(map[conn.Endpoint][][]byte)
-	for i, buf := range buffers {
-		bufByEp[eps[i]] = append(bufByEp[eps[i]], buf)
-	}
+	// optimization, if multiple contiguous buffers share the same endpoint, send them in a single batch
+	prevIdx := 0
+	prevEp := eps[0]
 
-	var totalLen uint64
 	var anyError error
-	for ep, bufs := range bufByEp {
-		if ep == nil {
-			ep = endpoints[0] // default endpoint
-		}
-		err := peer.device.net.bind.Send(bufs, ep)
-		if err == nil {
-			for _, b := range bufs {
-				totalLen += uint64(len(b))
+	for i := 0; i <= len(buffers); i++ {
+		if i == len(buffers) || eps[i] != prevEp {
+			// send batch from prevIdx to i-1
+			if prevEp == nil {
+				prevEp = endpoints[0] // default endpoint
+			}
+			err := peer.device.net.bind.Send(buffers[prevIdx:i], prevEp)
+			if err != nil {
+				anyError = err
+			}
+			prevIdx = i
+			if i < len(buffers) {
+				prevEp = eps[i]
 			}
-		} else {
-			anyError = err
 		}
 	}
+
+	var totalLen uint64
+	for _, b := range buffers {
+		totalLen += uint64(len(b))
+	}
 	peer.txBytes.Add(totalLen)
 	return anyError
 }

polyamide/device/send.go

@@ -438,6 +438,54 @@ func (device *Device) RoutineEncryption(id int) {
 	}
 }
 
+/*
+This function arranges buffers and endpoints by endpoint, into contiguous
+segments of endpoints. It ensures that packets within the same endpoint group
+are stable-ordered.
+*/
+func partitionBuffersByEndpoint(bufs [][]byte, eps []conn.Endpoint, outBufs *[][]byte, outEps *[]conn.Endpoint, peer *Peer) error {
+	// use a naive O(n^2) algorithm since n is expected to be small
+
+	// replace nil endpoints with the first known endpoint
+	peer.endpoints.Lock()
+	for i := 0; i < len(eps); i++ {
+		if eps[i] == nil {
+			if len(peer.endpoints.val) == 0 {
+				peer.endpoints.Unlock()
+				return errors.New("no known endpoints for peer")
+			}
+			eps[i] = peer.endpoints.val[0]
+		}
+	}
+	peer.endpoints.Unlock()
+
+	for i := 0; i < len(bufs); i++ {
+		// If endpoint is nil, it has already been processed
+		// as part of an earlier group.
+		if eps[i] == nil {
+			continue
+		}
+
+		// This is the first item of a new group.
+		currentEp := eps[i]
+		*outBufs = append(*outBufs, bufs[i])
+		*outEps = append(*outEps, currentEp)
+		eps[i] = nil // Mark as visited
+
+		// Scan the rest of the array for other items in this group
+		for j := i + 1; j < len(bufs); j++ {
+			// Check if it's the same endpoint
+			if eps[j] == currentEp {
+				// It's part of the same group. Add it.
+				*outBufs = append(*outBufs, bufs[j])
+				*outEps = append(*outEps, currentEp)
+				eps[j] = nil // Mark as visited
+			}
+		}
+	}
+	return nil
+}
+
 func (peer *Peer) RoutineSequentialSender(maxBatchSize int) {
 	device := peer.device
 	defer func() {
@@ -447,11 +495,15 @@ func (peer *Peer) RoutineSequentialSender(maxBatchSize int) {
 	device.Log.Verbosef("%v - Routine: sequential sender - started", peer)
 
 	bufs := make([][]byte, 0, maxBatchSize)
+	parBufs := make([][]byte, 0, maxBatchSize)
 	eps := make([]conn.Endpoint, 0)
+	parEps := make([]conn.Endpoint, 0)
 
 	for elemsContainer := range peer.queue.outbound.c {
 		bufs = bufs[:0]
 		eps = eps[:0]
+		parBufs = parBufs[:0]
+		parEps = parEps[:0]
 		if elemsContainer == nil {
 			return
 		}
@@ -483,7 +535,12 @@ func (peer *Peer) RoutineSequentialSender(maxBatchSize int) {
 		peer.timersAnyAuthenticatedPacketTraversal(false)
 		peer.timersAnyAuthenticatedPacketSent()
 
-		err := peer.SendBuffers(bufs, eps)
+		err := partitionBuffersByEndpoint(bufs, eps, &parBufs, &parEps, peer)
+		if err != nil {
+			device.Log.Errorf("%v - Failed to send data packets: %v", peer, err)
+			continue
+		}
+		err = peer.SendBuffers(parBufs, parEps)
 		if dataSent {
 			peer.timersDataSent()
 		}

polyamide/device/traffic_control.go

@@ -1,8 +1,9 @@
 package device
 
 import (
-	"github.com/encodeous/nylon/polyamide/conn"
 	"slices"
+
+	"github.com/encodeous/nylon/polyamide/conn"
 )
 
 // polyamide traffic control provides a facility to re-order, manipulate, and redirect packets between nylon/polyamide nodes
@@ -200,6 +201,7 @@ func (device *Device) TCBatch(batch []*TCElement, tcs *TCState) {
 				obe.packet = elem.Packet
 				obe.buffer = elem.Buffer
 				obec.elems = append(obec.elems, obe)
+				device.PutTCElement(elem)
 				elems[i] = nil
 			}
 			peer.StagePackets(obec)