From 8502cb15cf8f5c63ddb7de550b7bc60ee455522c Mon Sep 17 00:00:00 2001 From: Kleissner Date: Fri, 5 Nov 2021 00:41:06 +0100 Subject: [PATCH] UDT: Fix the first piece of !@#$ heap code. Rewrite from scratch. --- udt/sendpacket_heap.go | 149 ++++++++++++++++------------------------- udt/udtsocket_send.go | 54 ++++++--------- 2 files changed, 78 insertions(+), 125 deletions(-) diff --git a/udt/sendpacket_heap.go b/udt/sendpacket_heap.go index b4930f0..165d5dc 100644 --- a/udt/sendpacket_heap.go +++ b/udt/sendpacket_heap.go @@ -1,7 +1,8 @@ package udt import ( - "container/heap" + "fmt" + "sync" "time" "github.com/PeernetOfficial/core/udt/packet" @@ -9,115 +10,83 @@ import ( type sendPacketEntry struct { pkt *packet.DataPacket + + // data specific to sending packets tim time.Time ttl time.Duration } -// receiveLossList defines a list of recvLossEntry records sorted by their packet ID -type sendPacketHeap []sendPacketEntry +// sendPacketHeap stores a list of packets. Packets are identified by their sequences. +// Access to the list via functions is thread safe. +// This isn't the fastest implementation on the planet since each operation iterates over the entire list. However, it works and is thread safe (the previous code was neither). +type sendPacketHeap struct { + // list contains all packets + list []sendPacketEntry -func (h sendPacketHeap) Len() int { - return len(h) + sync.RWMutex } -func (h sendPacketHeap) Less(i, j int) bool { - return h[i].pkt.Seq.Seq < h[j].pkt.Seq.Seq +func createSendPacketHeap() (heap *sendPacketHeap) { + return &sendPacketHeap{} } -func (h sendPacketHeap) Swap(i, j int) { - h[i], h[j] = h[j], h[i] +// Add adds a packet to the list. Deduplication is not performed. +func (heap *sendPacketHeap) Add(newPacket sendPacketEntry) { + heap.Lock() + defer heap.Unlock() + + fmt.Printf("sendPacketHeap add %d\n", newPacket.pkt.Seq.Seq) + + heap.list = append(heap.list, newPacket) } -func (h *sendPacketHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents. - *h = append(*h, x.(sendPacketEntry)) -} +// Remove removes all packets with the sequence from the list. +func (heap *sendPacketHeap) Remove(sequence uint32) { + heap.Lock() + defer heap.Unlock() -func (h *sendPacketHeap) Pop() interface{} { - old := *h - n := len(old) - x := old[n-1] - *h = old[0 : n-1] - return x -} + var newList []sendPacketEntry -// Find does a binary search of the heap for the specified packetID which is returned -func (h sendPacketHeap) Find(packetID packet.PacketID) (*sendPacketEntry, int) { - for n := 0; n < len(h); n++ { - if h[n].pkt.Seq == packetID { - return &h[n], n + for n := range heap.list { + if heap.list[n].pkt.Seq.Seq != sequence { + newList = append(newList, heap.list[n]) } } - // buggy crappy implementation - // len := len(h) - // idx := 0 - // for idx < len { - // pid := h[idx].pkt.Seq - // if pid == packetID { - // return &h[idx], idx - // } else if pid.Seq > packetID.Seq { - // idx = idx * 2 - // } else { - // idx = idx*2 + 1 - // } - // } - return nil, -1 + heap.list = newList } -// Min does a binary search of the heap for the entry with the lowest packetID greater than or equal to the specified value -func (h sendPacketHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (*packet.DataPacket, int) { - if len(h) == 0 { // none available! - return nil, -1 - } - return h[0].pkt, 0 - - len := len(h) - idx := 0 - wrapped := greaterEqual.Seq > lessEqual.Seq - for idx < len { - pid := h[idx].pkt.Seq - var next int - if pid.Seq == greaterEqual.Seq { - return h[idx].pkt, idx - } else if pid.Seq >= greaterEqual.Seq { - next = idx * 2 - } else { - next = idx*2 + 1 - } - if next >= len && h[idx].pkt.Seq.Seq > greaterEqual.Seq && (wrapped || h[idx].pkt.Seq.Seq <= lessEqual.Seq) { - return h[idx].pkt, idx - } - idx = next - } - - // can't find any packets with greater value, wrap around - if wrapped { - idx = 0 - for { - next := idx * 2 - if next >= len && h[idx].pkt.Seq.Seq <= lessEqual.Seq { - return h[idx].pkt, idx - } - idx = next - } - } - return nil, -1 +// Count returns the number of packets stored +func (heap *sendPacketHeap) Count() (count int) { + return len(heap.list) } -// Remove does a binary search of the heap for the specified packetID, which is removed -func (h *sendPacketHeap) Remove(packetID packet.PacketID) bool { - len := len(*h) - idx := 0 - for idx < len { - pid := (*h)[idx].pkt.Seq - if pid.Seq == packetID.Seq { - heap.Remove(h, idx) - return true - } else if pid.Seq > packetID.Seq { - idx = idx * 2 - } else { - idx = idx*2 + 1 +// Find searches for the packet +func (heap *sendPacketHeap) Find(sequence uint32) (result *sendPacketEntry) { + heap.RLock() + defer heap.RUnlock() + + for n := range heap.list { + if heap.list[n].pkt.Seq.Seq == sequence { + return &heap.list[n] } } - return false + + return nil // not found +} + +// RemoveRange removes all packets that are within the given range. Check is from >= and to <. +func (heap *sendPacketHeap) RemoveRange(sequenceFrom, sequenceTo uint32) { + heap.Lock() + defer heap.Unlock() + + var newList []sendPacketEntry + + for n := range heap.list { + if !(heap.list[n].pkt.Seq.Seq >= sequenceFrom && heap.list[n].pkt.Seq.Seq < sequenceTo) { + newList = append(newList, heap.list[n]) + } + } + + heap.list = newList } diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 7dcf9ac..c84b5ca 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -31,7 +31,7 @@ type udtSocketSend struct { socket *udtSocket sendState sendState // current sender state - sendPktPend sendPacketHeap // list of packets that have been sent but not yet acknowledged + sendPktPend *sendPacketHeap // list of packets that have been sent but not yet acknowledged sendPktSeq packet.PacketID // the current packet sequence number msgPartialSend *sendMessage // when a message can only partially fit in a socket, this is the remainder msgSeq uint32 // the current message sequence number @@ -61,6 +61,7 @@ func newUdtSocketSend(s *udtSocket) *udtSocketSend { flowWindowSize: s.maxFlowWinSize, sendPacket: s.sendPacket, shutdownEvent: s.shutdownEvent, + sendPktPend: createSendPacketHeap(), } ss.resetEXP(s.created) go ss.goSendEvent() @@ -155,17 +156,16 @@ func (s *udtSocketSend) reevalSendState() sendState { if s.sndEvent != nil { return sendStateSending } + // Do we have too many unacknowledged packets for us to send any more? - if s.sendPktPend != nil { - congestWindow := uint(s.congestWindow.get()) - cwnd := s.flowWindowSize - if cwnd > congestWindow { - cwnd = congestWindow - } - if uint(len(s.sendPktPend)) >= cwnd { - return sendStateWaiting - } + cwnd := uint(s.congestWindow.get()) + if cwnd > s.flowWindowSize { + cwnd = s.flowWindowSize } + if uint(s.sendPktPend.Count()) > cwnd { + return sendStateWaiting + } + return sendStateIdle } @@ -243,7 +243,7 @@ func (s *udtSocketSend) processDataMsg(isFirst bool, inChan <-chan sendMessage) // If the sender's loss list is not empty, retransmit the first packet in the list and remove it from the list. func (s *udtSocketSend) processSendLoss() bool { - if s.sendLossList == nil || s.sendPktPend == nil { + if s.sendLossList == nil || s.sendPktPend.Count() == 0 { return false } @@ -260,7 +260,7 @@ func (s *udtSocketSend) processSendLoss() bool { s.sendLossList = nil } - dp, _ = s.sendPktPend.Find(minLoss) + dp = s.sendPktPend.Find(minLoss.Seq) if dp == nil { // can't find record of this packet, not much we can do really continue @@ -280,12 +280,12 @@ func (s *udtSocketSend) processSendLoss() bool { // evaluate our pending packet list to see if we have any expired messages func (s *udtSocketSend) processSendExpire() bool { - if s.sendPktPend == nil { + if s.sendPktPend.Count() == 0 { return false } - pktPend := make([]sendPacketEntry, len(s.sendPktPend)) - copy(pktPend, s.sendPktPend) + pktPend := make([]sendPacketEntry, s.sendPktPend.Count()) + copy(pktPend, s.sendPktPend.list) for _, p := range pktPend { if p.ttl != 0 && time.Now().Add(p.ttl).After(p.tim) { // this message has expired, drop it @@ -330,12 +330,7 @@ func (s *udtSocketSend) sendDataPacket(dp sendPacketEntry, isResend bool) { // It would not make any sense and introduce race condition with potential endless packet resends/ACKs. // Once the remote peer ACKs a sent packet, it is removed from the list. if !isResend { - if s.sendPktPend == nil { - s.sendPktPend = sendPacketHeap{dp} - heap.Init(&s.sendPktPend) - } else { - heap.Push(&s.sendPktPend, dp) - } + s.sendPktPend.Add(dp) } s.socket.cong.onDataPktSent(dp.pkt.Seq) @@ -397,18 +392,7 @@ func (s *udtSocketSend) ingestAck(p *packet.AckPacket, now time.Time) { // Update estimated link capacity: B = (B * 7 + b) / 8, where b is the value carried in the ACK. // Update sender's list of packets that have been sent but not yet acknowledged - if s.sendPktPend != nil { - for { - minLoss, minLossIdx := s.sendPktPend.Min(oldAckSeq, s.sendPktSeq) - if p.PktSeqHi.BlindDiff(minLoss.Seq) >= 0 || minLossIdx < 0 { - break - } - heap.Remove(&s.sendPktPend, minLossIdx) - } - if len(s.sendPktPend) == 0 { - s.sendPktPend = nil - } - } + s.sendPktPend.RemoveRange(oldAckSeq.Seq, p.PktSeqHi.Seq) // Update sender's loss list (by removing all those that has been acknowledged). if s.sendLossList != nil { @@ -518,8 +502,8 @@ func (s *udtSocketSend) expEvent(currTime time.Time) { // sender: Insert all the packets sent after last received acknowledgement into the sender loss list. // recver: Send out a keep-alive packet - if s.sendPktPend != nil { - if s.sendPktPend != nil && s.sendLossList == nil { + if s.sendPktPend.Count() > 0 { + if s.sendLossList == nil { // resend all unacknowledged packets on timeout, but only if there is no packet in the loss list newLossList := make([]packet.PacketID, 0) for span := s.recvAckSeq.Add(1); span != s.sendPktSeq.Add(1); span.Incr() {