From 9de6dc32df775a70ba6e727ca4a9591a65296d07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Mon, 6 Jul 2026 14:17:42 +0800 Subject: [PATCH] Add batched InputPackets --- device/send.go | 61 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/device/send.go b/device/send.go index 264e964..7a94068 100644 --- a/device/send.go +++ b/device/send.go @@ -329,6 +329,15 @@ func (device *Device) InputPacket(destination []byte, packetSlices [][]byte) { } elem := device.NewOutboundElement() packet := elem.buffer[MessageEncapsulatingTransportSize+MessageTransportHeaderSize:] + var totalLength int + for _, packetSlice := range packetSlices { + totalLength += len(packetSlice) + } + if totalLength > len(packet) { + device.PutMessageBuffer(elem.buffer) + device.PutOutboundElement(elem) + return + } var n int for _, packetSlice := range packetSlices { n += copy(packet[n:], packetSlice) @@ -346,6 +355,58 @@ func (device *Device) InputPacket(destination []byte, packetSlices [][]byte) { } } +type InputPacketRef struct { + Destination []byte + PacketSlices [][]byte +} + +func (device *Device) InputPackets(packets []*InputPacketRef) []*InputPacketRef { + var unmatched []*InputPacketRef + elemsByPeer := make(map[*Peer]*QueueOutboundElementsContainer, len(packets)) + for _, packetRef := range packets { + peer := device.allowedips.Lookup(packetRef.Destination) + if peer == nil { + unmatched = append(unmatched, packetRef) + continue + } + elem := device.NewOutboundElement() + packet := elem.buffer[MessageEncapsulatingTransportSize+MessageTransportHeaderSize:] + var totalLength int + for _, packetSlice := range packetRef.PacketSlices { + totalLength += len(packetSlice) + } + if totalLength > len(packet) { + device.PutMessageBuffer(elem.buffer) + device.PutOutboundElement(elem) + continue + } + var n int + for _, packetSlice := range packetRef.PacketSlices { + n += copy(packet[n:], packetSlice) + } + elem.packet = packet[:n] + elemsForPeer, ok := elemsByPeer[peer] + if !ok { + elemsForPeer = device.GetOutboundElementsContainer() + elemsByPeer[peer] = elemsForPeer + } + elemsForPeer.elems = append(elemsForPeer.elems, elem) + } + for peer, elemsForPeer := range elemsByPeer { + if peer.isRunning.Load() { + peer.StagePackets(elemsForPeer) + peer.SendStagedPackets() + } else { + for _, elem := range elemsForPeer.elems { + device.PutMessageBuffer(elem.buffer) + device.PutOutboundElement(elem) + } + device.PutOutboundElementsContainer(elemsForPeer) + } + } + return unmatched +} + func (peer *Peer) StagePackets(elems *QueueOutboundElementsContainer) { for { select {