package node
import (
- "context"
- "encoding/json"
+ "encoding/hex"
"errors"
"net"
"net/http"
+ // debug tool
_ "net/http/pprof"
- "os"
"path/filepath"
- "strings"
- "time"
+ "reflect"
"github.com/prometheus/prometheus/util/flock"
log "github.com/sirupsen/logrus"
cmn "github.com/tendermint/tmlibs/common"
- dbm "github.com/tendermint/tmlibs/db"
browser "github.com/toqueteos/webbrowser"
- "github.com/vapor/accesstoken"
- "github.com/vapor/account"
- "github.com/vapor/api"
- "github.com/vapor/asset"
- "github.com/vapor/blockchain/pseudohsm"
- "github.com/vapor/blockchain/txfeed"
- "github.com/vapor/common"
- cfg "github.com/vapor/config"
- "github.com/vapor/consensus"
- "github.com/vapor/crypto/ed25519/chainkd"
- "github.com/vapor/database/leveldb"
- "github.com/vapor/env"
- "github.com/vapor/mining/miner"
- "github.com/vapor/net/websocket"
- "github.com/vapor/netsync"
- "github.com/vapor/protocol"
- "github.com/vapor/protocol/bc"
- "github.com/vapor/util"
- w "github.com/vapor/wallet"
+ "github.com/bytom/vapor/accesstoken"
+ "github.com/bytom/vapor/account"
+ "github.com/bytom/vapor/api"
+ "github.com/bytom/vapor/application/mov"
+ "github.com/bytom/vapor/asset"
+ "github.com/bytom/vapor/blockchain/pseudohsm"
+ cfg "github.com/bytom/vapor/config"
+ "github.com/bytom/vapor/consensus"
+ "github.com/bytom/vapor/database"
+ dbm "github.com/bytom/vapor/database/leveldb"
+ "github.com/bytom/vapor/env"
+ "github.com/bytom/vapor/event"
+ vaporLog "github.com/bytom/vapor/log"
+ "github.com/bytom/vapor/net/websocket"
+ "github.com/bytom/vapor/netsync"
+ "github.com/bytom/vapor/proposal/blockproposer"
+ "github.com/bytom/vapor/protocol"
+ "github.com/bytom/vapor/protocol/bc/types"
+ w "github.com/bytom/vapor/wallet"
)
const (
- webHost = "http://127.0.0.1"
- maxNewBlockChSize = 1024
+ webHost = "http://127.0.0.1"
+ logModule = "node"
)
+// Node represent bytom node
type Node struct {
cmn.BaseService
- // config
- config *cfg.Config
+ config *cfg.Config
+ eventDispatcher *event.Dispatcher
+ syncManager *netsync.SyncManager
- syncManager *netsync.SyncManager
-
- //bcReactor *bc.BlockchainReactor
wallet *w.Wallet
accessTokens *accesstoken.CredentialStore
notificationMgr *websocket.WSNotificationManager
api *api.API
chain *protocol.Chain
- txfeed *txfeed.Tracker
- //cpuMiner *cpuminer.CPUMiner
- miner *miner.Miner
-
- miningEnable bool
-
- newBlockCh chan *bc.Hash
+ cpuMiner *blockproposer.BlockProposer
+ miningEnable bool
}
+// NewNode create bytom node
func NewNode(config *cfg.Config) *Node {
- ctx := context.Background()
- if err := lockDataDirectory(config); err != nil {
- cmn.Exit("Error: " + err.Error())
+ initNodeConfig(config)
+
+ if err := vaporLog.InitLogFile(config); err != nil {
+ log.WithField("err", err).Fatalln("InitLogFile failed")
}
- initLogFile(config)
- initActiveNetParams(config)
- initConsensusConfig(config)
- initCommonConfig(config)
- util.MainchainConfig = config.MainChain
- util.ValidatePegin = config.ValidatePegin
+ log.WithFields(log.Fields{
+ "module": logModule,
+ "pubkey": config.PrivateKey().XPub(),
+ "fed_xpubs": config.Federation.Xpubs,
+ "fed_quorum": config.Federation.Quorum,
+ "fed_controlprogram": hex.EncodeToString(cfg.FederationWScript(config)),
+ }).Info()
+
// Get store
if config.DBBackend != "memdb" && config.DBBackend != "leveldb" {
cmn.Exit(cmn.Fmt("Param db_backend [%v] is invalid, use leveldb or memdb", config.DBBackend))
}
coreDB := dbm.NewDB("core", config.DBBackend, config.DBDir())
- store := leveldb.NewStore(coreDB)
+ store := database.NewStore(coreDB)
tokenDB := dbm.NewDB("accesstoken", config.DBBackend, config.DBDir())
accessTokens := accesstoken.NewStore(tokenDB)
- txPool := protocol.NewTxPool(store)
- chain, err := protocol.NewChain(store, txPool)
+ dispatcher := event.NewDispatcher()
+ movCore := mov.NewCore(config.DBBackend, config.DBDir(), consensus.ActiveNetParams.MovStartHeight)
+ assetFilter := protocol.NewAssetFilter(config.CrossChain.AssetWhitelist)
+ txPool := protocol.NewTxPool(store, []protocol.DustFilterer{movCore, assetFilter}, dispatcher)
+ chain, err := protocol.NewChain(store, txPool, []protocol.SubProtocol{movCore}, dispatcher)
if err != nil {
cmn.Exit(cmn.Fmt("Failed to create chain structure: %v", err))
}
- switch config.Consensus.Type {
- case "dpos":
- initDpos(chain, config)
+ if err := checkConfig(chain, config); err != nil {
+ panic(err)
}
- var accounts *account.Manager = nil
- var assets *asset.Registry = nil
- var wallet *w.Wallet = nil
- var txFeed *txfeed.Tracker = nil
-
- txFeedDB := dbm.NewDB("txfeeds", config.DBBackend, config.DBDir())
- txFeed = txfeed.NewTracker(txFeedDB, chain)
-
- if err = txFeed.Prepare(ctx); err != nil {
- log.WithField("error", err).Error("start txfeed")
- return nil
- }
+ var accounts *account.Manager
+ var assets *asset.Registry
+ var wallet *w.Wallet
hsm, err := pseudohsm.New(config.KeysDir())
if err != nil {
}
if !config.Wallet.Disable {
- address, err := common.DecodeAddress(config.Consensus.Coinbase, &consensus.ActiveNetParams)
- if err != nil {
- cmn.Exit(cmn.Fmt("DecodeAddress: %v", err))
- }
walletDB := dbm.NewDB("wallet", config.DBBackend, config.DBDir())
- accounts = account.NewManager(walletDB, chain)
+ walletStore := database.NewWalletStore(walletDB)
+ accountStore := database.NewAccountStore(walletDB)
+ accounts = account.NewManager(accountStore, chain)
assets = asset.NewRegistry(walletDB, chain)
- wallet, err = w.NewWallet(walletDB, accounts, assets, hsm, chain, address)
+ wallet, err = w.NewWallet(walletStore, accounts, assets, hsm, chain, dispatcher, config.Wallet.TxIndex)
if err != nil {
- log.WithField("error", err).Error("init NewWallet")
+ log.WithFields(log.Fields{"module": logModule, "error": err}).Error("init NewWallet")
+ }
+
+ if err = wallet.Run(); err != nil {
+ log.WithFields(log.Fields{"module": logModule, "error": err}).Error("init NewWallet work running thread")
}
// trigger rescan wallet
wallet.RescanBlocks()
}
}
- newBlockCh := make(chan *bc.Hash, maxNewBlockChSize)
-
- syncManager, _ := netsync.NewSyncManager(config, chain, txPool, newBlockCh)
-
- notificationMgr := websocket.NewWsNotificationManager(config.Websocket.MaxNumWebsockets, config.Websocket.MaxNumConcurrentReqs, chain)
+ fastSyncDB := dbm.NewDB("fastsync", config.DBBackend, config.DBDir())
+ syncManager, err := netsync.NewSyncManager(config, chain, txPool, dispatcher, fastSyncDB)
+ if err != nil {
+ cmn.Exit(cmn.Fmt("Failed to create sync manager: %v", err))
+ }
- // get transaction from txPool and send it to syncManager and wallet
- go newPoolTxListener(txPool, syncManager, wallet, notificationMgr)
+ notificationMgr := websocket.NewWsNotificationManager(config.Websocket.MaxNumWebsockets, config.Websocket.MaxNumConcurrentReqs, chain, dispatcher)
// run the profile server
profileHost := config.ProfListenAddress
if profileHost != "" {
- // Profiling bytomd programs.see (https://blog.golang.org/profiling-go-programs)
+ // Profiling vapord programs.see (https://blog.golang.org/profiling-go-programs)
// go tool pprof http://profileHose/debug/pprof/heap
go func() {
if err = http.ListenAndServe(profileHost, nil); err != nil {
}
}()
}
+
node := &Node{
- config: config,
- syncManager: syncManager,
- accessTokens: accessTokens,
- wallet: wallet,
- chain: chain,
- txfeed: txFeed,
- miningEnable: config.Mining,
-
- newBlockCh: newBlockCh,
+ eventDispatcher: dispatcher,
+ config: config,
+ syncManager: syncManager,
+ accessTokens: accessTokens,
+ wallet: wallet,
+ chain: chain,
+ miningEnable: config.Mining,
+
notificationMgr: notificationMgr,
}
- node.miner = miner.NewMiner(chain, accounts, txPool, newBlockCh)
+ node.cpuMiner = blockproposer.NewBlockProposer(chain, accounts, txPool, dispatcher)
node.BaseService = *cmn.NewBaseService(nil, "Node", node)
-
return node
}
-// newPoolTxListener listener transaction from txPool, and send it to syncManager and wallet
-func newPoolTxListener(txPool *protocol.TxPool, syncManager *netsync.SyncManager, wallet *w.Wallet, notificationMgr *websocket.WSNotificationManager) {
- txMsgCh := txPool.GetMsgCh()
- syncManagerTxCh := syncManager.GetNewTxCh()
-
- for {
- msg := <-txMsgCh
- switch msg.MsgType {
- case protocol.MsgNewTx:
- syncManagerTxCh <- msg.Tx
- if wallet != nil {
- wallet.AddUnconfirmedTx(msg.TxDesc)
- }
- notificationMgr.NotifyMempoolTx(msg.Tx)
- case protocol.MsgRemoveTx:
- if wallet != nil {
- wallet.RemoveUnconfirmedTx(msg.TxDesc)
- }
- default:
- log.Warn("got unknow message type from the txPool channel")
- }
+// Rollback rollback chain from one height to targetHeight
+func Rollback(config *cfg.Config, targetHeight uint64) error {
+ if err := initNodeConfig(config); err != nil {
+ return err
}
-}
-// Lock data directory after daemonization
-func lockDataDirectory(config *cfg.Config) error {
- _, _, err := flock.New(filepath.Join(config.RootDir, "LOCK"))
+ // Get store
+ if config.DBBackend != "leveldb" {
+ return errors.New("Param db_backend is invalid, use leveldb")
+ }
+
+ coreDB := dbm.NewDB("core", config.DBBackend, config.DBDir())
+ store := database.NewStore(coreDB)
+
+ dispatcher := event.NewDispatcher()
+ movCore := mov.NewCore(config.DBBackend, config.DBDir(), consensus.ActiveNetParams.MovStartHeight)
+ txPool := protocol.NewTxPool(store, []protocol.DustFilterer{movCore}, dispatcher)
+ chain, err := protocol.NewChain(store, txPool, []protocol.SubProtocol{movCore}, dispatcher)
if err != nil {
- return errors.New("datadir already used by another process")
+ return err
}
- return nil
+
+ hsm, err := pseudohsm.New(config.KeysDir())
+ if err != nil {
+ return err
+ }
+
+ walletDB := dbm.NewDB("wallet", config.DBBackend, config.DBDir())
+ walletStore := database.NewWalletStore(walletDB)
+ accountStore := database.NewAccountStore(walletDB)
+ accounts := account.NewManager(accountStore, chain)
+ assets := asset.NewRegistry(walletDB, chain)
+ wallet, err := w.NewWallet(walletStore, accounts, assets, hsm, chain, dispatcher, config.Wallet.TxIndex)
+ if err != nil {
+ return err
+ }
+
+ if err := wallet.Rollback(targetHeight); err != nil {
+ return err
+ }
+
+ return chain.Rollback(targetHeight)
}
-func initActiveNetParams(config *cfg.Config) {
- var exist bool
- consensus.ActiveNetParams, exist = consensus.NetParams[config.ChainID]
- if !exist {
- cmn.Exit(cmn.Fmt("chain_id[%v] don't exist", config.ChainID))
+func initNodeConfig(config *cfg.Config) error {
+ if err := lockDataDirectory(config); err != nil {
+ log.WithField("err", err).Info("Error: " + err.Error())
+ return err
}
- if config.Side.FedpegXPubs != "" {
- var federationRedeemXPubs []chainkd.XPub
- fedpegXPubs := strings.Split(config.Side.FedpegXPubs, ",")
- for _, xpubStr := range fedpegXPubs {
- var xpub chainkd.XPub
- xpub.UnmarshalText([]byte(xpubStr))
- federationRedeemXPubs = append(federationRedeemXPubs, xpub)
- }
- consensus.ActiveNetParams.FedpegXPubs = federationRedeemXPubs
+
+ if err := cfg.LoadFederationFile(config.FederationFile(), config); err != nil {
+ log.WithField("err", err).Info("Failed to load federated information")
+ return err
+ }
+
+ if err := consensus.InitActiveNetParams(config.ChainID); err != nil {
+ log.Fatalf("Failed to init ActiveNetParams:[%s]", err.Error())
}
- consensus.ActiveNetParams.Signer = config.Signer
- consensus.ActiveNetParams.PeginMinDepth = config.Side.PeginMinDepth
- consensus.ActiveNetParams.ParentGenesisBlockHash = config.Side.ParentGenesisBlockHash
+ cfg.CommonConfig = config
+ return nil
}
-func initLogFile(config *cfg.Config) {
- if config.LogFile == "" {
- return
+// find whether config xpubs equal genesis block xpubs
+func checkConfig(chain *protocol.Chain, config *cfg.Config) error {
+ fedpegScript := cfg.FederationWScript(config)
+ genesisBlock, err := chain.GetBlockByHeight(0)
+ if err != nil {
+ return err
}
- cmn.EnsureDir(filepath.Dir(config.LogFile), 0700)
- file, err := os.OpenFile(config.LogFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
- if err == nil {
- log.SetOutput(file)
- } else {
- log.WithField("err", err).Info("using default")
+ typedInput := genesisBlock.Transactions[0].Inputs[0].TypedInput
+ if v, ok := typedInput.(*types.CoinbaseInput); ok {
+ if !reflect.DeepEqual(fedpegScript, v.Arbitrary) {
+ return errors.New("config xpubs don't equal genesis block xpubs")
+ }
}
+ return nil
+}
+// Lock data directory after daemonization
+func lockDataDirectory(config *cfg.Config) error {
+ _, _, err := flock.New(filepath.Join(config.RootDir, "LOCK"))
+ if err != nil {
+ return errors.New("datadir already used by another process")
+ }
+ return nil
}
func initCommonConfig(config *cfg.Config) {
}
}
-func (n *Node) initAndstartApiServer() {
- n.api = api.NewAPI(n.syncManager, n.wallet, n.txfeed, n.miner, n.chain, n.config, n.accessTokens, n.newBlockCh, n.notificationMgr)
+func (n *Node) initAndstartAPIServer() {
+ n.api = api.NewAPI(n.syncManager, n.wallet, n.cpuMiner, n.chain, n.config, n.accessTokens, n.eventDispatcher, n.notificationMgr)
listenAddr := env.String("LISTEN", n.config.ApiAddress)
env.Parse()
n.api.StartServer(*listenAddr)
}
+// OnStart implements BaseService
func (n *Node) OnStart() error {
if n.miningEnable {
if _, err := n.wallet.AccountMgr.GetMiningAddress(); err != nil {
n.miningEnable = false
log.Error(err)
} else {
- //n.cpuMiner.Start()
- n.miner.Start()
+ n.cpuMiner.Start()
}
}
if !n.config.VaultMode {
- n.syncManager.Start()
+ if err := n.syncManager.Start(); err != nil {
+ return err
+ }
+ }
+
+ n.initAndstartAPIServer()
+ if err := n.notificationMgr.Start(); err != nil {
+ return err
}
- n.initAndstartApiServer()
- n.notificationMgr.Start()
+
if !n.config.Web.Closed {
_, port, err := net.SplitHostPort(n.config.ApiAddress)
if err != nil {
}
launchWebBrowser(port)
}
- go bytomdRPCCheck()
return nil
}
+// OnStop implements BaseService
func (n *Node) OnStop() {
- if err := n.chain.Engine.Finish(); err != nil {
- log.Errorf("OnStop: %v", err)
- }
-
n.notificationMgr.Shutdown()
n.notificationMgr.WaitForShutdown()
n.BaseService.OnStop()
if n.miningEnable {
- n.miner.Stop()
+ n.cpuMiner.Stop()
}
if !n.config.VaultMode {
n.syncManager.Stop()
}
+ n.eventDispatcher.Stop()
}
+// RunForever listen to the stop signal
func (n *Node) RunForever() {
// Sleep forever and then...
cmn.TrapSignal(func() {
n.Stop()
})
}
-
-func (n *Node) SyncManager() *netsync.SyncManager {
- return n.syncManager
-}
-
-/**bytomdRPCCheck Check if bytomd connection via RPC is correctly working*/
-func bytomdRPCCheck() bool {
- type Req struct {
- BlockHeight uint64 `json:"block_height"`
- }
- if util.ValidatePegin {
- for {
- resp, err := util.CallRPC("/get-merkle-proof", &Req{BlockHeight: 0})
- if err != nil {
- log.Error("Call mainchain interface get-block-header failed")
- time.Sleep(time.Millisecond * 1000)
- continue
- }
- tmp, _ := json.Marshal(resp)
- var blockHeader api.GetBlockHeaderResp
- json.Unmarshal(tmp, &blockHeader)
- hash := blockHeader.BlockHeader.Hash()
- if strings.Compare(consensus.ActiveNetParams.ParentGenesisBlockHash, hash.String()) != 0 {
- log.Error("Invalid parent genesis block hash response via RPC. Contacting wrong parent daemon?", consensus.ActiveNetParams.ParentGenesisBlockHash, ":", hash.String())
- return false
- }
- break
- }
- }
-
- return true
-}
-
-func initConsensusConfig(config *cfg.Config) {
- if config.ConsensusConfigFile == "" {
- // poa
- } else {
- //
- file, err := os.Open(config.ConsensusConfigFile)
- if err != nil {
- cmn.Exit(cmn.Fmt("Failed to read consensus file: %v", err))
- }
- defer file.Close()
-
- if err := json.NewDecoder(file).Decode(config); err != nil {
- cmn.Exit(cmn.Fmt("invalid consensus file: %v", err))
- }
-
- for _, v := range config.Consensus.SelfVoteSigners {
- address, err := common.DecodeAddress(v, &consensus.ActiveNetParams)
- if err != nil {
- cmn.Exit(cmn.Fmt("Address resolution failed: %v", err))
- }
- config.Consensus.Signers = append(config.Consensus.Signers, address)
- }
- }
-}
-
-func initDpos(chain *protocol.Chain, config *cfg.Config) {
- header := chain.BestBlockHeader()
- height := header.Height
- hash := header.Hash()
- maxSignerCount := config.Consensus.MaxSignerCount
- period := config.Consensus.Period
- err := chain.Engine.Init(chain, maxSignerCount, period, height, hash)
-
- if height > 0 {
- oldBlockHeight := chain.Engine.GetOldBlockHeight()
- oldBlockHash := chain.Engine.GetOldBlockHash()
- if err != nil {
- oldBlockHeight = 0
- header, _ = chain.GetHeaderByHeight(oldBlockHeight)
- oldBlockHash = header.Hash()
- }
-
- if err := chain.RepairDPoSData(oldBlockHeight, oldBlockHash); err != nil {
- cmn.Exit(cmn.Fmt("initVote failed: %v", err))
- }
- }
-}