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.

606 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
7 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
7 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
7 years ago
9 years ago
7 years ago
9 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)
  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 switche'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 peer to the switch
  183. // and to all registered reactors.
  184. // NOTE: This performs a blocking handshake before the peer is added.
  185. // CONTRACT: If error is returned, peer is nil, and conn is immediately closed.
  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. // permute the list, dial them in random order.
  269. perm := rand.Perm(len(netAddrs))
  270. for i := 0; i < len(perm); i++ {
  271. go func(i int) {
  272. time.Sleep(time.Duration(rand.Int63n(3000)) * time.Millisecond)
  273. j := perm[i]
  274. sw.dialSeed(netAddrs[j])
  275. }(i)
  276. }
  277. return nil
  278. }
  279. func (sw *Switch) dialSeed(addr *NetAddress) {
  280. peer, err := sw.DialPeerWithAddress(addr, true)
  281. if err != nil {
  282. sw.Logger.Error("Error dialing seed", "err", err)
  283. } else {
  284. sw.Logger.Info("Connected to seed", "peer", peer)
  285. }
  286. }
  287. // DialPeerWithAddress dials the given peer and runs sw.addPeer if it connects successfully.
  288. // If `persistent == true`, the switch will always try to reconnect to this peer if the connection ever fails.
  289. func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (Peer, error) {
  290. sw.dialing.Set(addr.IP.String(), addr)
  291. defer sw.dialing.Delete(addr.IP.String())
  292. sw.Logger.Info("Dialing peer", "address", addr)
  293. peer, err := newOutboundPeer(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  294. if err != nil {
  295. sw.Logger.Error("Failed to dial peer", "address", addr, "err", err)
  296. return nil, err
  297. }
  298. peer.SetLogger(sw.Logger.With("peer", addr))
  299. if persistent {
  300. peer.makePersistent()
  301. }
  302. err = sw.addPeer(peer)
  303. if err != nil {
  304. sw.Logger.Error("Failed to add peer", "address", addr, "err", err)
  305. peer.CloseConn()
  306. return nil, err
  307. }
  308. sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer)
  309. return peer, nil
  310. }
  311. // IsDialing returns true if the switch is currently dialing the given address.
  312. func (sw *Switch) IsDialing(addr *NetAddress) bool {
  313. return sw.dialing.Has(addr.IP.String())
  314. }
  315. // Broadcast runs a go routine for each attempted send, which will block
  316. // trying to send for defaultSendTimeoutSeconds. Returns a channel
  317. // which receives success values for each attempted send (false if times out)
  318. // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
  319. // TODO: Something more intelligent.
  320. func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
  321. successChan := make(chan bool, len(sw.peers.List()))
  322. sw.Logger.Debug("Broadcast", "channel", chID, "msg", msg)
  323. for _, peer := range sw.peers.List() {
  324. go func(peer Peer) {
  325. success := peer.Send(chID, msg)
  326. successChan <- success
  327. }(peer)
  328. }
  329. return successChan
  330. }
  331. // NumPeers returns the count of outbound/inbound and outbound-dialing peers.
  332. func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
  333. peers := sw.peers.List()
  334. for _, peer := range peers {
  335. if peer.IsOutbound() {
  336. outbound++
  337. } else {
  338. inbound++
  339. }
  340. }
  341. dialing = sw.dialing.Size()
  342. return
  343. }
  344. // Peers returns the set of peers the switch is connected to.
  345. func (sw *Switch) Peers() IPeerSet {
  346. return sw.peers
  347. }
  348. // StopPeerForError disconnects from a peer due to external error.
  349. // If the peer is persistent, it will attempt to reconnect.
  350. // TODO: make record depending on reason.
  351. func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) {
  352. addr, _ := NewNetAddressString(peer.NodeInfo().RemoteAddr)
  353. sw.Logger.Error("Stopping peer for error", "peer", peer, "err", reason)
  354. sw.stopAndRemovePeer(peer, reason)
  355. if peer.IsPersistent() {
  356. go func() {
  357. sw.Logger.Info("Reconnecting to peer", "peer", peer)
  358. for i := 1; i < reconnectAttempts; i++ {
  359. if !sw.IsRunning() {
  360. return
  361. }
  362. peer, err := sw.DialPeerWithAddress(addr, true)
  363. if err != nil {
  364. if i == reconnectAttempts {
  365. sw.Logger.Info("Error reconnecting to peer. Giving up", "tries", i, "err", err)
  366. return
  367. }
  368. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err)
  369. time.Sleep(reconnectInterval)
  370. continue
  371. }
  372. sw.Logger.Info("Reconnected to peer", "peer", peer)
  373. return
  374. }
  375. }()
  376. }
  377. }
  378. // StopPeerGracefully disconnects from a peer gracefully.
  379. // TODO: handle graceful disconnects.
  380. func (sw *Switch) StopPeerGracefully(peer Peer) {
  381. sw.Logger.Info("Stopping peer gracefully")
  382. sw.stopAndRemovePeer(peer, nil)
  383. }
  384. func (sw *Switch) stopAndRemovePeer(peer Peer, reason interface{}) {
  385. sw.peers.Remove(peer)
  386. peer.Stop()
  387. for _, reactor := range sw.reactors {
  388. reactor.RemovePeer(peer, reason)
  389. }
  390. }
  391. func (sw *Switch) listenerRoutine(l Listener) {
  392. for {
  393. inConn, ok := <-l.Connections()
  394. if !ok {
  395. break
  396. }
  397. // ignore connection if we already have enough
  398. maxPeers := sw.config.MaxNumPeers
  399. if maxPeers <= sw.peers.Size() {
  400. sw.Logger.Info("Ignoring inbound connection: already have enough peers", "address", inConn.RemoteAddr().String(), "numPeers", sw.peers.Size(), "max", maxPeers)
  401. continue
  402. }
  403. // New inbound connection!
  404. err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
  405. if err != nil {
  406. sw.Logger.Info("Ignoring inbound connection: error while adding peer", "address", inConn.RemoteAddr().String(), "err", err)
  407. continue
  408. }
  409. // NOTE: We don't yet have the listening port of the
  410. // remote (if they have a listener at all).
  411. // The peerHandshake will handle that
  412. }
  413. // cleanup
  414. }
  415. //-----------------------------------------------------------------------------
  416. type SwitchEventNewPeer struct {
  417. Peer Peer
  418. }
  419. type SwitchEventDonePeer struct {
  420. Peer Peer
  421. Error interface{}
  422. }
  423. //------------------------------------------------------------------
  424. // Switches connected via arbitrary net.Conn; useful for testing
  425. // MakeConnectedSwitches returns n switches, connected according to the connect func.
  426. // If connect==Connect2Switches, the switches will be fully connected.
  427. // initSwitch defines how the ith switch should be initialized (ie. with what reactors).
  428. // NOTE: panics if any switch fails to start.
  429. func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
  430. switches := make([]*Switch, n)
  431. for i := 0; i < n; i++ {
  432. switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
  433. }
  434. if err := StartSwitches(switches); err != nil {
  435. panic(err)
  436. }
  437. for i := 0; i < n; i++ {
  438. for j := i; j < n; j++ {
  439. connect(switches, i, j)
  440. }
  441. }
  442. return switches
  443. }
  444. var PanicOnAddPeerErr = false
  445. // Connect2Switches will connect switches i and j via net.Pipe()
  446. // Blocks until a conection is established.
  447. // NOTE: caller ensures i and j are within bounds
  448. func Connect2Switches(switches []*Switch, i, j int) {
  449. switchI := switches[i]
  450. switchJ := switches[j]
  451. c1, c2 := net.Pipe()
  452. doneCh := make(chan struct{})
  453. go func() {
  454. err := switchI.addPeerWithConnection(c1)
  455. if PanicOnAddPeerErr && err != nil {
  456. panic(err)
  457. }
  458. doneCh <- struct{}{}
  459. }()
  460. go func() {
  461. err := switchJ.addPeerWithConnection(c2)
  462. if PanicOnAddPeerErr && err != nil {
  463. panic(err)
  464. }
  465. doneCh <- struct{}{}
  466. }()
  467. <-doneCh
  468. <-doneCh
  469. }
  470. // StartSwitches calls sw.Start() for each given switch.
  471. // It returns the first encountered error.
  472. func StartSwitches(switches []*Switch) error {
  473. for _, s := range switches {
  474. _, err := s.Start() // start switch and reactors
  475. if err != nil {
  476. return err
  477. }
  478. }
  479. return nil
  480. }
  481. func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
  482. privKey := crypto.GenPrivKeyEd25519()
  483. // new switch, add reactors
  484. // TODO: let the config be passed in?
  485. s := initSwitch(i, NewSwitch(cfg))
  486. s.SetNodeInfo(&NodeInfo{
  487. PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  488. Moniker: cmn.Fmt("switch%d", i),
  489. Network: network,
  490. Version: version,
  491. RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  492. ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  493. })
  494. s.SetNodePrivKey(privKey)
  495. return s
  496. }
  497. func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
  498. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  499. if err != nil {
  500. conn.Close()
  501. return err
  502. }
  503. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  504. if err = sw.addPeer(peer); err != nil {
  505. conn.Close()
  506. return err
  507. }
  508. return nil
  509. }
  510. func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
  511. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
  512. if err != nil {
  513. conn.Close()
  514. return err
  515. }
  516. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  517. if err = sw.addPeer(peer); err != nil {
  518. conn.Close()
  519. return err
  520. }
  521. return nil
  522. }