From 2924f0074dd7217160580d57ef274db973dbbb91 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 8 Nov 2021 00:14:56 +0100 Subject: [PATCH] UDT: Redeveloped receiving code to fix the packet reassembly bugs. --- udt/packet/pktseq.go | 20 +++ udt/readme.md | 4 + udt/recvloss_heap.go | 20 +-- udt/sendpacket_heap.go | 8 +- udt/udtsocket_recv.go | 315 ++++++++++++++++++++--------------------- udt/udtsocket_send.go | 4 +- 6 files changed, 188 insertions(+), 183 deletions(-) diff --git a/udt/packet/pktseq.go b/udt/packet/pktseq.go index 130a634..3ecf6e0 100644 --- a/udt/packet/pktseq.go +++ b/udt/packet/pktseq.go @@ -31,6 +31,26 @@ func (p PacketID) BlindDiff(rhs PacketID) int32 { return int32(result) } +// IsBiggerEqual checks if the current packet sequence is bigger or equal than the parameter +func (p PacketID) IsBiggerEqual(other PacketID) bool { + return p.BlindDiff(other) >= 0 +} + +// IsBigger checks if the current packet sequence is bigger than the parameter +func (p PacketID) IsBigger(other PacketID) bool { + return p.BlindDiff(other) > 0 +} + +// IsLessEqual checks if the current packet sequence is less or equal than the parameter +func (p PacketID) IsLessEqual(other PacketID) bool { + return p.BlindDiff(other) <= 0 +} + +// IsLess checks if the current packet sequence is less than the parameter +func (p PacketID) IsLess(other PacketID) bool { + return p.BlindDiff(other) < 0 +} + // RandomPacketSequence returns a random packet sequence func RandomPacketSequence() PacketID { return PacketID{rand.Uint32() & 0x7FFFFFFF} diff --git a/udt/readme.md b/udt/readme.md index 9bd8133..4c3b6f6 100644 --- a/udt/readme.md +++ b/udt/readme.md @@ -29,6 +29,8 @@ From `udtSocket.Read`: // fill up the passed buffer as far as we can without blocking again ``` +According to `DataPacket.SetMessageData`, datagram messages do not set the order flag (bit 29). + ## Deviations MTU negotiation is disabled. Peernet uses a hardcoded max packet size (see protocol package). Packets may be routed through any network adapter, therefore pinning a MTU specific to a network adapter would not make much sense. @@ -36,3 +38,5 @@ MTU negotiation is disabled. Peernet uses a hardcoded max packet size (see proto The "rendezvous" functionality has been removed since Peernet supports native Traverse messages for UDP hole punching. Multiplexing multiple UDT sockets to a single UDT connection is removed. It added complexity without benefits in this case. Peernet uses a single UDP port and UDP connection between two peers. Multiplexing has no effect other than breaking the concept and the security of Peernet message sequences. + +The order flag (bit 29) set for datagram messages is ignored for security reasons; the behavior whether incoming packets must be ordered or not is hardcoded to whether it is in streaming or datagram mode. diff --git a/udt/recvloss_heap.go b/udt/recvloss_heap.go index 837d619..91ba93d 100644 --- a/udt/recvloss_heap.go +++ b/udt/recvloss_heap.go @@ -74,31 +74,15 @@ func (heap *receiveLossHeap) Find(sequence uint32) (result *recvLossEntry) { 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 - } - } - } - - return result -} - // RemoveRange removes all packets that are within the given range. Check is from >= and to <. -func (heap *receiveLossHeap) RemoveRange(sequenceFrom, sequenceTo uint32) { +func (heap *receiveLossHeap) RemoveRange(sequenceFrom, sequenceTo packet.PacketID) { 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) { + if !(heap.list[n].packetID.IsBiggerEqual(sequenceFrom) && heap.list[n].packetID.IsLess(sequenceTo)) { newList = append(newList, heap.list[n]) } } diff --git a/udt/sendpacket_heap.go b/udt/sendpacket_heap.go index cd8e175..39aa81e 100644 --- a/udt/sendpacket_heap.go +++ b/udt/sendpacket_heap.go @@ -73,14 +73,14 @@ func (heap *sendPacketHeap) Find(sequence uint32) (result *sendPacketEntry) { } // RemoveRange removes all packets that are within the given range. Check is from >= and to <. -func (heap *sendPacketHeap) RemoveRange(sequenceFrom, sequenceTo uint32) { +func (heap *sendPacketHeap) RemoveRange(sequenceFrom, sequenceTo packet.PacketID) { 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) { + if !(heap.list[n].pkt.Seq.IsBiggerEqual(sequenceFrom) && heap.list[n].pkt.Seq.IsLess(sequenceTo)) { newList = append(newList, heap.list[n]) } } @@ -89,12 +89,12 @@ func (heap *sendPacketHeap) RemoveRange(sequenceFrom, sequenceTo uint32) { } // Range returns all packets that are within the given range. Check is from >= and to <. -func (heap *sendPacketHeap) Range(sequenceFrom, sequenceTo uint32) (result []sendPacketEntry) { +func (heap *sendPacketHeap) Range(sequenceFrom, sequenceTo packet.PacketID) (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 { + if heap.list[n].pkt.Seq.IsBiggerEqual(sequenceFrom) && heap.list[n].pkt.Seq.IsLess(sequenceTo) { result = append(result, heap.list[n]) } } diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index d15e243..4045817 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -14,8 +14,8 @@ 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) + nextSequenceExpect packet.PacketID // the peer's next largest packet ID expected. + lastSequence 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. @@ -29,10 +29,7 @@ type udtSocketRecv struct { 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 - ackSentEvent <-chan time.Time // if an ACK packet has recently sent, wait before resending it + ackLinkInfoSent time.Time // when link info was sent in ACK packet last time } func newUdtSocketRecv(s *udtSocket) *udtSocketRecv { @@ -51,8 +48,8 @@ func newUdtSocketRecv(s *udtSocket) *udtSocketRecv { } func (s *udtSocketRecv) configureHandshake(p *packet.HandshakePacket) { - s.farNextPktSeq = p.InitPktSeq - s.farRecdPktSeq = p.InitPktSeq.Add(-1) + s.nextSequenceExpect = p.InitPktSeq + s.lastSequence = p.InitPktSeq.Add(-1) s.sentAck = p.InitPktSeq s.recvAck2 = p.InitPktSeq } @@ -78,10 +75,6 @@ func (s *udtSocketRecv) goReceiveEvent() { } case _, _ = <-sockClosed: // socket is closed, leave now return - case <-s.ackSentEvent: - s.ackSentEvent = nil - case <-s.ackSentEvent2: - s.ackSentEvent2 = nil } } } @@ -126,6 +119,7 @@ func (s *udtSocketRecv) ingestAck2(p *packet.Ack2Packet, now time.Time) { } // ingestMsgDropReq is called to process an message drop request packet +// This function only makes sense for datagram messages that are OK to be lost. For streaming a file, this makes no sense. func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Time) { stopSeq := p.LastSeq.Add(1) for pktID := p.FirstSeq; pktID != stopSeq; pktID.Incr() { @@ -136,15 +130,15 @@ func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Ti s.recvPktPend.Remove(pktID.Seq) } - if p.FirstSeq == s.farRecdPktSeq.Add(1) { - s.farRecdPktSeq = p.LastSeq + if p.FirstSeq == s.lastSequence.Add(1) { + s.lastSequence = p.LastSeq + } + if s.recvLossList.Count() == 0 { + s.lastSequence = s.nextSequenceExpect.Add(-1) } - // 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) { + for _, nextPkt := range s.recvPktPend.Range(stopSeq, s.nextSequenceExpect) { if !s.attemptProcessPacket(nextPkt.pkt, false) { break } @@ -155,12 +149,10 @@ func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Ti func (s *udtSocketRecv) ingestData(p *packet.DataPacket, now time.Time) { s.socket.cong.onPktRecv(*p) - seq := p.Seq - /* If the sequence number of the current data packet is 16n + 1, where n is an integer, record the time interval between this packet and the last data packet in the Packet Pair Window. */ - if (seq.Seq-1)&0xf == 0 { + if (p.Seq.Seq-1)&0xf == 0 { if !s.recvLastProbe.IsZero() { if s.recvPktPairHistory == nil { s.recvPktPairHistory = []time.Duration{now.Sub(s.recvLastProbe)} @@ -187,135 +179,45 @@ func (s *udtSocketRecv) ingestData(p *packet.DataPacket, now time.Time) { } s.recvLastArrival = now - /* If the sequence number of the current data packet is greater - than LRSN + 1, put all the sequence numbers between (but - excluding) these two values into the receiver's loss list and - send them to the sender in an NAK packet. */ - seqDiff := seq.BlindDiff(s.farNextPktSeq) + // If the incoming sequence number is greater than the expected one, treat all sequence numbers in the middle as lost (add to lost list) and send a NAK. + seqDiff := p.Seq.BlindDiff(s.nextSequenceExpect) if seqDiff > 0 { + // Sequence is out of order. Received a higher sequence number than what is expected next. for n := uint32(0); n < uint32(seqDiff); n++ { - s.recvLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: (s.farNextPktSeq.Seq + n) & 0x7FFFFFFF}}) + s.recvLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: (s.nextSequenceExpect.Seq + n) & 0x7FFFFFFF}}) } - s.sendNAK(s.farNextPktSeq.Seq, uint32(seqDiff)) - s.farNextPktSeq = seq.Add(1) + s.sendNAK(s.nextSequenceExpect.Seq, uint32(seqDiff)) + s.nextSequenceExpect = p.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.Seq) { + if !s.recvLossList.Remove(p.Seq.Seq) { return // already previously received packet -- ignore } - - if s.recvLossList.Count() == 0 { - s.farRecdPktSeq = s.farNextPktSeq.Add(-1) - } else { - 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) + s.nextSequenceExpect = p.Seq.Add(1) + } + + if s.socket.isDatagram && p.Seq == s.lastSequence.Add(1) { + s.lastSequence = p.Seq + s.ackEvent() // Need special sending for datagram, otherwise below code would only send it out after all pieces are received. } s.attemptProcessPacket(p, true) } func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) bool { - seq := p.Seq + var pieces []*packet.DataPacket + var success bool - // can we process this packet? - boundary, mustOrder, msgID := p.GetMessageData() - 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}) - } - return false + if s.socket.isDatagram { + pieces, success = s.reassemblePacketPiecesDatagram(p) + } else { + pieces, success = s.reassemblePacketPiecesStream(p) } - // can we find the start of this message? - pieces := make([]*packet.DataPacket, 0) - cannotContinue := false - switch boundary { - case packet.MbLast, packet.MbMiddle: - // we need prior packets, let's make sure we have them - if s.recvPktPend.Count() > 0 { - pieceSeq := seq.Add(-1) - for { - prevPiece := s.recvPktPend.Find(pieceSeq.Seq) - if prevPiece == nil { - // we don't have the previous piece, is it missing? - 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 - break - } - prevBoundary, _, prevMsg := prevPiece.pkt.GetMessageData() - if prevMsg != msgID { - // ...oops? previous piece isn't in the same message - break - } - pieces = append([]*packet.DataPacket{prevPiece.pkt}, pieces...) - if prevBoundary == packet.MbFirst { - break - } - pieceSeq.Decr() - } - } - } - if !cannotContinue { - pieces = append(pieces, p) - - switch boundary { - case packet.MbFirst, packet.MbMiddle: - // we need following packets, let's make sure we have them - if s.recvPktPend.Count() > 0 { - pieceSeq := seq.Add(1) - for { - nextPiece := s.recvPktPend.Find(pieceSeq.Seq) - if nextPiece == nil { - // we don't have the previous piece, is it missing? - if pieceSeq == s.farNextPktSeq { - // hasn't been received yet - cannotContinue = true - } else if s.recvLossList.Count() > 0 { - if s.recvLossList.Find(pieceSeq.Seq) != nil { - // it's missing, stop processing - cannotContinue = true - } - } else { - } - // in any case we can't continue with this - break - } - nextBoundary, _, nextMsg := nextPiece.pkt.GetMessageData() - if nextMsg != msgID { - // ...oops? previous piece isn't in the same message - break - } - pieces = append(pieces, nextPiece.pkt) - if nextBoundary == packet.MbLast { - break - } - } - } - } - } - - // Acknowledge the packet if the threshold is reached. This used to be a parameter s.ackInterval supposed to be set by congestion control, but never was. - // 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++ - // DEBUG: Always send ack for now - //if s.unackPktCount >= s.socket.cong.GetCongestionWindowSize()/4 { - s.ackEvent() - //} - - if cannotContinue { + if !success { // we need to wait for more packets, store and return if isNew { s.recvPktPend.Add(sendPacketEntry{pkt: p}) @@ -323,14 +225,18 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b 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.Count() > 0 { + // If pieces were pulled from the list of packets that were waiting to be processed, remove it now. + if len(pieces) > 1 { for _, piece := range pieces { s.recvPktPend.Remove(piece.Seq.Seq) } } - msg := make([]byte, 0) + s.lastSequence = pieces[len(pieces)-1].Seq + s.ackEvent() + + // reassemble the data by appending it from all the pieces + var msg []byte for _, piece := range pieces { msg = append(msg, piece.Data...) } @@ -338,6 +244,96 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b return true } +// reassemblePacketPiecesDatagram attempts to reassemble a datagram message from multiple pieces +func (s *udtSocketRecv) reassemblePacketPiecesDatagram(p *packet.DataPacket) (pieces []*packet.DataPacket, success bool) { + boundary, _, msgID := p.GetMessageData() + + // First check if prior packets are needed. + switch boundary { + case packet.MbLast, packet.MbMiddle: + pieceSeq := p.Seq.Add(-1) + for { + prevPiece := s.recvPktPend.Find(pieceSeq.Seq) + if prevPiece == nil { + // we don't have the previous piece, is it missing? + if s.recvLossList.Find(pieceSeq.Seq) != nil { + // it's missing, stop processing + return nil, false + } else { + } + // in any case we can't continue with this + return nil, false + } + prevBoundary, _, prevMsg := prevPiece.pkt.GetMessageData() + if prevMsg != msgID { + // ...oops? previous piece isn't in the same message + return nil, false + } + pieces = append([]*packet.DataPacket{prevPiece.pkt}, pieces...) + if prevBoundary == packet.MbFirst { + break + } + pieceSeq.Decr() + } + } + + pieces = append(pieces, p) + + // If more packets are needed, make sure they are available. + switch boundary { + case packet.MbFirst, packet.MbMiddle: + pieceSeq := p.Seq.Add(1) + for { + nextPiece := s.recvPktPend.Find(pieceSeq.Seq) + if nextPiece == nil { + // we don't have the previous piece, is it missing? + if pieceSeq == s.nextSequenceExpect { + // hasn't been received yet + return nil, false + } else if s.recvLossList.Find(pieceSeq.Seq) != nil { + // it's missing, stop processing + return nil, false + } else { + } + // in any case we can't continue with this + return nil, false + } + nextBoundary, _, nextMsg := nextPiece.pkt.GetMessageData() + if nextMsg != msgID { + // ...oops? previous piece isn't in the same message + break + } + pieces = append(pieces, nextPiece.pkt) + if nextBoundary == packet.MbLast { + break + } + } + } + + return pieces, true +} + +// reassemblePacketPiecesStream tries to see if all remaining packets since the last verified one are buffered (as well as immediately following ones). +func (s *udtSocketRecv) reassemblePacketPiecesStream(p *packet.DataPacket) (pieces []*packet.DataPacket, success bool) { + // for streams this can continue only if the incoming packet is immediately the next one + if p.Seq != s.lastSequence.Add(1) { + return nil, false + } + + pieces = append(pieces, p) + + // find any other packets that are already buffered + for nextSeq := p.Seq.Add(1); ; nextSeq.Incr() { + if nextPacket := s.recvPktPend.Find(nextSeq.Seq); nextPacket != nil { + pieces = append(pieces, nextPacket.pkt) + } else { + break + } + } + + return pieces, true +} + func (s *udtSocketRecv) getRcvSpeeds() (recvSpeed, bandwidth int) { // get median value, but cannot change the original value order in the window @@ -403,25 +399,8 @@ func (s *udtSocketRecv) getRcvSpeeds() (recvSpeed, bandwidth int) { return } -func (s *udtSocketRecv) sendACK() { - var ack packet.PacketID - - // 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.Count() == 0 { - ack = s.farNextPktSeq - } else { - ack = s.farRecdPktSeq.Add(1) - } - - if ack == s.recvAck2 { - return - } - - // only send out an ACK if we either are saying something new or the ackSentEvent has expired - if ack == s.sentAck && s.ackSentEvent != nil { - return - } +// sendACK sends an ACK with the given sequence number. +func (s *udtSocketRecv) sendACK(ack packet.PacketID) { s.sentAck = ack s.lastACK++ @@ -433,7 +412,7 @@ func (s *udtSocketRecv) sendACK() { rtt, rttVar := s.socket.getRTT() - numPendPackets := int(s.farNextPktSeq.BlindDiff(s.farRecdPktSeq) - 1) + numPendPackets := int(s.nextSequenceExpect.BlindDiff(s.lastSequence) - 1) availWindow := int(s.socket.maxFlowWinSize) - numPendPackets if availWindow < 2 { availWindow = 2 @@ -446,15 +425,16 @@ func (s *udtSocketRecv) sendACK() { RttVar: uint32(rttVar), BuffAvail: uint32(availWindow), } - if s.ackSentEvent2 == nil { + + // Send the link info only every SynTime. In theory this should use a mutex, but it does not matter if the link info is sent out multiple times. + if s.ackLinkInfoSent.IsZero() || time.Since(s.ackLinkInfoSent) >= s.socket.Config.SynTime { + s.ackLinkInfoSent = time.Now() recvSpeed, bandwidth := s.getRcvSpeeds() p.IncludeLink = true p.PktRecvRate = uint32(recvSpeed) p.EstLinkCap = uint32(bandwidth) - s.ackSentEvent2 = time.After(s.socket.Config.SynTime) } s.sendPacket <- p - s.ackSentEvent = time.After(time.Duration(rtt+4*rttVar) * time.Microsecond) } func (s *udtSocketRecv) sendNAK(sequenceFrom uint32, count uint32) { @@ -471,8 +451,25 @@ func (s *udtSocketRecv) ingestError(p *packet.ErrPacket) { // TODO: umm something } -// assuming some condition has occured (ACK timer expired, ACK interval), send an ACK and reset the appropriate timer +// ackEvent sends an ACK message if appropriate. It informs the remote peer about the last packet received without loss. func (s *udtSocketRecv) ackEvent() { - s.sendACK() + // Acknowledge the packet if the threshold is reached. This used to be a parameter s.ackInterval supposed to be set by congestion control, but never was. + // 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++ + // DEBUG: Always send ack for now. Turns out the remote congestion window changes without the local one? + //if s.unackPktCount < s.socket.cong.GetCongestionWindowSize()/4 { + // return + //} + + // The ack number is excluding. + ack := s.lastSequence.Add(1) + + // Only send out the ACK if it represents new information to the remote, i.e. bigger than the last reported number. + if ack.IsLessEqual(s.sentAck) { + return + } + + s.sendACK(ack) s.unackPktCount = 0 } diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 1c1b155..3482bf4 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -372,10 +372,10 @@ 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 - s.sendPktPend.RemoveRange(oldAckSeq.Seq, p.PktSeqHi.Seq) + s.sendPktPend.RemoveRange(oldAckSeq, p.PktSeqHi) // Update sender's loss list (by removing all those that has been acknowledged). - s.sendLossList.RemoveRange(oldAckSeq.Seq, p.PktSeqHi.Seq) + s.sendLossList.RemoveRange(oldAckSeq, p.PktSeqHi) } // ingestNak is called to process an NAK packet