10 cfg "github.com/bytom/config"
11 log "github.com/sirupsen/logrus"
12 crypto "github.com/tendermint/go-crypto"
13 cmn "github.com/tendermint/tmlibs/common"
17 reconnectAttempts = 30
18 reconnectInterval = 3 * time.Second
21 type Reactor interface {
22 cmn.Service // Start, Stop
25 GetChannels() []*ChannelDescriptor
27 RemovePeer(peer *Peer, reason interface{})
28 Receive(chID byte, peer *Peer, msgBytes []byte)
31 //--------------------------------------
33 type BaseReactor struct {
34 cmn.BaseService // Provides Start, Stop, .Quit
38 func NewBaseReactor(name string, impl Reactor) *BaseReactor {
40 BaseService: *cmn.NewBaseService(nil, name, impl),
45 func (br *BaseReactor) SetSwitch(sw *Switch) {
48 func (_ *BaseReactor) GetChannels() []*ChannelDescriptor { return nil }
49 func (_ *BaseReactor) AddPeer(peer *Peer) {}
50 func (_ *BaseReactor) RemovePeer(peer *Peer, reason interface{}) {}
51 func (_ *BaseReactor) Receive(chID byte, peer *Peer, msgBytes []byte) {}
53 //-----------------------------------------------------------------------------
56 The `Switch` handles peer connections and exposes an API to receive incoming messages
57 on `Reactors`. Each `Reactor` is responsible for handling incoming messages of one
58 or more `Channels`. So while sending outgoing messages is typically performed on the peer,
59 incoming messages are received on the reactor.
65 peerConfig *PeerConfig
67 reactors map[string]Reactor
68 chDescs []*ChannelDescriptor
69 reactorsByCh map[byte]Reactor
72 nodeInfo *NodeInfo // our node info
73 nodePrivKey crypto.PrivKeyEd25519 // our node privkey
75 filterConnByAddr func(net.Addr) error
76 filterConnByPubKey func(crypto.PubKeyEd25519) error
80 ErrSwitchDuplicatePeer = errors.New("Duplicate peer")
83 func NewSwitch(config *cfg.P2PConfig) *Switch {
86 peerConfig: DefaultPeerConfig(),
87 reactors: make(map[string]Reactor),
88 chDescs: make([]*ChannelDescriptor, 0),
89 reactorsByCh: make(map[byte]Reactor),
91 dialing: cmn.NewCMap(),
94 sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
98 // Not goroutine safe.
99 func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
100 // Validate the reactor.
101 // No two reactors can share the same channel.
102 reactorChannels := reactor.GetChannels()
103 for _, chDesc := range reactorChannels {
105 if sw.reactorsByCh[chID] != nil {
106 cmn.PanicSanity(fmt.Sprintf("Channel %X has multiple reactors %v & %v", chID, sw.reactorsByCh[chID], reactor))
108 sw.chDescs = append(sw.chDescs, chDesc)
109 sw.reactorsByCh[chID] = reactor
111 sw.reactors[name] = reactor
112 reactor.SetSwitch(sw)
116 // Not goroutine safe.
117 func (sw *Switch) Reactors() map[string]Reactor {
121 // Not goroutine safe.
122 func (sw *Switch) Reactor(name string) Reactor {
123 return sw.reactors[name]
126 // Not goroutine safe.
127 func (sw *Switch) AddListener(l Listener) {
128 sw.listeners = append(sw.listeners, l)
131 // Not goroutine safe.
132 func (sw *Switch) Listeners() []Listener {
136 // Not goroutine safe.
137 func (sw *Switch) IsListening() bool {
138 return len(sw.listeners) > 0
141 // Not goroutine safe.
142 func (sw *Switch) SetNodeInfo(nodeInfo *NodeInfo) {
143 sw.nodeInfo = nodeInfo
146 // Not goroutine safe.
147 func (sw *Switch) NodeInfo() *NodeInfo {
151 // Not goroutine safe.
152 // NOTE: Overwrites sw.nodeInfo.PubKey
153 func (sw *Switch) SetNodePrivKey(nodePrivKey crypto.PrivKeyEd25519) {
154 sw.nodePrivKey = nodePrivKey
155 if sw.nodeInfo != nil {
156 sw.nodeInfo.PubKey = nodePrivKey.PubKey().Unwrap().(crypto.PubKeyEd25519)
160 // Switch.Start() starts all the reactors, peers, and listeners.
161 func (sw *Switch) OnStart() error {
162 sw.BaseService.OnStart()
164 for _, reactor := range sw.reactors {
165 _, err := reactor.Start()
171 for _, peer := range sw.peers.List() {
172 sw.startInitPeer(peer)
175 for _, listener := range sw.listeners {
176 go sw.listenerRoutine(listener)
181 func (sw *Switch) OnStop() {
182 sw.BaseService.OnStop()
184 for _, listener := range sw.listeners {
189 for _, peer := range sw.peers.List() {
191 sw.peers.Remove(peer)
194 for _, reactor := range sw.reactors {
199 // NOTE: This performs a blocking handshake before the peer is added.
200 // CONTRACT: If error is returned, peer is nil, and conn is immediately closed.
201 func (sw *Switch) AddPeer(peer *Peer) error {
202 if err := sw.FilterConnByAddr(peer.Addr()); err != nil {
206 if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil {
210 if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil {
215 if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
216 return errors.New("Ignoring connection from self")
219 // Check version, chain id
220 if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo); err != nil {
224 // Check for duplicate peer
225 if sw.peers.Has(peer.Key) {
226 return ErrSwitchDuplicatePeer
232 sw.startInitPeer(peer)
235 // Add the peer to .peers.
236 // We start it first so that a peer in the list is safe to Stop.
237 // It should not err since we already checked peers.Has()
238 if err := sw.peers.Add(peer); err != nil {
242 log.WithField("peer", peer).Info("Added peer")
246 func (sw *Switch) FilterConnByAddr(addr net.Addr) error {
247 if sw.filterConnByAddr != nil {
248 return sw.filterConnByAddr(addr)
253 func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error {
254 if sw.filterConnByPubKey != nil {
255 return sw.filterConnByPubKey(pubkey)
261 func (sw *Switch) SetAddrFilter(f func(net.Addr) error) {
262 sw.filterConnByAddr = f
265 func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) {
266 sw.filterConnByPubKey = f
269 func (sw *Switch) startInitPeer(peer *Peer) {
270 peer.Start() // spawn send/recv routines
271 for _, reactor := range sw.reactors {
272 reactor.AddPeer(peer)
276 // Dial a list of seeds asynchronously in random order
277 func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error {
279 netAddrs, err := NewNetAddressStrings(seeds)
285 // add seeds to `addrBook`
286 ourAddrS := sw.nodeInfo.ListenAddr
287 ourAddr, _ := NewNetAddressString(ourAddrS)
288 for _, netAddr := range netAddrs {
289 // do not add ourselves
290 if netAddr.Equals(ourAddr) {
293 addrBook.AddAddress(netAddr, ourAddr)
298 // permute the list, dial them in random order.
299 perm := rand.Perm(len(netAddrs))
300 for i := 0; i < len(perm); i++ {
302 time.Sleep(time.Duration(rand.Int63n(3000)) * time.Millisecond)
304 sw.dialSeed(netAddrs[j])
310 func (sw *Switch) dialSeed(addr *NetAddress) {
311 peer, err := sw.DialPeerWithAddress(addr, true)
313 log.WithField("error", err).Error("Error dialing seed")
315 log.WithField("peer", peer).Info("Connected to seed")
319 func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (*Peer, error) {
320 sw.dialing.Set(addr.IP.String(), addr)
321 defer sw.dialing.Delete(addr.IP.String())
323 log.WithField("address", addr).Info("Dialing peer")
324 peer, err := newOutboundPeerWithConfig(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
326 log.WithFields(log.Fields{
329 }).Info("Failed to dial peer")
332 peer.SetLogger(sw.Logger.With("peer", addr))
334 peer.makePersistent()
336 err = sw.AddPeer(peer)
338 log.WithFields(log.Fields{
341 }).Info("Failed to add peer")
345 log.WithFields(log.Fields{
348 }).Info("Dialed and added peer")
352 func (sw *Switch) IsDialing(addr *NetAddress) bool {
353 return sw.dialing.Has(addr.IP.String())
356 // Broadcast runs a go routine for each attempted send, which will block
357 // trying to send for defaultSendTimeoutSeconds. Returns a channel
358 // which receives success values for each attempted send (false if times out)
359 // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
360 func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
361 successChan := make(chan bool, len(sw.peers.List()))
362 log.WithFields(log.Fields{
365 }).Debug("Broadcast")
366 for _, peer := range sw.peers.List() {
367 go func(peer *Peer) {
368 success := peer.Send(chID, msg)
369 successChan <- success
375 // Returns the count of outbound/inbound and outbound-dialing peers.
376 func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
377 peers := sw.peers.List()
378 for _, peer := range peers {
385 dialing = sw.dialing.Size()
389 func (sw *Switch) Peers() IPeerSet {
393 // Disconnect from a peer due to external error, retry if it is a persistent peer.
394 // TODO: make record depending on reason.
395 func (sw *Switch) StopPeerForError(peer *Peer, reason interface{}) {
396 addr := NewNetAddress(peer.Addr())
397 log.WithFields(log.Fields{
400 }).Info("Stopping peer due to error")
401 sw.stopAndRemovePeer(peer, reason)
403 if peer.IsPersistent() {
405 log.WithField("peer", peer).Info("Reconnecting to peer")
406 for i := 1; i < reconnectAttempts; i++ {
411 peer, err := sw.DialPeerWithAddress(addr, true)
413 if i == reconnectAttempts {
414 log.WithFields(log.Fields{
417 }).Info("Error reconnecting to peer. Giving up")
420 log.WithFields(log.Fields{
423 }).Info("Error reconnecting to peer. Trying again")
424 time.Sleep(reconnectInterval)
428 log.WithField("peer", peer).Info("Reconnected to peer")
435 // Disconnect from a peer gracefully.
436 // TODO: handle graceful disconnects.
437 func (sw *Switch) StopPeerGracefully(peer *Peer) {
438 log.Info("Stopping peer gracefully")
439 sw.stopAndRemovePeer(peer, nil)
442 func (sw *Switch) stopAndRemovePeer(peer *Peer, reason interface{}) {
443 sw.peers.Remove(peer)
445 for _, reactor := range sw.reactors {
446 reactor.RemovePeer(peer, reason)
450 func (sw *Switch) listenerRoutine(l Listener) {
452 inConn, ok := <-l.Connections()
457 // ignore connection if we already have enough
458 maxPeers := sw.config.MaxNumPeers
459 if maxPeers <= sw.peers.Size() {
460 log.WithFields(log.Fields{
461 "address": inConn.RemoteAddr().String(),
462 "numPeers": sw.peers.Size(),
464 }).Info("Ignoring inbound connection: already have enough peers")
468 // New inbound connection!
469 err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
471 log.WithFields(log.Fields{
472 "address": inConn.RemoteAddr().String(),
474 }).Info("Ignoring inbound connection: error while adding peer")
478 // NOTE: We don't yet have the listening port of the
479 // remote (if they have a listener at all).
480 // The peerHandshake will handle that
486 //-----------------------------------------------------------------------------
488 type SwitchEventNewPeer struct {
492 type SwitchEventDonePeer struct {
497 //------------------------------------------------------------------
498 // Switches connected via arbitrary net.Conn; useful for testing
500 // Returns n switches, connected according to the connect func.
501 // If connect==Connect2Switches, the switches will be fully connected.
502 // initSwitch defines how the ith switch should be initialized (ie. with what reactors).
503 // NOTE: panics if any switch fails to start.
504 func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
505 switches := make([]*Switch, n)
506 for i := 0; i < n; i++ {
507 switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
510 if err := StartSwitches(switches); err != nil {
514 for i := 0; i < n; i++ {
515 for j := i; j < n; j++ {
516 connect(switches, i, j)
523 var PanicOnAddPeerErr = false
525 // Will connect switches i and j via net.Pipe()
526 // Blocks until a conection is established.
527 // NOTE: caller ensures i and j are within bounds
528 func Connect2Switches(switches []*Switch, i, j int) {
529 switchI := switches[i]
530 switchJ := switches[j]
532 doneCh := make(chan struct{})
534 err := switchI.addPeerWithConnection(c1)
535 if PanicOnAddPeerErr && err != nil {
541 err := switchJ.addPeerWithConnection(c2)
542 if PanicOnAddPeerErr && err != nil {
551 func StartSwitches(switches []*Switch) error {
552 for _, s := range switches {
553 _, err := s.Start() // start switch and reactors
561 func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
562 privKey := crypto.GenPrivKeyEd25519()
563 // new switch, add reactors
564 // TODO: let the config be passed in?
565 s := initSwitch(i, NewSwitch(cfg))
566 s.SetNodeInfo(&NodeInfo{
567 PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
568 Moniker: cmn.Fmt("switch%d", i),
571 RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
572 ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
574 s.SetNodePrivKey(privKey)
578 func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
579 peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey)
584 peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
585 if err = sw.AddPeer(peer); err != nil {
593 func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
594 peer, err := newInboundPeerWithConfig(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
599 peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
600 if err = sw.AddPeer(peer); err != nil {