UDT: Fix the first piece of !@#$ heap code. Rewrite from scratch.

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
 }

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() {