gbn: add timeout manager

To make the gbn package more modular, we move the responsibility of
managing the timeout values to a new struct, the `TimeoutManager`.

This make it easier to manage all timeout values used by the different
components of the gbn package.

Author: ViktorT-11

gbn/gbn_client.go

@@ -122,9 +122,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{}
@@ -109,6 +90,10 @@ type GoBackNConn struct {
 	quit      chan struct{}
 	closeOnce sync.Once
 	wg        sync.WaitGroup
+
+	// timeoutManager is used to manage all the timeouts used by the
+	// GoBackNConn.
+	timeoutManager *TimeoutManager
 }
 
 // newGoBackNConn creates a GoBackNConn instance with all the members which
@@ -118,51 +103,46 @@ func newGoBackNConn(ctx context.Context, sendFunc sendBytesFunc,
 
 	ctxc, cancel := context.WithCancel(ctx)
 
+	timeoutManager := NewTimeOutManager(isServer)
+
+	queue := newQueue(n+1, timeoutManager)
+
 	return &GoBackNConn{
 		n:                 n,
 		s:                 n + 1,
-		resendTimeout:     defaultResendTimeout,
 		recvFromStream:    recvFunc,
 		sendToStream:      sendFunc,
 		recvDataChan:      make(chan *PacketData, n),
 		sendDataChan:      make(chan *PacketData),
 		isServer:          isServer,
-		sendQueue:         newQueue(n+1, defaultHandshakeTimeout),
-		handshakeTimeout:  defaultHandshakeTimeout,
-		recvTimeout:       DefaultRecvTimeout,
-		sendTimeout:       DefaultSendTimeout,
+		sendQueue:         queue,
 		receivedACKSignal: make(chan struct{}),
 		resendSignal:      make(chan struct{}, 1),
 		remoteClosed:      make(chan struct{}),
 		ctx:               ctxc,
 		cancel:            cancel,
 		quit:              make(chan struct{}),
+		timeoutManager:    timeoutManager,
 	}
 }
 
-// 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(n+1, defaultHandshakeTimeout)
-}
-
 // 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
+	g.timeoutManager.SetSendTimeout(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.timeoutManager.SetRecvTimeout(timeout)
+}
 
-	g.recvTimeout = 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(n+1, g.timeoutManager)
 }
 
 // Send blocks until an ack is received for the packet sent N packets before.
@@ -174,9 +154,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 {
@@ -235,9 +213,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 {
@@ -279,7 +255,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() {
@@ -335,7 +311,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 {
@@ -388,7 +364,7 @@ func (g *GoBackNConn) sendPacketsForever() error {
 	// resendQueue re-sends the current contents of the queue.
 	resendQueue := func() error {
 		err := g.sendQueue.resend(
-			g.resendTimeout, g.quit,
+			g.quit,
 			func(packet *PacketData) error {
 				return g.sendPacket(g.ctx, packet)
 			},
@@ -400,7 +376,7 @@ func (g *GoBackNConn) sendPacketsForever() error {
 		// 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
 	}
@@ -520,7 +496,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:
@@ -578,7 +554,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")
@@ -603,9 +581,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

@@ -140,7 +140,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.SetResendTimeout(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

@@ -102,18 +102,19 @@ type queue struct {
 	// topMtx is used to guard sequenceTop.
 	topMtx sync.RWMutex
 
-	lastResend       time.Time
-	handshakeTimeout time.Duration
+	lastResend time.Time
+
+	timeoutManager *TimeoutManager
 }
 
 // newQueue creates a new queue.
-func newQueue(s uint8, handshakeTimeout time.Duration) *queue {
+func newQueue(s uint8, timeoutManager *TimeoutManager) *queue {
 	return &queue{
 		content:           make([]*PacketData, s),
 		s:                 s,
-		handshakeTimeout:  handshakeTimeout,
 		awaitedACKSignal:  make(chan struct{}, 1),
 		awaitedNACKSignal: make(chan struct{}, 1),
+		timeoutManager:    timeoutManager,
 	}
 }
 
@@ -199,10 +200,10 @@ 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, quit chan struct{},
+func (q *queue) resend(quit chan struct{},
 	sendFunc func(packet *PacketData) error,
 ) error {
-	if time.Since(q.lastResend) < q.handshakeTimeout {
+	if time.Since(q.lastResend) < q.timeoutManager.GetHandshakeTimeout() {
 		log.Tracef("Resent the queue recently.")
 
 		return nil
@@ -280,7 +281,7 @@ func (q *queue) resend(resendTimeout time.Duration, quit chan struct{},
 	q.awaitingCatchUpMu.Unlock()
 
 	// Then await until we know that both parties are in sync.
-	q.awaitCatchUp(resendTimeout, quit)
+	q.awaitCatchUp(quit)
 
 	return nil
 }
@@ -290,8 +291,10 @@ func (q *queue) resend(resendTimeout time.Duration, quit chan struct{},
 // 3X the resend timeout, the function will also return.
 // See the docs for the resend function for more details on why we need to await
 // the awaited ACK or NACK signal.
-func (q *queue) awaitCatchUp(resendTimeout time.Duration, quit chan struct{}) {
-	ticker := time.NewTimer(resendTimeout * awaitingTimeoutMultiplier)
+func (q *queue) awaitCatchUp(quit chan struct{}) {
+	ticker := time.NewTimer(
+		q.timeoutManager.GetResendTimeout() * awaitingTimeoutMultiplier,
+	)
 	defer ticker.Stop()
 
 	log.Tracef("Awaiting catchup after resending the queue")
@@ -351,7 +354,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)
@@ -366,7 +369,7 @@ func (q *queue) processACK(seq uint8, resendTimeout time.Duration) bool {
 	if seq == q.awaitedACK && q.awaitingCatchUp {
 		log.Tracef("Got awaited ACK")
 
-		q.proceedAfterTime(q.catchUpID, resendTimeout)
+		q.proceedAfterTime(q.catchUpID)
 	}
 	q.awaitingCatchUpMu.RUnlock()
 
@@ -474,7 +477,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(catchUpID int64, resendTimeout time.Duration) {
+func (q *queue) proceedAfterTime(catchUpID int64) {
 	processAwaitedACK := func() {
 		log.Tracef("Executing proceedAfterTime")
 		q.awaitingCatchUpMu.Lock()
@@ -508,7 +511,7 @@ func (q *queue) proceedAfterTime(catchUpID int64, 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,15 +7,17 @@ import (
 )
 
 func TestQueueSize(t *testing.T) {
-	q := newQueue(4, 0)
+	timeoutManager := NewTimeOutManager(false)
 
-	require.Equal(t, uint8(0), q.size())
+	queue := newQueue(4, timeoutManager)
 
-	q.sequenceBase = 2
-	q.sequenceTop = 3
-	require.Equal(t, uint8(1), q.size())
+	require.Equal(t, uint8(0), queue.size())
 
-	q.sequenceBase = 3
-	q.sequenceTop = 2
-	require.Equal(t, uint8(3), q.size())
+	queue.sequenceBase = 2
+	queue.sequenceTop = 3
+	require.Equal(t, uint8(1), queue.size())
+
+	queue.sequenceBase = 3
+	queue.sequenceTop = 2
+	require.Equal(t, uint8(3), queue.size())
 }