Merge branch 'master' into callum/4728-check-evidence

blockchain/v0/reactor.go

@@ -41,7 +41,7 @@ const (
 type consensusReactor interface {
 	// for when we switch from blockchain reactor and fast sync to
 	// the consensus machine
-	SwitchToConsensus(sm.State, uint64)
+	SwitchToConsensus(state sm.State, skipWAL bool)
 }
 
 type peerError struct {
@@ -115,11 +115,25 @@ func (bcR *BlockchainReactor) OnStart() error {
 		if err != nil {
 			return err
 		}
-		go bcR.poolRoutine()
+		go bcR.poolRoutine(false)
 	}
 	return nil
 }
 
+// SwitchToFastSync is called by the state sync reactor when switching to fast sync.
+func (bcR *BlockchainReactor) SwitchToFastSync(state sm.State) error {
+	bcR.fastSync = true
+	bcR.initialState = state
+
+	bcR.pool.height = state.LastBlockHeight + 1
+	err := bcR.pool.Start()
+	if err != nil {
+		return err
+	}
+	go bcR.poolRoutine(true)
+	return nil
+}
+
 // OnStop implements service.Service.
 func (bcR *BlockchainReactor) OnStop() {
 	bcR.pool.Stop()
@@ -213,7 +227,7 @@ func (bcR *BlockchainReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte)
 
 // Handle messages from the poolReactor telling the reactor what to do.
 // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
-func (bcR *BlockchainReactor) poolRoutine() {
+func (bcR *BlockchainReactor) poolRoutine(stateSynced bool) {
 
 	trySyncTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond)
 	statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
@@ -273,7 +287,7 @@ FOR_LOOP:
 				bcR.pool.Stop()
 				conR, ok := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
 				if ok {
-					conR.SwitchToConsensus(state, blocksSynced)
+					conR.SwitchToConsensus(state, blocksSynced > 0 || stateSynced)
 				}
 				// else {
 				// should only happen during testing

blockchain/v1/reactor.go

@@ -44,7 +44,7 @@ var (
 type consensusReactor interface {
 	// for when we switch from blockchain reactor and fast sync to
 	// the consensus machine
-	SwitchToConsensus(sm.State, uint64)
+	SwitchToConsensus(state sm.State, skipWAL bool)
 }
 
 // BlockchainReactor handles long-term catchup syncing.
@@ -57,7 +57,8 @@ type BlockchainReactor struct {
 	blockExec *sm.BlockExecutor
 	store     *store.BlockStore
 
-	fastSync bool
+	fastSync    bool
+	stateSynced bool
 
 	fsm          *BcReactorFSM
 	blocksSynced uint64
@@ -154,6 +155,19 @@ func (bcR *BlockchainReactor) OnStop() {
 	_ = bcR.Stop()
 }
 
+// SwitchToFastSync is called by the state sync reactor when switching to fast sync.
+func (bcR *BlockchainReactor) SwitchToFastSync(state sm.State) error {
+	bcR.fastSync = true
+	bcR.initialState = state
+	bcR.state = state
+	bcR.stateSynced = true
+
+	bcR.fsm = NewFSM(state.LastBlockHeight+1, bcR)
+	bcR.fsm.SetLogger(bcR.Logger)
+	go bcR.poolRoutine()
+	return nil
+}
+
 // GetChannels implements Reactor
 func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
 	return []*p2p.ChannelDescriptor{
@@ -474,7 +488,7 @@ func (bcR *BlockchainReactor) sendBlockRequest(peerID p2p.ID, height int64) erro
 func (bcR *BlockchainReactor) switchToConsensus() {
 	conR, ok := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
 	if ok {
-		conR.SwitchToConsensus(bcR.state, bcR.blocksSynced)
+		conR.SwitchToConsensus(bcR.state, bcR.blocksSynced > 0 || bcR.stateSynced)
 		bcR.eventsFromFSMCh <- bcFsmMessage{event: syncFinishedEv}
 	}
 	// else {

blockchain/v1/reactor_test.go

@@ -167,7 +167,7 @@ type consensusReactorTest struct {
 	mtx                 sync.Mutex
 }
 
-func (conR *consensusReactorTest) SwitchToConsensus(state sm.State, blocksSynced uint64) {
+func (conR *consensusReactorTest) SwitchToConsensus(state sm.State, blocksSynced bool) {
 	conR.mtx.Lock()
 	defer conR.mtx.Unlock()
 	conR.switchedToConsensus = true

blockchain/v2/io.go

@@ -16,7 +16,7 @@ type iIO interface {
 
 	broadcastStatusRequest(base int64, height int64)
 
-	trySwitchToConsensus(state state.State, blocksSynced int)
+	trySwitchToConsensus(state state.State, skipWAL bool)
 }
 
 type switchIO struct {
@@ -37,7 +37,7 @@ const (
 type consensusReactor interface {
 	// for when we switch from blockchain reactor and fast sync to
 	// the consensus machine
-	SwitchToConsensus(state.State, int)
+	SwitchToConsensus(state state.State, skipWAL bool)
 }
 
 func (sio *switchIO) sendBlockRequest(peerID p2p.ID, height int64) error {
@@ -97,10 +97,10 @@ func (sio *switchIO) sendBlockNotFound(height int64, peerID p2p.ID) error {
 	return nil
 }
 
-func (sio *switchIO) trySwitchToConsensus(state state.State, blocksSynced int) {
+func (sio *switchIO) trySwitchToConsensus(state state.State, skipWAL bool) {
 	conR, ok := sio.sw.Reactor("CONSENSUS").(consensusReactor)
 	if ok {
-		conR.SwitchToConsensus(state, blocksSynced)
+		conR.SwitchToConsensus(state, skipWAL)
 	}
 }
 

blockchain/v2/reactor.go

@@ -129,8 +129,8 @@ type blockStore interface {
 type BlockchainReactor struct {
 	p2p.BaseReactor
 
+	fastSync  bool       // if true, enable fast sync on start
 	events    chan Event // XXX: Rename eventsFromPeers
-	stopDemux chan struct{}
 	scheduler *Routine
 	processor *Routine
 	logger    log.Logger
@@ -157,7 +157,7 @@ type blockApplier interface {
 // XXX: unify naming in this package around tmState
 // XXX: V1 stores a copy of state as initialState, which is never mutated. Is that nessesary?
 func newReactor(state state.State, store blockStore, reporter behaviour.Reporter,
-	blockApplier blockApplier, bufferSize int) *BlockchainReactor {
+	blockApplier blockApplier, bufferSize int, fastSync bool) *BlockchainReactor {
 	scheduler := newScheduler(state.LastBlockHeight, time.Now())
 	pContext := newProcessorContext(store, blockApplier, state)
 	// TODO: Fix naming to just newProcesssor
@@ -166,12 +166,12 @@ func newReactor(state state.State, store blockStore, reporter behaviour.Reporter
 
 	return &BlockchainReactor{
 		events:    make(chan Event, bufferSize),
-		stopDemux: make(chan struct{}),
 		scheduler: newRoutine("scheduler", scheduler.handle, bufferSize),
 		processor: newRoutine("processor", processor.handle, bufferSize),
 		store:     store,
 		reporter:  reporter,
 		logger:    log.NewNopLogger(),
+		fastSync:  fastSync,
 	}
 }
 
@@ -182,7 +182,7 @@ func NewBlockchainReactor(
 	store blockStore,
 	fastSync bool) *BlockchainReactor {
 	reporter := behaviour.NewMockReporter()
-	return newReactor(state, store, reporter, blockApplier, 1000)
+	return newReactor(state, store, reporter, blockApplier, 1000, fastSync)
 }
 
 // SetSwitch implements Reactor interface.
@@ -226,9 +226,11 @@ func (r *BlockchainReactor) SetLogger(logger log.Logger) {
 // Start implements cmn.Service interface
 func (r *BlockchainReactor) Start() error {
 	r.reporter = behaviour.NewSwitchReporter(r.BaseReactor.Switch)
-	go r.scheduler.start()
-	go r.processor.start()
-	go r.demux()
+	if r.fastSync {
+		go r.scheduler.start()
+		go r.processor.start()
+		go r.demux()
+	}
 	return nil
 }
 
@@ -306,19 +308,29 @@ func (r *BlockchainReactor) demux() {
 		processBlockFreq = 20 * time.Millisecond
 		doProcessBlockCh = make(chan struct{}, 1)
 		doProcessBlockTk = time.NewTicker(processBlockFreq)
+	)
+	defer doProcessBlockTk.Stop()
 
+	var (
 		prunePeerFreq = 1 * time.Second
 		doPrunePeerCh = make(chan struct{}, 1)
 		doPrunePeerTk = time.NewTicker(prunePeerFreq)
+	)
+	defer doPrunePeerTk.Stop()
 
+	var (
 		scheduleFreq = 20 * time.Millisecond
 		doScheduleCh = make(chan struct{}, 1)
 		doScheduleTk = time.NewTicker(scheduleFreq)
+	)
+	defer doScheduleTk.Stop()
 
+	var (
 		statusFreq = 10 * time.Second
 		doStatusCh = make(chan struct{}, 1)
 		doStatusTk = time.NewTicker(statusFreq)
 	)
+	defer doStatusTk.Stop()
 
 	// XXX: Extract timers to make testing atemporal
 	for {
@@ -355,14 +367,20 @@ func (r *BlockchainReactor) demux() {
 		case <-doStatusCh:
 			r.io.broadcastStatusRequest(r.store.Base(), r.SyncHeight())
 
-		// Events from peers
-		case event := <-r.events:
+		// Events from peers. Closing the channel signals event loop termination.
+		case event, ok := <-r.events:
+			if !ok {
+				r.logger.Info("Stopping event processing")
+				return
+			}
 			switch event := event.(type) {
 			case bcStatusResponse:
 				r.setMaxPeerHeight(event.height)
 				r.scheduler.send(event)
 			case bcAddNewPeer, bcRemovePeer, bcBlockResponse, bcNoBlockResponse:
 				r.scheduler.send(event)
+			default:
+				r.logger.Error("Received unknown event", "event", fmt.Sprintf("%T", event))
 			}
 
 		// Incremental events form scheduler
@@ -378,6 +396,9 @@ func (r *BlockchainReactor) demux() {
 			case scFinishedEv:
 				r.processor.send(event)
 				r.scheduler.stop()
+			case noOpEvent:
+			default:
+				r.logger.Error("Received unknown scheduler event", "event", fmt.Sprintf("%T", event))
 			}
 
 		// Incremental events from processor
@@ -395,22 +416,30 @@ func (r *BlockchainReactor) demux() {
 			case pcBlockVerificationFailure:
 				r.scheduler.send(event)
 			case pcFinished:
-				r.io.trySwitchToConsensus(event.tmState, event.blocksSynced)
+				r.io.trySwitchToConsensus(event.tmState, event.blocksSynced > 0)
 				r.processor.stop()
+			case noOpEvent:
+			default:
+				r.logger.Error("Received unknown processor event", "event", fmt.Sprintf("%T", event))
 			}
 
-		// Terminal events from scheduler
+		// Terminal event from scheduler
 		case err := <-r.scheduler.final():
-			r.logger.Info(fmt.Sprintf("scheduler final %s", err))
-			// send the processor stop?
+			switch err {
+			case nil:
+				r.logger.Info("Scheduler stopped")
+			default:
+				r.logger.Error("Scheduler aborted with error", "err", err)
+			}
 
 		// Terminal event from processor
-		case event := <-r.processor.final():
-			r.logger.Info(fmt.Sprintf("processor final %s", event))
-
-		case <-r.stopDemux:
-			r.logger.Info("demuxing stopped")
-			return
+		case err := <-r.processor.final():
+			switch err {
+			case nil:
+				r.logger.Info("Processor stopped")
+			default:
+				r.logger.Error("Processor aborted with error", "err", err)
+			}
 		}
 	}
 }
@@ -421,7 +450,6 @@ func (r *BlockchainReactor) Stop() error {
 
 	r.scheduler.stop()
 	r.processor.stop()
-	close(r.stopDemux)
 	close(r.events)
 
 	r.logger.Info("reactor stopped")

blockchain/v2/reactor_test.go

@@ -118,7 +118,7 @@ func (sio *mockSwitchIo) sendBlockNotFound(height int64, peerID p2p.ID) error {
 	return nil
 }
 
-func (sio *mockSwitchIo) trySwitchToConsensus(state sm.State, blocksSynced int) {
+func (sio *mockSwitchIo) trySwitchToConsensus(state sm.State, skipWAL bool) {
 	sio.mtx.Lock()
 	defer sio.mtx.Unlock()
 	sio.switchedToConsensus = true
@@ -157,7 +157,7 @@ func newTestReactor(p testReactorParams) *BlockchainReactor {
 		sm.SaveState(db, state)
 	}
 
-	r := newReactor(state, store, reporter, appl, p.bufferSize)
+	r := newReactor(state, store, reporter, appl, p.bufferSize, true)
 	logger := log.TestingLogger()
 	r.SetLogger(logger.With("module", "blockchain"))
 

blockchain/v2/routine.go

@@ -69,9 +69,11 @@ func (rt *Routine) start() {
 
 	for {
 		events, err := rt.queue.Get(1)
-		if err != nil {
-			rt.logger.Info(fmt.Sprintf("%s: stopping\n", rt.name))
-			rt.terminate(fmt.Errorf("stopped"))
+		if err == queue.ErrDisposed {
+			rt.terminate(nil)
+			return
+		} else if err != nil {
+			rt.terminate(err)
 			return
 		}
 		oEvent, err := rt.handle(events[0].(Event))
@@ -131,6 +133,7 @@ func (rt *Routine) final() chan error {
 
 // XXX: Maybe get rid of this
 func (rt *Routine) terminate(reason error) {
-	close(rt.out)
+	// We don't close the rt.out channel here, to avoid spinning on the closed channel
+	// in the event loop.
 	rt.fin <- reason
 }