You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

376 lines
11 KiB

10 years ago
10 years ago
10 years ago
8 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
9 years ago
  1. package node
  2. import (
  3. "bytes"
  4. "errors"
  5. "net"
  6. "net/http"
  7. "strings"
  8. abci "github.com/tendermint/abci/types"
  9. cmn "github.com/tendermint/go-common"
  10. cfg "github.com/tendermint/go-config"
  11. "github.com/tendermint/go-crypto"
  12. dbm "github.com/tendermint/go-db"
  13. "github.com/tendermint/go-p2p"
  14. "github.com/tendermint/go-rpc"
  15. "github.com/tendermint/go-rpc/server"
  16. "github.com/tendermint/go-wire"
  17. bc "github.com/tendermint/tendermint/blockchain"
  18. "github.com/tendermint/tendermint/consensus"
  19. mempl "github.com/tendermint/tendermint/mempool"
  20. "github.com/tendermint/tendermint/proxy"
  21. rpccore "github.com/tendermint/tendermint/rpc/core"
  22. grpccore "github.com/tendermint/tendermint/rpc/grpc"
  23. sm "github.com/tendermint/tendermint/state"
  24. "github.com/tendermint/tendermint/types"
  25. "github.com/tendermint/tendermint/version"
  26. _ "net/http/pprof"
  27. )
  28. type Node struct {
  29. cmn.BaseService
  30. config cfg.Config
  31. sw *p2p.Switch
  32. evsw types.EventSwitch
  33. blockStore *bc.BlockStore
  34. bcReactor *bc.BlockchainReactor
  35. mempoolReactor *mempl.MempoolReactor
  36. consensusState *consensus.ConsensusState
  37. consensusReactor *consensus.ConsensusReactor
  38. privValidator *types.PrivValidator
  39. genesisDoc *types.GenesisDoc
  40. privKey crypto.PrivKeyEd25519
  41. proxyApp proxy.AppConns
  42. }
  43. func NewNodeDefault(config cfg.Config) *Node {
  44. // Get PrivValidator
  45. privValidatorFile := config.GetString("priv_validator_file")
  46. privValidator := types.LoadOrGenPrivValidator(privValidatorFile)
  47. return NewNode(config, privValidator, proxy.DefaultClientCreator(config))
  48. }
  49. func NewNode(config cfg.Config, privValidator *types.PrivValidator, clientCreator proxy.ClientCreator) *Node {
  50. // Get BlockStore
  51. blockStoreDB := dbm.NewDB("blockstore", config.GetString("db_backend"), config.GetString("db_dir"))
  52. blockStore := bc.NewBlockStore(blockStoreDB)
  53. // Get State
  54. stateDB := dbm.NewDB("state", config.GetString("db_backend"), config.GetString("db_dir"))
  55. state := sm.GetState(config, stateDB)
  56. // Create the proxyApp, which manages connections (consensus, mempool, query)
  57. proxyApp := proxy.NewAppConns(config, clientCreator, sm.NewHandshaker(config, state, blockStore))
  58. if _, err := proxyApp.Start(); err != nil {
  59. cmn.Exit(cmn.Fmt("Error starting proxy app connections: %v", err))
  60. }
  61. // add the chainid and number of validators to the global config
  62. config.Set("chain_id", state.ChainID)
  63. config.Set("num_vals", state.Validators.Size())
  64. // Generate node PrivKey
  65. privKey := crypto.GenPrivKeyEd25519()
  66. // Make event switch
  67. eventSwitch := types.NewEventSwitch()
  68. _, err := eventSwitch.Start()
  69. if err != nil {
  70. cmn.Exit(cmn.Fmt("Failed to start switch: %v", err))
  71. }
  72. // Decide whether to fast-sync or not
  73. // We don't fast-sync when the only validator is us.
  74. fastSync := config.GetBool("fast_sync")
  75. if state.Validators.Size() == 1 {
  76. addr, _ := state.Validators.GetByIndex(0)
  77. if bytes.Equal(privValidator.Address, addr) {
  78. fastSync = false
  79. }
  80. }
  81. // Make BlockchainReactor
  82. bcReactor := bc.NewBlockchainReactor(config, state.Copy(), proxyApp.Consensus(), blockStore, fastSync)
  83. // Make MempoolReactor
  84. mempool := mempl.NewMempool(config, proxyApp.Mempool())
  85. mempoolReactor := mempl.NewMempoolReactor(config, mempool)
  86. // Make ConsensusReactor
  87. consensusState := consensus.NewConsensusState(config, state.Copy(), proxyApp.Consensus(), blockStore, mempool)
  88. if privValidator != nil {
  89. consensusState.SetPrivValidator(privValidator)
  90. }
  91. consensusReactor := consensus.NewConsensusReactor(consensusState, fastSync)
  92. // Make p2p network switch
  93. sw := p2p.NewSwitch(config.GetConfig("p2p"))
  94. sw.AddReactor("MEMPOOL", mempoolReactor)
  95. sw.AddReactor("BLOCKCHAIN", bcReactor)
  96. sw.AddReactor("CONSENSUS", consensusReactor)
  97. // Optionally, start the pex reactor
  98. // TODO: this is a dev feature, it needs some love
  99. if config.GetBool("pex_reactor") {
  100. addrBook := p2p.NewAddrBook(config.GetString("addrbook_file"), config.GetBool("addrbook_strict"))
  101. addrBook.Start()
  102. pexReactor := p2p.NewPEXReactor(addrBook)
  103. sw.AddReactor("PEX", pexReactor)
  104. }
  105. // Filter peers by addr or pubkey with an ABCI query.
  106. // If the query return code is OK, add peer.
  107. // XXX: Query format subject to change
  108. if config.GetBool("filter_peers") {
  109. // NOTE: addr is ip:port
  110. sw.SetAddrFilter(func(addr net.Addr) error {
  111. resQuery, err := proxyApp.Query().QuerySync(abci.RequestQuery{Path: cmn.Fmt("/p2p/filter/addr/%s", addr.String())})
  112. if err != nil {
  113. return err
  114. }
  115. if resQuery.Code.IsOK() {
  116. return nil
  117. }
  118. return errors.New(resQuery.Code.String())
  119. })
  120. sw.SetPubKeyFilter(func(pubkey crypto.PubKeyEd25519) error {
  121. resQuery, err := proxyApp.Query().QuerySync(abci.RequestQuery{Path: cmn.Fmt("/p2p/filter/pubkey/%X", pubkey.Bytes())})
  122. if err != nil {
  123. return err
  124. }
  125. if resQuery.Code.IsOK() {
  126. return nil
  127. }
  128. return errors.New(resQuery.Code.String())
  129. })
  130. }
  131. // add the event switch to all services
  132. // they should all satisfy events.Eventable
  133. SetEventSwitch(eventSwitch, bcReactor, mempoolReactor, consensusReactor)
  134. // run the profile server
  135. profileHost := config.GetString("prof_laddr")
  136. if profileHost != "" {
  137. go func() {
  138. log.Warn("Profile server", "error", http.ListenAndServe(profileHost, nil))
  139. }()
  140. }
  141. node := &Node{
  142. config: config,
  143. sw: sw,
  144. evsw: eventSwitch,
  145. blockStore: blockStore,
  146. bcReactor: bcReactor,
  147. mempoolReactor: mempoolReactor,
  148. consensusState: consensusState,
  149. consensusReactor: consensusReactor,
  150. privValidator: privValidator,
  151. genesisDoc: state.GenesisDoc,
  152. privKey: privKey,
  153. proxyApp: proxyApp,
  154. }
  155. node.BaseService = *cmn.NewBaseService(log, "Node", node)
  156. return node
  157. }
  158. func (n *Node) OnStart() error {
  159. n.BaseService.OnStart()
  160. // Create & add listener
  161. protocol, address := ProtocolAndAddress(n.config.GetString("node_laddr"))
  162. l := p2p.NewDefaultListener(protocol, address, n.config.GetBool("skip_upnp"))
  163. n.sw.AddListener(l)
  164. // Start the switch
  165. n.sw.SetNodeInfo(makeNodeInfo(n.config, n.sw, n.privKey))
  166. n.sw.SetNodePrivKey(n.privKey)
  167. _, err := n.sw.Start()
  168. if err != nil {
  169. return err
  170. }
  171. // Dial out of seed nodes exist
  172. if n.config.GetString("seeds") != "" {
  173. seeds := strings.Split(n.config.GetString("seeds"), ",")
  174. n.sw.DialSeeds(seeds)
  175. }
  176. // Run the RPC server
  177. if n.config.GetString("rpc_laddr") != "" {
  178. _, err := n.startRPC()
  179. if err != nil {
  180. return err
  181. }
  182. }
  183. return nil
  184. }
  185. func (n *Node) RunForever() {
  186. // Sleep forever and then...
  187. cmn.TrapSignal(func() {
  188. n.Stop()
  189. })
  190. }
  191. func (n *Node) OnStop() {
  192. n.BaseService.OnStop()
  193. log.Notice("Stopping Node")
  194. // TODO: gracefully disconnect from peers.
  195. n.sw.Stop()
  196. }
  197. // Add the event switch to reactors, mempool, etc.
  198. func SetEventSwitch(evsw types.EventSwitch, eventables ...types.Eventable) {
  199. for _, e := range eventables {
  200. e.SetEventSwitch(evsw)
  201. }
  202. }
  203. // Add a Listener to accept inbound peer connections.
  204. // Add listeners before starting the Node.
  205. // The first listener is the primary listener (in NodeInfo)
  206. func (n *Node) AddListener(l p2p.Listener) {
  207. n.sw.AddListener(l)
  208. }
  209. func (n *Node) startRPC() ([]net.Listener, error) {
  210. rpccore.SetConfig(n.config)
  211. rpccore.SetEventSwitch(n.evsw)
  212. rpccore.SetBlockStore(n.blockStore)
  213. rpccore.SetConsensusState(n.consensusState)
  214. rpccore.SetMempool(n.mempoolReactor.Mempool)
  215. rpccore.SetSwitch(n.sw)
  216. rpccore.SetPubKey(n.privValidator.PubKey)
  217. rpccore.SetGenesisDoc(n.genesisDoc)
  218. rpccore.SetProxyAppQuery(n.proxyApp.Query())
  219. listenAddrs := strings.Split(n.config.GetString("rpc_laddr"), ",")
  220. // we may expose the rpc over both a unix and tcp socket
  221. listeners := make([]net.Listener, len(listenAddrs))
  222. for i, listenAddr := range listenAddrs {
  223. mux := http.NewServeMux()
  224. wm := rpcserver.NewWebsocketManager(rpccore.Routes, n.evsw)
  225. mux.HandleFunc("/websocket", wm.WebsocketHandler)
  226. rpcserver.RegisterRPCFuncs(mux, rpccore.Routes)
  227. listener, err := rpcserver.StartHTTPServer(listenAddr, mux)
  228. if err != nil {
  229. return nil, err
  230. }
  231. listeners[i] = listener
  232. }
  233. // we expose a simplified api over grpc for convenience to app devs
  234. grpcListenAddr := n.config.GetString("grpc_laddr")
  235. if grpcListenAddr != "" {
  236. listener, err := grpccore.StartGRPCServer(grpcListenAddr)
  237. if err != nil {
  238. return nil, err
  239. }
  240. listeners = append(listeners, listener)
  241. }
  242. return listeners, nil
  243. }
  244. func (n *Node) Switch() *p2p.Switch {
  245. return n.sw
  246. }
  247. func (n *Node) BlockStore() *bc.BlockStore {
  248. return n.blockStore
  249. }
  250. func (n *Node) ConsensusState() *consensus.ConsensusState {
  251. return n.consensusState
  252. }
  253. func (n *Node) ConsensusReactor() *consensus.ConsensusReactor {
  254. return n.consensusReactor
  255. }
  256. func (n *Node) MempoolReactor() *mempl.MempoolReactor {
  257. return n.mempoolReactor
  258. }
  259. func (n *Node) EventSwitch() types.EventSwitch {
  260. return n.evsw
  261. }
  262. // XXX: for convenience
  263. func (n *Node) PrivValidator() *types.PrivValidator {
  264. return n.privValidator
  265. }
  266. func (n *Node) GenesisDoc() *types.GenesisDoc {
  267. return n.genesisDoc
  268. }
  269. func (n *Node) ProxyApp() proxy.AppConns {
  270. return n.proxyApp
  271. }
  272. func makeNodeInfo(config cfg.Config, sw *p2p.Switch, privKey crypto.PrivKeyEd25519) *p2p.NodeInfo {
  273. nodeInfo := &p2p.NodeInfo{
  274. PubKey: privKey.PubKey().(crypto.PubKeyEd25519),
  275. Moniker: config.GetString("moniker"),
  276. Network: config.GetString("chain_id"),
  277. Version: version.Version,
  278. Other: []string{
  279. cmn.Fmt("wire_version=%v", wire.Version),
  280. cmn.Fmt("p2p_version=%v", p2p.Version),
  281. cmn.Fmt("consensus_version=%v", consensus.Version),
  282. cmn.Fmt("rpc_version=%v/%v", rpc.Version, rpccore.Version),
  283. },
  284. }
  285. // include git hash in the nodeInfo if available
  286. if rev, err := cmn.ReadFile(config.GetString("revision_file")); err == nil {
  287. nodeInfo.Other = append(nodeInfo.Other, cmn.Fmt("revision=%v", string(rev)))
  288. }
  289. if !sw.IsListening() {
  290. return nodeInfo
  291. }
  292. p2pListener := sw.Listeners()[0]
  293. p2pHost := p2pListener.ExternalAddress().IP.String()
  294. p2pPort := p2pListener.ExternalAddress().Port
  295. rpcListenAddr := config.GetString("rpc_laddr")
  296. // We assume that the rpcListener has the same ExternalAddress.
  297. // This is probably true because both P2P and RPC listeners use UPnP,
  298. // except of course if the rpc is only bound to localhost
  299. nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
  300. nodeInfo.Other = append(nodeInfo.Other, cmn.Fmt("rpc_addr=%v", rpcListenAddr))
  301. return nodeInfo
  302. }
  303. func (n *Node) NodeInfo() *p2p.NodeInfo {
  304. return n.sw.NodeInfo()
  305. }
  306. func (n *Node) DialSeeds(seeds []string) {
  307. n.sw.DialSeeds(seeds)
  308. }
  309. // Defaults to tcp
  310. func ProtocolAndAddress(listenAddr string) (string, string) {
  311. protocol, address := "tcp", listenAddr
  312. parts := strings.SplitN(address, "://", 2)
  313. if len(parts) == 2 {
  314. protocol, address = parts[0], parts[1]
  315. }
  316. return protocol, address
  317. }