8a7b73e62b
ok @davecgh
1527 lines
48 KiB
Go
1527 lines
48 KiB
Go
// Copyright (c) 2013-2014 Conformal Systems LLC.
|
|
// Use of this source code is governed by an ISC
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package main
|
|
|
|
import (
|
|
"container/list"
|
|
"net"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/conformal/btcchain"
|
|
"github.com/conformal/btcdb"
|
|
"github.com/conformal/btcnet"
|
|
"github.com/conformal/btcutil"
|
|
"github.com/conformal/btcwire"
|
|
)
|
|
|
|
const (
|
|
chanBufferSize = 50
|
|
|
|
// minInFlightBlocks is the minimum number of blocks that should be
|
|
// in the request queue for headers-first mode before requesting
|
|
// more.
|
|
minInFlightBlocks = 10
|
|
|
|
// blockDbNamePrefix is the prefix for the block database name. The
|
|
// database type is appended to this value to form the full block
|
|
// database name.
|
|
blockDbNamePrefix = "blocks"
|
|
)
|
|
|
|
// newPeerMsg signifies a newly connected peer to the block handler.
|
|
type newPeerMsg struct {
|
|
peer *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 *peer
|
|
}
|
|
|
|
// 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 *btcwire.MsgInv
|
|
peer *peer
|
|
}
|
|
|
|
// blockMsg packages a bitcoin block message and the peer it came from together
|
|
// so the block handler has access to that information.
|
|
type headersMsg struct {
|
|
headers *btcwire.MsgHeaders
|
|
peer *peer
|
|
}
|
|
|
|
// donePeerMsg signifies a newly disconnected peer to the block handler.
|
|
type donePeerMsg struct {
|
|
peer *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 *peer
|
|
}
|
|
|
|
// getSyncPeerMsg is a message type to be sent across the message channel for
|
|
// retrieving the current sync peer.
|
|
type getSyncPeerMsg struct {
|
|
reply chan *peer
|
|
}
|
|
|
|
// checkConnectBlockMsg is a message type to be sent across the message channel
|
|
// for requesting chain to check if a block connects to the end of the current
|
|
// main chain.
|
|
type checkConnectBlockMsg struct {
|
|
block *btcutil.Block
|
|
reply chan error
|
|
}
|
|
|
|
// calcNextReqDifficultyResponse is a response sent to the reply channel of a
|
|
// calcNextReqDifficultyMsg query.
|
|
type calcNextReqDifficultyResponse struct {
|
|
difficulty uint32
|
|
err error
|
|
}
|
|
|
|
// calcNextReqDifficultyMsg is a message type to be sent across the message
|
|
// channel for requesting the required difficulty of the next block.
|
|
type calcNextReqDifficultyMsg struct {
|
|
timestamp time.Time
|
|
reply chan calcNextReqDifficultyResponse
|
|
}
|
|
|
|
// 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 can 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 btcchain.BehaviorFlags
|
|
reply chan processBlockResponse
|
|
}
|
|
|
|
// isCurrentMsg is a message type to be sent across the message channel for
|
|
// requesting whether or not the block manager believes it is synced with
|
|
// the currently connected peers.
|
|
type isCurrentMsg struct {
|
|
reply chan bool
|
|
}
|
|
|
|
// headerNode is used as a node in a list of headers that are linked together
|
|
// between checkpoints.
|
|
type headerNode struct {
|
|
height int64
|
|
sha *btcwire.ShaHash
|
|
}
|
|
|
|
// chainState tracks the state of the best chain as blocks are inserted. This
|
|
// is done because btcchain is currently not safe for concurrent access and the
|
|
// block manager is typically quite busy processing block and inventory.
|
|
// Therefore, requesting this information from chain through the block manager
|
|
// would not be anywhere near as efficient as simply updating it as each block
|
|
// is inserted and protecting it with a mutex.
|
|
type chainState struct {
|
|
sync.Mutex
|
|
newestHash *btcwire.ShaHash
|
|
newestHeight int64
|
|
pastMedianTime time.Time
|
|
pastMedianTimeErr error
|
|
}
|
|
|
|
// Best returns the block hash and height known for the tip of the best known
|
|
// chain.
|
|
//
|
|
// This function is safe for concurrent access.
|
|
func (c *chainState) Best() (*btcwire.ShaHash, int64) {
|
|
c.Lock()
|
|
defer c.Unlock()
|
|
|
|
return c.newestHash, c.newestHeight
|
|
}
|
|
|
|
// blockManager provides a concurrency safe block manager for handling all
|
|
// incoming blocks.
|
|
type blockManager struct {
|
|
server *server
|
|
started int32
|
|
shutdown int32
|
|
blockChain *btcchain.BlockChain
|
|
requestedTxns map[btcwire.ShaHash]struct{}
|
|
requestedBlocks map[btcwire.ShaHash]struct{}
|
|
receivedLogBlocks int64
|
|
receivedLogTx int64
|
|
lastBlockLogTime time.Time
|
|
processingReqs bool
|
|
syncPeer *peer
|
|
msgChan chan interface{}
|
|
chainState chainState
|
|
wg sync.WaitGroup
|
|
quit chan struct{}
|
|
|
|
// The following fields are used for headers-first mode.
|
|
headersFirstMode bool
|
|
headerList *list.List
|
|
startHeader *list.Element
|
|
nextCheckpoint *btcnet.Checkpoint
|
|
}
|
|
|
|
// resetHeaderState sets the headers-first mode state to values appropriate for
|
|
// syncing from a new peer.
|
|
func (b *blockManager) resetHeaderState(newestHash *btcwire.ShaHash, newestHeight int64) {
|
|
b.headersFirstMode = false
|
|
b.headerList.Init()
|
|
b.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 b.nextCheckpoint != nil {
|
|
node := headerNode{height: newestHeight, sha: newestHash}
|
|
b.headerList.PushBack(&node)
|
|
}
|
|
}
|
|
|
|
// updateChainState updates the chain state associated with the block manager.
|
|
// This allows fast access to chain information since btcchain is currently not
|
|
// safe for concurrent access and the block manager is typically quite busy
|
|
// processing block and inventory.
|
|
func (b *blockManager) updateChainState(newestHash *btcwire.ShaHash, newestHeight int64) {
|
|
b.chainState.Lock()
|
|
defer b.chainState.Unlock()
|
|
|
|
b.chainState.newestHash = newestHash
|
|
b.chainState.newestHeight = newestHeight
|
|
medianTime, err := b.blockChain.CalcPastMedianTime()
|
|
if err != nil {
|
|
b.chainState.pastMedianTimeErr = err
|
|
} else {
|
|
b.chainState.pastMedianTime = medianTime
|
|
}
|
|
}
|
|
|
|
// 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 (b *blockManager) findNextHeaderCheckpoint(height int64) *btcnet.Checkpoint {
|
|
// There is no next checkpoint if checkpoints are disabled or there are
|
|
// none for this current network.
|
|
if cfg.DisableCheckpoints {
|
|
return nil
|
|
}
|
|
checkpoints := b.server.netParams.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 (b *blockManager) startSync(peers *list.List) {
|
|
// Return now if we're already syncing.
|
|
if b.syncPeer != nil {
|
|
return
|
|
}
|
|
|
|
// Find the height of the current known best block.
|
|
_, height, err := b.server.db.NewestSha()
|
|
if err != nil {
|
|
bmgrLog.Errorf("%v", err)
|
|
return
|
|
}
|
|
|
|
var bestPeer *peer
|
|
var enext *list.Element
|
|
for e := peers.Front(); e != nil; e = enext {
|
|
enext = e.Next()
|
|
p := e.Value.(*peer)
|
|
|
|
// Remove sync candidate peers that are no longer candidates due
|
|
// to passing their latest known block. NOTE: The < is
|
|
// intentional as opposed to <=. While techcnically 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 p.lastBlock < int32(height) {
|
|
peers.Remove(e)
|
|
continue
|
|
}
|
|
|
|
// TODO(davec): Use a better algorithm to choose the best peer.
|
|
// For now, just pick the first available candidate.
|
|
bestPeer = p
|
|
}
|
|
|
|
// Start syncing from the best peer if one was selected.
|
|
if bestPeer != nil {
|
|
locator, err := b.blockChain.LatestBlockLocator()
|
|
if err != nil {
|
|
bmgrLog.Errorf("Failed to get block locator for the "+
|
|
"latest block: %v", err)
|
|
return
|
|
}
|
|
|
|
bmgrLog.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 b.nextCheckpoint != nil && height < b.nextCheckpoint.Height &&
|
|
!cfg.RegressionTest && !cfg.DisableCheckpoints {
|
|
|
|
bestPeer.PushGetHeadersMsg(locator, b.nextCheckpoint.Hash)
|
|
b.headersFirstMode = true
|
|
bmgrLog.Infof("Downloading headers for blocks %d to "+
|
|
"%d from peer %s", height+1,
|
|
b.nextCheckpoint.Height, bestPeer.addr)
|
|
} else {
|
|
bestPeer.PushGetBlocksMsg(locator, &zeroHash)
|
|
}
|
|
b.syncPeer = bestPeer
|
|
} else {
|
|
bmgrLog.Warnf("No sync peer candidates available")
|
|
}
|
|
}
|
|
|
|
// isSyncCandidate returns whether or not the peer is a candidate to consider
|
|
// syncing from.
|
|
func (b *blockManager) isSyncCandidate(p *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 cfg.RegressionTest {
|
|
// 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(p.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.
|
|
if p.services&btcwire.SFNodeNetwork != btcwire.SFNodeNetwork {
|
|
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 (b *blockManager) handleNewPeerMsg(peers *list.List, p *peer) {
|
|
// Ignore if in the process of shutting down.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
return
|
|
}
|
|
|
|
bmgrLog.Infof("New valid peer %s (%s)", p, p.userAgent)
|
|
|
|
// Ignore the peer if it's not a sync candidate.
|
|
if !b.isSyncCandidate(p) {
|
|
return
|
|
}
|
|
|
|
// Add the peer as a candidate to sync from.
|
|
peers.PushBack(p)
|
|
|
|
// Start syncing by choosing the best candidate if needed.
|
|
b.startSync(peers)
|
|
}
|
|
|
|
// 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 (b *blockManager) handleDonePeerMsg(peers *list.List, p *peer) {
|
|
// Remove the peer from the list of candidate peers.
|
|
for e := peers.Front(); e != nil; e = e.Next() {
|
|
if e.Value == p {
|
|
peers.Remove(e)
|
|
break
|
|
}
|
|
}
|
|
|
|
bmgrLog.Infof("Lost peer %s", p)
|
|
|
|
// Remove requested transactions from the global map so that they will
|
|
// be fetched from elsewhere next time we get an inv.
|
|
for k := range p.requestedTxns {
|
|
delete(b.requestedTxns, k)
|
|
}
|
|
|
|
// Remove requested blocks from the global map so that they will be
|
|
// fetched from elsewhere next time we get an inv.
|
|
// TODO(oga) we could possibly here check which peers have these blocks
|
|
// and request them now to speed things up a little.
|
|
for k := range p.requestedBlocks {
|
|
delete(b.requestedBlocks, k)
|
|
}
|
|
|
|
// 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 b.syncPeer != nil && b.syncPeer == p {
|
|
b.syncPeer = nil
|
|
if b.headersFirstMode {
|
|
// This really shouldn't fail. We have a fairly
|
|
// unrecoverable database issue if it does.
|
|
newestHash, height, err := b.server.db.NewestSha()
|
|
if err != nil {
|
|
bmgrLog.Warnf("Unable to obtain latest "+
|
|
"block information from the database: "+
|
|
"%v", err)
|
|
return
|
|
}
|
|
b.resetHeaderState(newestHash, height)
|
|
}
|
|
b.startSync(peers)
|
|
}
|
|
}
|
|
|
|
// logBlockHeight logs a new block height as an information message to show
|
|
// progress to the user. In order to prevent spam, it limits logging to one
|
|
// message every 10 seconds with duration and totals included.
|
|
func (b *blockManager) logBlockHeight(block *btcutil.Block) {
|
|
b.receivedLogBlocks++
|
|
b.receivedLogTx += int64(len(block.MsgBlock().Transactions))
|
|
|
|
now := time.Now()
|
|
duration := now.Sub(b.lastBlockLogTime)
|
|
if duration < time.Second*10 {
|
|
return
|
|
}
|
|
|
|
// Truncate the duration to 10s of milliseconds.
|
|
durationMillis := int64(duration / time.Millisecond)
|
|
tDuration := 10 * time.Millisecond * time.Duration(durationMillis/10)
|
|
|
|
// Log information about new block height.
|
|
blockStr := "blocks"
|
|
if b.receivedLogBlocks == 1 {
|
|
blockStr = "block"
|
|
}
|
|
txStr := "transactions"
|
|
if b.receivedLogTx == 1 {
|
|
txStr = "transaction"
|
|
}
|
|
bmgrLog.Infof("Processed %d %s in the last %s (%d %s, height %d, %s)",
|
|
b.receivedLogBlocks, blockStr, tDuration, b.receivedLogTx,
|
|
txStr, block.Height(), block.MsgBlock().Header.Timestamp)
|
|
|
|
b.receivedLogBlocks = 0
|
|
b.receivedLogTx = 0
|
|
b.lastBlockLogTime = now
|
|
}
|
|
|
|
// handleTxMsg handles transaction messages from all peers.
|
|
func (b *blockManager) handleTxMsg(tmsg *txMsg) {
|
|
// 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. Unfortuantely 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.
|
|
|
|
// Process the transaction to include validation, insertion in the
|
|
// memory pool, orphan handling, etc.
|
|
err := tmsg.peer.server.txMemPool.ProcessTransaction(tmsg.tx, true, true)
|
|
|
|
// 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.
|
|
txHash := tmsg.tx.Sha()
|
|
delete(tmsg.peer.requestedTxns, *txHash)
|
|
delete(b.requestedTxns, *txHash)
|
|
|
|
if err != nil {
|
|
// 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.(RuleError); ok {
|
|
bmgrLog.Debugf("Rejected transaction %v from %s: %v",
|
|
txHash, tmsg.peer, err)
|
|
} else {
|
|
bmgrLog.Errorf("Failed to process transaction %v: %v",
|
|
txHash, err)
|
|
}
|
|
|
|
// Convert the error into an appropriate reject message and
|
|
// send it.
|
|
code, reason := errToRejectErr(err)
|
|
tmsg.peer.PushRejectMsg(btcwire.CmdBlock, code, reason, txHash,
|
|
false)
|
|
return
|
|
}
|
|
}
|
|
|
|
// current returns true if we believe we are synced with our peers, false if we
|
|
// still have blocks to check
|
|
func (b *blockManager) current() bool {
|
|
if !b.blockChain.IsCurrent() {
|
|
return false
|
|
}
|
|
|
|
// if blockChain thinks we are current and we have no syncPeer it
|
|
// is probably right.
|
|
if b.syncPeer == nil {
|
|
return true
|
|
}
|
|
|
|
_, height, err := b.server.db.NewestSha()
|
|
// No matter what chain thinks, if we are below the block we are
|
|
// syncing to we are not current.
|
|
// TODO(oga) we can get chain to return the height of each block when we
|
|
// parse an orphan, which would allow us to update the height of peers
|
|
// from what it was at initial handshake.
|
|
if err != nil || height < int64(b.syncPeer.lastBlock) {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// handleBlockMsg handles block messages from all peers.
|
|
func (b *blockManager) handleBlockMsg(bmsg *blockMsg) {
|
|
// If we didn't ask for this block then the peer is misbehaving.
|
|
blockSha, _ := bmsg.block.Sha()
|
|
if _, ok := bmsg.peer.requestedBlocks[*blockSha]; !ok {
|
|
// 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 !cfg.RegressionTest {
|
|
bmgrLog.Warnf("Got unrequested block %v from %s -- "+
|
|
"disconnecting", blockSha, bmsg.peer.addr)
|
|
bmsg.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 := btcchain.BFNone
|
|
if b.headersFirstMode {
|
|
firstNodeEl := b.headerList.Front()
|
|
if firstNodeEl != nil {
|
|
firstNode := firstNodeEl.Value.(*headerNode)
|
|
if blockSha.IsEqual(firstNode.sha) {
|
|
behaviorFlags |= btcchain.BFFastAdd
|
|
if firstNode.sha.IsEqual(b.nextCheckpoint.Hash) {
|
|
isCheckpointBlock = true
|
|
} else {
|
|
b.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(bmsg.peer.requestedBlocks, *blockSha)
|
|
delete(b.requestedBlocks, *blockSha)
|
|
|
|
// Process the block to include validation, best chain selection, orphan
|
|
// handling, etc.
|
|
isOrphan, err := b.blockChain.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.(btcchain.RuleError); ok {
|
|
bmgrLog.Infof("Rejected block %v from %s: %v", blockSha,
|
|
bmsg.peer, err)
|
|
} else {
|
|
bmgrLog.Errorf("Failed to process block %v: %v",
|
|
blockSha, err)
|
|
}
|
|
|
|
// Convert the error into an appropriate reject message and
|
|
// send it.
|
|
code, reason := errToRejectErr(err)
|
|
bmsg.peer.PushRejectMsg(btcwire.CmdBlock, code, reason,
|
|
blockSha, false)
|
|
return
|
|
}
|
|
|
|
// Request the parents for the orphan block from the peer that sent it.
|
|
if isOrphan {
|
|
orphanRoot := b.blockChain.GetOrphanRoot(blockSha)
|
|
locator, err := b.blockChain.LatestBlockLocator()
|
|
if err != nil {
|
|
bmgrLog.Warnf("Failed to get block locator for the "+
|
|
"latest block: %v", err)
|
|
} else {
|
|
bmsg.peer.PushGetBlocksMsg(locator, orphanRoot)
|
|
}
|
|
} else {
|
|
// When the block is not an orphan, log information about it and
|
|
// update the chain state.
|
|
|
|
b.logBlockHeight(bmsg.block)
|
|
|
|
// Query the db for the latest best block since the block
|
|
// that was processed could be on a side chain or have caused
|
|
// a reorg.
|
|
newestSha, newestHeight, _ := b.server.db.NewestSha()
|
|
b.updateChainState(newestSha, newestHeight)
|
|
|
|
// Allow any clients performing long polling via the
|
|
// getblocktemplate RPC to be notified when the new block causes
|
|
// their old block template to become stale.
|
|
rpcServer := b.server.rpcServer
|
|
if rpcServer != nil {
|
|
rpcServer.gbtWorkState.NotifyBlockConnected(blockSha)
|
|
}
|
|
}
|
|
|
|
// Sync the db to disk.
|
|
b.server.db.Sync()
|
|
|
|
// Nothing more to do if we aren't in headers-first mode.
|
|
if !b.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 b.startHeader != nil &&
|
|
len(bmsg.peer.requestedBlocks) < minInFlightBlocks {
|
|
b.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 := b.nextCheckpoint.Height
|
|
prevHash := b.nextCheckpoint.Hash
|
|
b.nextCheckpoint = b.findNextHeaderCheckpoint(prevHeight)
|
|
if b.nextCheckpoint != nil {
|
|
locator := btcchain.BlockLocator([]*btcwire.ShaHash{prevHash})
|
|
err := bmsg.peer.PushGetHeadersMsg(locator, b.nextCheckpoint.Hash)
|
|
if err != nil {
|
|
bmgrLog.Warnf("Failed to send getheaders message to "+
|
|
"peer %s: %v", bmsg.peer.addr, err)
|
|
return
|
|
}
|
|
bmgrLog.Infof("Downloading headers for blocks %d to %d from "+
|
|
"peer %s", prevHeight+1, b.nextCheckpoint.Height,
|
|
b.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).
|
|
b.headersFirstMode = false
|
|
b.headerList.Init()
|
|
bmgrLog.Infof("Reached the final checkpoint -- switching to normal mode")
|
|
locator := btcchain.BlockLocator([]*btcwire.ShaHash{blockSha})
|
|
err = bmsg.peer.PushGetBlocksMsg(locator, &zeroHash)
|
|
if err != nil {
|
|
bmgrLog.Warnf("Failed to send getblocks message to peer %s: %v",
|
|
bmsg.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 (b *blockManager) fetchHeaderBlocks() {
|
|
// Nothing to do if there is no start header.
|
|
if b.startHeader == nil {
|
|
bmgrLog.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 btcwire.MaxInvPerMsg by
|
|
// the function, so no need to double check it here.
|
|
gdmsg := btcwire.NewMsgGetDataSizeHint(uint(b.headerList.Len()))
|
|
numRequested := 0
|
|
for e := b.startHeader; e != nil; e = e.Next() {
|
|
node, ok := e.Value.(*headerNode)
|
|
if !ok {
|
|
bmgrLog.Warn("Header list node type is not a headerNode")
|
|
continue
|
|
}
|
|
|
|
iv := btcwire.NewInvVect(btcwire.InvTypeBlock, node.sha)
|
|
haveInv, err := b.haveInventory(iv)
|
|
if err != nil {
|
|
bmgrLog.Warnf("Unexpected failure when checking for "+
|
|
"existing inventory during header block "+
|
|
"fetch: %v", err)
|
|
}
|
|
if !haveInv {
|
|
b.requestedBlocks[*node.sha] = struct{}{}
|
|
b.syncPeer.requestedBlocks[*node.sha] = struct{}{}
|
|
gdmsg.AddInvVect(iv)
|
|
numRequested++
|
|
}
|
|
b.startHeader = e.Next()
|
|
if numRequested >= btcwire.MaxInvPerMsg {
|
|
break
|
|
}
|
|
}
|
|
if len(gdmsg.InvList) > 0 {
|
|
b.syncPeer.QueueMessage(gdmsg, nil)
|
|
}
|
|
}
|
|
|
|
// handleHeadersMsghandles headers messages from all peers.
|
|
func (b *blockManager) handleHeadersMsg(hmsg *headersMsg) {
|
|
// The remote peer is misbehaving if we didn't request headers.
|
|
msg := hmsg.headers
|
|
numHeaders := len(msg.Headers)
|
|
if !b.headersFirstMode {
|
|
bmgrLog.Warnf("Got %d unrequested headers from %s -- "+
|
|
"disconnecting", numHeaders, hmsg.peer.addr)
|
|
hmsg.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 *btcwire.ShaHash
|
|
for _, blockHeader := range msg.Headers {
|
|
blockHash, err := blockHeader.BlockSha()
|
|
if err != nil {
|
|
bmgrLog.Warnf("Failed to compute hash of header "+
|
|
"received from peer %s -- disconnecting",
|
|
hmsg.peer.addr)
|
|
hmsg.peer.Disconnect()
|
|
return
|
|
}
|
|
finalHash = &blockHash
|
|
|
|
// Ensure there is a previous header to compare against.
|
|
prevNodeEl := b.headerList.Back()
|
|
if prevNodeEl == nil {
|
|
bmgrLog.Warnf("Header list does not contain a previous" +
|
|
"element as expected -- disconnecting peer")
|
|
hmsg.peer.Disconnect()
|
|
return
|
|
}
|
|
|
|
// Ensure the header properly connects to the previous one and
|
|
// add it to the list of headers.
|
|
node := headerNode{sha: &blockHash}
|
|
prevNode := prevNodeEl.Value.(*headerNode)
|
|
if prevNode.sha.IsEqual(&blockHeader.PrevBlock) {
|
|
node.height = prevNode.height + 1
|
|
e := b.headerList.PushBack(&node)
|
|
if b.startHeader == nil {
|
|
b.startHeader = e
|
|
}
|
|
} else {
|
|
bmgrLog.Warnf("Received block header that does not"+
|
|
"properly connect to the chain from peer %s "+
|
|
"-- disconnecting", hmsg.peer.addr)
|
|
hmsg.peer.Disconnect()
|
|
return
|
|
}
|
|
|
|
// Verify the header at the next checkpoint height matches.
|
|
if node.height == b.nextCheckpoint.Height {
|
|
if node.sha.IsEqual(b.nextCheckpoint.Hash) {
|
|
receivedCheckpoint = true
|
|
bmgrLog.Infof("Verified downloaded block "+
|
|
"header against checkpoint at height "+
|
|
"%d/hash %s", node.height, node.sha)
|
|
} else {
|
|
bmgrLog.Warnf("Block header at height %d/hash "+
|
|
"%s from peer %s does NOT match "+
|
|
"expected checkpoint hash of %s -- "+
|
|
"disconnecting", node.height,
|
|
node.sha, hmsg.peer.addr,
|
|
b.nextCheckpoint.Hash)
|
|
hmsg.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.
|
|
b.headerList.Remove(b.headerList.Front())
|
|
bmgrLog.Infof("Received %v block headers: Fetching blocks",
|
|
b.headerList.Len())
|
|
b.lastBlockLogTime = time.Now()
|
|
b.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 := btcchain.BlockLocator([]*btcwire.ShaHash{finalHash})
|
|
err := hmsg.peer.PushGetHeadersMsg(locator, b.nextCheckpoint.Hash)
|
|
if err != nil {
|
|
bmgrLog.Warnf("Failed to send getheaders message to "+
|
|
"peer %s: %v", hmsg.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 (b *blockManager) haveInventory(invVect *btcwire.InvVect) (bool, error) {
|
|
switch invVect.Type {
|
|
case btcwire.InvTypeBlock:
|
|
// Ask chain if the block is known to it in any form (main
|
|
// chain, side chain, or orphan).
|
|
return b.blockChain.HaveBlock(&invVect.Hash)
|
|
|
|
case btcwire.InvTypeTx:
|
|
// Ask the transaction memory pool if the transaction is known
|
|
// to it in any form (main pool or orphan).
|
|
if b.server.txMemPool.HaveTransaction(&invVect.Hash) {
|
|
return true, nil
|
|
}
|
|
|
|
// Check if the transaction exists from the point of view of the
|
|
// end of the main chain.
|
|
return b.server.db.ExistsTxSha(&invVect.Hash)
|
|
}
|
|
|
|
// 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 (b *blockManager) handleInvMsg(imsg *invMsg) {
|
|
// Ignore invs from peers that aren't the sync if we are not current.
|
|
// Helps prevent fetching a mass of orphans.
|
|
if imsg.peer != b.syncPeer && !b.current() {
|
|
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 == btcwire.InvTypeBlock {
|
|
lastBlock = i
|
|
break
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
chain := b.blockChain
|
|
for i, iv := range invVects {
|
|
// Ignore unsupported inventory types.
|
|
if iv.Type != btcwire.InvTypeBlock && iv.Type != btcwire.InvTypeTx {
|
|
continue
|
|
}
|
|
|
|
// Add the inventory to the cache of known inventory
|
|
// for the peer.
|
|
imsg.peer.AddKnownInventory(iv)
|
|
|
|
// Ignore inventory when we're in headers-first mode.
|
|
if b.headersFirstMode {
|
|
continue
|
|
}
|
|
|
|
// Request the inventory if we don't already have it.
|
|
haveInv, err := b.haveInventory(iv)
|
|
if err != nil {
|
|
bmgrLog.Warnf("Unexpected failure when checking for "+
|
|
"existing inventory during inv message "+
|
|
"processing: %v", err)
|
|
continue
|
|
}
|
|
if !haveInv {
|
|
// Add it to the request queue.
|
|
imsg.peer.requestQueue.PushBack(iv)
|
|
continue
|
|
}
|
|
|
|
if iv.Type == btcwire.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 chain.IsKnownOrphan(&iv.Hash) {
|
|
// Request blocks starting at the latest known
|
|
// up to the root of the orphan that just came
|
|
// in.
|
|
orphanRoot := chain.GetOrphanRoot(&iv.Hash)
|
|
locator, err := chain.LatestBlockLocator()
|
|
if err != nil {
|
|
bmgrLog.Errorf("PEER: Failed to get block "+
|
|
"locator for the latest block: "+
|
|
"%v", err)
|
|
continue
|
|
}
|
|
imsg.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 := chain.BlockLocatorFromHash(&iv.Hash)
|
|
imsg.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 := btcwire.NewMsgGetData()
|
|
requestQueue := imsg.peer.requestQueue
|
|
for e := requestQueue.Front(); e != nil; e = requestQueue.Front() {
|
|
iv := e.Value.(*btcwire.InvVect)
|
|
imsg.peer.requestQueue.Remove(e)
|
|
|
|
switch iv.Type {
|
|
case btcwire.InvTypeBlock:
|
|
// Request the block if there is not already a pending
|
|
// request.
|
|
if _, exists := b.requestedBlocks[iv.Hash]; !exists {
|
|
b.requestedBlocks[iv.Hash] = struct{}{}
|
|
imsg.peer.requestedBlocks[iv.Hash] = struct{}{}
|
|
gdmsg.AddInvVect(iv)
|
|
numRequested++
|
|
}
|
|
|
|
case btcwire.InvTypeTx:
|
|
// Request the transaction if there is not already a
|
|
// pending request.
|
|
if _, exists := b.requestedTxns[iv.Hash]; !exists {
|
|
b.requestedTxns[iv.Hash] = struct{}{}
|
|
imsg.peer.requestedTxns[iv.Hash] = struct{}{}
|
|
gdmsg.AddInvVect(iv)
|
|
numRequested++
|
|
}
|
|
}
|
|
|
|
if numRequested >= btcwire.MaxInvPerMsg {
|
|
break
|
|
}
|
|
}
|
|
if len(gdmsg.InvList) > 0 {
|
|
imsg.peer.QueueMessage(gdmsg, nil)
|
|
}
|
|
}
|
|
|
|
// blockHandler is the main handler for the block 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 block manager controls which blocks are needed and how
|
|
// the fetching should proceed.
|
|
func (b *blockManager) blockHandler() {
|
|
candidatePeers := list.New()
|
|
out:
|
|
for {
|
|
select {
|
|
case m := <-b.msgChan:
|
|
switch msg := m.(type) {
|
|
case *newPeerMsg:
|
|
b.handleNewPeerMsg(candidatePeers, msg.peer)
|
|
|
|
case *txMsg:
|
|
b.handleTxMsg(msg)
|
|
msg.peer.txProcessed <- struct{}{}
|
|
|
|
case *blockMsg:
|
|
b.handleBlockMsg(msg)
|
|
msg.peer.blockProcessed <- struct{}{}
|
|
|
|
case *invMsg:
|
|
b.handleInvMsg(msg)
|
|
|
|
case *headersMsg:
|
|
b.handleHeadersMsg(msg)
|
|
|
|
case *donePeerMsg:
|
|
b.handleDonePeerMsg(candidatePeers, msg.peer)
|
|
|
|
case getSyncPeerMsg:
|
|
msg.reply <- b.syncPeer
|
|
|
|
case checkConnectBlockMsg:
|
|
err := b.blockChain.CheckConnectBlock(msg.block)
|
|
msg.reply <- err
|
|
|
|
case calcNextReqDifficultyMsg:
|
|
difficulty, err :=
|
|
b.blockChain.CalcNextRequiredDifficulty(
|
|
msg.timestamp)
|
|
msg.reply <- calcNextReqDifficultyResponse{
|
|
difficulty: difficulty,
|
|
err: err,
|
|
}
|
|
|
|
case processBlockMsg:
|
|
isOrphan, err := b.blockChain.ProcessBlock(
|
|
msg.block, msg.flags)
|
|
if err != nil {
|
|
msg.reply <- processBlockResponse{
|
|
isOrphan: false,
|
|
err: err,
|
|
}
|
|
}
|
|
|
|
// Query the db for the latest best block since
|
|
// the block that was processed could be on a
|
|
// side chain or have caused a reorg.
|
|
newestSha, newestHeight, _ := b.server.db.NewestSha()
|
|
b.updateChainState(newestSha, newestHeight)
|
|
|
|
msg.reply <- processBlockResponse{
|
|
isOrphan: isOrphan,
|
|
err: nil,
|
|
}
|
|
|
|
case isCurrentMsg:
|
|
msg.reply <- b.current()
|
|
|
|
default:
|
|
bmgrLog.Warnf("Invalid message type in block "+
|
|
"handler: %T", msg)
|
|
}
|
|
|
|
case <-b.quit:
|
|
break out
|
|
}
|
|
}
|
|
|
|
b.wg.Done()
|
|
bmgrLog.Trace("Block handler done")
|
|
}
|
|
|
|
// handleNotifyMsg handles notifications from btcchain. It does things such
|
|
// as request orphan block parents and relay accepted blocks to connected peers.
|
|
func (b *blockManager) handleNotifyMsg(notification *btcchain.Notification) {
|
|
switch notification.Type {
|
|
// A block has been accepted into the block chain. Relay it to other
|
|
// peers.
|
|
case btcchain.NTBlockAccepted:
|
|
// Don't relay if we are not current. Other peers that are
|
|
// current should already know about it.
|
|
|
|
if !b.current() {
|
|
return
|
|
}
|
|
|
|
block, ok := notification.Data.(*btcutil.Block)
|
|
if !ok {
|
|
bmgrLog.Warnf("Chain accepted notification is not a block.")
|
|
break
|
|
}
|
|
|
|
// It's ok to ignore the error here since the notification is
|
|
// coming from the chain code which has already cached the hash.
|
|
hash, _ := block.Sha()
|
|
|
|
// Generate the inventory vector and relay it.
|
|
iv := btcwire.NewInvVect(btcwire.InvTypeBlock, hash)
|
|
b.server.RelayInventory(iv)
|
|
|
|
// A block has been connected to the main block chain.
|
|
case btcchain.NTBlockConnected:
|
|
block, ok := notification.Data.(*btcutil.Block)
|
|
if !ok {
|
|
bmgrLog.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. Also, remove any
|
|
// transactions which are now double spends as a result of these
|
|
// new transactions. Note that removing a transaction from
|
|
// pool also removes any transactions which depend on it,
|
|
// recursively.
|
|
for _, tx := range block.Transactions()[1:] {
|
|
b.server.txMemPool.RemoveTransaction(tx)
|
|
b.server.txMemPool.RemoveDoubleSpends(tx)
|
|
}
|
|
|
|
if r := b.server.rpcServer; r != nil {
|
|
// Now that this block is in the blockchain we can mark
|
|
// all the transactions (except the coinbase) as no
|
|
// longer needing rebroadcasting.
|
|
for _, tx := range block.Transactions()[1:] {
|
|
iv := btcwire.NewInvVect(btcwire.InvTypeTx, tx.Sha())
|
|
b.server.RemoveRebroadcastInventory(iv)
|
|
}
|
|
|
|
// Notify registered websocket clients of incoming block.
|
|
r.ntfnMgr.NotifyBlockConnected(block)
|
|
}
|
|
|
|
// A block has been disconnected from the main block chain.
|
|
case btcchain.NTBlockDisconnected:
|
|
block, ok := notification.Data.(*btcutil.Block)
|
|
if !ok {
|
|
bmgrLog.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 := b.server.txMemPool.MaybeAcceptTransaction(tx, nil, false, true)
|
|
if err != nil {
|
|
// Remove the transaction and all transactions
|
|
// that depend on it if it wasn't accepted into
|
|
// the transaction pool.
|
|
b.server.txMemPool.RemoveTransaction(tx)
|
|
}
|
|
}
|
|
|
|
// Notify registered websocket clients.
|
|
if r := b.server.rpcServer; r != nil {
|
|
r.ntfnMgr.NotifyBlockDisconnected(block)
|
|
}
|
|
}
|
|
}
|
|
|
|
// NewPeer informs the block manager of a newly active peer.
|
|
func (b *blockManager) NewPeer(p *peer) {
|
|
// Ignore if we are shutting down.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
return
|
|
}
|
|
|
|
b.msgChan <- &newPeerMsg{peer: p}
|
|
}
|
|
|
|
// QueueTx adds the passed transaction message and peer to the block handling
|
|
// queue.
|
|
func (b *blockManager) QueueTx(tx *btcutil.Tx, p *peer) {
|
|
// Don't accept more transactions if we're shutting down.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
p.txProcessed <- struct{}{}
|
|
return
|
|
}
|
|
|
|
b.msgChan <- &txMsg{tx: tx, peer: p}
|
|
}
|
|
|
|
// QueueBlock adds the passed block message and peer to the block handling queue.
|
|
func (b *blockManager) QueueBlock(block *btcutil.Block, p *peer) {
|
|
// Don't accept more blocks if we're shutting down.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
p.blockProcessed <- struct{}{}
|
|
return
|
|
}
|
|
|
|
b.msgChan <- &blockMsg{block: block, peer: p}
|
|
}
|
|
|
|
// QueueInv adds the passed inv message and peer to the block handling queue.
|
|
func (b *blockManager) QueueInv(inv *btcwire.MsgInv, p *peer) {
|
|
// No channel handling here because peers do not need to block on inv
|
|
// messages.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
return
|
|
}
|
|
|
|
b.msgChan <- &invMsg{inv: inv, peer: p}
|
|
}
|
|
|
|
// QueueHeaders adds the passed headers message and peer to the block handling
|
|
// queue.
|
|
func (b *blockManager) QueueHeaders(headers *btcwire.MsgHeaders, p *peer) {
|
|
// No channel handling here because peers do not need to block on
|
|
// headers messages.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
return
|
|
}
|
|
|
|
b.msgChan <- &headersMsg{headers: headers, peer: p}
|
|
}
|
|
|
|
// DonePeer informs the blockmanager that a peer has disconnected.
|
|
func (b *blockManager) DonePeer(p *peer) {
|
|
// Ignore if we are shutting down.
|
|
if atomic.LoadInt32(&b.shutdown) != 0 {
|
|
return
|
|
}
|
|
|
|
b.msgChan <- &donePeerMsg{peer: p}
|
|
}
|
|
|
|
// Start begins the core block handler which processes block and inv messages.
|
|
func (b *blockManager) Start() {
|
|
// Already started?
|
|
if atomic.AddInt32(&b.started, 1) != 1 {
|
|
return
|
|
}
|
|
|
|
bmgrLog.Trace("Starting block manager")
|
|
b.wg.Add(1)
|
|
go b.blockHandler()
|
|
}
|
|
|
|
// Stop gracefully shuts down the block manager by stopping all asynchronous
|
|
// handlers and waiting for them to finish.
|
|
func (b *blockManager) Stop() error {
|
|
if atomic.AddInt32(&b.shutdown, 1) != 1 {
|
|
bmgrLog.Warnf("Block manager is already in the process of " +
|
|
"shutting down")
|
|
return nil
|
|
}
|
|
|
|
bmgrLog.Infof("Block manager shutting down")
|
|
close(b.quit)
|
|
b.wg.Wait()
|
|
return nil
|
|
}
|
|
|
|
// SyncPeer returns the current sync peer.
|
|
func (b *blockManager) SyncPeer() *peer {
|
|
reply := make(chan *peer)
|
|
b.msgChan <- getSyncPeerMsg{reply: reply}
|
|
return <-reply
|
|
}
|
|
|
|
// CheckConnectBlock performs several checks to confirm connecting the passed
|
|
// block to the main chain does not violate any rules. This function makes use
|
|
// of CheckConnectBlock on an internal instance of a block chain. It is funneled
|
|
// through the block manager since btcchain is not safe for concurrent access.
|
|
func (b *blockManager) CheckConnectBlock(block *btcutil.Block) error {
|
|
reply := make(chan error)
|
|
b.msgChan <- checkConnectBlockMsg{block: block, reply: reply}
|
|
return <-reply
|
|
}
|
|
|
|
// CalcNextRequiredDifficulty calculates the required difficulty for the next
|
|
// block after the current main chain. This function makes use of
|
|
// CalcNextRequiredDifficulty on an internal instance of a block chain. It is
|
|
// funneled through the block manager since btcchain is not safe for concurrent
|
|
// access.
|
|
func (b *blockManager) CalcNextRequiredDifficulty(timestamp time.Time) (uint32, error) {
|
|
reply := make(chan calcNextReqDifficultyResponse)
|
|
b.msgChan <- calcNextReqDifficultyMsg{timestamp: timestamp, reply: reply}
|
|
response := <-reply
|
|
return response.difficulty, response.err
|
|
}
|
|
|
|
// ProcessBlock makes use of ProcessBlock on an internal instance of a block
|
|
// chain. It is funneled through the block manager since btcchain is not safe
|
|
// for concurrent access.
|
|
func (b *blockManager) ProcessBlock(block *btcutil.Block, flags btcchain.BehaviorFlags) (bool, error) {
|
|
reply := make(chan processBlockResponse, 1)
|
|
b.msgChan <- processBlockMsg{block: block, flags: flags, reply: reply}
|
|
response := <-reply
|
|
return response.isOrphan, response.err
|
|
}
|
|
|
|
// IsCurrent returns whether or not the block manager believes it is synced with
|
|
// the connected peers.
|
|
func (b *blockManager) IsCurrent() bool {
|
|
reply := make(chan bool)
|
|
b.msgChan <- isCurrentMsg{reply: reply}
|
|
return <-reply
|
|
}
|
|
|
|
// newBlockManager returns a new bitcoin block manager.
|
|
// Use Start to begin processing asynchronous block and inv updates.
|
|
func newBlockManager(s *server) (*blockManager, error) {
|
|
newestHash, height, err := s.db.NewestSha()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
bm := blockManager{
|
|
server: s,
|
|
requestedTxns: make(map[btcwire.ShaHash]struct{}),
|
|
requestedBlocks: make(map[btcwire.ShaHash]struct{}),
|
|
lastBlockLogTime: time.Now(),
|
|
msgChan: make(chan interface{}, cfg.MaxPeers*3),
|
|
headerList: list.New(),
|
|
quit: make(chan struct{}),
|
|
}
|
|
bm.blockChain = btcchain.New(s.db, s.netParams, bm.handleNotifyMsg)
|
|
bm.blockChain.DisableCheckpoints(cfg.DisableCheckpoints)
|
|
if !cfg.DisableCheckpoints {
|
|
// Initialize the next checkpoint based on the current height.
|
|
bm.nextCheckpoint = bm.findNextHeaderCheckpoint(height)
|
|
if bm.nextCheckpoint != nil {
|
|
bm.resetHeaderState(newestHash, height)
|
|
}
|
|
} else {
|
|
bmgrLog.Info("Checkpoints are disabled")
|
|
}
|
|
|
|
bmgrLog.Infof("Generating initial block node index. This may " +
|
|
"take a while...")
|
|
err = bm.blockChain.GenerateInitialIndex()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
bmgrLog.Infof("Block index generation complete")
|
|
|
|
// Initialize the chain state now that the intial block node index has
|
|
// been generated.
|
|
bm.updateChainState(newestHash, height)
|
|
|
|
return &bm, nil
|
|
}
|
|
|
|
// removeRegressionDB removes the existing regression test database if running
|
|
// in regression test mode and it already exists.
|
|
func removeRegressionDB(dbPath string) error {
|
|
// Dont do anything if not in regression test mode.
|
|
if !cfg.RegressionTest {
|
|
return nil
|
|
}
|
|
|
|
// Remove the old regression test database if it already exists.
|
|
fi, err := os.Stat(dbPath)
|
|
if err == nil {
|
|
btcdLog.Infof("Removing regression test database from '%s'", dbPath)
|
|
if fi.IsDir() {
|
|
err := os.RemoveAll(dbPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
} else {
|
|
err := os.Remove(dbPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// dbPath returns the path to the block database given a database type.
|
|
func blockDbPath(dbType string) string {
|
|
// The database name is based on the database type.
|
|
dbName := blockDbNamePrefix + "_" + dbType
|
|
if dbType == "sqlite" {
|
|
dbName = dbName + ".db"
|
|
}
|
|
dbPath := filepath.Join(cfg.DataDir, dbName)
|
|
return dbPath
|
|
}
|
|
|
|
// warnMultipeDBs shows a warning if multiple block database types are detected.
|
|
// This is not a situation most users want. It is handy for development however
|
|
// to support multiple side-by-side databases.
|
|
func warnMultipeDBs() {
|
|
// This is intentionally not using the known db types which depend
|
|
// on the database types compiled into the binary since we want to
|
|
// detect legacy db types as well.
|
|
dbTypes := []string{"leveldb", "sqlite"}
|
|
duplicateDbPaths := make([]string, 0, len(dbTypes)-1)
|
|
for _, dbType := range dbTypes {
|
|
if dbType == cfg.DbType {
|
|
continue
|
|
}
|
|
|
|
// Store db path as a duplicate db if it exists.
|
|
dbPath := blockDbPath(dbType)
|
|
if fileExists(dbPath) {
|
|
duplicateDbPaths = append(duplicateDbPaths, dbPath)
|
|
}
|
|
}
|
|
|
|
// Warn if there are extra databases.
|
|
if len(duplicateDbPaths) > 0 {
|
|
selectedDbPath := blockDbPath(cfg.DbType)
|
|
btcdLog.Warnf("WARNING: There are multiple block chain databases "+
|
|
"using different database types.\nYou probably don't "+
|
|
"want to waste disk space by having more than one.\n"+
|
|
"Your current database is located at [%v].\nThe "+
|
|
"additional database is located at %v", selectedDbPath,
|
|
duplicateDbPaths)
|
|
}
|
|
}
|
|
|
|
// setupBlockDB loads (or creates when needed) the block database taking into
|
|
// account the selected database backend. It also contains additional logic
|
|
// such warning the user if there are multiple databases which consume space on
|
|
// the file system and ensuring the regression test database is clean when in
|
|
// regression test mode.
|
|
func setupBlockDB() (btcdb.Db, error) {
|
|
// The memdb backend does not have a file path associated with it, so
|
|
// handle it uniquely. We also don't want to worry about the multiple
|
|
// database type warnings when running with the memory database.
|
|
if cfg.DbType == "memdb" {
|
|
btcdLog.Infof("Creating block database in memory.")
|
|
db, err := btcdb.CreateDB(cfg.DbType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return db, nil
|
|
}
|
|
|
|
warnMultipeDBs()
|
|
|
|
// The database name is based on the database type.
|
|
dbPath := blockDbPath(cfg.DbType)
|
|
|
|
// The regression test is special in that it needs a clean database for
|
|
// each run, so remove it now if it already exists.
|
|
removeRegressionDB(dbPath)
|
|
|
|
btcdLog.Infof("Loading block database from '%s'", dbPath)
|
|
db, err := btcdb.OpenDB(cfg.DbType, dbPath)
|
|
if err != nil {
|
|
// Return the error if it's not because the database
|
|
// doesn't exist.
|
|
if err != btcdb.DbDoesNotExist {
|
|
return nil, err
|
|
}
|
|
|
|
// Create the db if it does not exist.
|
|
err = os.MkdirAll(cfg.DataDir, 0700)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db, err = btcdb.CreateDB(cfg.DbType, dbPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
return db, nil
|
|
}
|
|
|
|
// loadBlockDB opens the block database and returns a handle to it.
|
|
func loadBlockDB() (btcdb.Db, error) {
|
|
db, err := setupBlockDB()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Get the latest block height from the database.
|
|
_, height, err := db.NewestSha()
|
|
if err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
|
|
// Insert the appropriate genesis block for the bitcoin network being
|
|
// connected to if needed.
|
|
if height == -1 {
|
|
genesis := btcutil.NewBlock(activeNetParams.GenesisBlock)
|
|
_, err := db.InsertBlock(genesis)
|
|
if err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
btcdLog.Infof("Inserted genesis block %v",
|
|
activeNetParams.GenesisHash)
|
|
height = 0
|
|
}
|
|
|
|
btcdLog.Infof("Block database loaded with block height %d", height)
|
|
return db, nil
|
|
}
|