diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..feaae4be --- /dev/null +++ b/.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 \ No newline at end of file diff --git a/.gitignore b/.gitignore index 2c6e6792..4adf9f92 100644 --- a/.gitignore +++ b/.gitignore @@ -36,4 +36,6 @@ log log.csv -vendor \ No newline at end of file +vendor + +workdir \ No newline at end of file diff --git a/polyamide/device/peer.go b/polyamide/device/peer.go index 2b88064b..1c090caa 100644 --- a/polyamide/device/peer.go +++ b/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 } diff --git a/polyamide/device/send.go b/polyamide/device/send.go index fa699c5d..1a50ee06 100644 --- a/polyamide/device/send.go +++ b/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() } diff --git a/polyamide/device/traffic_control.go b/polyamide/device/traffic_control.go index ca97ffb9..04147042 100644 --- a/polyamide/device/traffic_control.go +++ b/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)