gbn: use timeout manager to handle timeouts

This commit moves the responsibility of managing the timeout values
throughout the gbn package to the `TimeoutManager` struct.

Author: ViktorT-11

gbn/gbn_client.go

@@ -120,9 +120,11 @@ handshake:
 			default:
 			}
 
+			timeout := g.timeoutManager.GetHandshakeTimeout()
+
 			var b []byte
 			select {
-			case <-time.After(g.handshakeTimeout):
+			case <-time.After(timeout):
 				log.Debugf("SYN resendTimeout. Resending SYN.")
 				continue handshake
 			case <-g.quit:

gbn/gbn_conn.go

@@ -18,12 +18,7 @@ var (
 )
 
 const (
-	DefaultN                = 20
-	defaultHandshakeTimeout = 100 * time.Millisecond
-	defaultResendTimeout    = 100 * time.Millisecond
-	finSendTimeout          = 1000 * time.Millisecond
-	DefaultSendTimeout      = math.MaxInt64
-	DefaultRecvTimeout      = math.MaxInt64
+	DefaultN = 20
 )
 
 type sendBytesFunc func(ctx context.Context, b []byte) error
@@ -58,30 +53,16 @@ type GoBackNConn struct {
 	// sequence that we have received.
 	recvSeq uint8
 
-	// resendTimeout is the duration that will be waited before resending
-	// the packets in the current queue.
-	resendTimeout time.Duration
-	resendTicker  *time.Ticker
+	resendTicker *time.Ticker
 
 	recvFromStream recvBytesFunc
 	sendToStream   sendBytesFunc
 
 	recvDataChan chan *PacketData
 	sendDataChan chan *PacketData
 
-	sendTimeout   time.Duration
-	sendTimeoutMu sync.RWMutex
-
-	recvTimeout   time.Duration
-	recvTimeoutMu sync.RWMutex
-
 	isServer bool
 
-	// handshakeTimeout is the time after which the server or client
-	// will abort and restart the handshake if the expected response is
-	// not received from the peer.
-	handshakeTimeout time.Duration
-
 	// receivedACKSignal channel is used to signal that the queue size has
 	// been decreased.
 	receivedACKSignal chan struct{}
@@ -103,6 +84,10 @@ type GoBackNConn struct {
 	// remoteClosed is closed if the remote party initiated the FIN sequence.
 	remoteClosed chan struct{}
 
+	// timeoutManager is used to manage all the timeouts used by the
+	// GoBackNConn.
+	timeoutManager *TimeoutManager
+
 	// quit is used to stop the normal operations of the connection.
 	// Once closed, the send and receive streams will still be available
 	// for the FIN sequence.
@@ -121,21 +106,20 @@ func newGoBackNConn(ctx context.Context, sendFunc sendBytesFunc,
 
 	ctxc, cancel := context.WithCancel(ctx)
 
+	timeoutManager := NewTimeOutManager(isServer)
+
 	gbn := &GoBackNConn{
-		resendTimeout:     defaultResendTimeout,
 		recvFromStream:    recvFunc,
 		sendToStream:      sendFunc,
 		sendDataChan:      make(chan *PacketData),
 		isServer:          isServer,
-		handshakeTimeout:  defaultHandshakeTimeout,
-		recvTimeout:       DefaultRecvTimeout,
-		sendTimeout:       DefaultSendTimeout,
 		receivedACKSignal: make(chan struct{}),
 		resendSignal:      make(chan struct{}, 1),
 		remoteClosed:      make(chan struct{}),
 		ctx:               ctxc,
 		cancel:            cancel,
 		quit:              make(chan struct{}),
+		timeoutManager:    timeoutManager,
 	}
 
 	for _, o := range opts {
@@ -147,34 +131,31 @@ func newGoBackNConn(ctx context.Context, sendFunc sendBytesFunc,
 	return gbn
 }
 
+// SetSendTimeout sets the timeout used in the Send function.
+func (g *GoBackNConn) SetSendTimeout(timeout time.Duration) {
+	g.timeoutManager.SetSendTimeout(timeout)
+}
+
+// SetRecvTimeout sets the timeout used in the Recv function.
+func (g *GoBackNConn) SetRecvTimeout(timeout time.Duration) {
+	g.timeoutManager.SetRecvTimeout(timeout)
+}
+
 // setN sets the current N to use. This _must_ be set before the handshake is
 // completed.
 func (g *GoBackNConn) setN(n uint8) {
 	g.n = n
 	g.s = n + 1
 	g.recvDataChan = make(chan *PacketData, n)
-	g.sendQueue = newQueue(&queueConfig{
+
+	cfg := &queueConfig{
 		s: g.s,
 		sendPkt: func(packet *PacketData) error {
 			return g.sendPacket(g.ctx, packet)
 		},
-	})
-}
-
-// SetSendTimeout sets the timeout used in the Send function.
-func (g *GoBackNConn) SetSendTimeout(timeout time.Duration) {
-	g.sendTimeoutMu.Lock()
-	defer g.sendTimeoutMu.Unlock()
-
-	g.sendTimeout = timeout
-}
-
-// SetRecvTimeout sets the timeout used in the Recv function.
-func (g *GoBackNConn) SetRecvTimeout(timeout time.Duration) {
-	g.recvTimeoutMu.Lock()
-	defer g.recvTimeoutMu.Unlock()
+	}
 
-	g.recvTimeout = timeout
+	g.sendQueue = newQueue(cfg, g.timeoutManager)
 }
 
 // Send blocks until an ack is received for the packet sent N packets before.
@@ -186,9 +167,7 @@ func (g *GoBackNConn) Send(data []byte) error {
 	default:
 	}
 
-	g.sendTimeoutMu.RLock()
-	ticker := time.NewTimer(g.sendTimeout)
-	g.sendTimeoutMu.RUnlock()
+	ticker := time.NewTimer(g.timeoutManager.GetSendTimeout())
 	defer ticker.Stop()
 
 	sendPacket := func(packet *PacketData) error {
@@ -247,9 +226,7 @@ func (g *GoBackNConn) Recv() ([]byte, error) {
 		msg *PacketData
 	)
 
-	g.recvTimeoutMu.RLock()
-	ticker := time.NewTimer(g.recvTimeout)
-	g.recvTimeoutMu.RUnlock()
+	ticker := time.NewTimer(g.timeoutManager.GetRecvTimeout())
 	defer ticker.Stop()
 
 	for {
@@ -291,7 +268,7 @@ func (g *GoBackNConn) start() {
 
 	g.pongTicker = NewIntervalAwareForceTicker(pongTime)
 
-	g.resendTicker = time.NewTicker(g.resendTimeout)
+	g.resendTicker = time.NewTicker(g.timeoutManager.GetResendTimeout())
 
 	g.wg.Add(1)
 	go func() {
@@ -347,7 +324,7 @@ func (g *GoBackNConn) Close() error {
 		default:
 			log.Tracef("Try sending FIN, isServer=%v", g.isServer)
 			ctxc, cancel := context.WithTimeout(
-				g.ctx, finSendTimeout,
+				g.ctx, g.timeoutManager.GetFinSendTimeout(),
 			)
 			defer cancel()
 			if err := g.sendPacket(ctxc, &PacketFIN{}); err != nil {
@@ -401,15 +378,15 @@ func (g *GoBackNConn) sendPacket(ctx context.Context, msg Message) error {
 func (g *GoBackNConn) sendPacketsForever() error {
 	// resendQueue re-sends the current contents of the queue.
 	resendQueue := func() error {
-		err := g.sendQueue.resend(g.resendTimeout)
+		err := g.sendQueue.resend()
 
 		// After resending the queue, we reset the resend ticker.
 		// This is so that we don't immediately resend the queue again,
 		// if the sendQueue.resend call above took a long time to
 		// execute. That can happen if the function was awaiting the
 		// expected ACK for a long time, or times out while awaiting the
 		// catch up.
-		g.resendTicker.Reset(g.resendTimeout)
+		g.resendTicker.Reset(g.timeoutManager.GetResendTimeout())
 
 		return err
 	}
@@ -529,7 +506,7 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 			g.pongTicker.Pause()
 		}
 
-		g.resendTicker.Reset(g.resendTimeout)
+		g.resendTicker.Reset(g.timeoutManager.GetResendTimeout())
 
 		switch m := msg.(type) {
 		case *PacketData:
@@ -587,7 +564,9 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 				// the resend, and therefore won't react to the
 				// NACK we send here in time.
 				sinceSent := time.Since(lastNackTime)
-				recentlySent := sinceSent < g.resendTimeout*2 //nolint:gomnd
+
+				timeout := g.timeoutManager.GetResendTimeout()
+				recentlySent := sinceSent < timeout*2 //nolint:gomnd
 
 				if lastNackSeq == g.recvSeq && recentlySent {
 					log.Tracef("Recently sent NACK")
@@ -612,9 +591,7 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 			}
 
 		case *PacketACK:
-			gotValidACK := g.sendQueue.processACK(
-				m.Seq, g.resendTimeout,
-			)
+			gotValidACK := g.sendQueue.processACK(m.Seq)
 
 			if gotValidACK {
 				// Send a signal to indicate that new

gbn/gbn_server.go

@@ -135,7 +135,7 @@ func (g *GoBackNConn) serverHandshake() error { // nolint:gocyclo
 		}
 
 		select {
-		case <-time.After(g.handshakeTimeout):
+		case <-time.After(g.timeoutManager.GetHandshakeTimeout()):
 			log.Debugf("SYNCACK resendTimeout. Abort and wait " +
 				"for client to re-initiate")
 			continue

gbn/options.go

@@ -18,7 +18,7 @@ func WithMaxSendSize(size int) Option {
 // for ACKs before resending the queue.
 func WithTimeout(timeout time.Duration) Option {
 	return func(conn *GoBackNConn) {
-		conn.resendTimeout = timeout
+		conn.timeoutManager.SetStaticResendTimeout(timeout)
 	}
 }
 
@@ -27,7 +27,7 @@ func WithTimeout(timeout time.Duration) Option {
 // will be aborted and restarted.
 func WithHandshakeTimeout(timeout time.Duration) Option {
 	return func(conn *GoBackNConn) {
-		conn.handshakeTimeout = timeout
+		conn.timeoutManager.SetHandshakeTimeout(timeout)
 	}
 }
 

gbn/queue.go

@@ -106,16 +106,19 @@ type queue struct {
 	lastResend time.Time
 
 	quit chan struct{}
+
+	timeoutManager *TimeoutManager
 }
 
 // newQueue creates a new queue.
-func newQueue(cfg *queueConfig) *queue {
+func newQueue(cfg *queueConfig, timeoutManager *TimeoutManager) *queue {
 	return &queue{
 		cfg:               cfg,
 		content:           make([]*PacketData, cfg.s),
 		awaitedACKSignal:  make(chan struct{}, 1),
 		awaitedNACKSignal: make(chan struct{}, 1),
 		quit:              make(chan struct{}),
+		timeoutManager:    timeoutManager,
 	}
 }
 
@@ -205,8 +208,8 @@ func (q *queue) addPacket(packet *PacketData) {
 //
 // When either of the 2 conditions above are met, we will consider both parties
 // to be in sync, and we can proceed to send new packets.
-func (q *queue) resend(resendTimeout time.Duration) error {
-	if time.Since(q.lastResend) < resendTimeout {
+func (q *queue) resend() error {
+	if time.Since(q.lastResend) < q.timeoutManager.GetHandshakeTimeout() {
 		log.Tracef("Resent the queue recently.")
 
 		return nil
@@ -284,7 +287,7 @@ func (q *queue) resend(resendTimeout time.Duration) error {
 	q.awaitingCatchUpMu.Unlock()
 
 	// Then await until we know that both parties are in sync.
-	q.awaitCatchUp(resendTimeout)
+	q.awaitCatchUp()
 
 	return nil
 }
@@ -296,8 +299,10 @@ func (q *queue) resend(resendTimeout time.Duration) error {
 // the awaited ACK or NACK signal.
 //
 //nolint:cyclop
-func (q *queue) awaitCatchUp(resendTimeout time.Duration) {
-	ticker := time.NewTimer(resendTimeout * awaitingTimeoutMultiplier)
+func (q *queue) awaitCatchUp() {
+	ticker := time.NewTimer(
+		q.timeoutManager.GetResendTimeout() * awaitingTimeoutMultiplier,
+	)
 	defer ticker.Stop()
 
 	log.Tracef("Awaiting catchup after resending the queue")
@@ -362,7 +367,7 @@ func (q *queue) noPingPackets(base, top uint8) bool {
 }
 
 // processACK processes an incoming ACK of a given sequence number.
-func (q *queue) processACK(seq uint8, resendTimeout time.Duration) bool {
+func (q *queue) processACK(seq uint8) bool {
 	// If our queue is empty, an ACK should not have any effect.
 	if q.size() == 0 {
 		log.Tracef("Received ack %d, but queue is empty. Ignoring.", seq)
@@ -377,7 +382,7 @@ func (q *queue) processACK(seq uint8, resendTimeout time.Duration) bool {
 	if seq == q.awaitedACK && q.awaitingCatchUp {
 		log.Tracef("Got awaited ACK")
 
-		q.proceedAfterTime(resendTimeout)
+		q.proceedAfterTime()
 	}
 	q.awaitingCatchUpMu.RUnlock()
 
@@ -485,7 +490,7 @@ func (q *queue) processNACK(seq uint8) (bool, bool) {
 
 // proceedAfterTime will wait for the resendTimeout and then send an
 // awaitedACKSignal, if we're still awaiting the resend catch up.
-func (q *queue) proceedAfterTime(resendTimeout time.Duration) {
+func (q *queue) proceedAfterTime() {
 	caughtUpSignal := q.caughtUpSignal
 
 	processAwaitedACK := func() {
@@ -522,7 +527,7 @@ func (q *queue) proceedAfterTime(resendTimeout time.Duration) {
 	// proceedAfterTime callback, as that's the time we'd expect it to take
 	// for the other party to respond with a NACK, if the resent last packet
 	// in the queue would lead to a NACK.
-	time.AfterFunc(resendTimeout, processAwaitedACK)
+	time.AfterFunc(q.timeoutManager.GetResendTimeout(), processAwaitedACK)
 }
 
 // containsSequence is used to determine if a number, seq, is between two other

gbn/queue_test.go

@@ -7,12 +7,15 @@ import (
 )
 
 func TestQueueSize(t *testing.T) {
-	queue := newQueue(&queueConfig{
+	timeoutManager := NewTimeOutManager(false)
+	cfg := &queueConfig{
 		s: 4,
 		sendPkt: func(packet *PacketData) error {
 			return nil
 		},
-	})
+	}
+
+	queue := newQueue(cfg,  timeoutManager)
 
 	require.Equal(t, uint8(0), queue.size())