Problem: async fireEvents could overlap

consensus/state.go

@@ -42,7 +42,10 @@ var (
 	errPubKeyIsNotSet = errors.New("pubkey is not set. Look for \"Can't get private validator pubkey\" errors")
 )
 
-var msgQueueSize = 1000
+var (
+	msgQueueSize  = 1000
+	taskQueueSize = 128
+)
 
 // msgs from the reactor which may update the state
 type msgInfo struct {
@@ -145,6 +148,9 @@ type State struct {
 
 	// offline state sync height indicating to which height the node synced offline
 	offlineStateSyncHeight int64
+
+	// run tasks asynchronously to avoid blocking block executor, currently mainly for firing tx/block events.
+	taskRunner *TaskRunner
 }
 
 // StateOption sets an optional parameter on the State.
@@ -160,6 +166,8 @@ func NewState(
 	evpool evidencePool,
 	options ...StateOption,
 ) *State {
+	taskRunner := StartTaskRunner(taskQueueSize)
+	blockExec.SetTaskRunner(taskRunner.RunTask)
 	cs := &State{
 		config:           config,
 		blockExec:        blockExec,
@@ -175,6 +183,7 @@ func NewState(
 		evpool:           evpool,
 		evsw:             cmtevents.NewEventSwitch(),
 		metrics:          NopMetrics(),
+		taskRunner:       taskRunner,
 	}
 	for _, option := range options {
 		option(cs)
@@ -438,6 +447,9 @@ func (cs *State) OnStop() {
 		cs.Logger.Error("failed trying to stop timeoutTicket", "error", err)
 	}
 	// WAL is stopped in receiveRoutine.
+
+	// Stop the task runner
+	cs.taskRunner.StopAndWait()
 }
 
 // Wait waits for the the main routine to return.

consensus/task.go

@@ -0,0 +1,31 @@
+package consensus
+
+import "sync"
+
+type TaskRunner struct {
+	wg     sync.WaitGroup
+	taskCh chan func()
+}
+
+func StartTaskRunner(buf int) *TaskRunner {
+	tr := &TaskRunner{
+		taskCh: make(chan func(), buf),
+	}
+	tr.wg.Add(1)
+	go func() {
+		defer tr.wg.Done()
+		for f := range tr.taskCh {
+			f()
+		}
+	}()
+	return tr
+}
+
+func (tr *TaskRunner) StopAndWait() {
+	close(tr.taskCh)
+	tr.wg.Wait()
+}
+
+func (tr *TaskRunner) RunTask(f func()) {
+	tr.taskCh <- f
+}

state/execution.go

@@ -43,6 +43,8 @@ type BlockExecutor struct {
 	logger log.Logger
 
 	metrics *Metrics
+
+	asyncRunner func(func())
 }
 
 type BlockExecutorOption func(executor *BlockExecutor)
@@ -53,6 +55,12 @@ func BlockExecutorWithMetrics(metrics *Metrics) BlockExecutorOption {
 	}
 }
 
+func BlockExecutorWithAsyncRunner(runner func(func())) BlockExecutorOption {
+	return func(blockExec *BlockExecutor) {
+		blockExec.asyncRunner = runner
+	}
+}
+
 // NewBlockExecutor returns a new BlockExecutor with a NopEventBus.
 // Call SetEventBus to provide one.
 func NewBlockExecutor(
@@ -92,6 +100,10 @@ func (blockExec *BlockExecutor) SetEventBus(eventBus types.BlockEventPublisher)
 	blockExec.eventBus = eventBus
 }
 
+func (blockExec *BlockExecutor) SetTaskRunner(runner func(func())) {
+	blockExec.asyncRunner = runner
+}
+
 // CreateProposalBlock calls state.MakeBlock with evidence from the evpool
 // and txs from the mempool. The max bytes must be big enough to fit the commit.
 // The block space is first allocated to outstanding evidence.
@@ -318,7 +330,15 @@ func (blockExec *BlockExecutor) applyBlock(state State, blockID types.BlockID, b
 	if _, ok := blockExec.eventBus.(types.NopEventBus); !ok {
 		// Events are fired after everything else.
 		// NOTE: if we crash between Commit and Save, events wont be fired during replay
-		go fireEvents(blockExec.logger, blockExec.eventBus, block, blockID, abciResponse, validatorUpdates)
+		task := func() {
+			fireEvents(blockExec.logger, blockExec.eventBus, block, blockID, abciResponse, validatorUpdates)
+		}
+
+		if blockExec.asyncRunner != nil {
+			blockExec.asyncRunner(task)
+		} else {
+			task()
+		}
 	}
 
 	return state, nil