node/node.go

   1 package node
   2
   3 import (
   4         "encoding/hex"
   5         "errors"
   6         "net"
   7         "net/http"
   8         _ "net/http/pprof"
   9         "path/filepath"
  10         "reflect"
  11
  12         "github.com/prometheus/prometheus/util/flock"
  13         log "github.com/sirupsen/logrus"
  14         cmn "github.com/tendermint/tmlibs/common"
  15         browser "github.com/toqueteos/webbrowser"
  16
  17         "github.com/vapor/accesstoken"
  18         "github.com/vapor/account"
  19         "github.com/vapor/api"
  20         "github.com/vapor/asset"
  21         "github.com/vapor/blockchain/pseudohsm"
  22         cfg "github.com/vapor/config"
  23         "github.com/vapor/consensus"
  24         "github.com/vapor/database"
  25         dbm "github.com/vapor/database/leveldb"
  26         "github.com/vapor/env"
  27         "github.com/vapor/event"
  28         vaporLog "github.com/vapor/log"
  29         "github.com/vapor/net/websocket"
  30         "github.com/vapor/netsync"
  31         "github.com/vapor/proposal/blockproposer"
  32         "github.com/vapor/protocol"
  33         "github.com/vapor/protocol/bc/types"
  34         w "github.com/vapor/wallet"
  35 )
  36
  37 const (
  38         webHost   = "http://127.0.0.1"
  39         logModule = "node"
  40 )
  41
  42 // Node represent bytom node
  43 type Node struct {
  44         cmn.BaseService
  45
  46         config          *cfg.Config
  47         eventDispatcher *event.Dispatcher
  48         syncManager     *netsync.SyncManager
  49
  50         wallet          *w.Wallet
  51         accessTokens    *accesstoken.CredentialStore
  52         notificationMgr *websocket.WSNotificationManager
  53         api             *api.API
  54         chain           *protocol.Chain
  55         cpuMiner        *blockproposer.BlockProposer
  56         miningEnable    bool
  57 }
  58
  59 // NewNode create bytom node
  60 func NewNode(config *cfg.Config) *Node {
  61         if err := lockDataDirectory(config); err != nil {
  62                 cmn.Exit("Error: " + err.Error())
  63         }
  64
  65         if err := cfg.LoadFederationFile(config.FederationFile(), config); err != nil {
  66                 cmn.Exit(cmn.Fmt("Failed to load federated information:[%s]", err.Error()))
  67         }
  68
  69         vaporLog.InitLogFile(config)
  70
  71         log.WithFields(log.Fields{
  72                 "module":             logModule,
  73                 "pubkey":             config.PrivateKey().XPub(),
  74                 "fed_xpubs":          config.Federation.Xpubs,
  75                 "fed_quorum":         config.Federation.Quorum,
  76                 "fed_controlprogram": hex.EncodeToString(cfg.FederationWScript(config)),
  77         }).Info()
  78
  79         if err := consensus.InitActiveNetParams(config.ChainID); err != nil {
  80                 log.Fatalf("Failed to init ActiveNetParams:[%s]", err.Error())
  81         }
  82
  83         initCommonConfig(config)
  84
  85         // Get store
  86         if config.DBBackend != "memdb" && config.DBBackend != "leveldb" {
  87                 cmn.Exit(cmn.Fmt("Param db_backend [%v] is invalid, use leveldb or memdb", config.DBBackend))
  88         }
  89         coreDB := dbm.NewDB("core", config.DBBackend, config.DBDir())
  90         store := database.NewStore(coreDB)
  91
  92         tokenDB := dbm.NewDB("accesstoken", config.DBBackend, config.DBDir())
  93         accessTokens := accesstoken.NewStore(tokenDB)
  94
  95         dispatcher := event.NewDispatcher()
  96         txPool := protocol.NewTxPool(store, dispatcher)
  97         chain, err := protocol.NewChain(store, txPool, dispatcher)
  98         if err != nil {
  99                 cmn.Exit(cmn.Fmt("Failed to create chain structure: %v", err))
 100         }
 101
 102         if err := checkConfig(chain, config); err != nil {
 103                 panic(err)
 104         }
 105
 106         var accounts *account.Manager
 107         var assets *asset.Registry
 108         var wallet *w.Wallet
 109
 110         hsm, err := pseudohsm.New(config.KeysDir())
 111         if err != nil {
 112                 cmn.Exit(cmn.Fmt("initialize HSM failed: %v", err))
 113         }
 114
 115         if !config.Wallet.Disable {
 116                 walletDB := dbm.NewDB("wallet", config.DBBackend, config.DBDir())
 117                 walletStore := database.NewWalletStore(walletDB)
 118                 accountStore := database.NewAccountStore(walletDB)
 119                 accounts = account.NewManager(accountStore, chain)
 120                 assets = asset.NewRegistry(walletDB, chain)
 121                 wallet, err = w.NewWallet(walletStore, accounts, assets, hsm, chain, dispatcher, config.Wallet.TxIndex)
 122                 if err != nil {
 123                         log.WithFields(log.Fields{"module": logModule, "error": err}).Error("init NewWallet")
 124                 }
 125
 126                 // trigger rescan wallet
 127                 if config.Wallet.Rescan {
 128                         wallet.RescanBlocks()
 129                 }
 130         }
 131         fastSyncDB := dbm.NewDB("fastsync", config.DBBackend, config.DBDir())
 132         syncManager, err := netsync.NewSyncManager(config, chain, txPool, dispatcher, fastSyncDB)
 133         if err != nil {
 134                 cmn.Exit(cmn.Fmt("Failed to create sync manager: %v", err))
 135         }
 136
 137         notificationMgr := websocket.NewWsNotificationManager(config.Websocket.MaxNumWebsockets, config.Websocket.MaxNumConcurrentReqs, chain, dispatcher)
 138
 139         // run the profile server
 140         profileHost := config.ProfListenAddress
 141         if profileHost != "" {
 142                 // Profiling vapord programs.see (https://blog.golang.org/profiling-go-programs)
 143                 // go tool pprof http://profileHose/debug/pprof/heap
 144                 go func() {
 145                         if err = http.ListenAndServe(profileHost, nil); err != nil {
 146                                 cmn.Exit(cmn.Fmt("Failed to register tcp profileHost: %v", err))
 147                         }
 148                 }()
 149         }
 150
 151         node := &Node{
 152                 eventDispatcher: dispatcher,
 153                 config:          config,
 154                 syncManager:     syncManager,
 155                 accessTokens:    accessTokens,
 156                 wallet:          wallet,
 157                 chain:           chain,
 158                 miningEnable:    config.Mining,
 159
 160                 notificationMgr: notificationMgr,
 161         }
 162
 163         node.cpuMiner = blockproposer.NewBlockProposer(chain, accounts, txPool, dispatcher)
 164         node.BaseService = *cmn.NewBaseService(nil, "Node", node)
 165         return node
 166 }
 167
 168 // find whether config xpubs equal genesis block xpubs
 169 func checkConfig(chain *protocol.Chain, config *cfg.Config) error {
 170         fedpegScript := cfg.FederationWScript(config)
 171         genesisBlock, err := chain.GetBlockByHeight(0)
 172         if err != nil {
 173                 return err
 174         }
 175         typedInput := genesisBlock.Transactions[0].Inputs[0].TypedInput
 176         if v, ok := typedInput.(*types.CoinbaseInput); ok {
 177                 if !reflect.DeepEqual(fedpegScript, v.Arbitrary) {
 178                         return errors.New("config xpubs don't equal genesis block xpubs.")
 179                 }
 180         }
 181         return nil
 182 }
 183
 184 // Lock data directory after daemonization
 185 func lockDataDirectory(config *cfg.Config) error {
 186         _, _, err := flock.New(filepath.Join(config.RootDir, "LOCK"))
 187         if err != nil {
 188                 return errors.New("datadir already used by another process")
 189         }
 190         return nil
 191 }
 192
 193 func initCommonConfig(config *cfg.Config) {
 194         cfg.CommonConfig = config
 195 }
 196
 197 // Lanch web broser or not
 198 func launchWebBrowser(port string) {
 199         webAddress := webHost + ":" + port
 200         log.Info("Launching System Browser with :", webAddress)
 201         if err := browser.Open(webAddress); err != nil {
 202                 log.Error(err.Error())
 203                 return
 204         }
 205 }
 206
 207 func (n *Node) initAndstartAPIServer() {
 208         n.api = api.NewAPI(n.syncManager, n.wallet, n.cpuMiner, n.chain, n.config, n.accessTokens, n.eventDispatcher, n.notificationMgr)
 209
 210         listenAddr := env.String("LISTEN", n.config.ApiAddress)
 211         env.Parse()
 212         n.api.StartServer(*listenAddr)
 213 }
 214
 215 func (n *Node) OnStart() error {
 216         if n.miningEnable {
 217                 if _, err := n.wallet.AccountMgr.GetMiningAddress(); err != nil {
 218                         n.miningEnable = false
 219                         log.Error(err)
 220                 } else {
 221                         n.cpuMiner.Start()
 222                 }
 223         }
 224         if !n.config.VaultMode {
 225                 if err := n.syncManager.Start(); err != nil {
 226                         return err
 227                 }
 228         }
 229
 230         n.initAndstartAPIServer()
 231         if err := n.notificationMgr.Start(); err != nil {
 232                 return err
 233         }
 234
 235         if !n.config.Web.Closed {
 236                 _, port, err := net.SplitHostPort(n.config.ApiAddress)
 237                 if err != nil {
 238                         log.Error("Invalid api address")
 239                         return err
 240                 }
 241                 launchWebBrowser(port)
 242         }
 243         return nil
 244 }
 245
 246 func (n *Node) OnStop() {
 247         n.notificationMgr.Shutdown()
 248         n.notificationMgr.WaitForShutdown()
 249         n.BaseService.OnStop()
 250         if n.miningEnable {
 251                 n.cpuMiner.Stop()
 252         }
 253         if !n.config.VaultMode {
 254                 n.syncManager.Stop()
 255         }
 256         n.eventDispatcher.Stop()
 257 }
 258
 259 func (n *Node) RunForever() {
 260         // Sleep forever and then...
 261         cmn.TrapSignal(func() {
 262                 n.Stop()
 263         })
 264 }