From 0d62c215eee705de6439ec0be9218c0332ef4e0d Mon Sep 17 00:00:00 2001 From: Kleissner Date: Fri, 5 Nov 2021 03:19:29 +0100 Subject: [PATCH] UDT: Fixing #43 #49 #51. File transfer works in test environment. More tests needed for production testing. Internet MTU needs to be tested. --- udt/ack_history_heap.go | 68 ++++++++-------- udt/packetid_heap.go | 107 ------------------------- udt/recvloss_heap.go | 173 ++++++++++++++++++++-------------------- udt/sendpacket_heap.go | 3 - udt/udtsocket_cc.go | 5 +- udt/udtsocket_recv.go | 159 ++++++++++++------------------------ udt/udtsocket_send.go | 98 +++++++---------------- 7 files changed, 200 insertions(+), 413 deletions(-) delete mode 100644 udt/packetid_heap.go diff --git a/udt/ack_history_heap.go b/udt/ack_history_heap.go index 3b3efeb..ab7d0c4 100644 --- a/udt/ack_history_heap.go +++ b/udt/ack_history_heap.go @@ -1,6 +1,7 @@ package udt import ( + "sync" "time" "github.com/PeernetOfficial/core/udt/packet" @@ -12,52 +13,47 @@ type ackHistoryEntry struct { sendTime time.Time } -// receiveLossList defines a list of ACK records sorted by their ACK id -type ackHistoryHeap []*ackHistoryEntry +// receiveLossList defines a list of recvLossEntry records +type ackHistoryHeap struct { + // list contains all entries + list []ackHistoryEntry -func (h ackHistoryHeap) Len() int { - return len(h) + sync.RWMutex } -func (h ackHistoryHeap) Less(i, j int) bool { - return h[i].ackID < h[j].ackID +func createHistoryHeap() (heap *ackHistoryHeap) { + return &ackHistoryHeap{} } -func (h ackHistoryHeap) Swap(i, j int) { - h[i], h[j] = h[j], h[i] +// Add adds an entry to the list. Deduplication is not performed. +func (heap *ackHistoryHeap) Add(newEntry ackHistoryEntry) { + heap.Lock() + defer heap.Unlock() + + heap.list = append(heap.list, newEntry) } -func (h *ackHistoryHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents. - *h = append(*h, x.(*ackHistoryEntry)) -} +// Remove removes all IDs matching from the list. +func (heap *ackHistoryHeap) Remove(sequence uint32) (found *ackHistoryEntry) { + heap.Lock() + defer heap.Unlock() -func (h *ackHistoryHeap) Pop() interface{} { - old := *h - n := len(old) - x := old[n-1] - *h = old[0 : n-1] - return x -} + var newList []ackHistoryEntry -// Find does a binary search of the heap for the specified ackID which is returned -func (h ackHistoryHeap) Find(ackID uint32) (*ackHistoryEntry, int) { - for n := 0; n < len(h); n++ { - if h[n].ackID == ackID { - return h[n], n + for n := range heap.list { + if heap.list[n].ackID != sequence { + newList = append(newList, heap.list[n]) + } else { + found = &heap.list[n] } } - // len := len(h) - // idx := 0 - // for idx < len { - // here := h[idx].ackID - // if here == ackID { - // return h[idx], idx - // } else if here > ackID { - // idx = idx * 2 - // } else { - // idx = idx*2 + 1 - // } - // } - return nil, -1 + heap.list = newList + + return found +} + +// Count returns the number of packets stored +func (heap *ackHistoryHeap) Count() (count int) { + return len(heap.list) } diff --git a/udt/packetid_heap.go b/udt/packetid_heap.go deleted file mode 100644 index 301951e..0000000 --- a/udt/packetid_heap.go +++ /dev/null @@ -1,107 +0,0 @@ -package udt - -import ( - "github.com/PeernetOfficial/core/udt/packet" -) - -// packetIdHeap defines a list of sorted packet IDs -type packetIDHeap []packet.PacketID - -func (h packetIDHeap) Len() int { - return len(h) -} - -func (h packetIDHeap) Less(i, j int) bool { - return h[i].Seq < h[j].Seq -} - -func (h packetIDHeap) Swap(i, j int) { - h[i], h[j] = h[j], h[i] -} - -func (h *packetIDHeap) 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.PacketID)) -} - -func (h *packetIDHeap) Pop() interface{} { - old := *h - n := len(old) - x := old[n-1] - *h = old[0 : n-1] - return x -} - -// 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 packetIDHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (packet.PacketID, int) { - if len(h) == 0 { // none available! - return packet.PacketID{Seq: 0}, -1 - } - return h[0], 0 - - // Disable below buggy code. The second for loop is an infinite loop. - // This whole function probably makes 0 sense! - - // len := len(h) - // wrapped := greaterEqual.Seq > lessEqual.Seq - - // for i := 0; i < len; { - // pid := h[i] - // var next int - // if pid.Seq == greaterEqual.Seq { - // return h[i], i - // } else if pid.Seq >= greaterEqual.Seq { - // next = i * 2 - // } else { - // next = i*2 + 1 - // } - // if next >= len && h[i].Seq > greaterEqual.Seq && (wrapped || h[i].Seq <= lessEqual.Seq) { - // return h[i], i - // } - // i = next - // } - - // // can't find any packets with greater value, wrap around - // if wrapped { - // for i := 0; ; { - // next := i * 2 - // if next >= len && h[i].Seq <= lessEqual.Seq { - // return h[i], i - // } - // i = next - // } - // } - // return packet.PacketID{Seq: 0}, -1 -} - -func (h packetIDHeap) compare(pktID packet.PacketID, idx int) int { - if pktID.Seq < h[idx].Seq { - return -1 - } - if pktID.Seq > h[idx].Seq { - return +1 - } - return 0 -} - -// Find does a binary search of the heap for the specified packetID which is returned -func (h packetIDHeap) Find(pktID packet.PacketID) (*packet.PacketID, int) { - for n := 0; n < len(h); n++ { - if h[n].Seq == pktID.Seq { - return &h[n], n - } - } - - // len := len(h) - // idx := 0 - // for idx < len { - // cmp := h.compare(pktID, idx) - // if cmp == 0 { - // return &h[idx], idx - // } else if cmp > 0 { - // idx = idx * 2 - // } else { - // idx = idx*2 + 1 - // } - // } - return nil, -1 -} diff --git a/udt/recvloss_heap.go b/udt/recvloss_heap.go index 1ea566f..837d619 100644 --- a/udt/recvloss_heap.go +++ b/udt/recvloss_heap.go @@ -1,122 +1,121 @@ package udt import ( - "container/heap" + "sync" "time" "github.com/PeernetOfficial/core/udt/packet" ) type recvLossEntry struct { - packetID packet.PacketID + packetID packet.PacketID + + // data specific to loss entries lastFeedback time.Time numNAK uint } -// receiveLossList defines a list of recvLossEntry records sorted by their packet ID -type receiveLossHeap []recvLossEntry +// receiveLossList defines a list of recvLossEntry records +type receiveLossHeap struct { + // list contains all entries + list []recvLossEntry -func (h receiveLossHeap) Len() int { - return len(h) + sync.RWMutex } -func (h receiveLossHeap) Less(i, j int) bool { - return h[i].packetID.Seq < h[j].packetID.Seq +func createPacketIDHeap() (heap *receiveLossHeap) { + return &receiveLossHeap{} } -func (h receiveLossHeap) Swap(i, j int) { - h[i], h[j] = h[j], h[i] +// Add adds an entry to the list. Deduplication is not performed. +func (heap *receiveLossHeap) Add(newEntry recvLossEntry) { + heap.Lock() + defer heap.Unlock() + + heap.list = append(heap.list, newEntry) } -func (h *receiveLossHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents. - *h = append(*h, x.(recvLossEntry)) -} +// Remove removes all IDs matching from the list. +func (heap *receiveLossHeap) Remove(sequence uint32) (found bool) { + heap.Lock() + defer heap.Unlock() -func (h *receiveLossHeap) Pop() interface{} { - old := *h - n := len(old) - x := old[n-1] - *h = old[0 : n-1] - return x -} + var newList []recvLossEntry -// 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 receiveLossHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (packet.PacketID, int) { - if len(h) == 0 { // none available! - return packet.PacketID{Seq: 0}, -1 - } - return h[0].packetID, 0 - - len := len(h) - idx := 0 - wrapped := greaterEqual.Seq > lessEqual.Seq - for idx < len { - pid := h[idx].packetID - var next int - if pid.Seq == greaterEqual.Seq { - return h[idx].packetID, idx - } else if pid.Seq >= greaterEqual.Seq { - next = idx * 2 + for n := range heap.list { + if heap.list[n].packetID.Seq != sequence { + newList = append(newList, heap.list[n]) } else { - next = idx*2 + 1 + found = true } - if next >= len && h[idx].packetID.Seq > greaterEqual.Seq && (wrapped || h[idx].packetID.Seq <= lessEqual.Seq) { - return h[idx].packetID, 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].packetID.Seq <= lessEqual.Seq { - return h[idx].packetID, idx + heap.list = newList + + return found +} + +// Count returns the number of packets stored +func (heap *receiveLossHeap) Count() (count int) { + return len(heap.list) +} + +// Find searches for the packet +func (heap *receiveLossHeap) Find(sequence uint32) (result *recvLossEntry) { + heap.RLock() + defer heap.RUnlock() + + for n := range heap.list { + if heap.list[n].packetID.Seq == sequence { + return &heap.list[n] + } + } + + return nil // not found +} + +// Min returns the lowest matching value, if available. Otherwise returns first value. +func (heap *receiveLossHeap) Min(sequenceFrom, sequenceTo uint32) (result *packet.PacketID) { + heap.RLock() + defer heap.RUnlock() + + for n := range heap.list { + if heap.list[n].packetID.Seq >= sequenceFrom && heap.list[n].packetID.Seq < sequenceTo { + if result == nil || heap.list[n].packetID.Seq < result.Seq { + result = &heap.list[n].packetID } - idx = next - } - } - return packet.PacketID{Seq: 0}, -1 -} - -// Find does a binary search of the heap for the specified packetID which is returned -func (h receiveLossHeap) Find(packetID packet.PacketID) (*recvLossEntry, int) { - for n := 0; n < len(h); n++ { - if h[n].packetID == packetID { - return &h[n], n } } - // len := len(h) - // idx := 0 - // for idx < len { - // pid := h[idx].packetID - // if pid == packetID { - // return &h[idx], idx - // } else if pid.Seq > packetID.Seq { - // idx = idx * 2 - // } else { - // idx = idx*2 + 1 - // } - // } - return nil, -1 + return result } -// Remove does a binary search of the heap for the specified packetID, which is removed -func (h *receiveLossHeap) Remove(packetID packet.PacketID) bool { - len := len(*h) - idx := 0 - for idx < len { - pid := (*h)[idx].packetID - if pid == packetID { - heap.Remove(h, idx) - return true - } else if pid.Seq > packetID.Seq { - idx = idx * 2 - } else { - idx = idx*2 + 1 +// RemoveRange removes all packets that are within the given range. Check is from >= and to <. +func (heap *receiveLossHeap) RemoveRange(sequenceFrom, sequenceTo uint32) { + heap.Lock() + defer heap.Unlock() + + var newList []recvLossEntry + + for n := range heap.list { + if !(heap.list[n].packetID.Seq >= sequenceFrom && heap.list[n].packetID.Seq < sequenceTo) { + newList = append(newList, heap.list[n]) } } - return false + + heap.list = newList +} + +// Range returns all packets that are within the given range. Check is from >= and to <. +func (heap *receiveLossHeap) Range(sequenceFrom, sequenceTo uint32) (result []recvLossEntry) { + heap.RLock() + defer heap.RUnlock() + + for n := range heap.list { + if heap.list[n].packetID.Seq >= sequenceFrom && heap.list[n].packetID.Seq < sequenceTo { + result = append(result, heap.list[n]) + } + } + + return result } diff --git a/udt/sendpacket_heap.go b/udt/sendpacket_heap.go index 0e3a8a3..cd8e175 100644 --- a/udt/sendpacket_heap.go +++ b/udt/sendpacket_heap.go @@ -1,7 +1,6 @@ package udt import ( - "fmt" "sync" "time" @@ -35,8 +34,6 @@ 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) } diff --git a/udt/udtsocket_cc.go b/udt/udtsocket_cc.go index c730c3e..82f3ed0 100644 --- a/udt/udtsocket_cc.go +++ b/udt/udtsocket_cc.go @@ -115,12 +115,9 @@ func (s *udtSocketCc) onACK(pktID packet.PacketID) { // OnNAK to be called when a loss report is received func (s *udtSocketCc) onNAK(loss []packet.PacketID) { - var ourLoss = make([]packet.PacketID, len(loss)) - copy(ourLoss, loss) - s.msgs <- congMsg{ mtyp: congOnNAK, - arg: ourLoss, + arg: loss, } } diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index a70dfec..d15e243 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -1,8 +1,6 @@ package udt import ( - "container/heap" - "fmt" "time" "github.com/PeernetOfficial/core/udt/packet" @@ -16,21 +14,21 @@ type udtSocketRecv struct { sendPacket chan<- packet.Packet // send a packet out on the wire socket *udtSocket - farNextPktSeq packet.PacketID // the peer's next largest packet ID expected. - 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 *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 - recvAck2 packet.PacketID // largest packetID we've received an ACK2 from - recvLastArrival time.Time // time of the most recent data packet arrival - recvLastProbe time.Time // time of the most recent data packet probe packet - ackPeriod atomicDuration // (set by congestion control) delay between sending ACKs - unackPktCount uint // number of packets we've received that we haven't sent an ACK for - recvPktHistory []time.Duration // list of recently received packets. - recvPktPairHistory []time.Duration // probing packet window. + farNextPktSeq packet.PacketID // the peer's next largest packet ID expected. + 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 *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 + recvAck2 packet.PacketID // largest packetID we've received an ACK2 from + recvLastArrival time.Time // time of the most recent data packet arrival + recvLastProbe time.Time // time of the most recent data packet probe packet + ackPeriod atomicDuration // (set by congestion control) delay between sending ACKs + unackPktCount uint // number of packets we've received that we haven't sent an ACK for + recvPktHistory []time.Duration // list of recently received packets. + recvPktPairHistory []time.Duration // probing packet window. // timers ackSentEvent2 <-chan time.Time // if an ACK packet has recently sent, don't include link information in the next one @@ -39,12 +37,14 @@ type udtSocketRecv struct { func newUdtSocketRecv(s *udtSocket) *udtSocketRecv { sr := &udtSocketRecv{ - socket: s, - sockClosed: s.sockClosed, - recvEvent: s.recvEvent, - messageIn: s.messageIn, - sendPacket: s.sendPacket, - recvPktPend: createPacketHeap(), + socket: s, + sockClosed: s.sockClosed, + recvEvent: s.recvEvent, + messageIn: s.messageIn, + sendPacket: s.sendPacket, + recvPktPend: createPacketHeap(), + recvLossList: createPacketIDHeap(), + ackHistory: createHistoryHeap(), } go sr.goReceiveEvent() return sr @@ -107,23 +107,17 @@ ACK is used to trigger an acknowledgement (ACK). Its period is set by // ingestAck2 is called to process an ACK2 packet func (s *udtSocketRecv) ingestAck2(p *packet.Ack2Packet, now time.Time) { - ackSeq := p.AckSeqNo - if s.ackHistory == nil { - return // no ACKs to search - } - - ackHistEntry, ackIdx := s.ackHistory.Find(ackSeq) + ackHistEntry := s.ackHistory.Remove(p.AckSeqNo) if ackHistEntry == nil { return // this ACK not found } if s.recvAck2.BlindDiff(ackHistEntry.lastPacket) < 0 { s.recvAck2 = ackHistEntry.lastPacket } - heap.Remove(&s.ackHistory, ackIdx) // Update the largest ACK number ever been acknowledged. - if s.largestACK < ackSeq { - s.largestACK = ackSeq + if s.largestACK < p.AckSeqNo { + s.largestACK = p.AckSeqNo } s.socket.applyRTT(uint(now.Sub(ackHistEntry.sendTime) / time.Microsecond)) @@ -136,24 +130,18 @@ func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Ti stopSeq := p.LastSeq.Add(1) for pktID := p.FirstSeq; pktID != stopSeq; pktID.Incr() { // remove all these packets from the loss list - if s.recvLossList != nil { - if lossEntry, idx := s.recvLossList.Find(pktID); lossEntry != nil { - heap.Remove(&s.recvLossList, idx) - } - } + s.recvLossList.Remove(pktID.Seq) // remove all pending packets with this message s.recvPktPend.Remove(pktID.Seq) - } if p.FirstSeq == s.farRecdPktSeq.Add(1) { s.farRecdPktSeq = p.LastSeq } - if s.recvLossList != nil && len(s.recvLossList) == 0 { - s.farRecdPktSeq = s.farNextPktSeq.Add(-1) - s.recvLossList = nil - } + // if s.recvLossList.Count() == 0 { + // s.farRecdPktSeq = s.farNextPktSeq.Add(-1) + // } // try to push any pending packets out, now that we have dropped any blocking packets for _, nextPkt := range s.recvPktPend.Range(stopSeq.Seq, s.farNextPktSeq.Seq) { @@ -205,36 +193,25 @@ func (s *udtSocketRecv) ingestData(p *packet.DataPacket, now time.Time) { send them to the sender in an NAK packet. */ seqDiff := seq.BlindDiff(s.farNextPktSeq) if seqDiff > 0 { - fmt.Printf("Warning sequence out of order :( Code that follows will crash. Expected %d but received is %d\n", s.farNextPktSeq, p.Seq) - newLoss := make(receiveLossHeap, 0, seqDiff) - for idx := s.farNextPktSeq; idx != seq; idx.Incr() { - newLoss = append(newLoss, recvLossEntry{packetID: idx}) + for n := uint32(0); n < uint32(seqDiff); n++ { + s.recvLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: (s.farNextPktSeq.Seq + n) & 0x7FFFFFFF}}) } - if s.recvLossList == nil { - s.recvLossList = newLoss - heap.Init(&s.recvLossList) - } else { - for idx := s.farNextPktSeq; idx != seq; idx.Incr() { - heap.Push(&s.recvLossList, recvLossEntry{packetID: idx}) - } - heap.Init(&newLoss) - } - - s.sendNAK(newLoss) + s.sendNAK(s.farNextPktSeq.Seq, uint32(seqDiff)) s.farNextPktSeq = seq.Add(1) } else if seqDiff < 0 { // If the sequence number is less than LRSN, remove it from the receiver's loss list. - if !s.recvLossList.Remove(seq) { + if !s.recvLossList.Remove(seq.Seq) { return // already previously received packet -- ignore } - if len(s.recvLossList) == 0 { + if s.recvLossList.Count() == 0 { s.farRecdPktSeq = s.farNextPktSeq.Add(-1) - s.recvLossList = nil } else { - s.farRecdPktSeq, _ = s.recvLossList.Min(s.farRecdPktSeq, s.farNextPktSeq) + if minR := s.recvLossList.Min(s.farRecdPktSeq.Seq, s.farNextPktSeq.Seq); minR != nil { + s.farRecdPktSeq = packet.PacketID{Seq: minR.Seq} + } } } else { s.farNextPktSeq = seq.Add(1) @@ -248,7 +225,7 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b // can we process this packet? boundary, mustOrder, msgID := p.GetMessageData() - if s.recvLossList != nil && mustOrder && s.farRecdPktSeq.Add(1) != seq { + if s.recvLossList.Count() > 0 && 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 { s.recvPktPend.Add(sendPacketEntry{pkt: p}) @@ -268,20 +245,18 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b prevPiece := s.recvPktPend.Find(pieceSeq.Seq) if prevPiece == nil { // we don't have the previous piece, is it missing? - if s.recvLossList != nil { - if lossEntry, _ := s.recvLossList.Find(pieceSeq); lossEntry != nil { + if s.recvLossList.Count() > 0 { + if s.recvLossList.Find(pieceSeq.Seq) != nil { // it's missing, stop processing cannotContinue = true } } // in any case we can't continue with this - fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID) break } 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.pkt}, pieces...) @@ -307,13 +282,12 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b if pieceSeq == s.farNextPktSeq { // hasn't been received yet cannotContinue = true - } else if s.recvLossList != nil { - if lossEntry, _ := s.recvLossList.Find(pieceSeq); lossEntry != nil { + } else if s.recvLossList.Count() > 0 { + if s.recvLossList.Find(pieceSeq.Seq) != nil { // it's missing, stop processing cannotContinue = true } } else { - fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID) } // in any case we can't continue with this break @@ -321,7 +295,6 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b 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.pkt) @@ -337,9 +310,10 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b // Before, there was both the (unused) ACK interval s.ackInterval and s.ackTimerEvent which fired at SynTime, which was way too often and basically a ddos. // It makes more sense to just send the ACK x split of the congestion window. s.unackPktCount++ - if s.unackPktCount >= s.socket.cong.GetCongestionWindowSize()/4 { - s.ackEvent() - } + // DEBUG: Always send ack for now + //if s.unackPktCount >= s.socket.cong.GetCongestionWindowSize()/4 { + s.ackEvent() + //} if cannotContinue { // we need to wait for more packets, store and return @@ -434,7 +408,7 @@ func (s *udtSocketRecv) sendACK() { // If there is no loss, the ACK is the current largest sequence number plus 1; // Otherwise it is the smallest sequence number in the receiver loss list. - if s.recvLossList == nil { + if s.recvLossList.Count() == 0 { ack = s.farNextPktSeq } else { ack = s.farRecdPktSeq.Add(1) @@ -451,17 +425,11 @@ func (s *udtSocketRecv) sendACK() { s.sentAck = ack s.lastACK++ - ackHist := &ackHistoryEntry{ + s.ackHistory.Add(ackHistoryEntry{ ackID: s.lastACK, lastPacket: ack, sendTime: time.Now(), - } - if s.ackHistory == nil { - s.ackHistory = ackHistoryHeap{ackHist} - heap.Init(&s.ackHistory) - } else { - heap.Push(&s.ackHistory, ackHist) - } + }) rtt, rttVar := s.socket.getRTT() @@ -489,31 +457,10 @@ func (s *udtSocketRecv) sendACK() { s.ackSentEvent = time.After(time.Duration(rtt+4*rttVar) * time.Microsecond) } -func (s *udtSocketRecv) sendNAK(rl receiveLossHeap) { +func (s *udtSocketRecv) sendNAK(sequenceFrom uint32, count uint32) { lossInfo := make([]uint32, 0) - - curPkt := s.farRecdPktSeq - for curPkt != s.farNextPktSeq { - minPkt, idx := rl.Min(curPkt, s.farRecdPktSeq) - if idx < 0 { - break - } - - lastPkt := minPkt - for { - nextPkt := lastPkt.Add(1) - _, idx = rl.Find(nextPkt) - if idx < 0 { - break - } - lastPkt = nextPkt - } - - if lastPkt == minPkt { - lossInfo = append(lossInfo, minPkt.Seq&0x7FFFFFFF) - } else { - lossInfo = append(lossInfo, minPkt.Seq|0x80000000, lastPkt.Seq&0x7FFFFFFF) - } + for n := uint32(0); n < count; n++ { + lossInfo = append(lossInfo, (sequenceFrom+n)&0x7FFFFFFF) } s.sendPacket <- &packet.NakPacket{CmpLossInfo: lossInfo} diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index efd22c3..1c1b155 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -1,7 +1,6 @@ package udt import ( - "container/heap" "fmt" "time" @@ -30,19 +29,19 @@ type udtSocketSend struct { shutdownEvent chan<- shutdownMessage // channel signals the connection to be shutdown socket *udtSocket - sendState sendState // current sender state - 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 - expCount uint // number of continuous EXP timeouts. - lastRecvTime time.Time // the last time we've heard something from the remote system - recvAckSeq packet.PacketID // largest packetID we've received an ACK from - sendLossList packetIDHeap // loss list - sndPeriod atomicDuration // (set by congestion control) delay between sending packets - rtoPeriod atomicDuration // (set by congestion control) override of EXP timer calculations - congestWindow atomicUint32 // (set by congestion control) size of the current congestion window (in packets) - flowWindowSize uint // negotiated maximum number of unacknowledged packets (in packets) + sendState sendState // current sender state + 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 + expCount uint // number of continuous EXP timeouts. + lastRecvTime time.Time // the last time we've heard something from the remote system + recvAckSeq packet.PacketID // largest packetID we've received an ACK from + sendLossList *receiveLossHeap // loss list + sndPeriod atomicDuration // (set by congestion control) delay between sending packets + rtoPeriod atomicDuration // (set by congestion control) override of EXP timer calculations + congestWindow atomicUint32 // (set by congestion control) size of the current congestion window (in packets) + flowWindowSize uint // negotiated maximum number of unacknowledged packets (in packets) // timers sndEvent <-chan time.Time // if a packet is recently sent, this timer fires when SND completes @@ -62,6 +61,7 @@ func newUdtSocketSend(s *udtSocket) *udtSocketSend { sendPacket: s.sendPacket, shutdownEvent: s.shutdownEvent, sendPktPend: createPacketHeap(), + sendLossList: createPacketIDHeap(), } ss.resetEXP(s.created) go ss.goSendEvent() @@ -243,24 +243,12 @@ 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.Count() == 0 { + if s.sendLossList.Count() == 0 || s.sendPktPend.Count() == 0 { return false } - var dp *sendPacketEntry - for { - minLoss, minLossIdx := s.sendLossList.Min(s.recvAckSeq, s.sendPktSeq) - if minLossIdx < 0 { - // empty loss list? shouldn't really happen as we don't keep empty lists, but check for it anyhow - return false - } - - heap.Remove(&s.sendLossList, minLossIdx) - if len(s.sendLossList) == 0 { - s.sendLossList = nil - } - - dp = s.sendPktPend.Find(minLoss.Seq) + for _, entry := range s.sendLossList.Range(s.recvAckSeq.Seq, s.sendPktSeq.Seq) { + dp := s.sendPktPend.Find(entry.packetID.Seq) if dp == nil { // can't find record of this packet, not much we can do really continue @@ -271,10 +259,9 @@ func (s *udtSocketSend) processSendLoss() bool { continue } - break + s.sendDataPacket(*dp, true) } - s.sendDataPacket(*dp, true) return true } @@ -307,14 +294,7 @@ func (s *udtSocketSend) processSendExpire() bool { dropMsg.LastSeq = p.pkt.Seq } } - if s.sendLossList != nil { - if _, slIdx := s.sendLossList.Find(p.pkt.Seq); slIdx >= 0 { - heap.Remove(&s.sendLossList, slIdx) - } - } - } - if s.sendLossList != nil && len(s.sendLossList) == 0 { - s.sendLossList = nil + s.sendLossList.Remove(p.pkt.Seq.Seq) } s.sendPacket <- dropMsg @@ -395,23 +375,12 @@ func (s *udtSocketSend) ingestAck(p *packet.AckPacket, now time.Time) { 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 { - for { - minLoss, minLossIdx := s.sendLossList.Min(oldAckSeq, s.sendPktSeq) - if p.PktSeqHi.BlindDiff(minLoss) >= 0 || minLossIdx < 0 { - break - } - heap.Remove(&s.sendLossList, minLossIdx) - } - if len(s.sendLossList) == 0 { - s.sendLossList = nil - } - } + s.sendLossList.RemoveRange(oldAckSeq.Seq, p.PktSeqHi.Seq) } // ingestNak is called to process an NAK packet func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) { - newLossList := make([]packet.PacketID, 0) + var lossList []packet.PacketID clen := len(p.CmpLossInfo) for idx := 0; idx < clen; idx++ { thisEntry := p.CmpLossInfo[idx] @@ -437,28 +406,20 @@ func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) { } idx++ for span := thisPktID; span != lastPktID; span.Incr() { - newLossList = append(newLossList, span) + s.sendLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: span.Seq}}) + lossList = append(lossList, packet.PacketID{Seq: span.Seq}) } } else { thisPktID := packet.PacketID{Seq: thisEntry} if !s.assertValidSentPktID("NAK", thisPktID) { return } - newLossList = append(newLossList, thisPktID) + s.sendLossList.Add(recvLossEntry{packetID: thisPktID}) + lossList = append(lossList, thisPktID) } } - s.socket.cong.onNAK(newLossList) - - if s.sendLossList == nil { - s.sendLossList = newLossList - heap.Init(&s.sendLossList) - } else { - llen := len(newLossList) - for idx := 0; idx < llen; idx++ { - heap.Push(&s.sendLossList, newLossList[idx]) - } - } + s.socket.cong.onNAK(lossList) s.sendState = sendStateProcessDrop // immediately restart transmission } @@ -503,14 +464,11 @@ 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.Count() > 0 { - if s.sendLossList == nil { + if s.sendLossList.Count() == 0 { // 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() { - newLossList = append(newLossList, span) + s.sendLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: span.Seq}}) } - s.sendLossList = newLossList - heap.Init(&s.sendLossList) } s.socket.cong.onTimeout() s.sendState = sendStateProcessDrop // immediately restart transmission