gbn: support dynamic resend timeouts

This commit adds support for setting the resend timeout dynamically for
a connection. Prior to this commit, the timeout before a client or
server resends the queue of packets, was always set to fixed value.
A fixed timeout isn't suitable for all connections though, as the
latency for different clients varies.

With this commit, we instead add support for setting the resend timeout
based on:
* How long it took for the other party to respond during the
handshake process
* How long it took for the other party to respond with the connect ACK
for a sent data packet.

The timeout is set then to the time it took for the server to respond,
multiplied by the resendMultiplier, unless the duration is shorter than
the default resend timeout.

Note though that if a connection's resend timeout is set manually set
through the WithStaticResendTimeout, the resend timeout will always be
set to that value, and won't be dynamically updated.

Author: ViktorTigerstrom

gbn/gbn_client.go

@@ -99,6 +99,7 @@ func (g *GoBackNConn) clientHandshake() error {
 	var (
 		resp    Message
 		respSYN *PacketSYN
+		resent  bool
 	)
 handshake:
 	for {
@@ -115,6 +116,9 @@ handshake:
 			return err
 		}
 
+		// Notify the timeout manager that we sent a SYN.
+		g.timeoutManager.Sent(msg, resent)
+
 		for {
 			// Wait for SYN
 			g.log.Debugf("Waiting for SYN")
@@ -135,6 +139,7 @@ handshake:
 			case <-time.After(timeout):
 				g.log.Debugf("SYN resendTimeout. Resending " +
 					"SYN.")
+				resent = true
 
 				continue handshake
 			case <-g.quit:
@@ -173,6 +178,10 @@ handshake:
 		return io.EOF
 	}
 
+	// Notify the timeout manager we've received the SYN response from the
+	// counterparty.
+	g.timeoutManager.Received(resp)
+
 	// Send SYNACK
 	g.log.Debugf("Sending SYNACK")
 	synack, err := new(PacketSYNACK).Serialize()

gbn/gbn_conn.go

@@ -104,7 +104,7 @@ func newGoBackNConn(ctx context.Context, cfg *config,
 			s:   cfg.n + 1,
 			log: plog,
 			sendPkt: func(packet *PacketData) error {
-				return g.sendPacket(g.ctx, packet)
+				return g.sendPacket(g.ctx, packet, true)
 			},
 		},
 		timeoutManager,
@@ -134,7 +134,7 @@ func (g *GoBackNConn) setN(n uint8) {
 			s:   n + 1,
 			log: g.log,
 			sendPkt: func(packet *PacketData) error {
-				return g.sendPacket(g.ctx, packet)
+				return g.sendPacket(g.ctx, packet, true)
 			},
 		},
 		g.timeoutManager,
@@ -308,7 +308,9 @@ func (g *GoBackNConn) Close() error {
 				g.ctx, g.timeoutManager.GetFinSendTimeout(),
 			)
 			defer cancel()
-			if err := g.sendPacket(ctxc, &PacketFIN{}); err != nil {
+
+			err := g.sendPacket(ctxc, &PacketFIN{}, false)
+			if err != nil {
 				g.log.Errorf("Error sending FIN: %v", err)
 			}
 		}
@@ -336,7 +338,9 @@ func (g *GoBackNConn) Close() error {
 }
 
 // sendPacket serializes a message and writes it to the underlying send stream.
-func (g *GoBackNConn) sendPacket(ctx context.Context, msg Message) error {
+func (g *GoBackNConn) sendPacket(ctx context.Context, msg Message,
+	isResend bool) error {
+
 	b, err := msg.Serialize()
 	if err != nil {
 		return fmt.Errorf("serialize error: %s", err)
@@ -347,6 +351,9 @@ func (g *GoBackNConn) sendPacket(ctx context.Context, msg Message) error {
 		return fmt.Errorf("error calling sendToStream: %s", err)
 	}
 
+	// Notify the timeout manager that a message has been sent.
+	g.timeoutManager.Sent(msg, isResend)
+
 	return nil
 }
 
@@ -428,7 +435,7 @@ func (g *GoBackNConn) sendPacketsForever() error {
 		g.sendQueue.addPacket(packet)
 
 		g.log.Tracef("Sending data %d", packet.Seq)
-		if err := g.sendPacket(g.ctx, packet); err != nil {
+		if err := g.sendPacket(g.ctx, packet, false); err != nil {
 			return err
 		}
 
@@ -490,6 +497,9 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 			return fmt.Errorf("deserialize error: %s", err)
 		}
 
+		// Notify the timeout manager that a message has been received.
+		g.timeoutManager.Received(msg)
+
 		// Reset the ping & pong timer if any packet is received.
 		// If ping/pong is disabled, this is a no-op.
 		g.pingTicker.Reset()
@@ -514,7 +524,8 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 					Seq: m.Seq,
 				}
 
-				if err = g.sendPacket(g.ctx, ack); err != nil {
+				err = g.sendPacket(g.ctx, ack, false)
+				if err != nil {
 					return err
 				}
 
@@ -573,7 +584,8 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 					Seq: g.recvSeq,
 				}
 
-				if err = g.sendPacket(g.ctx, nack); err != nil {
+				err = g.sendPacket(g.ctx, nack, false)
+				if err != nil {
 					return err
 				}
 

gbn/gbn_server.go

@@ -82,6 +82,7 @@ func (g *GoBackNConn) serverHandshake() error { // nolint:gocyclo
 	}()
 
 	var n uint8
+	var resent bool
 
 	for {
 		g.log.Debugf("Waiting for client SYN")
@@ -131,6 +132,9 @@ func (g *GoBackNConn) serverHandshake() error { // nolint:gocyclo
 			return err
 		}
 
+		// Notify the timeout manager that we sent a SYN.
+		g.timeoutManager.Sent(msg, resent)
+
 		// Wait for SYNACK
 		g.log.Debugf("Waiting for client SYNACK")
 		select {
@@ -146,6 +150,8 @@ func (g *GoBackNConn) serverHandshake() error { // nolint:gocyclo
 		case <-time.After(g.timeoutManager.GetHandshakeTimeout()):
 			g.log.Debugf("SYNCACK resendTimeout. Abort and wait " +
 				"for client to re-initiate")
+			resent = true
+
 			continue
 		case err := <-errChan:
 			return err
@@ -163,9 +169,15 @@ func (g *GoBackNConn) serverHandshake() error { // nolint:gocyclo
 
 		switch msg.(type) {
 		case *PacketSYNACK:
+			// Notify the timeout manager we've received the SYNACK
+			// response from the counterparty.
+			g.timeoutManager.Received(msg)
+
 			break
 		case *PacketSYN:
 			g.log.Debugf("Received SYN. Resend SYN.")
+			resent = true
+
 			goto recvClientSYN
 		default:
 			return io.EOF