From a7c8be4d69b644dce77b7caa06d40c511a09bbb5 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Thu, 5 Jul 2018 23:07:01 -0500 Subject: [PATCH 1/6] base backpressure decisions on queue size in bytes, instead of packet counts --- src/yggdrasil/switch.go | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index 5b72620c..e7b25a29 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -595,7 +595,7 @@ type switch_packetInfo struct { // Used to keep track of buffered packets type switch_buffer struct { packets []switch_packetInfo // Currently buffered packets, which may be dropped if it grows too large - count uint64 // Total queue size, including dropped packets + size uint64 // Total queue size in bytes } func (b *switch_buffer) dropTimedOut() { @@ -603,8 +603,10 @@ func (b *switch_buffer) dropTimedOut() { const timeout = 25 * time.Millisecond now := time.Now() for len(b.packets) > 0 && now.Sub(b.packets[0].time) > timeout { - util_putBytes(b.packets[0].bytes) - b.packets = b.packets[1:] + var packet switch_packetInfo + packet, b.packets = b.packets[0], b.packets[1:] + b.size -= uint64(len(packet.bytes)) + util_putBytes(packet.bytes) } } @@ -629,9 +631,9 @@ func (t *switchTable) handleIdle(port switchPort, buffs map[string]switch_buffer buffs[streamID] = buf packet := buf.packets[0] coords := switch_getPacketCoords(packet.bytes) - if (bestSize == 0 || buf.count < bestSize) && t.portIsCloser(coords, port) { + if (bestSize == 0 || buf.size < bestSize) && t.portIsCloser(coords, port) { best = streamID - bestSize = buf.count + bestSize = buf.size } } if bestSize != 0 { @@ -639,7 +641,7 @@ func (t *switchTable) handleIdle(port switchPort, buffs map[string]switch_buffer var packet switch_packetInfo // TODO decide if this should be LIFO or FIFO packet, buf.packets = buf.packets[0], buf.packets[1:] - buf.count-- + buf.size -= uint64(len(packet.bytes)) if len(buf.packets) == 0 { delete(buffs, best) } else { @@ -658,16 +660,16 @@ func (t *switchTable) doWorker() { idle := make(map[switchPort]struct{}) // this is to deduplicate things for { select { - case packet := <-t.packetIn: + case bytes := <-t.packetIn: // Try to send it somewhere (or drop it if it's corrupt or at a dead end) - if !t.handleIn(packet, idle) { + if !t.handleIn(bytes, idle) { // There's nobody free to take it right now, so queue it for later - streamID := switch_getPacketStreamID(packet) + packet := switch_packetInfo{bytes, time.Now()} + streamID := switch_getPacketStreamID(packet.bytes) buf := buffs[streamID] buf.dropTimedOut() - pinfo := switch_packetInfo{packet, time.Now()} - buf.packets = append(buf.packets, pinfo) - buf.count++ + buf.packets = append(buf.packets, packet) + buf.size += uint64(len(packet.bytes)) buffs[streamID] = buf } case port := <-t.idleIn: From 7da4967f5e67b3110be27dc8f49f9c9c7e915a75 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Thu, 5 Jul 2018 23:39:41 -0500 Subject: [PATCH 2/6] Limit maximum queue size to 4 MB --- src/yggdrasil/switch.go | 71 ++++++++++++++++++++++++++--------------- 1 file changed, 45 insertions(+), 26 deletions(-) diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index e7b25a29..20b5783d 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -598,37 +598,53 @@ type switch_buffer struct { size uint64 // Total queue size in bytes } -func (b *switch_buffer) dropTimedOut() { - // TODO figure out what timeout makes sense - const timeout = 25 * time.Millisecond - now := time.Now() - for len(b.packets) > 0 && now.Sub(b.packets[0].time) > timeout { - var packet switch_packetInfo - packet, b.packets = b.packets[0], b.packets[1:] - b.size -= uint64(len(packet.bytes)) - util_putBytes(packet.bytes) +type switch_buffers struct { + bufs map[string]switch_buffer // Buffers indexed by StreamID + size uint64 // Total size of all buffers, in bytes +} + +func (b *switch_buffers) cleanup(t *switchTable) { + remove := func(streamID string) { + // Helper function to drop a queue + buf := b.bufs[streamID] + for _, packet := range buf.packets { + util_putBytes(packet.bytes) + } + b.size -= buf.size + delete(b.bufs, streamID) + } + for streamID, buf := range b.bufs { + // Remove queues for which we have no next hop + packet := buf.packets[0] + coords := switch_getPacketCoords(packet.bytes) + if t.selfIsClosest(coords) { + remove(streamID) + } + } + const maxSize = 4 * 1048576 // Maximum 4 MB + for b.size > maxSize { + // Drop a random queue + for streamID := range b.bufs { + remove(streamID) + break + } } } // Handles incoming idle notifications // Loops over packets and sends the newest one that's OK for this peer to send // Returns true if the peer is no longer idle, false if it should be added to the idle list -func (t *switchTable) handleIdle(port switchPort, buffs map[string]switch_buffer) bool { +func (t *switchTable) handleIdle(port switchPort, bufs *switch_buffers) bool { to := t.core.peers.getPorts()[port] if to == nil { return true } var best string var bestSize uint64 - for streamID, buf := range buffs { + bufs.cleanup(t) + for streamID, buf := range bufs.bufs { // Filter over the streams that this node is closer to // Keep the one with the smallest queue - buf.dropTimedOut() - if len(buf.packets) == 0 { - delete(buffs, streamID) - continue - } - buffs[streamID] = buf packet := buf.packets[0] coords := switch_getPacketCoords(packet.bytes) if (bestSize == 0 || buf.size < bestSize) && t.portIsCloser(coords, port) { @@ -637,15 +653,16 @@ func (t *switchTable) handleIdle(port switchPort, buffs map[string]switch_buffer } } if bestSize != 0 { - buf := buffs[best] + buf := bufs.bufs[best] var packet switch_packetInfo // TODO decide if this should be LIFO or FIFO packet, buf.packets = buf.packets[0], buf.packets[1:] buf.size -= uint64(len(packet.bytes)) + bufs.size -= uint64(len(packet.bytes)) if len(buf.packets) == 0 { - delete(buffs, best) + delete(bufs.bufs, best) } else { - buffs[best] = buf + bufs.bufs[best] = buf } to.sendPacket(packet.bytes) return true @@ -656,8 +673,9 @@ func (t *switchTable) handleIdle(port switchPort, buffs map[string]switch_buffer // The switch worker does routing lookups and sends packets to where they need to be func (t *switchTable) doWorker() { - buffs := make(map[string]switch_buffer) // Packets per PacketStreamID (string) - idle := make(map[switchPort]struct{}) // this is to deduplicate things + var bufs switch_buffers + bufs.bufs = make(map[string]switch_buffer) // Packets per PacketStreamID (string) + idle := make(map[switchPort]struct{}) // this is to deduplicate things for { select { case bytes := <-t.packetIn: @@ -666,15 +684,16 @@ func (t *switchTable) doWorker() { // There's nobody free to take it right now, so queue it for later packet := switch_packetInfo{bytes, time.Now()} streamID := switch_getPacketStreamID(packet.bytes) - buf := buffs[streamID] - buf.dropTimedOut() + buf := bufs.bufs[streamID] buf.packets = append(buf.packets, packet) buf.size += uint64(len(packet.bytes)) - buffs[streamID] = buf + bufs.size += uint64(len(packet.bytes)) + bufs.bufs[streamID] = buf + bufs.cleanup(t) } case port := <-t.idleIn: // Try to find something to send to this peer - if !t.handleIdle(port, buffs) { + if !t.handleIdle(port, &bufs) { // Didn't find anything ready to send yet, so stay idle idle[port] = struct{}{} } From 1a65c065d07f8034c9162740cc81ad9d4bbe9420 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Thu, 5 Jul 2018 23:56:37 -0500 Subject: [PATCH 3/6] prioritize sending from small queues that have been blocked for a long time --- src/yggdrasil/switch.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index 20b5783d..1a0e7818 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -640,19 +640,21 @@ func (t *switchTable) handleIdle(port switchPort, bufs *switch_buffers) bool { return true } var best string - var bestSize uint64 + var bestPriority float64 bufs.cleanup(t) + now := time.Now() for streamID, buf := range bufs.bufs { // Filter over the streams that this node is closer to // Keep the one with the smallest queue packet := buf.packets[0] coords := switch_getPacketCoords(packet.bytes) - if (bestSize == 0 || buf.size < bestSize) && t.portIsCloser(coords, port) { + priority := float64(now.Sub(packet.time)) / float64(buf.size) + if priority > bestPriority && t.portIsCloser(coords, port) { best = streamID - bestSize = buf.size + bestPriority = priority } } - if bestSize != 0 { + if bestPriority != 0 { buf := bufs.bufs[best] var packet switch_packetInfo // TODO decide if this should be LIFO or FIFO From e6a47f705da746372e53620132b37e0f43101525 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Fri, 6 Jul 2018 00:11:36 -0500 Subject: [PATCH 4/6] when dropping a queue, select one at random based on queue size in bytes --- src/yggdrasil/switch.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index 1a0e7818..f47526f2 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -12,6 +12,7 @@ package yggdrasil // A little annoying to do with constant changes from backpressure import ( + "math/rand" "sync" "sync/atomic" "time" @@ -624,7 +625,13 @@ func (b *switch_buffers) cleanup(t *switchTable) { const maxSize = 4 * 1048576 // Maximum 4 MB for b.size > maxSize { // Drop a random queue - for streamID := range b.bufs { + target := rand.Uint64() % b.size + var size uint64 // running total + for streamID, buf := range b.bufs { + size += buf.size + if size < target { + continue + } remove(streamID) break } From ad5dc9ea87afbde2aee36c9a7f377460105a8589 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Fri, 6 Jul 2018 00:55:00 -0500 Subject: [PATCH 5/6] Drop 1 packet instead of a whole queue when overflowing --- src/yggdrasil/switch.go | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index f47526f2..ea52fdc9 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -605,21 +605,16 @@ type switch_buffers struct { } func (b *switch_buffers) cleanup(t *switchTable) { - remove := func(streamID string) { - // Helper function to drop a queue - buf := b.bufs[streamID] - for _, packet := range buf.packets { - util_putBytes(packet.bytes) - } - b.size -= buf.size - delete(b.bufs, streamID) - } for streamID, buf := range b.bufs { // Remove queues for which we have no next hop packet := buf.packets[0] coords := switch_getPacketCoords(packet.bytes) if t.selfIsClosest(coords) { - remove(streamID) + for _, packet := range buf.packets { + util_putBytes(packet.bytes) + } + b.size -= buf.size + delete(b.bufs, streamID) } } const maxSize = 4 * 1048576 // Maximum 4 MB @@ -632,7 +627,13 @@ func (b *switch_buffers) cleanup(t *switchTable) { if size < target { continue } - remove(streamID) + var packet switch_packetInfo + packet, buf.packets = buf.packets[0], buf.packets[1:] + buf.size -= uint64(len(packet.bytes)) + b.size -= uint64(len(packet.bytes)) + if len(buf.packets) == 0 { + delete(b.bufs, streamID) + } break } } From ba4047b51a7db87a0e13df7221f39a6fc66e048b Mon Sep 17 00:00:00 2001 From: Arceliar Date: Fri, 6 Jul 2018 17:27:04 -0500 Subject: [PATCH 6/6] correctly update buffer sizs when buffers overflow, and returned freed packets to the byte store --- src/yggdrasil/switch.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index ea52fdc9..6fe50bf3 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -631,8 +631,12 @@ func (b *switch_buffers) cleanup(t *switchTable) { packet, buf.packets = buf.packets[0], buf.packets[1:] buf.size -= uint64(len(packet.bytes)) b.size -= uint64(len(packet.bytes)) + util_putBytes(packet.bytes) if len(buf.packets) == 0 { delete(b.bufs, streamID) + } else { + // Need to update the map, since buf was retrieved by value + b.bufs[streamID] = buf } break } @@ -672,6 +676,7 @@ func (t *switchTable) handleIdle(port switchPort, bufs *switch_buffers) bool { if len(buf.packets) == 0 { delete(bufs.bufs, best) } else { + // Need to update the map, since buf was retrieved by value bufs.bufs[best] = buf } to.sendPacket(packet.bytes)