From f76426aa4827040e1b17fae66370ba203c240462 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Fri, 5 Nov 2021 01:06:02 +0100 Subject: [PATCH] UDT: Remove the datapacket_heap with the new heap code --- udt/datapacket_heap.go | 115 ----------------------------------------- udt/sendpacket_heap.go | 16 +++++- udt/udtsocket_recv.go | 64 ++++++++--------------- udt/udtsocket_send.go | 2 +- 4 files changed, 38 insertions(+), 159 deletions(-) delete mode 100644 udt/datapacket_heap.go diff --git a/udt/datapacket_heap.go b/udt/datapacket_heap.go deleted file mode 100644 index 19e7cad..0000000 --- a/udt/datapacket_heap.go +++ /dev/null @@ -1,115 +0,0 @@ -package udt - -import ( - "container/heap" - - "github.com/PeernetOfficial/core/udt/packet" -) - -// receiveLossList defines a list of recvLossEntry records sorted by their packet ID -type dataPacketHeap []*packet.DataPacket - -func (h dataPacketHeap) Len() int { - return len(h) -} - -func (h dataPacketHeap) Less(i, j int) bool { - return h[i].Seq.Seq < h[j].Seq.Seq -} - -func (h dataPacketHeap) Swap(i, j int) { - h[i], h[j] = h[j], h[i] -} - -func (h *dataPacketHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents. - *h = append(*h, x.(*packet.DataPacket)) -} - -func (h *dataPacketHeap) Pop() interface{} { - old := *h - n := len(old) - x := old[n-1] - *h = old[0 : n-1] - return x -} - -// Find does a binary search of the heap for the specified packetID which is returned -func (h dataPacketHeap) Find(packetID packet.PacketID) (*packet.DataPacket, int) { - for n := 0; n < len(h); n++ { - if h[n].Seq == packetID { - return h[n], n - } - } - - // len := len(h) - // idx := 0 - // for idx < len { - // pid := h[idx].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 -} - -// 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 dataPacketHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (*packet.DataPacket, int) { - if len(h) == 0 { // none available! - return nil, -1 - } - return h[0], 0 - - len := len(h) - idx := 0 - wrapped := greaterEqual.Seq > lessEqual.Seq - for idx < len { - pid := h[idx].Seq - var next int - if pid.Seq == greaterEqual.Seq { - return h[idx], idx - } else if pid.Seq >= greaterEqual.Seq { - next = idx * 2 - } else { - next = idx*2 + 1 - } - if next >= len && h[idx].Seq.Seq > greaterEqual.Seq && (wrapped || h[idx].Seq.Seq <= lessEqual.Seq) { - return h[idx], 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].Seq.Seq <= lessEqual.Seq { - return h[idx], idx - } - idx = next - } - } - return nil, -1 -} - -// Remove does a binary search of the heap for the specified packetID, which is removed -func (h *dataPacketHeap) Remove(packetID packet.PacketID) bool { - len := len(*h) - idx := 0 - for idx < len { - pid := (*h)[idx].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 - } - } - return false -} diff --git a/udt/sendpacket_heap.go b/udt/sendpacket_heap.go index 165d5dc..0e3a8a3 100644 --- a/udt/sendpacket_heap.go +++ b/udt/sendpacket_heap.go @@ -26,7 +26,7 @@ type sendPacketHeap struct { sync.RWMutex } -func createSendPacketHeap() (heap *sendPacketHeap) { +func createPacketHeap() (heap *sendPacketHeap) { return &sendPacketHeap{} } @@ -90,3 +90,17 @@ func (heap *sendPacketHeap) RemoveRange(sequenceFrom, sequenceTo uint32) { heap.list = newList } + +// Range returns all packets that are within the given range. Check is from >= and to <. +func (heap *sendPacketHeap) Range(sequenceFrom, sequenceTo uint32) (result []sendPacketEntry) { + heap.RLock() + defer heap.RUnlock() + + for n := range heap.list { + if heap.list[n].pkt.Seq.Seq >= sequenceFrom && heap.list[n].pkt.Seq.Seq < sequenceTo { + result = append(result, heap.list[n]) + } + } + + return result +} diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index 8a1a472..a70dfec 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -20,7 +20,7 @@ type udtSocketRecv struct { farRecdPktSeq packet.PacketID // the peer's last "received" packet ID (before any loss events) lastACK uint32 // last ACK packet we've sent largestACK uint32 // largest ACK packet we've sent that has been acknowledged (by an ACK2). - recvPktPend dataPacketHeap // list of packets that are waiting to be processed. + recvPktPend *sendPacketHeap // list of packets that are waiting to be processed. recvLossList receiveLossHeap // loss list. ackHistory ackHistoryHeap // list of sent ACKs. sentAck packet.PacketID // largest packetID we've sent an ACK regarding @@ -39,11 +39,12 @@ type udtSocketRecv struct { func newUdtSocketRecv(s *udtSocket) *udtSocketRecv { sr := &udtSocketRecv{ - socket: s, - sockClosed: s.sockClosed, - recvEvent: s.recvEvent, - messageIn: s.messageIn, - sendPacket: s.sendPacket, + socket: s, + sockClosed: s.sockClosed, + recvEvent: s.recvEvent, + messageIn: s.messageIn, + sendPacket: s.sendPacket, + recvPktPend: createPacketHeap(), } go sr.goReceiveEvent() return sr @@ -142,11 +143,7 @@ func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Ti } // remove all pending packets with this message - if s.recvPktPend != nil { - if lossEntry, idx := s.recvPktPend.Find(pktID); lossEntry != nil { - heap.Remove(&s.recvPktPend, idx) - } - } + s.recvPktPend.Remove(pktID.Seq) } @@ -157,14 +154,10 @@ func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Ti s.farRecdPktSeq = s.farNextPktSeq.Add(-1) s.recvLossList = nil } - if s.recvPktPend != nil && len(s.recvPktPend) == 0 { - s.recvPktPend = nil - } // try to push any pending packets out, now that we have dropped any blocking packets - for s.recvPktPend != nil && stopSeq != s.farNextPktSeq { - nextPkt, _ := s.recvPktPend.Min(stopSeq, s.farNextPktSeq) - if nextPkt == nil || !s.attemptProcessPacket(nextPkt, false) { + for _, nextPkt := range s.recvPktPend.Range(stopSeq.Seq, s.farNextPktSeq.Seq) { + if !s.attemptProcessPacket(nextPkt.pkt, false) { break } } @@ -258,12 +251,7 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b if s.recvLossList != nil && mustOrder && s.farRecdPktSeq.Add(1) != seq { // we're required to order these packets and we're missing prior packets, so push and return if isNew { - if s.recvPktPend == nil { - s.recvPktPend = dataPacketHeap{p} - heap.Init(&s.recvPktPend) - } else { - heap.Push(&s.recvPktPend, p) - } + s.recvPktPend.Add(sendPacketEntry{pkt: p}) } return false } @@ -274,10 +262,10 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b switch boundary { case packet.MbLast, packet.MbMiddle: // we need prior packets, let's make sure we have them - if s.recvPktPend != nil { + if s.recvPktPend.Count() > 0 { pieceSeq := seq.Add(-1) for { - prevPiece, _ := s.recvPktPend.Find(pieceSeq) + prevPiece := s.recvPktPend.Find(pieceSeq.Seq) if prevPiece == nil { // we don't have the previous piece, is it missing? if s.recvLossList != nil { @@ -290,13 +278,13 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID) break } - prevBoundary, _, prevMsg := prevPiece.GetMessageData() + prevBoundary, _, prevMsg := prevPiece.pkt.GetMessageData() if prevMsg != msgID { // ...oops? previous piece isn't in the same message fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID) break } - pieces = append([]*packet.DataPacket{prevPiece}, pieces...) + pieces = append([]*packet.DataPacket{prevPiece.pkt}, pieces...) if prevBoundary == packet.MbFirst { break } @@ -310,10 +298,10 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b switch boundary { case packet.MbFirst, packet.MbMiddle: // we need following packets, let's make sure we have them - if s.recvPktPend != nil { + if s.recvPktPend.Count() > 0 { pieceSeq := seq.Add(1) for { - nextPiece, _ := s.recvPktPend.Find(pieceSeq) + nextPiece := s.recvPktPend.Find(pieceSeq.Seq) if nextPiece == nil { // we don't have the previous piece, is it missing? if pieceSeq == s.farNextPktSeq { @@ -330,13 +318,13 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b // in any case we can't continue with this break } - nextBoundary, _, nextMsg := nextPiece.GetMessageData() + nextBoundary, _, nextMsg := nextPiece.pkt.GetMessageData() if nextMsg != msgID { // ...oops? previous piece isn't in the same message fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID) break } - pieces = append(pieces, nextPiece) + pieces = append(pieces, nextPiece.pkt) if nextBoundary == packet.MbLast { break } @@ -356,23 +344,15 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b if cannotContinue { // we need to wait for more packets, store and return if isNew { - if s.recvPktPend == nil { - s.recvPktPend = dataPacketHeap{p} - heap.Init(&s.recvPktPend) - } else { - heap.Push(&s.recvPktPend, p) - } + s.recvPktPend.Add(sendPacketEntry{pkt: p}) } return false } // we have a message, pull it from the pending heap (if necessary), assemble it into a message, and return it - if s.recvPktPend != nil { + if s.recvPktPend.Count() > 0 { for _, piece := range pieces { - s.recvPktPend.Remove(piece.Seq) - } - if len(s.recvPktPend) == 0 { - s.recvPktPend = nil + s.recvPktPend.Remove(piece.Seq.Seq) } } diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index c84b5ca..efd22c3 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -61,7 +61,7 @@ func newUdtSocketSend(s *udtSocket) *udtSocketSend { flowWindowSize: s.maxFlowWinSize, sendPacket: s.sendPacket, shutdownEvent: s.shutdownEvent, - sendPktPend: createSendPacketHeap(), + sendPktPend: createPacketHeap(), } ss.resetEXP(s.created) go ss.goSendEvent()