+++ /dev/null
-// Copyright (c) 2013-2017 The btcsuite developers
-// Use of this source code is governed by an ISC
-// license that can be found in the LICENSE file.
-
-package netsync
-
-import (
- "container/list"
- "net"
- "sync"
- "sync/atomic"
- "time"
-
- "github.com/btcsuite/btcd/blockchain"
- "github.com/btcsuite/btcd/chaincfg"
- "github.com/btcsuite/btcd/chaincfg/chainhash"
- "github.com/btcsuite/btcd/database"
- "github.com/btcsuite/btcd/mempool"
- peerpkg "github.com/btcsuite/btcd/peer"
- "github.com/btcsuite/btcd/wire"
- "github.com/btcsuite/btcutil"
-)
-
-const (
- // minInFlightBlocks is the minimum number of blocks that should be
- // in the request queue for headers-first mode before requesting
- // more.
- minInFlightBlocks = 10
-
- // maxRejectedTxns is the maximum number of rejected transactions
- // hashes to store in memory.
- maxRejectedTxns = 1000
-
- // maxRequestedBlocks is the maximum number of requested block
- // hashes to store in memory.
- maxRequestedBlocks = wire.MaxInvPerMsg
-
- // maxRequestedTxns is the maximum number of requested transactions
- // hashes to store in memory.
- maxRequestedTxns = wire.MaxInvPerMsg
-)
-
-// zeroHash is the zero value hash (all zeros). It is defined as a convenience.
-var zeroHash chainhash.Hash
-
-// newPeerMsg signifies a newly connected peer to the block handler.
-type newPeerMsg struct {
- peer *peerpkg.Peer
-}
-
-// blockMsg packages a bitcoin block message and the peer it came from together
-// so the block handler has access to that information.
-type blockMsg struct {
- block *btcutil.Block
- peer *peerpkg.Peer
- reply chan struct{}
-}
-
-// invMsg packages a bitcoin inv message and the peer it came from together
-// so the block handler has access to that information.
-type invMsg struct {
- inv *wire.MsgInv
- peer *peerpkg.Peer
-}
-
-// headersMsg packages a bitcoin headers message and the peer it came from
-// together so the block handler has access to that information.
-type headersMsg struct {
- headers *wire.MsgHeaders
- peer *peerpkg.Peer
-}
-
-// donePeerMsg signifies a newly disconnected peer to the block handler.
-type donePeerMsg struct {
- peer *peerpkg.Peer
-}
-
-// txMsg packages a bitcoin tx message and the peer it came from together
-// so the block handler has access to that information.
-type txMsg struct {
- tx *btcutil.Tx
- peer *peerpkg.Peer
- reply chan struct{}
-}
-
-// getSyncPeerMsg is a message type to be sent across the message channel for
-// retrieving the current sync peer.
-type getSyncPeerMsg struct {
- reply chan int32
-}
-
-// processBlockResponse is a response sent to the reply channel of a
-// processBlockMsg.
-type processBlockResponse struct {
- isOrphan bool
- err error
-}
-
-// processBlockMsg is a message type to be sent across the message channel
-// for requested a block is processed. Note this call differs from blockMsg
-// above in that blockMsg is intended for blocks that came from peers and have
-// extra handling whereas this message essentially is just a concurrent safe
-// way to call ProcessBlock on the internal block chain instance.
-type processBlockMsg struct {
- block *btcutil.Block
- flags blockchain.BehaviorFlags
- reply chan processBlockResponse
-}
-
-// isCurrentMsg is a message type to be sent across the message channel for
-// requesting whether or not the sync manager believes it is synced with the
-// currently connected peers.
-type isCurrentMsg struct {
- reply chan bool
-}
-
-// pauseMsg is a message type to be sent across the message channel for
-// pausing the sync manager. This effectively provides the caller with
-// exclusive access over the manager until a receive is performed on the
-// unpause channel.
-type pauseMsg struct {
- unpause <-chan struct{}
-}
-
-// headerNode is used as a node in a list of headers that are linked together
-// between checkpoints.
-type headerNode struct {
- height int32
- hash *chainhash.Hash
-}
-
-// peerSyncState stores additional information that the SyncManager tracks
-// about a peer.
-type peerSyncState struct {
- syncCandidate bool
- requestQueue []*wire.InvVect
- requestedTxns map[chainhash.Hash]struct{}
- requestedBlocks map[chainhash.Hash]struct{}
-}
-
-// SyncManager is used to communicate block related messages with peers. The
-// SyncManager is started as by executing Start() in a goroutine. Once started,
-// it selects peers to sync from and starts the initial block download. Once the
-// chain is in sync, the SyncManager handles incoming block and header
-// notifications and relays announcements of new blocks to peers.
-type SyncManager struct {
- peerNotifier PeerNotifier
- started int32
- shutdown int32
- chain *blockchain.BlockChain
- txMemPool *mempool.TxPool
- chainParams *chaincfg.Params
- progressLogger *blockProgressLogger
- msgChan chan interface{}
- wg sync.WaitGroup
- quit chan struct{}
-
- // These fields should only be accessed from the blockHandler thread
- rejectedTxns map[chainhash.Hash]struct{}
- requestedTxns map[chainhash.Hash]struct{}
- requestedBlocks map[chainhash.Hash]struct{}
- syncPeer *peerpkg.Peer
- peerStates map[*peerpkg.Peer]*peerSyncState
-
- // The following fields are used for headers-first mode.
- headersFirstMode bool
- headerList *list.List
- startHeader *list.Element
- nextCheckpoint *chaincfg.Checkpoint
-}
-
-// resetHeaderState sets the headers-first mode state to values appropriate for
-// syncing from a new peer.
-func (sm *SyncManager) resetHeaderState(newestHash *chainhash.Hash, newestHeight int32) {
- sm.headersFirstMode = false
- sm.headerList.Init()
- sm.startHeader = nil
-
- // When there is a next checkpoint, add an entry for the latest known
- // block into the header pool. This allows the next downloaded header
- // to prove it links to the chain properly.
- if sm.nextCheckpoint != nil {
- node := headerNode{height: newestHeight, hash: newestHash}
- sm.headerList.PushBack(&node)
- }
-}
-
-// findNextHeaderCheckpoint returns the next checkpoint after the passed height.
-// It returns nil when there is not one either because the height is already
-// later than the final checkpoint or some other reason such as disabled
-// checkpoints.
-func (sm *SyncManager) findNextHeaderCheckpoint(height int32) *chaincfg.Checkpoint {
- checkpoints := sm.chain.Checkpoints()
- if len(checkpoints) == 0 {
- return nil
- }
-
- // There is no next checkpoint if the height is already after the final
- // checkpoint.
- finalCheckpoint := &checkpoints[len(checkpoints)-1]
- if height >= finalCheckpoint.Height {
- return nil
- }
-
- // Find the next checkpoint.
- nextCheckpoint := finalCheckpoint
- for i := len(checkpoints) - 2; i >= 0; i-- {
- if height >= checkpoints[i].Height {
- break
- }
- nextCheckpoint = &checkpoints[i]
- }
- return nextCheckpoint
-}
-
-// startSync will choose the best peer among the available candidate peers to
-// download/sync the blockchain from. When syncing is already running, it
-// simply returns. It also examines the candidates for any which are no longer
-// candidates and removes them as needed.
-func (sm *SyncManager) startSync() {
- // Return now if we're already syncing.
- if sm.syncPeer != nil {
- return
- }
-
- // Once the segwit soft-fork package has activated, we only
- // want to sync from peers which are witness enabled to ensure
- // that we fully validate all blockchain data.
- segwitActive, err := sm.chain.IsDeploymentActive(chaincfg.DeploymentSegwit)
- if err != nil {
- log.Errorf("Unable to query for segwit soft-fork state: %v", err)
- return
- }
-
- best := sm.chain.BestSnapshot()
- var bestPeer *peerpkg.Peer
- for peer, state := range sm.peerStates {
- if !state.syncCandidate {
- continue
- }
-
- if segwitActive && !peer.IsWitnessEnabled() {
- log.Debugf("peer %v not witness enabled, skipping", peer)
- continue
- }
-
- // Remove sync candidate peers that are no longer candidates due
- // to passing their latest known block. NOTE: The < is
- // intentional as opposed to <=. While technically the peer
- // doesn't have a later block when it's equal, it will likely
- // have one soon so it is a reasonable choice. It also allows
- // the case where both are at 0 such as during regression test.
- if peer.LastBlock() < best.Height {
- state.syncCandidate = false
- continue
- }
-
- // TODO(davec): Use a better algorithm to choose the best peer.
- // For now, just pick the first available candidate.
- bestPeer = peer
- }
-
- // Start syncing from the best peer if one was selected.
- if bestPeer != nil {
- // Clear the requestedBlocks if the sync peer changes, otherwise
- // we may ignore blocks we need that the last sync peer failed
- // to send.
- sm.requestedBlocks = make(map[chainhash.Hash]struct{})
-
- locator, err := sm.chain.LatestBlockLocator()
- if err != nil {
- log.Errorf("Failed to get block locator for the "+
- "latest block: %v", err)
- return
- }
-
- log.Infof("Syncing to block height %d from peer %v",
- bestPeer.LastBlock(), bestPeer.Addr())
-
- // When the current height is less than a known checkpoint we
- // can use block headers to learn about which blocks comprise
- // the chain up to the checkpoint and perform less validation
- // for them. This is possible since each header contains the
- // hash of the previous header and a merkle root. Therefore if
- // we validate all of the received headers link together
- // properly and the checkpoint hashes match, we can be sure the
- // hashes for the blocks in between are accurate. Further, once
- // the full blocks are downloaded, the merkle root is computed
- // and compared against the value in the header which proves the
- // full block hasn't been tampered with.
- //
- // Once we have passed the final checkpoint, or checkpoints are
- // disabled, use standard inv messages learn about the blocks
- // and fully validate them. Finally, regression test mode does
- // not support the headers-first approach so do normal block
- // downloads when in regression test mode.
- if sm.nextCheckpoint != nil &&
- best.Height < sm.nextCheckpoint.Height &&
- sm.chainParams != &chaincfg.RegressionNetParams {
-
- bestPeer.PushGetHeadersMsg(locator, sm.nextCheckpoint.Hash)
- sm.headersFirstMode = true
- log.Infof("Downloading headers for blocks %d to "+
- "%d from peer %s", best.Height+1,
- sm.nextCheckpoint.Height, bestPeer.Addr())
- } else {
- bestPeer.PushGetBlocksMsg(locator, &zeroHash)
- }
- sm.syncPeer = bestPeer
- } else {
- log.Warnf("No sync peer candidates available")
- }
-}
-
-// isSyncCandidate returns whether or not the peer is a candidate to consider
-// syncing from.
-func (sm *SyncManager) isSyncCandidate(peer *peerpkg.Peer) bool {
- // Typically a peer is not a candidate for sync if it's not a full node,
- // however regression test is special in that the regression tool is
- // not a full node and still needs to be considered a sync candidate.
- if sm.chainParams == &chaincfg.RegressionNetParams {
- // The peer is not a candidate if it's not coming from localhost
- // or the hostname can't be determined for some reason.
- host, _, err := net.SplitHostPort(peer.Addr())
- if err != nil {
- return false
- }
-
- if host != "127.0.0.1" && host != "localhost" {
- return false
- }
- } else {
- // The peer is not a candidate for sync if it's not a full
- // node. Additionally, if the segwit soft-fork package has
- // activated, then the peer must also be upgraded.
- segwitActive, err := sm.chain.IsDeploymentActive(chaincfg.DeploymentSegwit)
- if err != nil {
- log.Errorf("Unable to query for segwit "+
- "soft-fork state: %v", err)
- }
- nodeServices := peer.Services()
- if nodeServices&wire.SFNodeNetwork != wire.SFNodeNetwork ||
- (segwitActive && !peer.IsWitnessEnabled()) {
- return false
- }
- }
-
- // Candidate if all checks passed.
- return true
-}
-
-// handleNewPeerMsg deals with new peers that have signalled they may
-// be considered as a sync peer (they have already successfully negotiated). It
-// also starts syncing if needed. It is invoked from the syncHandler goroutine.
-func (sm *SyncManager) handleNewPeerMsg(peer *peerpkg.Peer) {
- // Ignore if in the process of shutting down.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- return
- }
-
- log.Infof("New valid peer %s (%s)", peer, peer.UserAgent())
-
- // Initialize the peer state
- isSyncCandidate := sm.isSyncCandidate(peer)
- sm.peerStates[peer] = &peerSyncState{
- syncCandidate: isSyncCandidate,
- requestedTxns: make(map[chainhash.Hash]struct{}),
- requestedBlocks: make(map[chainhash.Hash]struct{}),
- }
-
- // Start syncing by choosing the best candidate if needed.
- if isSyncCandidate && sm.syncPeer == nil {
- sm.startSync()
- }
-}
-
-// handleDonePeerMsg deals with peers that have signalled they are done. It
-// removes the peer as a candidate for syncing and in the case where it was
-// the current sync peer, attempts to select a new best peer to sync from. It
-// is invoked from the syncHandler goroutine.
-func (sm *SyncManager) handleDonePeerMsg(peer *peerpkg.Peer) {
- state, exists := sm.peerStates[peer]
- if !exists {
- log.Warnf("Received done peer message for unknown peer %s", peer)
- return
- }
-
- // Remove the peer from the list of candidate peers.
- delete(sm.peerStates, peer)
-
- log.Infof("Lost peer %s", peer)
-
- // Remove requested transactions from the global map so that they will
- // be fetched from elsewhere next time we get an inv.
- for txHash := range state.requestedTxns {
- delete(sm.requestedTxns, txHash)
- }
-
- // Remove requested blocks from the global map so that they will be
- // fetched from elsewhere next time we get an inv.
- // TODO: we could possibly here check which peers have these blocks
- // and request them now to speed things up a little.
- for blockHash := range state.requestedBlocks {
- delete(sm.requestedBlocks, blockHash)
- }
-
- // Attempt to find a new peer to sync from if the quitting peer is the
- // sync peer. Also, reset the headers-first state if in headers-first
- // mode so
- if sm.syncPeer == peer {
- sm.syncPeer = nil
- if sm.headersFirstMode {
- best := sm.chain.BestSnapshot()
- sm.resetHeaderState(&best.Hash, best.Height)
- }
- sm.startSync()
- }
-}
-
-// handleTxMsg handles transaction messages from all peers.
-func (sm *SyncManager) handleTxMsg(tmsg *txMsg) {
- peer := tmsg.peer
- state, exists := sm.peerStates[peer]
- if !exists {
- log.Warnf("Received tx message from unknown peer %s", peer)
- return
- }
-
- // NOTE: BitcoinJ, and possibly other wallets, don't follow the spec of
- // sending an inventory message and allowing the remote peer to decide
- // whether or not they want to request the transaction via a getdata
- // message. Unfortunately, the reference implementation permits
- // unrequested data, so it has allowed wallets that don't follow the
- // spec to proliferate. While this is not ideal, there is no check here
- // to disconnect peers for sending unsolicited transactions to provide
- // interoperability.
- txHash := tmsg.tx.Hash()
-
- // Ignore transactions that we have already rejected. Do not
- // send a reject message here because if the transaction was already
- // rejected, the transaction was unsolicited.
- if _, exists = sm.rejectedTxns[*txHash]; exists {
- log.Debugf("Ignoring unsolicited previously rejected "+
- "transaction %v from %s", txHash, peer)
- return
- }
-
- // Process the transaction to include validation, insertion in the
- // memory pool, orphan handling, etc.
- acceptedTxs, err := sm.txMemPool.ProcessTransaction(tmsg.tx,
- true, true, mempool.Tag(peer.ID()))
-
- // Remove transaction from request maps. Either the mempool/chain
- // already knows about it and as such we shouldn't have any more
- // instances of trying to fetch it, or we failed to insert and thus
- // we'll retry next time we get an inv.
- delete(state.requestedTxns, *txHash)
- delete(sm.requestedTxns, *txHash)
-
- if err != nil {
- // Do not request this transaction again until a new block
- // has been processed.
- sm.rejectedTxns[*txHash] = struct{}{}
- sm.limitMap(sm.rejectedTxns, maxRejectedTxns)
-
- // When the error is a rule error, it means the transaction was
- // simply rejected as opposed to something actually going wrong,
- // so log it as such. Otherwise, something really did go wrong,
- // so log it as an actual error.
- if _, ok := err.(mempool.RuleError); ok {
- log.Debugf("Rejected transaction %v from %s: %v",
- txHash, peer, err)
- } else {
- log.Errorf("Failed to process transaction %v: %v",
- txHash, err)
- }
-
- // Convert the error into an appropriate reject message and
- // send it.
- code, reason := mempool.ErrToRejectErr(err)
- peer.PushRejectMsg(wire.CmdTx, code, reason, txHash, false)
- return
- }
-
- sm.peerNotifier.AnnounceNewTransactions(acceptedTxs)
-}
-
-// current returns true if we believe we are synced with our peers, false if we
-// still have blocks to check
-func (sm *SyncManager) current() bool {
- if !sm.chain.IsCurrent() {
- return false
- }
-
- // if blockChain thinks we are current and we have no syncPeer it
- // is probably right.
- if sm.syncPeer == nil {
- return true
- }
-
- // No matter what chain thinks, if we are below the block we are syncing
- // to we are not current.
- if sm.chain.BestSnapshot().Height < sm.syncPeer.LastBlock() {
- return false
- }
- return true
-}
-
-// handleBlockMsg handles block messages from all peers.
-func (sm *SyncManager) handleBlockMsg(bmsg *blockMsg) {
- peer := bmsg.peer
- state, exists := sm.peerStates[peer]
- if !exists {
- log.Warnf("Received block message from unknown peer %s", peer)
- return
- }
-
- // If we didn't ask for this block then the peer is misbehaving.
- blockHash := bmsg.block.Hash()
- if _, exists = state.requestedBlocks[*blockHash]; !exists {
- // The regression test intentionally sends some blocks twice
- // to test duplicate block insertion fails. Don't disconnect
- // the peer or ignore the block when we're in regression test
- // mode in this case so the chain code is actually fed the
- // duplicate blocks.
- if sm.chainParams != &chaincfg.RegressionNetParams {
- log.Warnf("Got unrequested block %v from %s -- "+
- "disconnecting", blockHash, peer.Addr())
- peer.Disconnect()
- return
- }
- }
-
- // When in headers-first mode, if the block matches the hash of the
- // first header in the list of headers that are being fetched, it's
- // eligible for less validation since the headers have already been
- // verified to link together and are valid up to the next checkpoint.
- // Also, remove the list entry for all blocks except the checkpoint
- // since it is needed to verify the next round of headers links
- // properly.
- isCheckpointBlock := false
- behaviorFlags := blockchain.BFNone
- if sm.headersFirstMode {
- firstNodeEl := sm.headerList.Front()
- if firstNodeEl != nil {
- firstNode := firstNodeEl.Value.(*headerNode)
- if blockHash.IsEqual(firstNode.hash) {
- behaviorFlags |= blockchain.BFFastAdd
- if firstNode.hash.IsEqual(sm.nextCheckpoint.Hash) {
- isCheckpointBlock = true
- } else {
- sm.headerList.Remove(firstNodeEl)
- }
- }
- }
- }
-
- // Remove block from request maps. Either chain will know about it and
- // so we shouldn't have any more instances of trying to fetch it, or we
- // will fail the insert and thus we'll retry next time we get an inv.
- delete(state.requestedBlocks, *blockHash)
- delete(sm.requestedBlocks, *blockHash)
-
- // Process the block to include validation, best chain selection, orphan
- // handling, etc.
- _, isOrphan, err := sm.chain.ProcessBlock(bmsg.block, behaviorFlags)
- if err != nil {
- // When the error is a rule error, it means the block was simply
- // rejected as opposed to something actually going wrong, so log
- // it as such. Otherwise, something really did go wrong, so log
- // it as an actual error.
- if _, ok := err.(blockchain.RuleError); ok {
- log.Infof("Rejected block %v from %s: %v", blockHash,
- peer, err)
- } else {
- log.Errorf("Failed to process block %v: %v",
- blockHash, err)
- }
- if dbErr, ok := err.(database.Error); ok && dbErr.ErrorCode ==
- database.ErrCorruption {
- panic(dbErr)
- }
-
- // Convert the error into an appropriate reject message and
- // send it.
- code, reason := mempool.ErrToRejectErr(err)
- peer.PushRejectMsg(wire.CmdBlock, code, reason, blockHash, false)
- return
- }
-
- // Meta-data about the new block this peer is reporting. We use this
- // below to update this peer's lastest block height and the heights of
- // other peers based on their last announced block hash. This allows us
- // to dynamically update the block heights of peers, avoiding stale
- // heights when looking for a new sync peer. Upon acceptance of a block
- // or recognition of an orphan, we also use this information to update
- // the block heights over other peers who's invs may have been ignored
- // if we are actively syncing while the chain is not yet current or
- // who may have lost the lock announcment race.
- var heightUpdate int32
- var blkHashUpdate *chainhash.Hash
-
- // Request the parents for the orphan block from the peer that sent it.
- if isOrphan {
- // We've just received an orphan block from a peer. In order
- // to update the height of the peer, we try to extract the
- // block height from the scriptSig of the coinbase transaction.
- // Extraction is only attempted if the block's version is
- // high enough (ver 2+).
- header := &bmsg.block.MsgBlock().Header
- if blockchain.ShouldHaveSerializedBlockHeight(header) {
- coinbaseTx := bmsg.block.Transactions()[0]
- cbHeight, err := blockchain.ExtractCoinbaseHeight(coinbaseTx)
- if err != nil {
- log.Warnf("Unable to extract height from "+
- "coinbase tx: %v", err)
- } else {
- log.Debugf("Extracted height of %v from "+
- "orphan block", cbHeight)
- heightUpdate = cbHeight
- blkHashUpdate = blockHash
- }
- }
-
- orphanRoot := sm.chain.GetOrphanRoot(blockHash)
- locator, err := sm.chain.LatestBlockLocator()
- if err != nil {
- log.Warnf("Failed to get block locator for the "+
- "latest block: %v", err)
- } else {
- peer.PushGetBlocksMsg(locator, orphanRoot)
- }
- } else {
- // When the block is not an orphan, log information about it and
- // update the chain state.
- sm.progressLogger.LogBlockHeight(bmsg.block)
-
- // Update this peer's latest block height, for future
- // potential sync node candidacy.
- best := sm.chain.BestSnapshot()
- heightUpdate = best.Height
- blkHashUpdate = &best.Hash
-
- // Clear the rejected transactions.
- sm.rejectedTxns = make(map[chainhash.Hash]struct{})
- }
-
- // Update the block height for this peer. But only send a message to
- // the server for updating peer heights if this is an orphan or our
- // chain is "current". This avoids sending a spammy amount of messages
- // if we're syncing the chain from scratch.
- if blkHashUpdate != nil && heightUpdate != 0 {
- peer.UpdateLastBlockHeight(heightUpdate)
- if isOrphan || sm.current() {
- go sm.peerNotifier.UpdatePeerHeights(blkHashUpdate, heightUpdate,
- peer)
- }
- }
-
- // Nothing more to do if we aren't in headers-first mode.
- if !sm.headersFirstMode {
- return
- }
-
- // This is headers-first mode, so if the block is not a checkpoint
- // request more blocks using the header list when the request queue is
- // getting short.
- if !isCheckpointBlock {
- if sm.startHeader != nil &&
- len(state.requestedBlocks) < minInFlightBlocks {
- sm.fetchHeaderBlocks()
- }
- return
- }
-
- // This is headers-first mode and the block is a checkpoint. When
- // there is a next checkpoint, get the next round of headers by asking
- // for headers starting from the block after this one up to the next
- // checkpoint.
- prevHeight := sm.nextCheckpoint.Height
- prevHash := sm.nextCheckpoint.Hash
- sm.nextCheckpoint = sm.findNextHeaderCheckpoint(prevHeight)
- if sm.nextCheckpoint != nil {
- locator := blockchain.BlockLocator([]*chainhash.Hash{prevHash})
- err := peer.PushGetHeadersMsg(locator, sm.nextCheckpoint.Hash)
- if err != nil {
- log.Warnf("Failed to send getheaders message to "+
- "peer %s: %v", peer.Addr(), err)
- return
- }
- log.Infof("Downloading headers for blocks %d to %d from "+
- "peer %s", prevHeight+1, sm.nextCheckpoint.Height,
- sm.syncPeer.Addr())
- return
- }
-
- // This is headers-first mode, the block is a checkpoint, and there are
- // no more checkpoints, so switch to normal mode by requesting blocks
- // from the block after this one up to the end of the chain (zero hash).
- sm.headersFirstMode = false
- sm.headerList.Init()
- log.Infof("Reached the final checkpoint -- switching to normal mode")
- locator := blockchain.BlockLocator([]*chainhash.Hash{blockHash})
- err = peer.PushGetBlocksMsg(locator, &zeroHash)
- if err != nil {
- log.Warnf("Failed to send getblocks message to peer %s: %v",
- peer.Addr(), err)
- return
- }
-}
-
-// fetchHeaderBlocks creates and sends a request to the syncPeer for the next
-// list of blocks to be downloaded based on the current list of headers.
-func (sm *SyncManager) fetchHeaderBlocks() {
- // Nothing to do if there is no start header.
- if sm.startHeader == nil {
- log.Warnf("fetchHeaderBlocks called with no start header")
- return
- }
-
- // Build up a getdata request for the list of blocks the headers
- // describe. The size hint will be limited to wire.MaxInvPerMsg by
- // the function, so no need to double check it here.
- gdmsg := wire.NewMsgGetDataSizeHint(uint(sm.headerList.Len()))
- numRequested := 0
- for e := sm.startHeader; e != nil; e = e.Next() {
- node, ok := e.Value.(*headerNode)
- if !ok {
- log.Warn("Header list node type is not a headerNode")
- continue
- }
-
- iv := wire.NewInvVect(wire.InvTypeBlock, node.hash)
- haveInv, err := sm.haveInventory(iv)
- if err != nil {
- log.Warnf("Unexpected failure when checking for "+
- "existing inventory during header block "+
- "fetch: %v", err)
- }
- if !haveInv {
- syncPeerState := sm.peerStates[sm.syncPeer]
-
- sm.requestedBlocks[*node.hash] = struct{}{}
- syncPeerState.requestedBlocks[*node.hash] = struct{}{}
-
- // If we're fetching from a witness enabled peer
- // post-fork, then ensure that we receive all the
- // witness data in the blocks.
- if sm.syncPeer.IsWitnessEnabled() {
- iv.Type = wire.InvTypeWitnessBlock
- }
-
- gdmsg.AddInvVect(iv)
- numRequested++
- }
- sm.startHeader = e.Next()
- if numRequested >= wire.MaxInvPerMsg {
- break
- }
- }
- if len(gdmsg.InvList) > 0 {
- sm.syncPeer.QueueMessage(gdmsg, nil)
- }
-}
-
-// handleHeadersMsg handles block header messages from all peers. Headers are
-// requested when performing a headers-first sync.
-func (sm *SyncManager) handleHeadersMsg(hmsg *headersMsg) {
- peer := hmsg.peer
- _, exists := sm.peerStates[peer]
- if !exists {
- log.Warnf("Received headers message from unknown peer %s", peer)
- return
- }
-
- // The remote peer is misbehaving if we didn't request headers.
- msg := hmsg.headers
- numHeaders := len(msg.Headers)
- if !sm.headersFirstMode {
- log.Warnf("Got %d unrequested headers from %s -- "+
- "disconnecting", numHeaders, peer.Addr())
- peer.Disconnect()
- return
- }
-
- // Nothing to do for an empty headers message.
- if numHeaders == 0 {
- return
- }
-
- // Process all of the received headers ensuring each one connects to the
- // previous and that checkpoints match.
- receivedCheckpoint := false
- var finalHash *chainhash.Hash
- for _, blockHeader := range msg.Headers {
- blockHash := blockHeader.BlockHash()
- finalHash = &blockHash
-
- // Ensure there is a previous header to compare against.
- prevNodeEl := sm.headerList.Back()
- if prevNodeEl == nil {
- log.Warnf("Header list does not contain a previous" +
- "element as expected -- disconnecting peer")
- peer.Disconnect()
- return
- }
-
- // Ensure the header properly connects to the previous one and
- // add it to the list of headers.
- node := headerNode{hash: &blockHash}
- prevNode := prevNodeEl.Value.(*headerNode)
- if prevNode.hash.IsEqual(&blockHeader.PrevBlock) {
- node.height = prevNode.height + 1
- e := sm.headerList.PushBack(&node)
- if sm.startHeader == nil {
- sm.startHeader = e
- }
- } else {
- log.Warnf("Received block header that does not "+
- "properly connect to the chain from peer %s "+
- "-- disconnecting", peer.Addr())
- peer.Disconnect()
- return
- }
-
- // Verify the header at the next checkpoint height matches.
- if node.height == sm.nextCheckpoint.Height {
- if node.hash.IsEqual(sm.nextCheckpoint.Hash) {
- receivedCheckpoint = true
- log.Infof("Verified downloaded block "+
- "header against checkpoint at height "+
- "%d/hash %s", node.height, node.hash)
- } else {
- log.Warnf("Block header at height %d/hash "+
- "%s from peer %s does NOT match "+
- "expected checkpoint hash of %s -- "+
- "disconnecting", node.height,
- node.hash, peer.Addr(),
- sm.nextCheckpoint.Hash)
- peer.Disconnect()
- return
- }
- break
- }
- }
-
- // When this header is a checkpoint, switch to fetching the blocks for
- // all of the headers since the last checkpoint.
- if receivedCheckpoint {
- // Since the first entry of the list is always the final block
- // that is already in the database and is only used to ensure
- // the next header links properly, it must be removed before
- // fetching the blocks.
- sm.headerList.Remove(sm.headerList.Front())
- log.Infof("Received %v block headers: Fetching blocks",
- sm.headerList.Len())
- sm.progressLogger.SetLastLogTime(time.Now())
- sm.fetchHeaderBlocks()
- return
- }
-
- // This header is not a checkpoint, so request the next batch of
- // headers starting from the latest known header and ending with the
- // next checkpoint.
- locator := blockchain.BlockLocator([]*chainhash.Hash{finalHash})
- err := peer.PushGetHeadersMsg(locator, sm.nextCheckpoint.Hash)
- if err != nil {
- log.Warnf("Failed to send getheaders message to "+
- "peer %s: %v", peer.Addr(), err)
- return
- }
-}
-
-// haveInventory returns whether or not the inventory represented by the passed
-// inventory vector is known. This includes checking all of the various places
-// inventory can be when it is in different states such as blocks that are part
-// of the main chain, on a side chain, in the orphan pool, and transactions that
-// are in the memory pool (either the main pool or orphan pool).
-func (sm *SyncManager) haveInventory(invVect *wire.InvVect) (bool, error) {
- switch invVect.Type {
- case wire.InvTypeWitnessBlock:
- fallthrough
- case wire.InvTypeBlock:
- // Ask chain if the block is known to it in any form (main
- // chain, side chain, or orphan).
- return sm.chain.HaveBlock(&invVect.Hash)
-
- case wire.InvTypeWitnessTx:
- fallthrough
- case wire.InvTypeTx:
- // Ask the transaction memory pool if the transaction is known
- // to it in any form (main pool or orphan).
- if sm.txMemPool.HaveTransaction(&invVect.Hash) {
- return true, nil
- }
-
- // Check if the transaction exists from the point of view of the
- // end of the main chain.
- entry, err := sm.chain.FetchUtxoEntry(&invVect.Hash)
- if err != nil {
- return false, err
- }
- return entry != nil && !entry.IsFullySpent(), nil
- }
-
- // The requested inventory is is an unsupported type, so just claim
- // it is known to avoid requesting it.
- return true, nil
-}
-
-// handleInvMsg handles inv messages from all peers.
-// We examine the inventory advertised by the remote peer and act accordingly.
-func (sm *SyncManager) handleInvMsg(imsg *invMsg) {
- peer := imsg.peer
- state, exists := sm.peerStates[peer]
- if !exists {
- log.Warnf("Received inv message from unknown peer %s", peer)
- return
- }
-
- // Attempt to find the final block in the inventory list. There may
- // not be one.
- lastBlock := -1
- invVects := imsg.inv.InvList
- for i := len(invVects) - 1; i >= 0; i-- {
- if invVects[i].Type == wire.InvTypeBlock {
- lastBlock = i
- break
- }
- }
-
- // If this inv contains a block announcement, and this isn't coming from
- // our current sync peer or we're current, then update the last
- // announced block for this peer. We'll use this information later to
- // update the heights of peers based on blocks we've accepted that they
- // previously announced.
- if lastBlock != -1 && (peer != sm.syncPeer || sm.current()) {
- peer.UpdateLastAnnouncedBlock(&invVects[lastBlock].Hash)
- }
-
- // Ignore invs from peers that aren't the sync if we are not current.
- // Helps prevent fetching a mass of orphans.
- if peer != sm.syncPeer && !sm.current() {
- return
- }
-
- // If our chain is current and a peer announces a block we already
- // know of, then update their current block height.
- if lastBlock != -1 && sm.current() {
- blkHeight, err := sm.chain.BlockHeightByHash(&invVects[lastBlock].Hash)
- if err == nil {
- peer.UpdateLastBlockHeight(blkHeight)
- }
- }
-
- // Request the advertised inventory if we don't already have it. Also,
- // request parent blocks of orphans if we receive one we already have.
- // Finally, attempt to detect potential stalls due to long side chains
- // we already have and request more blocks to prevent them.
- for i, iv := range invVects {
- // Ignore unsupported inventory types.
- switch iv.Type {
- case wire.InvTypeBlock:
- case wire.InvTypeTx:
- case wire.InvTypeWitnessBlock:
- case wire.InvTypeWitnessTx:
- default:
- continue
- }
-
- // Add the inventory to the cache of known inventory
- // for the peer.
- peer.AddKnownInventory(iv)
-
- // Ignore inventory when we're in headers-first mode.
- if sm.headersFirstMode {
- continue
- }
-
- // Request the inventory if we don't already have it.
- haveInv, err := sm.haveInventory(iv)
- if err != nil {
- log.Warnf("Unexpected failure when checking for "+
- "existing inventory during inv message "+
- "processing: %v", err)
- continue
- }
- if !haveInv {
- if iv.Type == wire.InvTypeTx {
- // Skip the transaction if it has already been
- // rejected.
- if _, exists := sm.rejectedTxns[iv.Hash]; exists {
- continue
- }
- }
-
- // Ignore invs block invs from non-witness enabled
- // peers, as after segwit activation we only want to
- // download from peers that can provide us full witness
- // data for blocks.
- if !peer.IsWitnessEnabled() && iv.Type == wire.InvTypeBlock {
- continue
- }
-
- // Add it to the request queue.
- state.requestQueue = append(state.requestQueue, iv)
- continue
- }
-
- if iv.Type == wire.InvTypeBlock {
- // The block is an orphan block that we already have.
- // When the existing orphan was processed, it requested
- // the missing parent blocks. When this scenario
- // happens, it means there were more blocks missing
- // than are allowed into a single inventory message. As
- // a result, once this peer requested the final
- // advertised block, the remote peer noticed and is now
- // resending the orphan block as an available block
- // to signal there are more missing blocks that need to
- // be requested.
- if sm.chain.IsKnownOrphan(&iv.Hash) {
- // Request blocks starting at the latest known
- // up to the root of the orphan that just came
- // in.
- orphanRoot := sm.chain.GetOrphanRoot(&iv.Hash)
- locator, err := sm.chain.LatestBlockLocator()
- if err != nil {
- log.Errorf("PEER: Failed to get block "+
- "locator for the latest block: "+
- "%v", err)
- continue
- }
- peer.PushGetBlocksMsg(locator, orphanRoot)
- continue
- }
-
- // We already have the final block advertised by this
- // inventory message, so force a request for more. This
- // should only happen if we're on a really long side
- // chain.
- if i == lastBlock {
- // Request blocks after this one up to the
- // final one the remote peer knows about (zero
- // stop hash).
- locator := sm.chain.BlockLocatorFromHash(&iv.Hash)
- peer.PushGetBlocksMsg(locator, &zeroHash)
- }
- }
- }
-
- // Request as much as possible at once. Anything that won't fit into
- // the request will be requested on the next inv message.
- numRequested := 0
- gdmsg := wire.NewMsgGetData()
- requestQueue := state.requestQueue
- for len(requestQueue) != 0 {
- iv := requestQueue[0]
- requestQueue[0] = nil
- requestQueue = requestQueue[1:]
-
- switch iv.Type {
- case wire.InvTypeWitnessBlock:
- fallthrough
- case wire.InvTypeBlock:
- // Request the block if there is not already a pending
- // request.
- if _, exists := sm.requestedBlocks[iv.Hash]; !exists {
- sm.requestedBlocks[iv.Hash] = struct{}{}
- sm.limitMap(sm.requestedBlocks, maxRequestedBlocks)
- state.requestedBlocks[iv.Hash] = struct{}{}
-
- if peer.IsWitnessEnabled() {
- iv.Type = wire.InvTypeWitnessBlock
- }
-
- gdmsg.AddInvVect(iv)
- numRequested++
- }
-
- case wire.InvTypeWitnessTx:
- fallthrough
- case wire.InvTypeTx:
- // Request the transaction if there is not already a
- // pending request.
- if _, exists := sm.requestedTxns[iv.Hash]; !exists {
- sm.requestedTxns[iv.Hash] = struct{}{}
- sm.limitMap(sm.requestedTxns, maxRequestedTxns)
- state.requestedTxns[iv.Hash] = struct{}{}
-
- // If the peer is capable, request the txn
- // including all witness data.
- if peer.IsWitnessEnabled() {
- iv.Type = wire.InvTypeWitnessTx
- }
-
- gdmsg.AddInvVect(iv)
- numRequested++
- }
- }
-
- if numRequested >= wire.MaxInvPerMsg {
- break
- }
- }
- state.requestQueue = requestQueue
- if len(gdmsg.InvList) > 0 {
- peer.QueueMessage(gdmsg, nil)
- }
-}
-
-// limitMap is a helper function for maps that require a maximum limit by
-// evicting a random transaction if adding a new value would cause it to
-// overflow the maximum allowed.
-func (sm *SyncManager) limitMap(m map[chainhash.Hash]struct{}, limit int) {
- if len(m)+1 > limit {
- // Remove a random entry from the map. For most compilers, Go's
- // range statement iterates starting at a random item although
- // that is not 100% guaranteed by the spec. The iteration order
- // is not important here because an adversary would have to be
- // able to pull off preimage attacks on the hashing function in
- // order to target eviction of specific entries anyways.
- for txHash := range m {
- delete(m, txHash)
- return
- }
- }
-}
-
-// blockHandler is the main handler for the sync manager. It must be run as a
-// goroutine. It processes block and inv messages in a separate goroutine
-// from the peer handlers so the block (MsgBlock) messages are handled by a
-// single thread without needing to lock memory data structures. This is
-// important because the sync manager controls which blocks are needed and how
-// the fetching should proceed.
-func (sm *SyncManager) blockHandler() {
-out:
- for {
- select {
- case m := <-sm.msgChan:
- switch msg := m.(type) {
- case *newPeerMsg:
- sm.handleNewPeerMsg(msg.peer)
-
- case *txMsg:
- sm.handleTxMsg(msg)
- msg.reply <- struct{}{}
-
- case *blockMsg:
- sm.handleBlockMsg(msg)
- msg.reply <- struct{}{}
-
- case *invMsg:
- sm.handleInvMsg(msg)
-
- case *headersMsg:
- sm.handleHeadersMsg(msg)
-
- case *donePeerMsg:
- sm.handleDonePeerMsg(msg.peer)
-
- case getSyncPeerMsg:
- var peerID int32
- if sm.syncPeer != nil {
- peerID = sm.syncPeer.ID()
- }
- msg.reply <- peerID
-
- case processBlockMsg:
- _, isOrphan, err := sm.chain.ProcessBlock(
- msg.block, msg.flags)
- if err != nil {
- msg.reply <- processBlockResponse{
- isOrphan: false,
- err: err,
- }
- }
-
- msg.reply <- processBlockResponse{
- isOrphan: isOrphan,
- err: nil,
- }
-
- case isCurrentMsg:
- msg.reply <- sm.current()
-
- case pauseMsg:
- // Wait until the sender unpauses the manager.
- <-msg.unpause
-
- default:
- log.Warnf("Invalid message type in block "+
- "handler: %T", msg)
- }
-
- case <-sm.quit:
- break out
- }
- }
-
- sm.wg.Done()
- log.Trace("Block handler done")
-}
-
-// handleBlockchainNotification handles notifications from blockchain. It does
-// things such as request orphan block parents and relay accepted blocks to
-// connected peers.
-func (sm *SyncManager) handleBlockchainNotification(notification *blockchain.Notification) {
- switch notification.Type {
- // A block has been accepted into the block chain. Relay it to other
- // peers.
- case blockchain.NTBlockAccepted:
- // Don't relay if we are not current. Other peers that are
- // current should already know about it.
- if !sm.current() {
- return
- }
-
- block, ok := notification.Data.(*btcutil.Block)
- if !ok {
- log.Warnf("Chain accepted notification is not a block.")
- break
- }
-
- // Generate the inventory vector and relay it.
- iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
- sm.peerNotifier.RelayInventory(iv, block.MsgBlock().Header)
-
- // A block has been connected to the main block chain.
- case blockchain.NTBlockConnected:
- block, ok := notification.Data.(*btcutil.Block)
- if !ok {
- log.Warnf("Chain connected notification is not a block.")
- break
- }
-
- // Remove all of the transactions (except the coinbase) in the
- // connected block from the transaction pool. Secondly, remove any
- // transactions which are now double spends as a result of these
- // new transactions. Finally, remove any transaction that is
- // no longer an orphan. Transactions which depend on a confirmed
- // transaction are NOT removed recursively because they are still
- // valid.
- for _, tx := range block.Transactions()[1:] {
- sm.txMemPool.RemoveTransaction(tx, false)
- sm.txMemPool.RemoveDoubleSpends(tx)
- sm.txMemPool.RemoveOrphan(tx)
- sm.peerNotifier.TransactionConfirmed(tx)
- acceptedTxs := sm.txMemPool.ProcessOrphans(tx)
- sm.peerNotifier.AnnounceNewTransactions(acceptedTxs)
- }
-
- // A block has been disconnected from the main block chain.
- case blockchain.NTBlockDisconnected:
- block, ok := notification.Data.(*btcutil.Block)
- if !ok {
- log.Warnf("Chain disconnected notification is not a block.")
- break
- }
-
- // Reinsert all of the transactions (except the coinbase) into
- // the transaction pool.
- for _, tx := range block.Transactions()[1:] {
- _, _, err := sm.txMemPool.MaybeAcceptTransaction(tx,
- false, false)
- if err != nil {
- // Remove the transaction and all transactions
- // that depend on it if it wasn't accepted into
- // the transaction pool.
- sm.txMemPool.RemoveTransaction(tx, true)
- }
- }
- }
-}
-
-// NewPeer informs the sync manager of a newly active peer.
-func (sm *SyncManager) NewPeer(peer *peerpkg.Peer) {
- // Ignore if we are shutting down.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- return
- }
- sm.msgChan <- &newPeerMsg{peer: peer}
-}
-
-// QueueTx adds the passed transaction message and peer to the block handling
-// queue. Responds to the done channel argument after the tx message is
-// processed.
-func (sm *SyncManager) QueueTx(tx *btcutil.Tx, peer *peerpkg.Peer, done chan struct{}) {
- // Don't accept more transactions if we're shutting down.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- done <- struct{}{}
- return
- }
-
- sm.msgChan <- &txMsg{tx: tx, peer: peer, reply: done}
-}
-
-// QueueBlock adds the passed block message and peer to the block handling
-// queue. Responds to the done channel argument after the block message is
-// processed.
-func (sm *SyncManager) QueueBlock(block *btcutil.Block, peer *peerpkg.Peer, done chan struct{}) {
- // Don't accept more blocks if we're shutting down.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- done <- struct{}{}
- return
- }
-
- sm.msgChan <- &blockMsg{block: block, peer: peer, reply: done}
-}
-
-// QueueInv adds the passed inv message and peer to the block handling queue.
-func (sm *SyncManager) QueueInv(inv *wire.MsgInv, peer *peerpkg.Peer) {
- // No channel handling here because peers do not need to block on inv
- // messages.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- return
- }
-
- sm.msgChan <- &invMsg{inv: inv, peer: peer}
-}
-
-// QueueHeaders adds the passed headers message and peer to the block handling
-// queue.
-func (sm *SyncManager) QueueHeaders(headers *wire.MsgHeaders, peer *peerpkg.Peer) {
- // No channel handling here because peers do not need to block on
- // headers messages.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- return
- }
-
- sm.msgChan <- &headersMsg{headers: headers, peer: peer}
-}
-
-// DonePeer informs the blockmanager that a peer has disconnected.
-func (sm *SyncManager) DonePeer(peer *peerpkg.Peer) {
- // Ignore if we are shutting down.
- if atomic.LoadInt32(&sm.shutdown) != 0 {
- return
- }
-
- sm.msgChan <- &donePeerMsg{peer: peer}
-}
-
-// Start begins the core block handler which processes block and inv messages.
-func (sm *SyncManager) Start() {
- // Already started?
- if atomic.AddInt32(&sm.started, 1) != 1 {
- return
- }
-
- log.Trace("Starting sync manager")
- sm.wg.Add(1)
- go sm.blockHandler()
-}
-
-// Stop gracefully shuts down the sync manager by stopping all asynchronous
-// handlers and waiting for them to finish.
-func (sm *SyncManager) Stop() error {
- if atomic.AddInt32(&sm.shutdown, 1) != 1 {
- log.Warnf("Sync manager is already in the process of " +
- "shutting down")
- return nil
- }
-
- log.Infof("Sync manager shutting down")
- close(sm.quit)
- sm.wg.Wait()
- return nil
-}
-
-// SyncPeerID returns the ID of the current sync peer, or 0 if there is none.
-func (sm *SyncManager) SyncPeerID() int32 {
- reply := make(chan int32)
- sm.msgChan <- getSyncPeerMsg{reply: reply}
- return <-reply
-}
-
-// ProcessBlock makes use of ProcessBlock on an internal instance of a block
-// chain.
-func (sm *SyncManager) ProcessBlock(block *btcutil.Block, flags blockchain.BehaviorFlags) (bool, error) {
- reply := make(chan processBlockResponse, 1)
- sm.msgChan <- processBlockMsg{block: block, flags: flags, reply: reply}
- response := <-reply
- return response.isOrphan, response.err
-}
-
-// IsCurrent returns whether or not the sync manager believes it is synced with
-// the connected peers.
-func (sm *SyncManager) IsCurrent() bool {
- reply := make(chan bool)
- sm.msgChan <- isCurrentMsg{reply: reply}
- return <-reply
-}
-
-// Pause pauses the sync manager until the returned channel is closed.
-//
-// Note that while paused, all peer and block processing is halted. The
-// message sender should avoid pausing the sync manager for long durations.
-func (sm *SyncManager) Pause() chan<- struct{} {
- c := make(chan struct{})
- sm.msgChan <- pauseMsg{c}
- return c
-}
-
-// New constructs a new SyncManager. Use Start to begin processing asynchronous
-// block, tx, and inv updates.
-func New(config *Config) (*SyncManager, error) {
- sm := SyncManager{
- peerNotifier: config.PeerNotifier,
- chain: config.Chain,
- txMemPool: config.TxMemPool,
- chainParams: config.ChainParams,
- rejectedTxns: make(map[chainhash.Hash]struct{}),
- requestedTxns: make(map[chainhash.Hash]struct{}),
- requestedBlocks: make(map[chainhash.Hash]struct{}),
- peerStates: make(map[*peerpkg.Peer]*peerSyncState),
- progressLogger: newBlockProgressLogger("Processed", log),
- msgChan: make(chan interface{}, config.MaxPeers*3),
- headerList: list.New(),
- quit: make(chan struct{}),
- }
-
- best := sm.chain.BestSnapshot()
- if !config.DisableCheckpoints {
- // Initialize the next checkpoint based on the current height.
- sm.nextCheckpoint = sm.findNextHeaderCheckpoint(best.Height)
- if sm.nextCheckpoint != nil {
- sm.resetHeaderState(&best.Hash, best.Height)
- }
- } else {
- log.Info("Checkpoints are disabled")
- }
-
- sm.chain.Subscribe(sm.handleBlockchainNotification)
-
- return &sm, nil
-}
OSDN Git Service
