Refactor timer management to event-driven model

consensus/istanbul/qbft/core/core.go

@@ -52,6 +52,7 @@ func New(backend istanbul.Backend, config *istanbul.Config) istanbul.Core {
 		pendingRequests:    prque.New(),
 		pendingRequestsMu:  new(sync.Mutex),
 		consensusTimestamp: time.Time{},
+		afterFunc:          func(d time.Duration, f func()) timer { return time.AfterFunc(d, f) },
 	}
 
 	c.validateFn = c.checkValidatorSignature
@@ -66,11 +67,9 @@ type core struct {
 	state   State
 	logger  log.Logger
 
-	backend               istanbul.Backend
-	events                *event.TypeMuxSubscription
-	finalCommittedSub     *event.TypeMuxSubscription
-	timeoutSub            *event.TypeMuxSubscription
-	futurePreprepareTimer *time.Timer
+	backend           istanbul.Backend
+	events            *event.TypeMuxSubscription
+	finalCommittedSub *event.TypeMuxSubscription
 
 	valSet     istanbul.ValidatorSet
 	validateFn func([]byte, []byte) (common.Address, error)
@@ -82,8 +81,7 @@ type core struct {
 	currentMutex sync.Mutex
 	handlerWg    *sync.WaitGroup
 
-	roundChangeSet   *roundChangeSet
-	roundChangeTimer *time.Timer
+	roundChangeSet *roundChangeSet
 
 	QBFTPreparedPrepares []*qbfttypes.Prepare
 
@@ -92,8 +90,12 @@ type core struct {
 
 	consensusTimestamp time.Time
 
-	newRoundMutex sync.Mutex
-	newRoundTimer *time.Timer
+	// Timer state belongs to the event loop. Callbacks only send immutable events.
+	timers      [timerKindCount]scheduledTimer
+	afterFunc   func(time.Duration, func()) timer
+	timerEvents chan timerEvent
+	timerDone   chan struct{}
+	stopOnce    sync.Once
 }
 
 func (c *core) currentView() *istanbul.View {
@@ -186,6 +188,8 @@ func (c *core) startNewRound(round *big.Int) {
 		c.valSet = c.backend.Validators(lastProposal)
 	}
 
+	c.stopTimer()
+
 	// New snapshot for new round
 	c.updateRoundState(newView, c.valSet, roundChange)
 
@@ -243,24 +247,17 @@ func (c *core) Address() common.Address {
 }
 
 func (c *core) stopFuturePreprepareTimer() {
-	if c.futurePreprepareTimer != nil {
-		c.futurePreprepareTimer.Stop()
-	}
+	c.cancelTimer(futurePreprepareTimer)
 }
 
 func (c *core) stopTimer() {
-	c.stopFuturePreprepareTimer()
-	if c.roundChangeTimer != nil {
-		c.roundChangeTimer.Stop()
+	for kind := timerKind(0); kind < timerKindCount; kind++ {
+		c.cancelTimer(kind)
 	}
 }
 
 func (c *core) newRoundChangeTimer() {
-	c.stopTimer()
-
-	for c.current == nil { // wait because it is asynchronous in handleRequest
-		time.Sleep(10 * time.Millisecond)
-	}
+	c.stopFuturePreprepareTimer()
 
 	// set timeout based on the round number
 	baseTimeout := time.Duration(c.config.GetConfig(c.current.Sequence()).RequestTimeout) * time.Millisecond
@@ -294,9 +291,7 @@ func (c *core) newRoundChangeTimer() {
 	}
 
 	c.currentLogger(true, nil).Trace("QBFT: start new ROUND-CHANGE timer", "timeout", timeout.Seconds())
-	c.roundChangeTimer = time.AfterFunc(timeout, func() {
-		c.sendEvent(timeoutEvent{})
-	})
+	c.scheduleTimer(roundChangeTimer, timeout, timerEvent{})
 }
 
 func (c *core) checkValidatorSignature(data []byte, sig []byte) (common.Address, error) {

consensus/istanbul/qbft/core/events.go

@@ -25,5 +25,3 @@ type backlogEvent struct {
 	src istanbul.Validator
 	msg qbfttypes.QBFTMessage
 }
-
-type timeoutEvent struct{}

consensus/istanbul/qbft/core/handler.go

@@ -19,6 +19,7 @@ package core
 import (
 	"fmt"
 	"math/big"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
@@ -32,20 +33,23 @@ func (c *core) Start() error {
 	c.logger.Info("QBFT: start")
 	// Tests will handle events itself, so we have to make subscribeEvents()
 	// be able to call in test.
+	c.timerEvents = make(chan timerEvent)
+	c.timerDone = make(chan struct{})
+	c.stopOnce = sync.Once{}
 	c.subscribeEvents()
-	c.handlerWg.Add(1)
-	go c.handleEvents()
 
-	// Start a new round from last sequence + 1
+	// Initialize consensus before the event loop can receive messages.
 	c.startNewRound(common.Big0)
+	c.handlerWg.Add(1)
+	go c.handleEvents()
 
 	return nil
 }
 
 // Stop implements core.Engine.Stop
 func (c *core) Stop() error {
 	c.logger.Info("QBFT: stopping...")
-	c.stopTimer()
+	c.stopOnce.Do(func() { close(c.timerDone) })
 	c.unsubscribeEvents()
 
 	// Make sure the handler goroutine exits
@@ -65,9 +69,6 @@ func (c *core) subscribeEvents() {
 		// internal events
 		backlogEvent{},
 	)
-	c.timeoutSub = c.backend.EventMux().Subscribe(
-		timeoutEvent{},
-	)
 	c.finalCommittedSub = c.backend.EventMux().Subscribe(
 		istanbul.FinalCommittedEvent{},
 	)
@@ -76,7 +77,6 @@ func (c *core) subscribeEvents() {
 // Unsubscribe all events
 func (c *core) unsubscribeEvents() {
 	c.events.Unsubscribe()
-	c.timeoutSub.Unsubscribe()
 	c.finalCommittedSub.Unsubscribe()
 }
 
@@ -94,12 +94,18 @@ func (c *core) unsubscribeEvents() {
 func (c *core) handleEvents() {
 	// Clear state
 	defer func() {
+		c.stopOnce.Do(func() { close(c.timerDone) })
+		c.stopTimer()
 		c.current = nil
 		c.handlerWg.Done()
 	}()
 
 	for {
 		select {
+		case <-c.timerDone:
+			return
+		case event := <-c.timerEvents:
+			c.handleTimerEvent(event)
 		case event, ok := <-c.events.Chan():
 			if !ok {
 				return
@@ -141,12 +147,6 @@ func (c *core) handleEvents() {
 				// if successfully processed, we gossip message to other validators
 				c.backend.Gossip(c.valSet, ev.msg.Code(), data)
 			}
-		case _, ok := <-c.timeoutSub.Chan():
-			// we received a round change timeout
-			if !ok {
-				return
-			}
-			c.handleTimeoutMsg()
 		case event, ok := <-c.finalCommittedSub.Chan():
 			// our block proposal got committed
 			if !ok {

consensus/istanbul/qbft/core/preprepare.go

@@ -132,15 +132,8 @@ func (c *core) handlePreprepareMsg(preprepare *qbfttypes.Preprepare) error {
 		if err == consensus.ErrFutureBlock {
 			logger.Info("QBFT: PRE-PREPARE block proposal is in the future (will be treated again later)", "duration", duration)
 
-			// start a timer to re-input PRE-PREPARE message as a backlog event
-			c.stopFuturePreprepareTimer()
-			c.futurePreprepareTimer = time.AfterFunc(duration, func() {
-				_, validator := c.valSet.GetByAddress(preprepare.Source())
-				c.sendEvent(backlogEvent{
-					src: validator,
-					msg: preprepare,
-				})
-			})
+			// Retry through the event loop only while this view and timer remain valid.
+			c.scheduleTimer(futurePreprepareTimer, duration, timerEvent{preprepare: preprepare})
 		} else {
 			logger.Warn("QBFT: invalid PRE-PREPARE block proposal", "err", err)
 		}
@@ -152,6 +145,8 @@ func (c *core) handlePreprepareMsg(preprepare *qbfttypes.Preprepare) error {
 	if c.state == StateAcceptRequest {
 		c.logger.Info("QBFT: accepted PRE-PREPARE message")
 
+		c.cancelTimer(emptyBlockTimer)
+
 		// Re-initialize ROUND-CHANGE timer
 		c.newRoundChangeTimer()
 		c.consensusTimestamp = time.Now()

consensus/istanbul/qbft/core/request.go

@@ -44,6 +44,7 @@ func (c *core) handleRequest(request *Request) error {
 		return err
 	}
 
+	c.cancelTimer(emptyBlockTimer)
 	c.current.pendingRequest = request
 	if c.state == StateAcceptRequest {
 		config := c.config.GetConfig(c.current.Sequence())
@@ -54,14 +55,6 @@ func (c *core) handleRequest(request *Request) error {
 			// Send PRE-PREPARE message to other validators
 			c.sendPreprepareMsg(request)
 		} else { // emptyBlockPeriod is set
-			c.newRoundMutex.Lock()
-			defer c.newRoundMutex.Unlock()
-
-			if c.newRoundTimer != nil {
-				c.newRoundTimer.Stop()
-				c.newRoundTimer = nil
-			}
-
 			delay := time.Duration(0)
 
 			block, ok := request.Proposal.(*types.Block)
@@ -77,14 +70,7 @@ func (c *core) handleRequest(request *Request) error {
 				}
 			}
 			if delay > 0 {
-				c.newRoundTimer = time.AfterFunc(delay, func() {
-					c.newRoundTimer = nil
-					// Start ROUND-CHANGE timer
-					c.newRoundChangeTimer()
-
-					// Send PRE-PREPARE message to other validators
-					c.sendPreprepareMsg(request)
-				})
+				c.scheduleTimer(emptyBlockTimer, delay, timerEvent{proposalHash: request.Proposal.Hash()})
 			} else {
 				// Start ROUND-CHANGE timer
 				c.newRoundChangeTimer()

consensus/istanbul/qbft/core/timer.go

@@ -0,0 +1,122 @@
+package core
+
+import (
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus/istanbul"
+	qbfttypes "github.com/ethereum/go-ethereum/consensus/istanbul/qbft/types"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+type timer interface {
+	Stop() bool
+}
+
+type timerKind uint8
+
+const (
+	roundChangeTimer timerKind = iota
+	emptyBlockTimer
+	futurePreprepareTimer
+	timerKindCount
+)
+
+func (kind timerKind) String() string {
+	switch kind {
+	case roundChangeTimer:
+		return "round-change"
+	case emptyBlockTimer:
+		return "empty-block"
+	case futurePreprepareTimer:
+		return "future-preprepare"
+	default:
+		return "unknown"
+	}
+}
+
+type scheduledTimer struct {
+	timer      timer
+	generation uint64
+}
+
+type timerEvent struct {
+	kind         timerKind
+	view         istanbul.View
+	generation   uint64
+	proposalHash common.Hash
+	preprepare   *qbfttypes.Preprepare
+}
+
+// cancelTimer also invalidates callbacks which have already fired. All timer
+// registration, cancellation and event handling runs on the consensus loop.
+func (c *core) cancelTimer(kind timerKind) {
+	slot := &c.timers[kind]
+	if slot.timer != nil {
+		slot.timer.Stop()
+		slot.timer = nil
+		c.currentLogger(true, nil).Trace("QBFT: stop timer", "kind", kind, "generation", slot.generation)
+	}
+	slot.generation++
+}
+
+func (c *core) scheduleTimer(kind timerKind, delay time.Duration, ev timerEvent) {
+	c.cancelTimer(kind)
+	slot := &c.timers[kind]
+	ev.kind = kind
+	ev.view = *c.currentView() // Copy big.Int values; do not capture mutable round state.
+	ev.generation = slot.generation
+	// Capture this execution's channels, so an old callback cannot enter a
+	// restarted core. Callbacks never read or mutate consensus or timer state.
+	events, done := c.timerEvents, c.timerDone
+	slot.timer = c.afterFunc(delay, func() {
+		select {
+		case events <- ev:
+		case <-done:
+		}
+	})
+	c.currentLogger(true, nil).Trace("QBFT: start timer", "kind", kind, "generation", ev.generation, "delay", delay)
+}
+
+func (c *core) handleTimerEvent(ev timerEvent) {
+	select {
+	case <-c.timerDone:
+		return
+	default:
+	}
+	if ev.kind >= timerKindCount {
+		return
+	}
+	slot := &c.timers[ev.kind]
+	logger := c.currentLogger(true, nil).New("kind", ev.kind, "generation", ev.generation,
+		"timer.sequence", ev.view.Sequence, "timer.round", ev.view.Round)
+	if slot.timer == nil || ev.generation != slot.generation || c.current == nil || ev.view.Cmp(c.currentView()) != 0 {
+		logger.Trace("QBFT: discard timer event", "reason", "inactive timer or stale generation/view")
+		return
+	}
+	c.cancelTimer(ev.kind) // Consume once, even if the callback is delivered twice.
+	logger.Trace("QBFT: handle timer event")
+	switch ev.kind {
+	case roundChangeTimer:
+		c.handleTimeoutMsg()
+	case emptyBlockTimer:
+		request := c.current.pendingRequest
+		if c.state != StateAcceptRequest || request == nil || request.Proposal.Hash() != ev.proposalHash {
+			logger.Trace("QBFT: discard timer event", "reason", "proposal or state changed")
+			return
+		}
+		c.newRoundChangeTimer()
+		c.sendPreprepareMsg(request)
+	case futurePreprepareTimer:
+		// Apply the normal view/state checks again before retrying verification.
+		if err := c.handleDecodedMessage(ev.preprepare); err != nil {
+			return
+		}
+		data, err := rlp.EncodeToBytes(ev.preprepare)
+		if err != nil {
+			logger.Error("QBFT: can not encode deferred PRE-PREPARE", "err", err)
+			return
+		}
+		c.backend.Gossip(c.valSet, ev.preprepare.Code(), data)
+	}
+}