mirror of
https://github.com/tendermint/tendermint.git
synced 2026-09-25 09:24:30 +00:00
Erik GrinakerandGitHub
eb443f4b77
Integrates the blockchain v2 reactor with state sync, fixes #4765. This mostly involves deferring fast syncing until after state sync completes. I tried a few different approaches, this was the least effort: * `Reactor.events` is `nil` if no fast sync is in progress, in which case events are not dispatched - most importantly `AddPeer`. * Accept status messages from unknown peers in the scheduler and register them as ready. On fast sync startup, broadcast status requests to all existing peers. * When switching from state sync, first send a `bcResetState` message to the processor and scheduler to update their states - most importantly the initial block height. * When fast sync completes, shut down event loop, scheduler and processor, and set `events` channel to `nil`.
119 lines
3.0 KiB
Go
119 lines
3.0 KiB
Go
package v2
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/tendermint/tendermint/p2p"
|
|
"github.com/tendermint/tendermint/state"
|
|
"github.com/tendermint/tendermint/types"
|
|
)
|
|
|
|
type iIO interface {
|
|
sendBlockRequest(peerID p2p.ID, height int64) error
|
|
sendBlockToPeer(block *types.Block, peerID p2p.ID) error
|
|
sendBlockNotFound(height int64, peerID p2p.ID) error
|
|
sendStatusResponse(height int64, peerID p2p.ID) error
|
|
|
|
broadcastStatusRequest(base int64, height int64)
|
|
|
|
trySwitchToConsensus(state state.State, skipWAL bool) bool
|
|
}
|
|
|
|
type switchIO struct {
|
|
sw *p2p.Switch
|
|
}
|
|
|
|
func newSwitchIo(sw *p2p.Switch) *switchIO {
|
|
return &switchIO{
|
|
sw: sw,
|
|
}
|
|
}
|
|
|
|
const (
|
|
// BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
|
|
BlockchainChannel = byte(0x40)
|
|
)
|
|
|
|
type consensusReactor interface {
|
|
// for when we switch from blockchain reactor and fast sync to
|
|
// the consensus machine
|
|
SwitchToConsensus(state state.State, skipWAL bool)
|
|
}
|
|
|
|
func (sio *switchIO) sendBlockRequest(peerID p2p.ID, height int64) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
|
|
msgBytes := cdc.MustMarshalBinaryBare(&bcBlockRequestMessage{Height: height})
|
|
queued := peer.TrySend(BlockchainChannel, msgBytes)
|
|
if !queued {
|
|
return fmt.Errorf("send queue full")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) sendStatusResponse(height int64, peerID p2p.ID) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
msgBytes := cdc.MustMarshalBinaryBare(&bcStatusResponseMessage{Height: height})
|
|
|
|
if queued := peer.TrySend(BlockchainChannel, msgBytes); !queued {
|
|
return fmt.Errorf("peer queue full")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) sendBlockToPeer(block *types.Block, peerID p2p.ID) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
if block == nil {
|
|
panic("trying to send nil block")
|
|
}
|
|
msgBytes := cdc.MustMarshalBinaryBare(&bcBlockResponseMessage{Block: block})
|
|
if queued := peer.TrySend(BlockchainChannel, msgBytes); !queued {
|
|
return fmt.Errorf("peer queue full")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) sendBlockNotFound(height int64, peerID p2p.ID) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
msgBytes := cdc.MustMarshalBinaryBare(&bcNoBlockResponseMessage{Height: height})
|
|
if queued := peer.TrySend(BlockchainChannel, msgBytes); !queued {
|
|
return fmt.Errorf("peer queue full")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) trySwitchToConsensus(state state.State, skipWAL bool) bool {
|
|
conR, ok := sio.sw.Reactor("CONSENSUS").(consensusReactor)
|
|
if ok {
|
|
conR.SwitchToConsensus(state, skipWAL)
|
|
}
|
|
return ok
|
|
}
|
|
|
|
func (sio *switchIO) broadcastStatusRequest(base int64, height int64) {
|
|
if height == 0 && base > 0 {
|
|
base = 0
|
|
}
|
|
msgBytes := cdc.MustMarshalBinaryBare(&bcStatusRequestMessage{
|
|
Base: base,
|
|
Height: height,
|
|
})
|
|
// XXX: maybe we should use an io specific peer list here
|
|
sio.sw.Broadcast(BlockchainChannel, msgBytes)
|
|
}
|