From ad8aa2bdc0ad40261ba5aac2359cc57c491af122 Mon Sep 17 00:00:00 2001 From: William Banfield Date: Mon, 31 Oct 2022 11:57:25 -0400 Subject: [PATCH] remove erroneously added blockchain reactors --- blockchain/v1/reactor.go | 568 -------------------------------------- blockchain/v2/reactor.go | 572 --------------------------------------- 2 files changed, 1140 deletions(-) delete mode 100644 blockchain/v1/reactor.go delete mode 100644 blockchain/v2/reactor.go diff --git a/blockchain/v1/reactor.go b/blockchain/v1/reactor.go deleted file mode 100644 index 133bf179d..000000000 --- a/blockchain/v1/reactor.go +++ /dev/null @@ -1,568 +0,0 @@ -package v1 - -import ( - "fmt" - "time" - - "github.com/cosmos/gogoproto/proto" - - "github.com/tendermint/tendermint/behaviour" - bc "github.com/tendermint/tendermint/blockchain" - "github.com/tendermint/tendermint/libs/log" - "github.com/tendermint/tendermint/p2p" - bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain" - sm "github.com/tendermint/tendermint/state" - "github.com/tendermint/tendermint/store" - "github.com/tendermint/tendermint/types" -) - -const ( - // BlockchainChannel is a channel for blocks and status updates (`BlockStore` height) - BlockchainChannel = byte(0x40) - trySyncIntervalMS = 10 - trySendIntervalMS = 10 - - // ask for best height every 10s - statusUpdateIntervalSeconds = 10 -) - -var ( - // Maximum number of requests that can be pending per peer, i.e. for which requests have been sent but blocks - // have not been received. - maxRequestsPerPeer = 20 - // Maximum number of block requests for the reactor, pending or for which blocks have been received. - maxNumRequests = 64 -) - -type consensusReactor interface { - // for when we switch from blockchain reactor and fast sync to - // the consensus machine - SwitchToConsensus(state sm.State, skipWAL bool) -} - -// BlockchainReactor handles long-term catchup syncing. -type BlockchainReactor struct { - p2p.BaseReactor - - initialState sm.State // immutable - state sm.State - - blockExec *sm.BlockExecutor - store *store.BlockStore - - fastSync bool - stateSynced bool - - fsm *BcReactorFSM - blocksSynced uint64 - - // Receive goroutine forwards messages to this channel to be processed in the context of the poolRoutine. - messagesForFSMCh chan bcReactorMessage - - // Switch goroutine may send RemovePeer to the blockchain reactor. This is an error message that is relayed - // to this channel to be processed in the context of the poolRoutine. - errorsForFSMCh chan bcReactorMessage - - // This channel is used by the FSM and indirectly the block pool to report errors to the blockchain reactor and - // the switch. - eventsFromFSMCh chan bcFsmMessage - - swReporter *behaviour.SwitchReporter -} - -// NewBlockchainReactor returns new reactor instance. -func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *store.BlockStore, - fastSync bool) *BlockchainReactor { - - if state.LastBlockHeight != store.Height() { - panic(fmt.Sprintf("state (%v) and store (%v) height mismatch", state.LastBlockHeight, - store.Height())) - } - - const capacity = 1000 - eventsFromFSMCh := make(chan bcFsmMessage, capacity) - messagesForFSMCh := make(chan bcReactorMessage, capacity) - errorsForFSMCh := make(chan bcReactorMessage, capacity) - - startHeight := store.Height() + 1 - if startHeight == 1 { - startHeight = state.InitialHeight - } - bcR := &BlockchainReactor{ - initialState: state, - state: state, - blockExec: blockExec, - fastSync: fastSync, - store: store, - messagesForFSMCh: messagesForFSMCh, - eventsFromFSMCh: eventsFromFSMCh, - errorsForFSMCh: errorsForFSMCh, - } - fsm := NewFSM(startHeight, bcR) - bcR.fsm = fsm - bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR) - // bcR.swReporter = behaviour.NewSwitchReporter(bcR.BaseReactor.Switch) - - return bcR -} - -// bcReactorMessage is used by the reactor to send messages to the FSM. -type bcReactorMessage struct { - event bReactorEvent - data bReactorEventData -} - -type bFsmEvent uint - -const ( - // message type events - peerErrorEv = iota + 1 - syncFinishedEv -) - -type bFsmEventData struct { - peerID p2p.ID - err error -} - -// bcFsmMessage is used by the FSM to send messages to the reactor -type bcFsmMessage struct { - event bFsmEvent - data bFsmEventData -} - -// SetLogger implements service.Service by setting the logger on reactor and pool. -func (bcR *BlockchainReactor) SetLogger(l log.Logger) { - bcR.BaseService.Logger = l - bcR.fsm.SetLogger(l) -} - -// OnStart implements service.Service. -func (bcR *BlockchainReactor) OnStart() error { - bcR.swReporter = behaviour.NewSwitchReporter(bcR.BaseReactor.Switch) - if bcR.fastSync { - go bcR.poolRoutine() - } - return nil -} - -// OnStop implements service.Service. -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{ - { - ID: BlockchainChannel, - Priority: 10, - SendQueueCapacity: 2000, - RecvBufferCapacity: 50 * 4096, - RecvMessageCapacity: bc.MaxMsgSize, - MessageType: &bcproto.Message{}, - }, - } -} - -// AddPeer implements Reactor by sending our state to peer. -func (bcR *BlockchainReactor) AddPeer(peer p2p.Peer) { - peer.NewSend(p2p.Envelope{ - ChannelID: BlockchainChannel, - Message: &bcproto.StatusResponse{ - Base: bcR.store.Base(), - Height: bcR.store.Height(), - }, - }) - // it's OK if send fails. will try later in poolRoutine - - // peer is added to the pool once we receive the first - // bcStatusResponseMessage from the peer and call pool.updatePeer() -} - -// sendBlockToPeer loads a block and sends it to the requesting peer. -// If the block doesn't exist a bcNoBlockResponseMessage is sent. -// If all nodes are honest, no node should be requesting for a block that doesn't exist. -func (bcR *BlockchainReactor) sendBlockToPeer(msg *bcproto.BlockRequest, - src p2p.Peer) (queued bool) { - - block := bcR.store.LoadBlock(msg.Height) - if block != nil { - pbbi, err := block.ToProto() - if err != nil { - bcR.Logger.Error("Could not send block message to peer", "err", err) - return false - } - return src.NewTrySend(p2p.Envelope{ - ChannelID: BlockchainChannel, - Message: &bcproto.BlockResponse{Block: pbbi}, - }) - } - - bcR.Logger.Info("peer asking for a block we don't have", "src", src, "height", msg.Height) - - return src.NewTrySend(p2p.Envelope{ - ChannelID: BlockchainChannel, - Message: &bcproto.NoBlockResponse{Height: msg.Height}, - }) -} - -func (bcR *BlockchainReactor) sendStatusResponseToPeer(msg *bcproto.StatusRequest, src p2p.Peer) (queued bool) { - return src.NewTrySend(p2p.Envelope{ - ChannelID: BlockchainChannel, - Message: &bcproto.StatusResponse{ - Base: bcR.store.Base(), - Height: bcR.store.Height(), - }, - }) -} - -// RemovePeer implements Reactor by removing peer from the pool. -func (bcR *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) { - msgData := bcReactorMessage{ - event: peerRemoveEv, - data: bReactorEventData{ - peerID: peer.ID(), - err: errSwitchRemovesPeer, - }, - } - bcR.errorsForFSMCh <- msgData -} - -// Receive implements Reactor by handling 4 types of messages (look below). -func (bcR *BlockchainReactor) NewReceive(e p2p.Envelope) { - if err := bc.ValidateMsg(e.Message); err != nil { - bcR.Logger.Error("peer sent us invalid msg", "peer", e.Src, "msg", e.Message, "err", err) - _ = bcR.swReporter.Report(behaviour.BadMessage(e.Src.ID(), err.Error())) - return - } - - bcR.Logger.Debug("Receive", "src", e.Src, "chID", e.ChannelID, "msg", e.Message) - - switch msg := e.Message.(type) { - case *bcproto.BlockRequest: - if queued := bcR.sendBlockToPeer(msg, e.Src); !queued { - // Unfortunately not queued since the queue is full. - bcR.Logger.Error("Could not send block message to peer", "src", e.Src, "height", msg.Height) - } - - case *bcproto.StatusRequest: - // Send peer our state. - if queued := bcR.sendStatusResponseToPeer(msg, e.Src); !queued { - // Unfortunately not queued since the queue is full. - bcR.Logger.Error("Could not send status message to peer", "src", e.Src) - } - - case *bcproto.BlockResponse: - bi, err := types.BlockFromProto(msg.Block) - if err != nil { - bcR.Logger.Error("error transition block from protobuf", "err", err) - return - } - msgForFSM := bcReactorMessage{ - event: blockResponseEv, - data: bReactorEventData{ - peerID: e.Src.ID(), - height: bi.Height, - block: bi, - }, - } - bcR.Logger.Info("Received", "src", e.Src, "height", bi.Height) - bcR.messagesForFSMCh <- msgForFSM - case *bcproto.NoBlockResponse: - msgForFSM := bcReactorMessage{ - event: noBlockResponseEv, - data: bReactorEventData{ - peerID: e.Src.ID(), - height: msg.Height, - }, - } - bcR.Logger.Debug("Peer does not have requested block", "peer", e.Src, "height", msg.Height) - bcR.messagesForFSMCh <- msgForFSM - - case *bcproto.StatusResponse: - // Got a peer status. Unverified. - msgForFSM := bcReactorMessage{ - event: statusResponseEv, - data: bReactorEventData{ - peerID: e.Src.ID(), - height: msg.Height, - }, - } - bcR.messagesForFSMCh <- msgForFSM - - default: - bcR.Logger.Error(fmt.Sprintf("unknown message type %T", msg)) - } -} - -func (bcR *BlockchainReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) { - var msg *bcproto.Message - err := proto.Unmarshal(msgBytes, msg) - if err != nil { - panic(err) - } - uw, err := msg.Unwrap() - if err != nil { - panic(err) - } - bcR.NewReceive(p2p.Envelope{ - ChannelID: chID, - Src: peer, - Message: uw, - }) -} - -// processBlocksRoutine processes blocks until signlaed to stop over the stopProcessing channel -func (bcR *BlockchainReactor) processBlocksRoutine(stopProcessing chan struct{}) { - - processReceivedBlockTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond) - doProcessBlockCh := make(chan struct{}, 1) - - lastHundred := time.Now() - lastRate := 0.0 - -ForLoop: - for { - select { - case <-stopProcessing: - bcR.Logger.Info("finishing block execution") - break ForLoop - case <-processReceivedBlockTicker.C: // try to execute blocks - select { - case doProcessBlockCh <- struct{}{}: - default: - } - case <-doProcessBlockCh: - for { - err := bcR.processBlock() - if err == errMissingBlock { - break - } - // Notify FSM of block processing result. - msgForFSM := bcReactorMessage{ - event: processedBlockEv, - data: bReactorEventData{ - err: err, - }, - } - _ = bcR.fsm.Handle(&msgForFSM) - - if err != nil { - break - } - - bcR.blocksSynced++ - if bcR.blocksSynced%100 == 0 { - lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds()) - height, maxPeerHeight := bcR.fsm.Status() - bcR.Logger.Info("Fast Sync Rate", "height", height, - "max_peer_height", maxPeerHeight, "blocks/s", lastRate) - lastHundred = time.Now() - } - } - } - } -} - -// poolRoutine receives and handles messages from the Receive() routine and from the FSM. -func (bcR *BlockchainReactor) poolRoutine() { - - bcR.fsm.Start() - - sendBlockRequestTicker := time.NewTicker(trySendIntervalMS * time.Millisecond) - statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second) - - stopProcessing := make(chan struct{}, 1) - go bcR.processBlocksRoutine(stopProcessing) - -ForLoop: - for { - select { - - case <-sendBlockRequestTicker.C: - if !bcR.fsm.NeedsBlocks() { - continue - } - _ = bcR.fsm.Handle(&bcReactorMessage{ - event: makeRequestsEv, - data: bReactorEventData{ - maxNumRequests: maxNumRequests}}) - - case <-statusUpdateTicker.C: - // Ask for status updates. - go bcR.sendStatusRequest() - - case msg := <-bcR.messagesForFSMCh: - // Sent from the Receive() routine when status (statusResponseEv) and - // block (blockResponseEv) response events are received - _ = bcR.fsm.Handle(&msg) - - case msg := <-bcR.errorsForFSMCh: - // Sent from the switch.RemovePeer() routine (RemovePeerEv) and - // FSM state timer expiry routine (stateTimeoutEv). - _ = bcR.fsm.Handle(&msg) - - case msg := <-bcR.eventsFromFSMCh: - switch msg.event { - case syncFinishedEv: - stopProcessing <- struct{}{} - // Sent from the FSM when it enters finished state. - break ForLoop - case peerErrorEv: - // Sent from the FSM when it detects peer error - bcR.reportPeerErrorToSwitch(msg.data.err, msg.data.peerID) - if msg.data.err == errNoPeerResponse { - // Sent from the peer timeout handler routine - _ = bcR.fsm.Handle(&bcReactorMessage{ - event: peerRemoveEv, - data: bReactorEventData{ - peerID: msg.data.peerID, - err: msg.data.err, - }, - }) - } - // else { - // For slow peers, or errors due to blocks received from wrong peer - // the FSM had already removed the peers - // } - default: - bcR.Logger.Error("Event from FSM not supported", "type", msg.event) - } - - case <-bcR.Quit(): - break ForLoop - } - } -} - -func (bcR *BlockchainReactor) reportPeerErrorToSwitch(err error, peerID p2p.ID) { - peer := bcR.Switch.Peers().Get(peerID) - if peer != nil { - _ = bcR.swReporter.Report(behaviour.BadMessage(peerID, err.Error())) - } -} - -func (bcR *BlockchainReactor) processBlock() error { - - first, second, err := bcR.fsm.FirstTwoBlocks() - if err != nil { - // We need both to sync the first block. - return err - } - - chainID := bcR.initialState.ChainID - - firstParts := first.MakePartSet(types.BlockPartSizeBytes) - firstPartSetHeader := firstParts.Header() - firstID := types.BlockID{Hash: first.Hash(), PartSetHeader: firstPartSetHeader} - // Finally, verify the first block using the second's commit - // NOTE: we can probably make this more efficient, but note that calling - // first.Hash() doesn't verify the tx contents, so MakePartSet() is - // currently necessary. - err = bcR.state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit) - if err != nil { - bcR.Logger.Error("error during commit verification", "err", err, - "first", first.Height, "second", second.Height) - return errBlockVerificationFailure - } - - bcR.store.SaveBlock(first, firstParts, second.LastCommit) - - bcR.state, _, err = bcR.blockExec.ApplyBlock(bcR.state, firstID, first) - if err != nil { - panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", first.Height, first.Hash(), err)) - } - - return nil -} - -// Implements bcRNotifier -// sendStatusRequest broadcasts `BlockStore` height. -func (bcR *BlockchainReactor) sendStatusRequest() { - bcR.Switch.NewBroadcast(p2p.Envelope{ - ChannelID: BlockchainChannel, - Message: &bcproto.StatusRequest{}, - }) -} - -// Implements bcRNotifier -// BlockRequest sends `BlockRequest` height. -func (bcR *BlockchainReactor) sendBlockRequest(peerID p2p.ID, height int64) error { - peer := bcR.Switch.Peers().Get(peerID) - if peer == nil { - return errNilPeerForBlockRequest - } - - queued := peer.NewTrySend(p2p.Envelope{ - ChannelID: BlockchainChannel, - Message: &bcproto.BlockRequest{Height: height}, - }) - if !queued { - return errSendQueueFull - } - return nil -} - -// Implements bcRNotifier -func (bcR *BlockchainReactor) switchToConsensus() { - conR, ok := bcR.Switch.Reactor("CONSENSUS").(consensusReactor) - if ok { - conR.SwitchToConsensus(bcR.state, bcR.blocksSynced > 0 || bcR.stateSynced) - bcR.eventsFromFSMCh <- bcFsmMessage{event: syncFinishedEv} - } - // else { - // Should only happen during testing. - // } -} - -// Implements bcRNotifier -// Called by FSM and pool: -// - pool calls when it detects slow peer or when peer times out -// - FSM calls when: -// - adding a block (addBlock) fails -// - reactor processing of a block reports failure and FSM sends back the peers of first and second blocks -func (bcR *BlockchainReactor) sendPeerError(err error, peerID p2p.ID) { - bcR.Logger.Info("sendPeerError:", "peer", peerID, "error", err) - msgData := bcFsmMessage{ - event: peerErrorEv, - data: bFsmEventData{ - peerID: peerID, - err: err, - }, - } - bcR.eventsFromFSMCh <- msgData -} - -// Implements bcRNotifier -func (bcR *BlockchainReactor) resetStateTimer(name string, timer **time.Timer, timeout time.Duration) { - if timer == nil { - panic("nil timer pointer parameter") - } - if *timer == nil { - *timer = time.AfterFunc(timeout, func() { - msg := bcReactorMessage{ - event: stateTimeoutEv, - data: bReactorEventData{ - stateName: name, - }, - } - bcR.errorsForFSMCh <- msg - }) - } else { - (*timer).Reset(timeout) - } -} diff --git a/blockchain/v2/reactor.go b/blockchain/v2/reactor.go deleted file mode 100644 index 9d3289ae5..000000000 --- a/blockchain/v2/reactor.go +++ /dev/null @@ -1,572 +0,0 @@ -package v2 - -import ( - "errors" - "fmt" - "time" - - "github.com/cosmos/gogoproto/proto" - - "github.com/tendermint/tendermint/behaviour" - bc "github.com/tendermint/tendermint/blockchain" - "github.com/tendermint/tendermint/libs/log" - tmsync "github.com/tendermint/tendermint/libs/sync" - "github.com/tendermint/tendermint/p2p" - bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain" - "github.com/tendermint/tendermint/state" - "github.com/tendermint/tendermint/types" -) - -const ( - // chBufferSize is the buffer size of all event channels. - chBufferSize int = 1000 -) - -type blockStore interface { - LoadBlock(height int64) *types.Block - SaveBlock(*types.Block, *types.PartSet, *types.Commit) - Base() int64 - Height() int64 -} - -// BlockchainReactor handles fast sync protocol. -type BlockchainReactor struct { - p2p.BaseReactor - - fastSync bool // if true, enable fast sync on start - stateSynced bool // set to true when SwitchToFastSync is called by state sync - scheduler *Routine - processor *Routine - logger log.Logger - - mtx tmsync.RWMutex - maxPeerHeight int64 - syncHeight int64 - events chan Event // non-nil during a fast sync - - reporter behaviour.Reporter - io iIO - store blockStore -} - -//nolint:unused,deadcode -type blockVerifier interface { - VerifyCommit(chainID string, blockID types.BlockID, height int64, commit *types.Commit) error -} - -type blockApplier interface { - ApplyBlock(state state.State, blockID types.BlockID, block *types.Block) (state.State, int64, error) -} - -// XXX: unify naming in this package around tmState -func newReactor(state state.State, store blockStore, reporter behaviour.Reporter, - blockApplier blockApplier, fastSync bool) *BlockchainReactor { - initHeight := state.LastBlockHeight + 1 - if initHeight == 1 { - initHeight = state.InitialHeight - } - scheduler := newScheduler(initHeight, time.Now()) - pContext := newProcessorContext(store, blockApplier, state) - // TODO: Fix naming to just newProcesssor - // newPcState requires a processorContext - processor := newPcState(pContext) - - return &BlockchainReactor{ - scheduler: newRoutine("scheduler", scheduler.handle, chBufferSize), - processor: newRoutine("processor", processor.handle, chBufferSize), - store: store, - reporter: reporter, - logger: log.NewNopLogger(), - fastSync: fastSync, - } -} - -// NewBlockchainReactor creates a new reactor instance. -func NewBlockchainReactor( - state state.State, - blockApplier blockApplier, - store blockStore, - fastSync bool) *BlockchainReactor { - reporter := behaviour.NewMockReporter() - return newReactor(state, store, reporter, blockApplier, fastSync) -} - -// SetSwitch implements Reactor interface. -func (r *BlockchainReactor) SetSwitch(sw *p2p.Switch) { - r.Switch = sw - if sw != nil { - r.io = newSwitchIo(sw) - } else { - r.io = nil - } -} - -func (r *BlockchainReactor) setMaxPeerHeight(height int64) { - r.mtx.Lock() - defer r.mtx.Unlock() - if height > r.maxPeerHeight { - r.maxPeerHeight = height - } -} - -func (r *BlockchainReactor) setSyncHeight(height int64) { - r.mtx.Lock() - defer r.mtx.Unlock() - r.syncHeight = height -} - -// SyncHeight returns the height to which the BlockchainReactor has synced. -func (r *BlockchainReactor) SyncHeight() int64 { - r.mtx.RLock() - defer r.mtx.RUnlock() - return r.syncHeight -} - -// SetLogger sets the logger of the reactor. -func (r *BlockchainReactor) SetLogger(logger log.Logger) { - r.logger = logger - r.scheduler.setLogger(logger) - r.processor.setLogger(logger) -} - -// Start implements cmn.Service interface -func (r *BlockchainReactor) Start() error { - r.reporter = behaviour.NewSwitchReporter(r.BaseReactor.Switch) - if r.fastSync { - err := r.startSync(nil) - if err != nil { - return fmt.Errorf("failed to start fast sync: %w", err) - } - } - return nil -} - -// startSync begins a fast sync, signalled by r.events being non-nil. If state is non-nil, -// the scheduler and processor is updated with this state on startup. -func (r *BlockchainReactor) startSync(state *state.State) error { - r.mtx.Lock() - defer r.mtx.Unlock() - if r.events != nil { - return errors.New("fast sync already in progress") - } - r.events = make(chan Event, chBufferSize) - go r.scheduler.start() - go r.processor.start() - if state != nil { - <-r.scheduler.ready() - <-r.processor.ready() - r.scheduler.send(bcResetState{state: *state}) - r.processor.send(bcResetState{state: *state}) - } - go r.demux(r.events) - return nil -} - -// endSync ends a fast sync -func (r *BlockchainReactor) endSync() { - r.mtx.Lock() - defer r.mtx.Unlock() - if r.events != nil { - close(r.events) - } - r.events = nil - r.scheduler.stop() - r.processor.stop() -} - -// SwitchToFastSync is called by the state sync reactor when switching to fast sync. -func (r *BlockchainReactor) SwitchToFastSync(state state.State) error { - r.stateSynced = true - state = state.Copy() - return r.startSync(&state) -} - -// reactor generated ticker events: -// ticker for cleaning peers -type rTryPrunePeer struct { - priorityHigh - time time.Time -} - -func (e rTryPrunePeer) String() string { - return fmt.Sprintf("rTryPrunePeer{%v}", e.time) -} - -// ticker event for scheduling block requests -type rTrySchedule struct { - priorityHigh - time time.Time -} - -func (e rTrySchedule) String() string { - return fmt.Sprintf("rTrySchedule{%v}", e.time) -} - -// ticker for block processing -type rProcessBlock struct { - priorityNormal -} - -func (e rProcessBlock) String() string { - return "rProcessBlock" -} - -// reactor generated events based on blockchain related messages from peers: -// blockResponse message received from a peer -type bcBlockResponse struct { - priorityNormal - time time.Time - peerID p2p.ID - block *types.Block -} - -func (resp bcBlockResponse) String() string { - return fmt.Sprintf("bcBlockResponse{%d#%X from %v at %v}", - resp.block.Height, resp.block.Hash(), resp.peerID, resp.time) -} - -// blockNoResponse message received from a peer -type bcNoBlockResponse struct { - priorityNormal - time time.Time - peerID p2p.ID - height int64 -} - -func (resp bcNoBlockResponse) String() string { - return fmt.Sprintf("bcNoBlockResponse{%v has no block at height %d at %v}", - resp.peerID, resp.height, resp.time) -} - -// statusResponse message received from a peer -type bcStatusResponse struct { - priorityNormal - time time.Time - peerID p2p.ID - base int64 - height int64 -} - -func (resp bcStatusResponse) String() string { - return fmt.Sprintf("bcStatusResponse{%v is at height %d (base: %d) at %v}", - resp.peerID, resp.height, resp.base, resp.time) -} - -// new peer is connected -type bcAddNewPeer struct { - priorityNormal - peerID p2p.ID -} - -func (resp bcAddNewPeer) String() string { - return fmt.Sprintf("bcAddNewPeer{%v}", resp.peerID) -} - -// existing peer is removed -type bcRemovePeer struct { - priorityHigh - peerID p2p.ID - reason interface{} -} - -func (resp bcRemovePeer) String() string { - return fmt.Sprintf("bcRemovePeer{%v due to %v}", resp.peerID, resp.reason) -} - -// resets the scheduler and processor state, e.g. following a switch from state syncing -type bcResetState struct { - priorityHigh - state state.State -} - -func (e bcResetState) String() string { - return fmt.Sprintf("bcResetState{%v}", e.state) -} - -// Takes the channel as a parameter to avoid race conditions on r.events. -func (r *BlockchainReactor) demux(events <-chan Event) { - var lastRate = 0.0 - var lastHundred = time.Now() - - var ( - 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() - doStatusCh <- struct{}{} // immediately broadcast to get status of existing peers - - // XXX: Extract timers to make testing atemporal - for { - select { - // Pacers: send at most per frequency but don't saturate - case <-doProcessBlockTk.C: - select { - case doProcessBlockCh <- struct{}{}: - default: - } - case <-doPrunePeerTk.C: - select { - case doPrunePeerCh <- struct{}{}: - default: - } - case <-doScheduleTk.C: - select { - case doScheduleCh <- struct{}{}: - default: - } - case <-doStatusTk.C: - select { - case doStatusCh <- struct{}{}: - default: - } - - // Tickers: perform tasks periodically - case <-doScheduleCh: - r.scheduler.send(rTrySchedule{time: time.Now()}) - case <-doPrunePeerCh: - r.scheduler.send(rTryPrunePeer{time: time.Now()}) - case <-doProcessBlockCh: - r.processor.send(rProcessBlock{}) - case <-doStatusCh: - r.io.broadcastStatusRequest() - - // Events from peers. Closing the channel signals event loop termination. - case event, ok := <-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 unexpected event", "event", fmt.Sprintf("%T", event)) - } - - // Incremental events from scheduler - case event := <-r.scheduler.next(): - switch event := event.(type) { - case scBlockReceived: - r.processor.send(event) - case scPeerError: - r.processor.send(event) - if err := r.reporter.Report(behaviour.BadMessage(event.peerID, "scPeerError")); err != nil { - r.logger.Error("Error reporting peer", "err", err) - } - case scBlockRequest: - if err := r.io.sendBlockRequest(event.peerID, event.height); err != nil { - r.logger.Error("Error sending block request", "err", err) - } - case scFinishedEv: - r.processor.send(event) - r.scheduler.stop() - case scSchedulerFail: - r.logger.Error("Scheduler failure", "err", event.reason.Error()) - case scPeersPruned: - // Remove peers from the processor. - for _, peerID := range event.peers { - r.processor.send(scPeerError{peerID: peerID, reason: errors.New("peer was pruned")}) - } - r.logger.Debug("Pruned peers", "count", len(event.peers)) - case noOpEvent: - default: - r.logger.Error("Received unexpected scheduler event", "event", fmt.Sprintf("%T", event)) - } - - // Incremental events from processor - case event := <-r.processor.next(): - switch event := event.(type) { - case pcBlockProcessed: - r.setSyncHeight(event.height) - if r.syncHeight%100 == 0 { - lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds()) - r.logger.Info("Fast Sync Rate", "height", r.syncHeight, - "max_peer_height", r.maxPeerHeight, "blocks/s", lastRate) - lastHundred = time.Now() - } - r.scheduler.send(event) - case pcBlockVerificationFailure: - r.scheduler.send(event) - case pcFinished: - r.logger.Info("Fast sync complete, switching to consensus") - if !r.io.trySwitchToConsensus(event.tmState, event.blocksSynced > 0 || r.stateSynced) { - r.logger.Error("Failed to switch to consensus reactor") - } - r.endSync() - return - case noOpEvent: - default: - r.logger.Error("Received unexpected processor event", "event", fmt.Sprintf("%T", event)) - } - - // Terminal event from scheduler - case err := <-r.scheduler.final(): - switch err { - case nil: - r.logger.Info("Scheduler stopped") - default: - r.logger.Error("Scheduler aborted with error", "err", err) - } - - // Terminal event from processor - case err := <-r.processor.final(): - switch err { - case nil: - r.logger.Info("Processor stopped") - default: - r.logger.Error("Processor aborted with error", "err", err) - } - } - } -} - -// Stop implements cmn.Service interface. -func (r *BlockchainReactor) Stop() error { - r.logger.Info("reactor stopping") - r.endSync() - r.logger.Info("reactor stopped") - return nil -} - -// Receive implements Reactor by handling different message types. -func (r *BlockchainReactor) NewReceive(e p2p.Envelope) { - if err := bc.ValidateMsg(e.Message); err != nil { - r.logger.Error("peer sent us invalid msg", "peer", e.Src, "msg", e.Message, "err", err) - _ = r.reporter.Report(behaviour.BadMessage(e.Src.ID(), err.Error())) - return - } - - r.logger.Debug("Receive", "src", e.Src.ID(), "chID", e.ChannelID, "msg", e.Message) - - switch msg := e.Message.(type) { - case *bcproto.StatusRequest: - if err := r.io.sendStatusResponse(r.store.Base(), r.store.Height(), e.Src.ID()); err != nil { - r.logger.Error("Could not send status message to peer", "src", e.Src) - } - - case *bcproto.BlockRequest: - block := r.store.LoadBlock(msg.Height) - if block != nil { - if err := r.io.sendBlockToPeer(block, e.Src.ID()); err != nil { - r.logger.Error("Could not send block message to peer: ", err) - } - } else { - r.logger.Info("peer asking for a block we don't have", "src", e.Src, "height", msg.Height) - peerID := e.Src.ID() - if err := r.io.sendBlockNotFound(msg.Height, peerID); err != nil { - r.logger.Error("Couldn't send block not found: ", err) - } - } - - case *bcproto.StatusResponse: - r.mtx.RLock() - if r.events != nil { - r.events <- bcStatusResponse{peerID: e.Src.ID(), base: msg.Base, height: msg.Height} - } - r.mtx.RUnlock() - - case *bcproto.BlockResponse: - bi, err := types.BlockFromProto(msg.Block) - if err != nil { - r.logger.Error("error transitioning block from protobuf", "err", err) - return - } - r.mtx.RLock() - if r.events != nil { - r.events <- bcBlockResponse{ - peerID: e.Src.ID(), - block: bi, - time: time.Now(), - } - } - r.mtx.RUnlock() - - case *bcproto.NoBlockResponse: - r.mtx.RLock() - if r.events != nil { - r.events <- bcNoBlockResponse{peerID: e.Src.ID(), height: msg.Height, time: time.Now()} - } - r.mtx.RUnlock() - } -} - -func (r *BlockchainReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) { - var msg *bcproto.Message - err := proto.Unmarshal(msgBytes, msg) - if err != nil { - panic(err) - } - uw, err := msg.Unwrap() - if err != nil { - panic(err) - } - r.NewReceive(p2p.Envelope{ - ChannelID: chID, - Src: peer, - Message: uw, - }) -} - -// AddPeer implements Reactor interface -func (r *BlockchainReactor) AddPeer(peer p2p.Peer) { - err := r.io.sendStatusResponse(r.store.Base(), r.store.Height(), peer.ID()) - if err != nil { - r.logger.Error("Could not send status message to peer new", "src", peer.ID, "height", r.SyncHeight()) - } - r.mtx.RLock() - defer r.mtx.RUnlock() - if r.events != nil { - r.events <- bcAddNewPeer{peerID: peer.ID()} - } -} - -// RemovePeer implements Reactor interface. -func (r *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) { - r.mtx.RLock() - defer r.mtx.RUnlock() - if r.events != nil { - r.events <- bcRemovePeer{ - peerID: peer.ID(), - reason: reason, - } - } -} - -// GetChannels implements Reactor -func (r *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor { - return []*p2p.ChannelDescriptor{ - { - ID: BlockchainChannel, - Priority: 5, - SendQueueCapacity: 2000, - RecvBufferCapacity: 50 * 4096, - RecvMessageCapacity: bc.MaxMsgSize, - MessageType: &bcproto.Message{}, - }, - } -}