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.

596 lines
17 KiB

9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
8 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
8 years ago
9 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
9 years ago
8 years ago
7 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
7 years ago
8 years ago
8 years ago
9 years ago
7 years ago
9 years ago
7 years ago
8 years ago
9 years ago
7 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
7 years ago
7 years ago
7 years ago
8 years ago
7 years ago
8 years ago
8 years ago
8 years ago
  1. package p2p
  2. import (
  3. "errors"
  4. "fmt"
  5. "math/rand"
  6. "net"
  7. "time"
  8. crypto "github.com/tendermint/go-crypto"
  9. cfg "github.com/tendermint/tendermint/config"
  10. cmn "github.com/tendermint/tmlibs/common"
  11. )
  12. const (
  13. reconnectAttempts = 30
  14. reconnectInterval = 3 * time.Second
  15. )
  16. type Reactor interface {
  17. cmn.Service // Start, Stop
  18. SetSwitch(*Switch)
  19. GetChannels() []*ChannelDescriptor
  20. AddPeer(peer Peer)
  21. RemovePeer(peer Peer, reason interface{})
  22. Receive(chID byte, peer Peer, msgBytes []byte) // CONTRACT: msgBytes are not nil
  23. }
  24. //--------------------------------------
  25. type BaseReactor struct {
  26. cmn.BaseService // Provides Start, Stop, .Quit
  27. Switch *Switch
  28. }
  29. func NewBaseReactor(name string, impl Reactor) *BaseReactor {
  30. return &BaseReactor{
  31. BaseService: *cmn.NewBaseService(nil, name, impl),
  32. Switch: nil,
  33. }
  34. }
  35. func (br *BaseReactor) SetSwitch(sw *Switch) {
  36. br.Switch = sw
  37. }
  38. func (_ *BaseReactor) GetChannels() []*ChannelDescriptor { return nil }
  39. func (_ *BaseReactor) AddPeer(peer Peer) {}
  40. func (_ *BaseReactor) RemovePeer(peer Peer, reason interface{}) {}
  41. func (_ *BaseReactor) Receive(chID byte, peer Peer, msgBytes []byte) {}
  42. //-----------------------------------------------------------------------------
  43. /*
  44. The `Switch` handles peer connections and exposes an API to receive incoming messages
  45. on `Reactors`. Each `Reactor` is responsible for handling incoming messages of one
  46. or more `Channels`. So while sending outgoing messages is typically performed on the peer,
  47. incoming messages are received on the reactor.
  48. */
  49. type Switch struct {
  50. cmn.BaseService
  51. config *cfg.P2PConfig
  52. peerConfig *PeerConfig
  53. listeners []Listener
  54. reactors map[string]Reactor
  55. chDescs []*ChannelDescriptor
  56. reactorsByCh map[byte]Reactor
  57. peers *PeerSet
  58. dialing *cmn.CMap
  59. nodeInfo *NodeInfo // our node info
  60. nodePrivKey crypto.PrivKeyEd25519 // our node privkey
  61. filterConnByAddr func(net.Addr) error
  62. filterConnByPubKey func(crypto.PubKeyEd25519) error
  63. }
  64. var (
  65. ErrSwitchDuplicatePeer = errors.New("Duplicate peer")
  66. )
  67. func NewSwitch(config *cfg.P2PConfig) *Switch {
  68. sw := &Switch{
  69. config: config,
  70. peerConfig: DefaultPeerConfig(),
  71. reactors: make(map[string]Reactor),
  72. chDescs: make([]*ChannelDescriptor, 0),
  73. reactorsByCh: make(map[byte]Reactor),
  74. peers: NewPeerSet(),
  75. dialing: cmn.NewCMap(),
  76. nodeInfo: nil,
  77. }
  78. // TODO: collapse the peerConfig into the config ?
  79. sw.peerConfig.MConfig.flushThrottle = time.Duration(config.FlushThrottleTimeout) * time.Millisecond
  80. sw.peerConfig.MConfig.SendRate = config.SendRate
  81. sw.peerConfig.MConfig.RecvRate = config.RecvRate
  82. sw.peerConfig.MConfig.maxMsgPacketPayloadSize = config.MaxMsgPacketPayloadSize
  83. sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
  84. return sw
  85. }
  86. // AddReactor adds the given reactor to the switch.
  87. // NOTE: Not goroutine safe.
  88. func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
  89. // Validate the reactor.
  90. // No two reactors can share the same channel.
  91. reactorChannels := reactor.GetChannels()
  92. for _, chDesc := range reactorChannels {
  93. chID := chDesc.ID
  94. if sw.reactorsByCh[chID] != nil {
  95. cmn.PanicSanity(fmt.Sprintf("Channel %X has multiple reactors %v & %v", chID, sw.reactorsByCh[chID], reactor))
  96. }
  97. sw.chDescs = append(sw.chDescs, chDesc)
  98. sw.reactorsByCh[chID] = reactor
  99. }
  100. sw.reactors[name] = reactor
  101. reactor.SetSwitch(sw)
  102. return reactor
  103. }
  104. // Reactors returns a map of reactors registered on the switch.
  105. // NOTE: Not goroutine safe.
  106. func (sw *Switch) Reactors() map[string]Reactor {
  107. return sw.reactors
  108. }
  109. // Reactor returns the reactor with the given name.
  110. // NOTE: Not goroutine safe.
  111. func (sw *Switch) Reactor(name string) Reactor {
  112. return sw.reactors[name]
  113. }
  114. // AddListener adds the given listener to the switch for listening to incoming peer connections.
  115. // NOTE: Not goroutine safe.
  116. func (sw *Switch) AddListener(l Listener) {
  117. sw.listeners = append(sw.listeners, l)
  118. }
  119. // Listeners returns the list of listeners the switch listens on.
  120. // NOTE: Not goroutine safe.
  121. func (sw *Switch) Listeners() []Listener {
  122. return sw.listeners
  123. }
  124. // IsListening returns true if the switch has at least one listener.
  125. // NOTE: Not goroutine safe.
  126. func (sw *Switch) IsListening() bool {
  127. return len(sw.listeners) > 0
  128. }
  129. // SetNodeInfo sets the switch's NodeInfo for checking compatibility and handshaking with other nodes.
  130. // NOTE: Not goroutine safe.
  131. func (sw *Switch) SetNodeInfo(nodeInfo *NodeInfo) {
  132. sw.nodeInfo = nodeInfo
  133. }
  134. // NodeInfo returns the switch's NodeInfo.
  135. // NOTE: Not goroutine safe.
  136. func (sw *Switch) NodeInfo() *NodeInfo {
  137. return sw.nodeInfo
  138. }
  139. // SetNodePrivKey sets the switch's private key for authenticated encryption.
  140. // NOTE: Overwrites sw.nodeInfo.PubKey.
  141. // NOTE: Not goroutine safe.
  142. func (sw *Switch) SetNodePrivKey(nodePrivKey crypto.PrivKeyEd25519) {
  143. sw.nodePrivKey = nodePrivKey
  144. if sw.nodeInfo != nil {
  145. sw.nodeInfo.PubKey = nodePrivKey.PubKey().Unwrap().(crypto.PubKeyEd25519)
  146. }
  147. }
  148. // OnStart implements BaseService. It starts all the reactors, peers, and listeners.
  149. func (sw *Switch) OnStart() error {
  150. sw.BaseService.OnStart()
  151. // Start reactors
  152. for _, reactor := range sw.reactors {
  153. _, err := reactor.Start()
  154. if err != nil {
  155. return err
  156. }
  157. }
  158. // Start listeners
  159. for _, listener := range sw.listeners {
  160. go sw.listenerRoutine(listener)
  161. }
  162. return nil
  163. }
  164. // OnStop implements BaseService. It stops all listeners, peers, and reactors.
  165. func (sw *Switch) OnStop() {
  166. sw.BaseService.OnStop()
  167. // Stop listeners
  168. for _, listener := range sw.listeners {
  169. listener.Stop()
  170. }
  171. sw.listeners = nil
  172. // Stop peers
  173. for _, peer := range sw.peers.List() {
  174. peer.Stop()
  175. sw.peers.Remove(peer)
  176. }
  177. // Stop reactors
  178. for _, reactor := range sw.reactors {
  179. reactor.Stop()
  180. }
  181. }
  182. // addPeer checks the given peer's validity, performs a handshake, and adds the
  183. // peer to the switch and to all registered reactors.
  184. // NOTE: This performs a blocking handshake before the peer is added.
  185. // NOTE: If error is returned, caller is responsible for calling peer.CloseConn()
  186. func (sw *Switch) addPeer(peer *peer) error {
  187. if err := sw.FilterConnByAddr(peer.Addr()); err != nil {
  188. return err
  189. }
  190. if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil {
  191. return err
  192. }
  193. if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil {
  194. return err
  195. }
  196. // Avoid self
  197. if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
  198. return errors.New("Ignoring connection from self")
  199. }
  200. // Check version, chain id
  201. if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo()); err != nil {
  202. return err
  203. }
  204. // Check for duplicate peer
  205. if sw.peers.Has(peer.Key()) {
  206. return ErrSwitchDuplicatePeer
  207. }
  208. // Start peer
  209. if sw.IsRunning() {
  210. sw.startInitPeer(peer)
  211. }
  212. // Add the peer to .peers.
  213. // We start it first so that a peer in the list is safe to Stop.
  214. // It should not err since we already checked peers.Has().
  215. if err := sw.peers.Add(peer); err != nil {
  216. return err
  217. }
  218. sw.Logger.Info("Added peer", "peer", peer)
  219. return nil
  220. }
  221. // FilterConnByAddr returns an error if connecting to the given address is forbidden.
  222. func (sw *Switch) FilterConnByAddr(addr net.Addr) error {
  223. if sw.filterConnByAddr != nil {
  224. return sw.filterConnByAddr(addr)
  225. }
  226. return nil
  227. }
  228. // FilterConnByPubKey returns an error if connecting to the given public key is forbidden.
  229. func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error {
  230. if sw.filterConnByPubKey != nil {
  231. return sw.filterConnByPubKey(pubkey)
  232. }
  233. return nil
  234. }
  235. // SetAddrFilter sets the function for filtering connections by address.
  236. func (sw *Switch) SetAddrFilter(f func(net.Addr) error) {
  237. sw.filterConnByAddr = f
  238. }
  239. // SetPubKeyFilter sets the function for filtering connections by public key.
  240. func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) {
  241. sw.filterConnByPubKey = f
  242. }
  243. func (sw *Switch) startInitPeer(peer *peer) {
  244. peer.Start() // spawn send/recv routines
  245. for _, reactor := range sw.reactors {
  246. reactor.AddPeer(peer)
  247. }
  248. }
  249. // DialSeeds dials a list of seeds asynchronously in random order.
  250. func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error {
  251. netAddrs, err := NewNetAddressStrings(seeds)
  252. if err != nil {
  253. return err
  254. }
  255. if addrBook != nil {
  256. // add seeds to `addrBook`
  257. ourAddrS := sw.nodeInfo.ListenAddr
  258. ourAddr, _ := NewNetAddressString(ourAddrS)
  259. for _, netAddr := range netAddrs {
  260. // do not add ourselves
  261. if netAddr.Equals(ourAddr) {
  262. continue
  263. }
  264. addrBook.AddAddress(netAddr, ourAddr)
  265. }
  266. addrBook.Save()
  267. }
  268. // Ensure we have a completely undeterministic PRNG. cmd.RandInt64() draws
  269. // from a seed that's initialized with OS entropy on process start.
  270. rng := rand.New(rand.NewSource(cmn.RandInt64()))
  271. // permute the list, dial them in random order.
  272. perm := rng.Perm(len(netAddrs))
  273. for i := 0; i < len(perm); i++ {
  274. go func(i int) {
  275. time.Sleep(time.Duration(rng.Int63n(3000)) * time.Millisecond)
  276. j := perm[i]
  277. sw.dialSeed(netAddrs[j])
  278. }(i)
  279. }
  280. return nil
  281. }
  282. func (sw *Switch) dialSeed(addr *NetAddress) {
  283. peer, err := sw.DialPeerWithAddress(addr, true)
  284. if err != nil {
  285. sw.Logger.Error("Error dialing seed", "err", err)
  286. } else {
  287. sw.Logger.Info("Connected to seed", "peer", peer)
  288. }
  289. }
  290. // DialPeerWithAddress dials the given peer and runs sw.addPeer if it connects successfully.
  291. // If `persistent == true`, the switch will always try to reconnect to this peer if the connection ever fails.
  292. func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (Peer, error) {
  293. sw.dialing.Set(addr.IP.String(), addr)
  294. defer sw.dialing.Delete(addr.IP.String())
  295. sw.Logger.Info("Dialing peer", "address", addr)
  296. peer, err := newOutboundPeer(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  297. if err != nil {
  298. sw.Logger.Error("Failed to dial peer", "address", addr, "err", err)
  299. return nil, err
  300. }
  301. peer.SetLogger(sw.Logger.With("peer", addr))
  302. if persistent {
  303. peer.makePersistent()
  304. }
  305. err = sw.addPeer(peer)
  306. if err != nil {
  307. sw.Logger.Error("Failed to add peer", "address", addr, "err", err)
  308. peer.CloseConn()
  309. return nil, err
  310. }
  311. sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer)
  312. return peer, nil
  313. }
  314. // IsDialing returns true if the switch is currently dialing the given address.
  315. func (sw *Switch) IsDialing(addr *NetAddress) bool {
  316. return sw.dialing.Has(addr.IP.String())
  317. }
  318. // Broadcast runs a go routine for each attempted send, which will block
  319. // trying to send for defaultSendTimeoutSeconds. Returns a channel
  320. // which receives success values for each attempted send (false if times out).
  321. // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
  322. // TODO: Something more intelligent.
  323. func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
  324. successChan := make(chan bool, len(sw.peers.List()))
  325. sw.Logger.Debug("Broadcast", "channel", chID, "msg", msg)
  326. for _, peer := range sw.peers.List() {
  327. go func(peer Peer) {
  328. success := peer.Send(chID, msg)
  329. successChan <- success
  330. }(peer)
  331. }
  332. return successChan
  333. }
  334. // NumPeers returns the count of outbound/inbound and outbound-dialing peers.
  335. func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
  336. peers := sw.peers.List()
  337. for _, peer := range peers {
  338. if peer.IsOutbound() {
  339. outbound++
  340. } else {
  341. inbound++
  342. }
  343. }
  344. dialing = sw.dialing.Size()
  345. return
  346. }
  347. // Peers returns the set of peers that are connected to the switch.
  348. func (sw *Switch) Peers() IPeerSet {
  349. return sw.peers
  350. }
  351. // StopPeerForError disconnects from a peer due to external error.
  352. // If the peer is persistent, it will attempt to reconnect.
  353. // TODO: make record depending on reason.
  354. func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) {
  355. addr, _ := NewNetAddressString(peer.NodeInfo().RemoteAddr)
  356. sw.Logger.Error("Stopping peer for error", "peer", peer, "err", reason)
  357. sw.stopAndRemovePeer(peer, reason)
  358. if peer.IsPersistent() {
  359. go func() {
  360. sw.Logger.Info("Reconnecting to peer", "peer", peer)
  361. for i := 1; i < reconnectAttempts; i++ {
  362. if !sw.IsRunning() {
  363. return
  364. }
  365. peer, err := sw.DialPeerWithAddress(addr, true)
  366. if err != nil {
  367. if i == reconnectAttempts {
  368. sw.Logger.Info("Error reconnecting to peer. Giving up", "tries", i, "err", err)
  369. return
  370. }
  371. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err)
  372. time.Sleep(reconnectInterval)
  373. continue
  374. }
  375. sw.Logger.Info("Reconnected to peer", "peer", peer)
  376. return
  377. }
  378. }()
  379. }
  380. }
  381. // StopPeerGracefully disconnects from a peer gracefully.
  382. // TODO: handle graceful disconnects.
  383. func (sw *Switch) StopPeerGracefully(peer Peer) {
  384. sw.Logger.Info("Stopping peer gracefully")
  385. sw.stopAndRemovePeer(peer, nil)
  386. }
  387. func (sw *Switch) stopAndRemovePeer(peer Peer, reason interface{}) {
  388. sw.peers.Remove(peer)
  389. peer.Stop()
  390. for _, reactor := range sw.reactors {
  391. reactor.RemovePeer(peer, reason)
  392. }
  393. }
  394. func (sw *Switch) listenerRoutine(l Listener) {
  395. for {
  396. inConn, ok := <-l.Connections()
  397. if !ok {
  398. break
  399. }
  400. // ignore connection if we already have enough
  401. maxPeers := sw.config.MaxNumPeers
  402. if maxPeers <= sw.peers.Size() {
  403. sw.Logger.Info("Ignoring inbound connection: already have enough peers", "address", inConn.RemoteAddr().String(), "numPeers", sw.peers.Size(), "max", maxPeers)
  404. continue
  405. }
  406. // New inbound connection!
  407. err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
  408. if err != nil {
  409. sw.Logger.Info("Ignoring inbound connection: error while adding peer", "address", inConn.RemoteAddr().String(), "err", err)
  410. continue
  411. }
  412. // NOTE: We don't yet have the listening port of the
  413. // remote (if they have a listener at all).
  414. // The peerHandshake will handle that.
  415. }
  416. // cleanup
  417. }
  418. //------------------------------------------------------------------
  419. // Connects switches via arbitrary net.Conn. Used for testing.
  420. // MakeConnectedSwitches returns n switches, connected according to the connect func.
  421. // If connect==Connect2Switches, the switches will be fully connected.
  422. // initSwitch defines how the i'th switch should be initialized (ie. with what reactors).
  423. // NOTE: panics if any switch fails to start.
  424. func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
  425. switches := make([]*Switch, n)
  426. for i := 0; i < n; i++ {
  427. switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
  428. }
  429. if err := StartSwitches(switches); err != nil {
  430. panic(err)
  431. }
  432. for i := 0; i < n; i++ {
  433. for j := i + 1; j < n; j++ {
  434. connect(switches, i, j)
  435. }
  436. }
  437. return switches
  438. }
  439. // Connect2Switches will connect switches i and j via net.Pipe().
  440. // Blocks until a conection is established.
  441. // NOTE: caller ensures i and j are within bounds.
  442. func Connect2Switches(switches []*Switch, i, j int) {
  443. switchI := switches[i]
  444. switchJ := switches[j]
  445. c1, c2 := netPipe()
  446. doneCh := make(chan struct{})
  447. go func() {
  448. err := switchI.addPeerWithConnection(c1)
  449. if err != nil {
  450. panic(err)
  451. }
  452. doneCh <- struct{}{}
  453. }()
  454. go func() {
  455. err := switchJ.addPeerWithConnection(c2)
  456. if err != nil {
  457. panic(err)
  458. }
  459. doneCh <- struct{}{}
  460. }()
  461. <-doneCh
  462. <-doneCh
  463. }
  464. // StartSwitches calls sw.Start() for each given switch.
  465. // It returns the first encountered error.
  466. func StartSwitches(switches []*Switch) error {
  467. for _, s := range switches {
  468. _, err := s.Start() // start switch and reactors
  469. if err != nil {
  470. return err
  471. }
  472. }
  473. return nil
  474. }
  475. func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
  476. privKey := crypto.GenPrivKeyEd25519()
  477. // new switch, add reactors
  478. // TODO: let the config be passed in?
  479. s := initSwitch(i, NewSwitch(cfg))
  480. s.SetNodeInfo(&NodeInfo{
  481. PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  482. Moniker: cmn.Fmt("switch%d", i),
  483. Network: network,
  484. Version: version,
  485. RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  486. ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  487. })
  488. s.SetNodePrivKey(privKey)
  489. return s
  490. }
  491. func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
  492. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  493. if err != nil {
  494. conn.Close()
  495. return err
  496. }
  497. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  498. if err = sw.addPeer(peer); err != nil {
  499. peer.CloseConn()
  500. return err
  501. }
  502. return nil
  503. }
  504. func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
  505. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
  506. if err != nil {
  507. conn.Close()
  508. return err
  509. }
  510. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  511. if err = sw.addPeer(peer); err != nil {
  512. peer.CloseConn()
  513. return err
  514. }
  515. return nil
  516. }