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

@@ -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)
 }

udt/packetid_heap.go

@@ -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
-}

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
 }

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)
 }
 

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,
 	}
 }
 

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}

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