From 014d4a34eef9266a7b40d79cfdb902df787eb121 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 8 Nov 2021 19:26:53 +0100 Subject: [PATCH] UDT: Fix in receiveLossHeap.Range --- udt/listener.go | 11 +++++------ udt/recvloss_heap.go | 4 ++-- udt/udtsocket_send.go | 15 ++++++++------- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/udt/listener.go b/udt/listener.go index 882b9ea..129aea6 100644 --- a/udt/listener.go +++ b/udt/listener.go @@ -4,7 +4,6 @@ import ( "container/heap" "encoding/binary" "errors" - "fmt" "net" "sync" "time" @@ -62,7 +61,7 @@ func (l *listener) checkValidHandshake(m *multiplexer, p *packet.HandshakePacket } func (l *listener) rejectHandshake(m *multiplexer, hsPacket *packet.HandshakePacket) { - fmt.Printf("(listener) sending handshake(reject) (id=%d)\n", hsPacket.SockID) + //fmt.Printf("(listener) sending handshake(reject) (id=%d)\n", hsPacket.SockID) m.sendPacket(hsPacket.SockID, 0, &packet.HandshakePacket{ UdtVer: hsPacket.UdtVer, SockType: hsPacket.SockType, @@ -72,7 +71,7 @@ func (l *listener) rejectHandshake(m *multiplexer, hsPacket *packet.HandshakePac func (l *listener) readHandshake(m *multiplexer, hsPacket *packet.HandshakePacket) bool { if hsPacket.ReqType == packet.HsRequest { - fmt.Printf("(listener) sending handshake(request) (id=%d)\n", hsPacket.SockID) + //fmt.Printf("(listener) sending handshake(request) (id=%d)\n", hsPacket.SockID) m.sendPacket(hsPacket.SockID, 0, &packet.HandshakePacket{ UdtVer: hsPacket.UdtVer, @@ -111,19 +110,19 @@ func (l *listener) readHandshake(m *multiplexer, hsPacket *packet.HandshakePacke l.acceptHistProt.Unlock() if !l.config.CanAcceptDgram && hsPacket.SockType == packet.TypeDGRAM { - fmt.Printf("Refusing new socket creation from listener requesting DGRAM\n") + //fmt.Printf("Refusing new socket creation from listener requesting DGRAM\n") l.rejectHandshake(m, hsPacket) return false } if !l.config.CanAcceptStream && hsPacket.SockType == packet.TypeSTREAM { - fmt.Printf("Refusing new socket creation from listener requesting STREAM\n") + //fmt.Printf("Refusing new socket creation from listener requesting STREAM\n") l.rejectHandshake(m, hsPacket) return false } if l.config.CanAccept != nil { err := l.config.CanAccept(hsPacket) if err != nil { - fmt.Printf("New socket creation from listener rejected by config: %s\n", err.Error()) + //fmt.Printf("New socket creation from listener rejected by config: %s\n", err.Error()) l.rejectHandshake(m, hsPacket) return false } diff --git a/udt/recvloss_heap.go b/udt/recvloss_heap.go index 91ba93d..6f38a54 100644 --- a/udt/recvloss_heap.go +++ b/udt/recvloss_heap.go @@ -91,12 +91,12 @@ func (heap *receiveLossHeap) RemoveRange(sequenceFrom, sequenceTo packet.PacketI } // Range returns all packets that are within the given range. Check is from >= and to <. -func (heap *receiveLossHeap) Range(sequenceFrom, sequenceTo uint32) (result []recvLossEntry) { +func (heap *receiveLossHeap) Range(sequenceFrom, sequenceTo packet.PacketID) (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 { + if heap.list[n].packetID.IsBiggerEqual(sequenceFrom) && heap.list[n].packetID.IsLess(sequenceTo) { result = append(result, heap.list[n]) } } diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index c662262..b09b5eb 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -238,7 +238,7 @@ func (s *udtSocketSend) processSendLoss() bool { return false } - for _, entry := range s.sendLossList.Range(s.recvAckSeq.Seq, s.sendPktSeq.Seq) { + for _, entry := range s.sendLossList.Range(s.recvAckSeq, s.sendPktSeq) { dp := s.sendPktPend.Find(entry.packetID.Seq) if dp == nil { // can't find record of this packet, not much we can do really @@ -372,12 +372,13 @@ func (s *udtSocketSend) ingestAck(p *packet.AckPacket, now time.Time) { // ingestNak is called to process an NAK packet func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) { var lossList []packet.PacketID - clen := len(p.CmpLossInfo) - for idx := 0; idx < clen; idx++ { - thisEntry := p.CmpLossInfo[idx] + + for n := 0; n < len(p.CmpLossInfo); n++ { + thisEntry := p.CmpLossInfo[n] + if thisEntry&0x80000000 != 0 { thisPktID := packet.PacketID{Seq: thisEntry & 0x7FFFFFFF} - if idx+1 == clen { + if n+1 == len(p.CmpLossInfo) { s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false, err: fmt.Errorf("FAULT: While unpacking a NAK, the last entry (%x) was describing a start-of-range", thisEntry), reason: TerminateReasonCorruptPacketNak} return @@ -385,7 +386,7 @@ func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) { if !s.assertValidSentPktID("NAK", thisPktID, TerminateReasonInvalidPacketIDNak) { return } - lastEntry := p.CmpLossInfo[idx+1] + lastEntry := p.CmpLossInfo[n+1] if lastEntry&0x80000000 != 0 { s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false, err: fmt.Errorf("FAULT: While unpacking a NAK, a start-of-range (%x) was followed by another start-of-range (%x)", thisEntry, lastEntry), reason: TerminateReasonCorruptPacketNak} @@ -395,7 +396,7 @@ func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) { if !s.assertValidSentPktID("NAK", lastPktID, TerminateReasonInvalidPacketIDNak) { return } - idx++ + n++ for span := thisPktID; span != lastPktID; span.Incr() { s.sendLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: span.Seq}}) lossList = append(lossList, packet.PacketID{Seq: span.Seq})