UDT: Revert removal of udtSocketRecv.ackInterval from b728b939ec close #55

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

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

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
 	}
 

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

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)