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.

Even though the `TimeoutManager` is now used to manage timeouts, the
resend timeout is not yet set dynamically for the connections of the gbn
package. Support and usage of that functionality will added in the
upcoming commits.

Author: ViktorTigerstrom

gbn/gbn_client.go

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

gbn/gbn_conn.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"sync"
 	"time"
 
@@ -41,10 +40,6 @@ type GoBackNConn struct {
 	recvDataChan chan *PacketData
 	sendDataChan chan *PacketData
 
-	sendTimeout time.Duration
-	recvTimeout time.Duration
-	timeoutsMu  sync.RWMutex
-
 	log btclog.Logger
 
 	// receivedACKSignal channel is used to signal that the queue size has
@@ -65,6 +60,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.
@@ -84,63 +83,62 @@ func newGoBackNConn(ctx context.Context, cfg *config,
 	prefix := fmt.Sprintf("(%s)", loggerPrefix)
 	plog := build.NewPrefixLog(prefix, log)
 
+	timeoutManager := NewTimeOutManager(plog)
+
 	g := &GoBackNConn{
 		cfg:               cfg,
 		recvDataChan:      make(chan *PacketData, cfg.n),
 		sendDataChan:      make(chan *PacketData),
-		recvTimeout:       DefaultRecvTimeout,
-		sendTimeout:       DefaultSendTimeout,
 		receivedACKSignal: make(chan struct{}),
 		resendSignal:      make(chan struct{}, 1),
 		remoteClosed:      make(chan struct{}),
 		ctx:               ctxc,
 		cancel:            cancel,
 		log:               plog,
 		quit:              make(chan struct{}),
+		timeoutManager:    timeoutManager,
 	}
 
-	g.sendQueue = newQueue(&queueCfg{
-		s:       cfg.n + 1,
-		timeout: cfg.resendTimeout,
-		log:     plog,
-		sendPkt: func(packet *PacketData) error {
-			return g.sendPacket(g.ctx, packet)
+	g.sendQueue = newQueue(
+		&queueCfg{
+			s:   cfg.n + 1,
+			log: plog,
+			sendPkt: func(packet *PacketData) error {
+				return g.sendPacket(g.ctx, packet)
+			},
 		},
-	})
+		timeoutManager,
+	)
 
 	return g
 }
 
-// setN sets the current N to use. This _must_ be set before the handshake is
-// completed.
-func (g *GoBackNConn) setN(n uint8) {
-	g.cfg.n = n
-	g.cfg.s = n + 1
-	g.recvDataChan = make(chan *PacketData, n)
-	g.sendQueue = newQueue(&queueCfg{
-		s:       n + 1,
-		timeout: g.cfg.resendTimeout,
-		log:     g.log,
-		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.timeoutsMu.Lock()
-	defer g.timeoutsMu.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.timeoutsMu.Lock()
-	defer g.timeoutsMu.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.cfg.n = n
+	g.cfg.s = n + 1
+	g.recvDataChan = make(chan *PacketData, n)
+	g.sendQueue = newQueue(
+		&queueCfg{
+			s:   n + 1,
+			log: g.log,
+			sendPkt: func(packet *PacketData) error {
+				return g.sendPacket(g.ctx, packet)
+			},
+		},
+		g.timeoutManager,
+	)
 }
 
 // Send blocks until an ack is received for the packet sent N packets before.
@@ -152,9 +150,7 @@ func (g *GoBackNConn) Send(data []byte) error {
 	default:
 	}
 
-	g.timeoutsMu.RLock()
-	ticker := time.NewTimer(g.sendTimeout)
-	g.timeoutsMu.RUnlock()
+	ticker := time.NewTimer(g.timeoutManager.GetSendTimeout())
 	defer ticker.Stop()
 
 	sendPacket := func(packet *PacketData) error {
@@ -216,9 +212,7 @@ func (g *GoBackNConn) Recv() ([]byte, error) {
 		msg *PacketData
 	)
 
-	g.timeoutsMu.RLock()
-	ticker := time.NewTimer(g.recvTimeout)
-	g.timeoutsMu.RUnlock()
+	ticker := time.NewTimer(g.timeoutManager.GetRecvTimeout())
 	defer ticker.Stop()
 
 	for {
@@ -245,22 +239,16 @@ func (g *GoBackNConn) Recv() ([]byte, error) {
 func (g *GoBackNConn) start() {
 	g.log.Debugf("Starting")
 
-	pingTime := time.Duration(math.MaxInt64)
-	if g.cfg.pingTime != 0 {
-		pingTime = g.cfg.pingTime
-	}
-
-	g.pingTicker = NewIntervalAwareForceTicker(pingTime)
+	g.pingTicker = NewIntervalAwareForceTicker(
+		g.timeoutManager.GetPingTime(),
+	)
 	g.pingTicker.Resume()
 
-	pongTime := time.Duration(math.MaxInt64)
-	if g.cfg.pongTime != 0 {
-		pongTime = g.cfg.pongTime
-	}
-
-	g.pongTicker = NewIntervalAwareForceTicker(pongTime)
+	g.pongTicker = NewIntervalAwareForceTicker(
+		g.timeoutManager.GetPongTime(),
+	)
 
-	g.resendTicker = time.NewTicker(g.cfg.resendTimeout)
+	g.resendTicker = time.NewTicker(g.timeoutManager.GetResendTimeout())
 
 	g.wg.Add(1)
 	go func() {
@@ -317,7 +305,7 @@ func (g *GoBackNConn) Close() error {
 			g.log.Tracef("Try sending FIN")
 
 			ctxc, cancel := context.WithTimeout(
-				g.ctx, defaultFinSendTimeout,
+				g.ctx, g.timeoutManager.GetFinSendTimeout(),
 			)
 			defer cancel()
 			if err := g.sendPacket(ctxc, &PacketFIN{}); err != nil {
@@ -382,7 +370,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.cfg.resendTimeout)
+		g.resendTicker.Reset(g.timeoutManager.GetResendTimeout())
 
 		// Also drain the resend signal channel, as resendTicker.Reset
 		// doesn't drain the channel if the ticker ticked during the
@@ -509,7 +497,7 @@ func (g *GoBackNConn) receivePacketsForever() error { // nolint:gocyclo
 			g.pongTicker.Pause()
 		}
 
-		g.resendTicker.Reset(g.cfg.resendTimeout)
+		g.resendTicker.Reset(g.timeoutManager.GetResendTimeout())
 
 		switch m := msg.(type) {
 		case *PacketData:
@@ -567,8 +555,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.cfg.resendTimeout*2
+
+				timeout := g.timeoutManager.GetResendTimeout()
+				recentlySent := sinceSent < timeout*2
 
 				if lastNackSeq == g.recvSeq && recentlySent {
 					g.log.Tracef("Recently sent NACK")

gbn/gbn_server.go

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

gbn/queue.go

@@ -17,8 +17,6 @@ type queueCfg struct {
 	// no way to tell.
 	s uint8
 
-	timeout time.Duration
-
 	log btclog.Logger
 
 	sendPkt func(packet *PacketData) error
@@ -29,6 +27,8 @@ type queueCfg struct {
 type queue struct {
 	cfg *queueCfg
 
+	timeoutManager *TimeoutManager
+
 	// content is the current content of the queue. This is always a slice
 	// of length s but can contain nil elements if the queue isn't full.
 	content []*PacketData
@@ -59,18 +59,19 @@ type queue struct {
 }
 
 // newQueue creates a new queue.
-func newQueue(cfg *queueCfg) *queue {
+func newQueue(cfg *queueCfg, timeoutManager *TimeoutManager) *queue {
 	if cfg.log == nil {
 		cfg.log = log
 	}
 
 	q := &queue{
-		cfg:     cfg,
-		content: make([]*PacketData, cfg.s),
-		quit:    make(chan struct{}),
+		cfg:            cfg,
+		content:        make([]*PacketData, cfg.s),
+		quit:           make(chan struct{}),
+		timeoutManager: timeoutManager,
 	}
 
-	q.syncer = newSyncer(cfg.s, cfg.log, cfg.timeout, q.quit)
+	q.syncer = newSyncer(cfg.s, cfg.log, timeoutManager, q.quit)
 
 	return q
 }
@@ -108,7 +109,7 @@ func (q *queue) addPacket(packet *PacketData) {
 // two parties to be seen as synced; this may fail in which case the caller is
 // expected to call resend again.
 func (q *queue) resend() error {
-	if time.Since(q.lastResend) < q.cfg.timeout {
+	if time.Since(q.lastResend) < q.timeoutManager.GetHandshakeTimeout() {
 		q.cfg.log.Tracef("Resent the queue recently.")
 
 		return nil

gbn/queue_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestQueueSize(t *testing.T) {
-	q := newQueue(&queueCfg{s: 4})
+	q := newQueue(&queueCfg{s: 4}, NewTimeOutManager(nil))
 
 	require.Equal(t, uint8(0), q.size())
 
@@ -33,16 +33,18 @@ func TestQueueResend(t *testing.T) {
 	resentPackets := make(map[uint8]struct{})
 	queueTimeout := time.Second * 1
 
+	tm := NewTimeOutManager(nil)
+	tm.resendTimeout = queueTimeout
+
 	cfg := &queueCfg{
-		s:       5,
-		timeout: queueTimeout,
+		s: 5,
 		sendPkt: func(packet *PacketData) error {
 			resentPackets[packet.Seq] = struct{}{}
 
 			return nil
 		},
 	}
-	q := newQueue(cfg)
+	q := newQueue(cfg, tm)
 
 	pkt1 := &PacketData{Seq: 1}
 	pkt2 := &PacketData{Seq: 2}

gbn/syncer.go

@@ -99,9 +99,9 @@ const (
 // When either of the 3 conditions above are met, we will consider both parties
 // to be in sync.
 type syncer struct {
-	s       uint8
-	log     btclog.Logger
-	timeout time.Duration
+	s              uint8
+	log            btclog.Logger
+	timeoutManager *TimeoutManager
 
 	state syncState
 
@@ -127,20 +127,20 @@ type syncer struct {
 }
 
 // newSyncer creates a new syncer instance.
-func newSyncer(s uint8, prefixLogger btclog.Logger, timeout time.Duration,
-	quit chan struct{}) *syncer {
+func newSyncer(s uint8, prefixLogger btclog.Logger,
+	timeoutManager *TimeoutManager, quit chan struct{}) *syncer {
 
 	if prefixLogger == nil {
 		prefixLogger = log
 	}
 
 	return &syncer{
-		s:       s,
-		log:     prefixLogger,
-		timeout: timeout,
-		state:   syncStateIdle,
-		cancel:  make(chan struct{}),
-		quit:    quit,
+		s:              s,
+		log:            prefixLogger,
+		timeoutManager: timeoutManager,
+		state:          syncStateIdle,
+		cancel:         make(chan struct{}),
+		quit:           quit,
 	}
 }
 
@@ -210,7 +210,9 @@ func (c *syncer) waitForSync() {
 	case <-c.cancel:
 		c.log.Tracef("sync canceled or reset")
 
-	case <-time.After(c.timeout * awaitingTimeoutMultiplier):
+	case <-time.After(
+		c.timeoutManager.GetResendTimeout() * awaitingTimeoutMultiplier,
+	):
 		c.log.Tracef("Timed out while waiting for sync")
 	}
 
@@ -291,7 +293,7 @@ func (c *syncer) proceedAfterTime() {
 
 		return
 
-	case <-time.After(c.timeout):
+	case <-time.After(c.timeoutManager.GetResendTimeout()):
 		c.mu.Lock()
 		defer c.mu.Unlock()
 

gbn/syncer_test.go

@@ -19,7 +19,10 @@ func TestSyncer(t *testing.T) {
 	syncTimeout := time.Second * 1
 	expectedNACK := uint8(3)
 
-	syncer := newSyncer(5, nil, syncTimeout, make(chan struct{}))
+	tm := NewTimeOutManager(nil)
+	tm.resendTimeout = syncTimeout
+
+	syncer := newSyncer(5, nil, tm, make(chan struct{}))
 
 	// Let's first test the scenario where we don't receive the expected
 	// ACK/NACK after initiating the resend. This should trigger a timeout