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.

577 lines
15 KiB

9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
8 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
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
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
8 years ago
8 years ago
9 years ago
8 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
8 years ago
8 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. sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
  79. return sw
  80. }
  81. // Not goroutine safe.
  82. func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
  83. // Validate the reactor.
  84. // No two reactors can share the same channel.
  85. reactorChannels := reactor.GetChannels()
  86. for _, chDesc := range reactorChannels {
  87. chID := chDesc.ID
  88. if sw.reactorsByCh[chID] != nil {
  89. cmn.PanicSanity(fmt.Sprintf("Channel %X has multiple reactors %v & %v", chID, sw.reactorsByCh[chID], reactor))
  90. }
  91. sw.chDescs = append(sw.chDescs, chDesc)
  92. sw.reactorsByCh[chID] = reactor
  93. }
  94. sw.reactors[name] = reactor
  95. reactor.SetSwitch(sw)
  96. return reactor
  97. }
  98. // Not goroutine safe.
  99. func (sw *Switch) Reactors() map[string]Reactor {
  100. return sw.reactors
  101. }
  102. // Not goroutine safe.
  103. func (sw *Switch) Reactor(name string) Reactor {
  104. return sw.reactors[name]
  105. }
  106. // Not goroutine safe.
  107. func (sw *Switch) AddListener(l Listener) {
  108. sw.listeners = append(sw.listeners, l)
  109. }
  110. // Not goroutine safe.
  111. func (sw *Switch) Listeners() []Listener {
  112. return sw.listeners
  113. }
  114. // Not goroutine safe.
  115. func (sw *Switch) IsListening() bool {
  116. return len(sw.listeners) > 0
  117. }
  118. // Not goroutine safe.
  119. func (sw *Switch) SetNodeInfo(nodeInfo *NodeInfo) {
  120. sw.nodeInfo = nodeInfo
  121. }
  122. // Not goroutine safe.
  123. func (sw *Switch) NodeInfo() *NodeInfo {
  124. return sw.nodeInfo
  125. }
  126. // Not goroutine safe.
  127. // NOTE: Overwrites sw.nodeInfo.PubKey
  128. func (sw *Switch) SetNodePrivKey(nodePrivKey crypto.PrivKeyEd25519) {
  129. sw.nodePrivKey = nodePrivKey
  130. if sw.nodeInfo != nil {
  131. sw.nodeInfo.PubKey = nodePrivKey.PubKey().Unwrap().(crypto.PubKeyEd25519)
  132. }
  133. }
  134. // Switch.Start() starts all the reactors, peers, and listeners.
  135. func (sw *Switch) OnStart() error {
  136. sw.BaseService.OnStart()
  137. // Start reactors
  138. for _, reactor := range sw.reactors {
  139. _, err := reactor.Start()
  140. if err != nil {
  141. return err
  142. }
  143. }
  144. // Start peers
  145. for _, peer := range sw.peers.List() {
  146. sw.startInitPeer(peer)
  147. }
  148. // Start listeners
  149. for _, listener := range sw.listeners {
  150. go sw.listenerRoutine(listener)
  151. }
  152. return nil
  153. }
  154. func (sw *Switch) OnStop() {
  155. sw.BaseService.OnStop()
  156. // Stop listeners
  157. for _, listener := range sw.listeners {
  158. listener.Stop()
  159. }
  160. sw.listeners = nil
  161. // Stop peers
  162. for _, peer := range sw.peers.List() {
  163. peer.Stop()
  164. sw.peers.Remove(peer)
  165. }
  166. // Stop reactors
  167. for _, reactor := range sw.reactors {
  168. reactor.Stop()
  169. }
  170. }
  171. // NOTE: This performs a blocking handshake before the peer is added.
  172. // CONTRACT: If error is returned, peer is nil, and conn is immediately closed.
  173. func (sw *Switch) AddPeer(peer *Peer) error {
  174. if err := sw.FilterConnByAddr(peer.Addr()); err != nil {
  175. return err
  176. }
  177. if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil {
  178. return err
  179. }
  180. if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil {
  181. return err
  182. }
  183. // Avoid self
  184. if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
  185. return errors.New("Ignoring connection from self")
  186. }
  187. // Check version, chain id
  188. if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo); err != nil {
  189. return err
  190. }
  191. // Check for duplicate peer
  192. if sw.peers.Has(peer.Key) {
  193. return ErrSwitchDuplicatePeer
  194. }
  195. // Start peer
  196. if sw.IsRunning() {
  197. sw.startInitPeer(peer)
  198. }
  199. // Add the peer to .peers.
  200. // We start it first so that a peer in the list is safe to Stop.
  201. // It should not err since we already checked peers.Has()
  202. if err := sw.peers.Add(peer); err != nil {
  203. return err
  204. }
  205. sw.Logger.Info("Added peer", "peer", peer)
  206. return nil
  207. }
  208. func (sw *Switch) FilterConnByAddr(addr net.Addr) error {
  209. if sw.filterConnByAddr != nil {
  210. return sw.filterConnByAddr(addr)
  211. }
  212. return nil
  213. }
  214. func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error {
  215. if sw.filterConnByPubKey != nil {
  216. return sw.filterConnByPubKey(pubkey)
  217. }
  218. return nil
  219. }
  220. func (sw *Switch) SetAddrFilter(f func(net.Addr) error) {
  221. sw.filterConnByAddr = f
  222. }
  223. func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) {
  224. sw.filterConnByPubKey = f
  225. }
  226. func (sw *Switch) startInitPeer(peer *Peer) {
  227. peer.Start() // spawn send/recv routines
  228. for _, reactor := range sw.reactors {
  229. reactor.AddPeer(peer)
  230. }
  231. }
  232. // Dial a list of seeds asynchronously in random order
  233. func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error {
  234. netAddrs, err := NewNetAddressStrings(seeds)
  235. if err != nil {
  236. return err
  237. }
  238. if addrBook != nil {
  239. // add seeds to `addrBook`
  240. ourAddrS := sw.nodeInfo.ListenAddr
  241. ourAddr, _ := NewNetAddressString(ourAddrS)
  242. for _, netAddr := range netAddrs {
  243. // do not add ourselves
  244. if netAddr.Equals(ourAddr) {
  245. continue
  246. }
  247. addrBook.AddAddress(netAddr, ourAddr)
  248. }
  249. addrBook.Save()
  250. }
  251. // permute the list, dial them in random order.
  252. perm := rand.Perm(len(netAddrs))
  253. for i := 0; i < len(perm); i++ {
  254. go func(i int) {
  255. time.Sleep(time.Duration(rand.Int63n(3000)) * time.Millisecond)
  256. j := perm[i]
  257. sw.dialSeed(netAddrs[j])
  258. }(i)
  259. }
  260. return nil
  261. }
  262. func (sw *Switch) dialSeed(addr *NetAddress) {
  263. peer, err := sw.DialPeerWithAddress(addr, true)
  264. if err != nil {
  265. sw.Logger.Error("Error dialing seed", "err", err)
  266. } else {
  267. sw.Logger.Info("Connected to seed", "peer", peer)
  268. }
  269. }
  270. func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (*Peer, error) {
  271. sw.dialing.Set(addr.IP.String(), addr)
  272. defer sw.dialing.Delete(addr.IP.String())
  273. sw.Logger.Info("Dialing peer", "address", addr)
  274. peer, err := newOutboundPeerWithConfig(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  275. if err != nil {
  276. sw.Logger.Error("Failed to dial peer", "address", addr, "err", err)
  277. return nil, err
  278. }
  279. peer.SetLogger(sw.Logger.With("peer", addr))
  280. if persistent {
  281. peer.makePersistent()
  282. }
  283. err = sw.AddPeer(peer)
  284. if err != nil {
  285. sw.Logger.Error("Failed to add peer", "address", addr, "err", err)
  286. peer.CloseConn()
  287. return nil, err
  288. }
  289. sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer)
  290. return peer, nil
  291. }
  292. func (sw *Switch) IsDialing(addr *NetAddress) bool {
  293. return sw.dialing.Has(addr.IP.String())
  294. }
  295. // Broadcast runs a go routine for each attempted send, which will block
  296. // trying to send for defaultSendTimeoutSeconds. Returns a channel
  297. // which receives success values for each attempted send (false if times out)
  298. // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
  299. func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
  300. successChan := make(chan bool, len(sw.peers.List()))
  301. sw.Logger.Debug("Broadcast", "channel", chID, "msg", msg)
  302. for _, peer := range sw.peers.List() {
  303. go func(peer *Peer) {
  304. success := peer.Send(chID, msg)
  305. successChan <- success
  306. }(peer)
  307. }
  308. return successChan
  309. }
  310. // Returns the count of outbound/inbound and outbound-dialing peers.
  311. func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
  312. peers := sw.peers.List()
  313. for _, peer := range peers {
  314. if peer.outbound {
  315. outbound++
  316. } else {
  317. inbound++
  318. }
  319. }
  320. dialing = sw.dialing.Size()
  321. return
  322. }
  323. func (sw *Switch) Peers() IPeerSet {
  324. return sw.peers
  325. }
  326. // Disconnect from a peer due to external error, retry if it is a persistent peer.
  327. // TODO: make record depending on reason.
  328. func (sw *Switch) StopPeerForError(peer *Peer, reason interface{}) {
  329. addr := NewNetAddress(peer.Addr())
  330. sw.Logger.Info("Stopping peer for error", "peer", peer, "err", reason)
  331. sw.stopAndRemovePeer(peer, reason)
  332. if peer.IsPersistent() {
  333. go func() {
  334. sw.Logger.Info("Reconnecting to peer", "peer", peer)
  335. for i := 1; i < reconnectAttempts; i++ {
  336. if !sw.IsRunning() {
  337. return
  338. }
  339. peer, err := sw.DialPeerWithAddress(addr, true)
  340. if err != nil {
  341. if i == reconnectAttempts {
  342. sw.Logger.Info("Error reconnecting to peer. Giving up", "tries", i, "err", err)
  343. return
  344. }
  345. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err)
  346. time.Sleep(reconnectInterval)
  347. continue
  348. }
  349. sw.Logger.Info("Reconnected to peer", "peer", peer)
  350. return
  351. }
  352. }()
  353. }
  354. }
  355. // Disconnect from a peer gracefully.
  356. // TODO: handle graceful disconnects.
  357. func (sw *Switch) StopPeerGracefully(peer *Peer) {
  358. sw.Logger.Info("Stopping peer gracefully")
  359. sw.stopAndRemovePeer(peer, nil)
  360. }
  361. func (sw *Switch) stopAndRemovePeer(peer *Peer, reason interface{}) {
  362. sw.peers.Remove(peer)
  363. peer.Stop()
  364. for _, reactor := range sw.reactors {
  365. reactor.RemovePeer(peer, reason)
  366. }
  367. }
  368. func (sw *Switch) listenerRoutine(l Listener) {
  369. for {
  370. inConn, ok := <-l.Connections()
  371. if !ok {
  372. break
  373. }
  374. // ignore connection if we already have enough
  375. maxPeers := sw.config.MaxNumPeers
  376. if maxPeers <= sw.peers.Size() {
  377. sw.Logger.Info("Ignoring inbound connection: already have enough peers", "address", inConn.RemoteAddr().String(), "numPeers", sw.peers.Size(), "max", maxPeers)
  378. continue
  379. }
  380. // New inbound connection!
  381. err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
  382. if err != nil {
  383. sw.Logger.Info("Ignoring inbound connection: error while adding peer", "address", inConn.RemoteAddr().String(), "err", err)
  384. continue
  385. }
  386. // NOTE: We don't yet have the listening port of the
  387. // remote (if they have a listener at all).
  388. // The peerHandshake will handle that
  389. }
  390. // cleanup
  391. }
  392. //-----------------------------------------------------------------------------
  393. type SwitchEventNewPeer struct {
  394. Peer *Peer
  395. }
  396. type SwitchEventDonePeer struct {
  397. Peer *Peer
  398. Error interface{}
  399. }
  400. //------------------------------------------------------------------
  401. // Switches connected via arbitrary net.Conn; useful for testing
  402. // Returns n switches, connected according to the connect func.
  403. // If connect==Connect2Switches, the switches will be fully connected.
  404. // initSwitch defines how the ith switch should be initialized (ie. with what reactors).
  405. // NOTE: panics if any switch fails to start.
  406. func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
  407. switches := make([]*Switch, n)
  408. for i := 0; i < n; i++ {
  409. switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
  410. }
  411. if err := StartSwitches(switches); err != nil {
  412. panic(err)
  413. }
  414. for i := 0; i < n; i++ {
  415. for j := i; j < n; j++ {
  416. connect(switches, i, j)
  417. }
  418. }
  419. return switches
  420. }
  421. var PanicOnAddPeerErr = false
  422. // Will connect switches i and j via net.Pipe()
  423. // Blocks until a conection is established.
  424. // NOTE: caller ensures i and j are within bounds
  425. func Connect2Switches(switches []*Switch, i, j int) {
  426. switchI := switches[i]
  427. switchJ := switches[j]
  428. c1, c2 := net.Pipe()
  429. doneCh := make(chan struct{})
  430. go func() {
  431. err := switchI.addPeerWithConnection(c1)
  432. if PanicOnAddPeerErr && err != nil {
  433. panic(err)
  434. }
  435. doneCh <- struct{}{}
  436. }()
  437. go func() {
  438. err := switchJ.addPeerWithConnection(c2)
  439. if PanicOnAddPeerErr && err != nil {
  440. panic(err)
  441. }
  442. doneCh <- struct{}{}
  443. }()
  444. <-doneCh
  445. <-doneCh
  446. }
  447. func StartSwitches(switches []*Switch) error {
  448. for _, s := range switches {
  449. _, err := s.Start() // start switch and reactors
  450. if err != nil {
  451. return err
  452. }
  453. }
  454. return nil
  455. }
  456. func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
  457. privKey := crypto.GenPrivKeyEd25519()
  458. // new switch, add reactors
  459. // TODO: let the config be passed in?
  460. s := initSwitch(i, NewSwitch(cfg))
  461. s.SetNodeInfo(&NodeInfo{
  462. PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  463. Moniker: cmn.Fmt("switch%d", i),
  464. Network: network,
  465. Version: version,
  466. RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  467. ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  468. })
  469. s.SetNodePrivKey(privKey)
  470. return s
  471. }
  472. func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
  473. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey)
  474. if err != nil {
  475. conn.Close()
  476. return err
  477. }
  478. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  479. if err = sw.AddPeer(peer); err != nil {
  480. conn.Close()
  481. return err
  482. }
  483. return nil
  484. }
  485. func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
  486. peer, err := newInboundPeerWithConfig(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
  487. if err != nil {
  488. conn.Close()
  489. return err
  490. }
  491. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  492. if err = sw.AddPeer(peer); err != nil {
  493. conn.Close()
  494. return err
  495. }
  496. return nil
  497. }