diff --git a/udt/congestion.go b/udt/congestion.go index 3914d6e..57ab096 100644 --- a/udt/congestion.go +++ b/udt/congestion.go @@ -35,8 +35,11 @@ type CongestionControlParms interface { // GetMSS is the largest packet size we can currently send (in bytes) GetMSS() uint - // SetACKPerid sets the time between ACKs sent to the peer + // SetACKPeriod sets the time between ACKs sent to the peer SetACKPeriod(time.Duration) + + // SetACKInterval sets the number of packets sent to the peer before sending an ACK + SetACKInterval(uint) } // CongestionControl controls how timing is handled and UDT connections tuned diff --git a/udt/congestion_native.go b/udt/congestion_native.go index a1e92da..84506b6 100644 --- a/udt/congestion_native.go +++ b/udt/congestion_native.go @@ -29,6 +29,9 @@ func (ncc NativeCongestionControl) Init(parms CongestionControlParms, synTime ti ncc.lastRCTime = time.Now() parms.SetACKPeriod(ncc.rcInterval) + // TODO: Once packet loss is reported, the ACK interval should fall back to 1, and increased if a stable connection is detected. + parms.SetACKInterval(4) + ncc.slowStart = true ncc.lastAck = parms.GetSndCurrSeqNo() ncc.loss = false diff --git a/udt/multiplexer.go b/udt/multiplexer.go index ca7aa96..a6789d2 100644 --- a/udt/multiplexer.go +++ b/udt/multiplexer.go @@ -3,7 +3,6 @@ package udt import ( - "fmt" "math/rand" "github.com/PeernetOfficial/core/udt/packet" @@ -57,7 +56,7 @@ func (m *multiplexer) goRead() { p, err := packet.DecodePacket(buf) if err != nil { - fmt.Printf("Error decoding UDT packet: %s\n", err) + //fmt.Printf("Error decoding UDT packet: %s\n", err) return } @@ -67,7 +66,7 @@ func (m *multiplexer) goRead() { var hsPacket *packet.HandshakePacket var ok bool if hsPacket, ok = p.(*packet.HandshakePacket); !ok { - fmt.Printf("Received non-handshake packet with destination socket = 0\n") + //fmt.Printf("Received non-handshake packet with destination socket = 0\n") return } @@ -85,7 +84,7 @@ func (m *multiplexer) sendPacket(destSockID uint32, ts uint32, p packet.Packet) p.SetHeader(destSockID, ts) if destSockID == 0 { if _, ok := p.(*packet.HandshakePacket); !ok { - fmt.Printf("Sending non-handshake packet with destination socket = 0\n") + //fmt.Printf("Sending non-handshake packet with destination socket = 0\n") return } } @@ -93,7 +92,7 @@ func (m *multiplexer) sendPacket(destSockID uint32, ts uint32, p packet.Packet) buf := make([]byte, m.maxPacketSize) plen, err := p.WriteTo(buf) // encode if err != nil { - fmt.Printf("Error encoding UDT packet: %s\n", err.Error()) + //fmt.Printf("Error encoding UDT packet: %s\n", err.Error()) return } diff --git a/udt/udtsocket_cc.go b/udt/udtsocket_cc.go index 3b050f2..2d1fd95 100644 --- a/udt/udtsocket_cc.go +++ b/udt/udtsocket_cc.go @@ -212,3 +212,8 @@ func (s *udtSocketCc) GetMSS() uint { func (s *udtSocketCc) SetACKPeriod(ack time.Duration) { s.socket.recv.ackPeriod.set(ack) } + +// SetACKInterval sets the number of packets sent to the peer before sending an ACK +func (s *udtSocketCc) SetACKInterval(ack uint) { + s.socket.recv.ackInterval.set(uint32(ack)) +} diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index 36f9efa..87164ca 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -24,7 +24,8 @@ type udtSocketRecv struct { 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 + ackPeriod atomicDuration // (set by congestion control) delay between sending ACKs. Currently not used. + ackInterval atomicUint32 // (set by congestion control) number of data packets to send before sending an ACK 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. @@ -464,14 +465,13 @@ func (s *udtSocketRecv) ingestError(p *packet.ErrPacket) { // ackEvent sends an ACK message if appropriate. It informs the remote peer about the last packet received without loss. func (s *udtSocketRecv) ackEvent() { - // 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 - //} + + // Check if the threshold to send is reached, if used. Note that sendACK is called revery SynTime. + ackInterval := uint(s.ackInterval.get()) + if (ackInterval > 0) && (ackInterval > s.unackPktCount) { + return + } // The ack number is excluding. ack := s.lastSequence.Add(1)