node/node.go

   1 package node
   2
   3 import (
   4         "context"
   5         "errors"
   6         "net"
   7         "net/http"
   8         "os"
   9         "path/filepath"
  10
  11         "github.com/prometheus/prometheus/util/flock"
  12         log "github.com/sirupsen/logrus"
  13         cmn "github.com/tendermint/tmlibs/common"
  14         dbm "github.com/tendermint/tmlibs/db"
  15         browser "github.com/toqueteos/webbrowser"
  16
  17         "github.com/bytom/accesstoken"
  18         "github.com/bytom/account"
  19         "github.com/bytom/api"
  20         "github.com/bytom/asset"
  21         "github.com/bytom/blockchain/pseudohsm"
  22         "github.com/bytom/blockchain/txfeed"
  23         cfg "github.com/bytom/config"
  24         "github.com/bytom/consensus"
  25         "github.com/bytom/database/leveldb"
  26         "github.com/bytom/env"
  27         "github.com/bytom/event"
  28         "github.com/bytom/mining/cpuminer"
  29         "github.com/bytom/mining/miningpool"
  30         "github.com/bytom/mining/tensority"
  31         "github.com/bytom/net/websocket"
  32         "github.com/bytom/netsync"
  33         "github.com/bytom/p2p"
  34         "github.com/bytom/protocol"
  35         w "github.com/bytom/wallet"
  36 )
  37
  38 const webHost = "http://127.0.0.1"
  39
  40 // Node represent bytom node
  41 type Node struct {
  42         cmn.BaseService
  43
  44         config          *cfg.Config
  45         eventDispatcher *event.Dispatcher
  46         syncManager     *netsync.SyncManager
  47
  48         wallet          *w.Wallet
  49         accessTokens    *accesstoken.CredentialStore
  50         notificationMgr *websocket.WSNotificationManager
  51         api             *api.API
  52         chain           *protocol.Chain
  53         txfeed          *txfeed.Tracker
  54         cpuMiner        *cpuminer.CPUMiner
  55         miningPool      *miningpool.MiningPool
  56         miningEnable    bool
  57 }
  58
  59 // NewNode create bytom node
  60 func NewNode(config *cfg.Config) *Node {
  61         ctx := context.Background()
  62         if err := lockDataDirectory(config); err != nil {
  63                 cmn.Exit("Error: " + err.Error())
  64         }
  65         initLogFile(config)
  66         initActiveNetParams(config)
  67         initCommonConfig(config)
  68
  69         // Get store
  70         if config.DBBackend != "memdb" && config.DBBackend != "leveldb" {
  71                 cmn.Exit(cmn.Fmt("Param db_backend [%v] is invalid, use leveldb or memdb", config.DBBackend))
  72         }
  73         coreDB := dbm.NewDB("core", config.DBBackend, config.DBDir())
  74         store := leveldb.NewStore(coreDB)
  75
  76         tokenDB := dbm.NewDB("accesstoken", config.DBBackend, config.DBDir())
  77         accessTokens := accesstoken.NewStore(tokenDB)
  78
  79         txPool := protocol.NewTxPool(store)
  80         chain, err := protocol.NewChain(store, txPool)
  81         if err != nil {
  82                 cmn.Exit(cmn.Fmt("Failed to create chain structure: %v", err))
  83         }
  84
  85         var accounts *account.Manager
  86         var assets *asset.Registry
  87         var wallet *w.Wallet
  88         var txFeed *txfeed.Tracker
  89
  90         txFeedDB := dbm.NewDB("txfeeds", config.DBBackend, config.DBDir())
  91         txFeed = txfeed.NewTracker(txFeedDB, chain)
  92
  93         if err = txFeed.Prepare(ctx); err != nil {
  94                 log.WithField("error", err).Error("start txfeed")
  95                 return nil
  96         }
  97
  98         hsm, err := pseudohsm.New(config.KeysDir())
  99         if err != nil {
 100                 cmn.Exit(cmn.Fmt("initialize HSM failed: %v", err))
 101         }
 102
 103         if !config.Wallet.Disable {
 104                 walletDB := dbm.NewDB("wallet", config.DBBackend, config.DBDir())
 105                 accounts = account.NewManager(walletDB, chain)
 106                 assets = asset.NewRegistry(walletDB, chain)
 107                 wallet, err = w.NewWallet(walletDB, accounts, assets, hsm, chain)
 108                 if err != nil {
 109                         log.WithField("error", err).Error("init NewWallet")
 110                 }
 111
 112                 // trigger rescan wallet
 113                 if config.Wallet.Rescan {
 114                         wallet.RescanBlocks()
 115                 }
 116         }
 117
 118         dispatcher := event.NewDispatcher()
 119         syncManager, err := netsync.NewSyncManager(config, chain, txPool, dispatcher)
 120         if err != nil {
 121                 cmn.Exit(cmn.Fmt("Failed to create sync manager: %v", err))
 122         }
 123
 124         notificationMgr := websocket.NewWsNotificationManager(config.Websocket.MaxNumWebsockets, config.Websocket.MaxNumConcurrentReqs, chain)
 125
 126         // get transaction from txPool and send it to syncManager and wallet
 127         go newPoolTxListener(txPool, syncManager, wallet, notificationMgr)
 128
 129         // run the profile server
 130         profileHost := config.ProfListenAddress
 131         if profileHost != "" {
 132                 // Profiling bytomd programs.see (https://blog.golang.org/profiling-go-programs)
 133                 // go tool pprof http://profileHose/debug/pprof/heap
 134                 go func() {
 135                         if err = http.ListenAndServe(profileHost, nil); err != nil {
 136                                 cmn.Exit(cmn.Fmt("Failed to register tcp profileHost: %v", err))
 137                         }
 138                 }()
 139         }
 140
 141         node := &Node{
 142                 eventDispatcher: dispatcher,
 143                 config:          config,
 144                 syncManager:     syncManager,
 145                 accessTokens:    accessTokens,
 146                 wallet:          wallet,
 147                 chain:           chain,
 148                 txfeed:          txFeed,
 149                 miningEnable:    config.Mining.Enable,
 150
 151                 notificationMgr: notificationMgr,
 152         }
 153
 154         node.cpuMiner = cpuminer.NewCPUMiner(chain, accounts, txPool, dispatcher)
 155         node.miningPool = miningpool.NewMiningPool(chain, accounts, txPool, dispatcher, config.Mining.RecommitInterval)
 156
 157         node.BaseService = *cmn.NewBaseService(nil, "Node", node)
 158
 159         if config.Simd.Enable {
 160                 tensority.UseSIMD = true
 161         }
 162
 163         return node
 164 }
 165
 166 // newPoolTxListener listener transaction from txPool, and send it to syncManager and wallet
 167 func newPoolTxListener(txPool *protocol.TxPool, syncManager *netsync.SyncManager, wallet *w.Wallet, notificationMgr *websocket.WSNotificationManager) {
 168         txMsgCh := txPool.GetMsgCh()
 169         syncManagerTxCh := syncManager.GetNewTxCh()
 170
 171         for {
 172                 msg := <-txMsgCh
 173                 switch msg.MsgType {
 174                 case protocol.MsgNewTx:
 175                         syncManagerTxCh <- msg.Tx
 176                         if wallet != nil {
 177                                 wallet.AddUnconfirmedTx(msg.TxDesc)
 178                         }
 179                         notificationMgr.NotifyMempoolTx(msg.TxDesc)
 180                 case protocol.MsgRemoveTx:
 181                         if wallet != nil {
 182                                 wallet.RemoveUnconfirmedTx(msg.TxDesc)
 183                         }
 184                 default:
 185                         log.Warn("got unknow message type from the txPool channel")
 186                 }
 187         }
 188 }
 189
 190 // Lock data directory after daemonization
 191 func lockDataDirectory(config *cfg.Config) error {
 192         _, _, err := flock.New(filepath.Join(config.RootDir, "LOCK"))
 193         if err != nil {
 194                 return errors.New("datadir already used by another process")
 195         }
 196         return nil
 197 }
 198
 199 func initActiveNetParams(config *cfg.Config) {
 200         var exist bool
 201         consensus.ActiveNetParams, exist = consensus.NetParams[config.ChainID]
 202         if !exist {
 203                 cmn.Exit(cmn.Fmt("chain_id[%v] don't exist", config.ChainID))
 204         }
 205 }
 206
 207 func initLogFile(config *cfg.Config) {
 208         if config.LogFile == "" {
 209                 return
 210         }
 211         cmn.EnsureDir(filepath.Dir(config.LogFile), 0700)
 212         file, err := os.OpenFile(config.LogFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
 213         if err == nil {
 214                 log.SetOutput(file)
 215         } else {
 216                 log.WithField("err", err).Info("using default")
 217         }
 218
 219 }
 220
 221 func initCommonConfig(config *cfg.Config) {
 222         cfg.CommonConfig = config
 223 }
 224
 225 // Lanch web broser or not
 226 func launchWebBrowser(port string) {
 227         webAddress := webHost + ":" + port
 228         log.Info("Launching System Browser with :", webAddress)
 229         if err := browser.Open(webAddress); err != nil {
 230                 log.Error(err.Error())
 231                 return
 232         }
 233 }
 234
 235 func (n *Node) initAndstartAPIServer() {
 236         n.api = api.NewAPI(n.syncManager, n.wallet, n.txfeed, n.cpuMiner, n.miningPool, n.chain, n.config, n.accessTokens, n.eventDispatcher, n.notificationMgr)
 237
 238         listenAddr := env.String("LISTEN", n.config.ApiAddress)
 239         env.Parse()
 240         n.api.StartServer(*listenAddr)
 241 }
 242
 243 func (n *Node) OnStart() error {
 244         if n.miningEnable {
 245                 if _, err := n.wallet.AccountMgr.GetMiningAddress(); err != nil {
 246                         n.miningEnable = false
 247                         log.Error(err)
 248                 } else {
 249                         n.cpuMiner.Start()
 250                 }
 251         }
 252         if !n.config.VaultMode {
 253                 if err := n.syncManager.Start(); err != nil {
 254                         return err
 255                 }
 256         }
 257
 258         n.initAndstartAPIServer()
 259         n.notificationMgr.Start()
 260         if !n.config.Web.Closed {
 261                 _, port, err := net.SplitHostPort(n.config.ApiAddress)
 262                 if err != nil {
 263                         log.Error("Invalid api address")
 264                         return err
 265                 }
 266                 launchWebBrowser(port)
 267         }
 268         return nil
 269 }
 270
 271 func (n *Node) OnStop() {
 272         n.notificationMgr.Shutdown()
 273         n.notificationMgr.WaitForShutdown()
 274         n.BaseService.OnStop()
 275         if n.miningEnable {
 276                 n.cpuMiner.Stop()
 277         }
 278         if !n.config.VaultMode {
 279                 n.syncManager.Stop()
 280         }
 281         n.eventDispatcher.Stop()
 282 }
 283
 284 func (n *Node) RunForever() {
 285         // Sleep forever and then...
 286         cmn.TrapSignal(func() {
 287                 n.Stop()
 288         })
 289 }
 290
 291 func (n *Node) NodeInfo() *p2p.NodeInfo {
 292         return n.syncManager.NodeInfo()
 293 }
 294
 295 func (n *Node) MiningPool() *miningpool.MiningPool {
 296         return n.miningPool
 297 }