|
|
|
@ -16,7 +16,7 @@ import ( |
|
|
|
|
|
|
|
const ( |
|
|
|
// wait a random amount of time from this interval
|
|
|
|
// before dialing seeds or reconnecting to help prevent DoS
|
|
|
|
// before dialing peers or reconnecting to help prevent DoS
|
|
|
|
dialRandomizerIntervalMilliseconds = 3000 |
|
|
|
|
|
|
|
// repeatedly try to reconnect for a few minutes
|
|
|
|
@ -30,46 +30,17 @@ const ( |
|
|
|
reconnectBackOffBaseSeconds = 3 |
|
|
|
) |
|
|
|
|
|
|
|
type Reactor interface { |
|
|
|
cmn.Service // Start, Stop
|
|
|
|
|
|
|
|
SetSwitch(*Switch) |
|
|
|
GetChannels() []*ChannelDescriptor |
|
|
|
AddPeer(peer Peer) |
|
|
|
RemovePeer(peer Peer, reason interface{}) |
|
|
|
Receive(chID byte, peer Peer, msgBytes []byte) // CONTRACT: msgBytes are not nil
|
|
|
|
} |
|
|
|
|
|
|
|
//--------------------------------------
|
|
|
|
|
|
|
|
type BaseReactor struct { |
|
|
|
cmn.BaseService // Provides Start, Stop, .Quit
|
|
|
|
Switch *Switch |
|
|
|
} |
|
|
|
|
|
|
|
func NewBaseReactor(name string, impl Reactor) *BaseReactor { |
|
|
|
return &BaseReactor{ |
|
|
|
BaseService: *cmn.NewBaseService(nil, name, impl), |
|
|
|
Switch: nil, |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func (br *BaseReactor) SetSwitch(sw *Switch) { |
|
|
|
br.Switch = sw |
|
|
|
} |
|
|
|
func (_ *BaseReactor) GetChannels() []*ChannelDescriptor { return nil } |
|
|
|
func (_ *BaseReactor) AddPeer(peer Peer) {} |
|
|
|
func (_ *BaseReactor) RemovePeer(peer Peer, reason interface{}) {} |
|
|
|
func (_ *BaseReactor) Receive(chID byte, peer Peer, msgBytes []byte) {} |
|
|
|
var ( |
|
|
|
ErrSwitchDuplicatePeer = errors.New("Duplicate peer") |
|
|
|
ErrSwitchConnectToSelf = errors.New("Connect to self") |
|
|
|
) |
|
|
|
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
|
|
|
|
/* |
|
|
|
The `Switch` handles peer connections and exposes an API to receive incoming messages |
|
|
|
on `Reactors`. Each `Reactor` is responsible for handling incoming messages of one |
|
|
|
or more `Channels`. So while sending outgoing messages is typically performed on the peer, |
|
|
|
incoming messages are received on the reactor. |
|
|
|
*/ |
|
|
|
// `Switch` handles peer connections and exposes an API to receive incoming messages
|
|
|
|
// on `Reactors`. Each `Reactor` is responsible for handling incoming messages of one
|
|
|
|
// or more `Channels`. So while sending outgoing messages is typically performed on the peer,
|
|
|
|
// incoming messages are received on the reactor.
|
|
|
|
type Switch struct { |
|
|
|
cmn.BaseService |
|
|
|
|
|
|
|
@ -81,19 +52,15 @@ type Switch struct { |
|
|
|
reactorsByCh map[byte]Reactor |
|
|
|
peers *PeerSet |
|
|
|
dialing *cmn.CMap |
|
|
|
nodeInfo *NodeInfo // our node info
|
|
|
|
nodePrivKey crypto.PrivKeyEd25519 // our node privkey
|
|
|
|
nodeInfo NodeInfo // our node info
|
|
|
|
nodeKey *NodeKey // our node privkey
|
|
|
|
|
|
|
|
filterConnByAddr func(net.Addr) error |
|
|
|
filterConnByPubKey func(crypto.PubKeyEd25519) error |
|
|
|
filterConnByPubKey func(crypto.PubKey) error |
|
|
|
|
|
|
|
rng *rand.Rand // seed for randomizing dial times and orders
|
|
|
|
} |
|
|
|
|
|
|
|
var ( |
|
|
|
ErrSwitchDuplicatePeer = errors.New("Duplicate peer") |
|
|
|
) |
|
|
|
|
|
|
|
func NewSwitch(config *cfg.P2PConfig) *Switch { |
|
|
|
sw := &Switch{ |
|
|
|
config: config, |
|
|
|
@ -103,7 +70,6 @@ func NewSwitch(config *cfg.P2PConfig) *Switch { |
|
|
|
reactorsByCh: make(map[byte]Reactor), |
|
|
|
peers: NewPeerSet(), |
|
|
|
dialing: cmn.NewCMap(), |
|
|
|
nodeInfo: nil, |
|
|
|
} |
|
|
|
|
|
|
|
// Ensure we have a completely undeterministic PRNG. cmd.RandInt64() draws
|
|
|
|
@ -120,6 +86,9 @@ func NewSwitch(config *cfg.P2PConfig) *Switch { |
|
|
|
return sw |
|
|
|
} |
|
|
|
|
|
|
|
//---------------------------------------------------------------------
|
|
|
|
// Switch setup
|
|
|
|
|
|
|
|
// AddReactor adds the given reactor to the switch.
|
|
|
|
// NOTE: Not goroutine safe.
|
|
|
|
func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor { |
|
|
|
@ -171,26 +140,25 @@ func (sw *Switch) IsListening() bool { |
|
|
|
|
|
|
|
// SetNodeInfo sets the switch's NodeInfo for checking compatibility and handshaking with other nodes.
|
|
|
|
// NOTE: Not goroutine safe.
|
|
|
|
func (sw *Switch) SetNodeInfo(nodeInfo *NodeInfo) { |
|
|
|
func (sw *Switch) SetNodeInfo(nodeInfo NodeInfo) { |
|
|
|
sw.nodeInfo = nodeInfo |
|
|
|
} |
|
|
|
|
|
|
|
// NodeInfo returns the switch's NodeInfo.
|
|
|
|
// NOTE: Not goroutine safe.
|
|
|
|
func (sw *Switch) NodeInfo() *NodeInfo { |
|
|
|
func (sw *Switch) NodeInfo() NodeInfo { |
|
|
|
return sw.nodeInfo |
|
|
|
} |
|
|
|
|
|
|
|
// SetNodePrivKey sets the switch's private key for authenticated encryption.
|
|
|
|
// NOTE: Overwrites sw.nodeInfo.PubKey.
|
|
|
|
// SetNodeKey sets the switch's private key for authenticated encryption.
|
|
|
|
// NOTE: Not goroutine safe.
|
|
|
|
func (sw *Switch) SetNodePrivKey(nodePrivKey crypto.PrivKeyEd25519) { |
|
|
|
sw.nodePrivKey = nodePrivKey |
|
|
|
if sw.nodeInfo != nil { |
|
|
|
sw.nodeInfo.PubKey = nodePrivKey.PubKey().Unwrap().(crypto.PubKeyEd25519) |
|
|
|
} |
|
|
|
func (sw *Switch) SetNodeKey(nodeKey *NodeKey) { |
|
|
|
sw.nodeKey = nodeKey |
|
|
|
} |
|
|
|
|
|
|
|
//---------------------------------------------------------------------
|
|
|
|
// Service start/stop
|
|
|
|
|
|
|
|
// OnStart implements BaseService. It starts all the reactors, peers, and listeners.
|
|
|
|
func (sw *Switch) OnStart() error { |
|
|
|
// Start reactors
|
|
|
|
@ -226,172 +194,26 @@ func (sw *Switch) OnStop() { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// addPeer checks the given peer's validity, performs a handshake, and adds the
|
|
|
|
// peer to the switch and to all registered reactors.
|
|
|
|
// NOTE: This performs a blocking handshake before the peer is added.
|
|
|
|
// NOTE: If error is returned, caller is responsible for calling peer.CloseConn()
|
|
|
|
func (sw *Switch) addPeer(peer *peer) error { |
|
|
|
|
|
|
|
if err := sw.FilterConnByAddr(peer.Addr()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
// Avoid self
|
|
|
|
if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) { |
|
|
|
return errors.New("Ignoring connection from self") |
|
|
|
} |
|
|
|
|
|
|
|
// Check version, chain id
|
|
|
|
if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
// Check for duplicate peer
|
|
|
|
if sw.peers.Has(peer.Key()) { |
|
|
|
return ErrSwitchDuplicatePeer |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
// Start peer
|
|
|
|
if sw.IsRunning() { |
|
|
|
sw.startInitPeer(peer) |
|
|
|
} |
|
|
|
|
|
|
|
// Add the peer to .peers.
|
|
|
|
// We start it first so that a peer in the list is safe to Stop.
|
|
|
|
// It should not err since we already checked peers.Has().
|
|
|
|
if err := sw.peers.Add(peer); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
sw.Logger.Info("Added peer", "peer", peer) |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
// FilterConnByAddr returns an error if connecting to the given address is forbidden.
|
|
|
|
func (sw *Switch) FilterConnByAddr(addr net.Addr) error { |
|
|
|
if sw.filterConnByAddr != nil { |
|
|
|
return sw.filterConnByAddr(addr) |
|
|
|
} |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
// FilterConnByPubKey returns an error if connecting to the given public key is forbidden.
|
|
|
|
func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error { |
|
|
|
if sw.filterConnByPubKey != nil { |
|
|
|
return sw.filterConnByPubKey(pubkey) |
|
|
|
} |
|
|
|
return nil |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
// SetAddrFilter sets the function for filtering connections by address.
|
|
|
|
func (sw *Switch) SetAddrFilter(f func(net.Addr) error) { |
|
|
|
sw.filterConnByAddr = f |
|
|
|
} |
|
|
|
|
|
|
|
// SetPubKeyFilter sets the function for filtering connections by public key.
|
|
|
|
func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) { |
|
|
|
sw.filterConnByPubKey = f |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) startInitPeer(peer *peer) { |
|
|
|
err := peer.Start() // spawn send/recv routines
|
|
|
|
if err != nil { |
|
|
|
// Should never happen
|
|
|
|
sw.Logger.Error("Error starting peer", "peer", peer, "err", err) |
|
|
|
} |
|
|
|
//---------------------------------------------------------------------
|
|
|
|
// Peers
|
|
|
|
|
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
reactor.AddPeer(peer) |
|
|
|
} |
|
|
|
// Peers returns the set of peers that are connected to the switch.
|
|
|
|
func (sw *Switch) Peers() IPeerSet { |
|
|
|
return sw.peers |
|
|
|
} |
|
|
|
|
|
|
|
// DialSeeds dials a list of seeds asynchronously in random order.
|
|
|
|
func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error { |
|
|
|
netAddrs, errs := NewNetAddressStrings(seeds) |
|
|
|
for _, err := range errs { |
|
|
|
sw.Logger.Error("Error in seed's address", "err", err) |
|
|
|
} |
|
|
|
|
|
|
|
if addrBook != nil { |
|
|
|
// add seeds to `addrBook`
|
|
|
|
ourAddrS := sw.nodeInfo.ListenAddr |
|
|
|
ourAddr, _ := NewNetAddressString(ourAddrS) |
|
|
|
for _, netAddr := range netAddrs { |
|
|
|
// do not add ourselves
|
|
|
|
if netAddr.Equals(ourAddr) { |
|
|
|
continue |
|
|
|
} |
|
|
|
addrBook.AddAddress(netAddr, ourAddr) |
|
|
|
// NumPeers returns the count of outbound/inbound and outbound-dialing peers.
|
|
|
|
func (sw *Switch) NumPeers() (outbound, inbound, dialing int) { |
|
|
|
peers := sw.peers.List() |
|
|
|
for _, peer := range peers { |
|
|
|
if peer.IsOutbound() { |
|
|
|
outbound++ |
|
|
|
} else { |
|
|
|
inbound++ |
|
|
|
} |
|
|
|
addrBook.Save() |
|
|
|
} |
|
|
|
|
|
|
|
// permute the list, dial them in random order.
|
|
|
|
perm := sw.rng.Perm(len(netAddrs)) |
|
|
|
for i := 0; i < len(perm); i++ { |
|
|
|
go func(i int) { |
|
|
|
sw.randomSleep(0) |
|
|
|
j := perm[i] |
|
|
|
sw.dialSeed(netAddrs[j]) |
|
|
|
}(i) |
|
|
|
} |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
// sleep for interval plus some random amount of ms on [0, dialRandomizerIntervalMilliseconds]
|
|
|
|
func (sw *Switch) randomSleep(interval time.Duration) { |
|
|
|
r := time.Duration(sw.rng.Int63n(dialRandomizerIntervalMilliseconds)) * time.Millisecond |
|
|
|
time.Sleep(r + interval) |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) dialSeed(addr *NetAddress) { |
|
|
|
peer, err := sw.DialPeerWithAddress(addr, true) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Error("Error dialing seed", "err", err) |
|
|
|
} else { |
|
|
|
sw.Logger.Info("Connected to seed", "peer", peer) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// DialPeerWithAddress dials the given peer and runs sw.addPeer if it connects successfully.
|
|
|
|
// If `persistent == true`, the switch will always try to reconnect to this peer if the connection ever fails.
|
|
|
|
func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (Peer, error) { |
|
|
|
sw.dialing.Set(addr.IP.String(), addr) |
|
|
|
defer sw.dialing.Delete(addr.IP.String()) |
|
|
|
|
|
|
|
sw.Logger.Info("Dialing peer", "address", addr) |
|
|
|
peer, err := newOutboundPeer(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Error("Failed to dial peer", "address", addr, "err", err) |
|
|
|
return nil, err |
|
|
|
} |
|
|
|
peer.SetLogger(sw.Logger.With("peer", addr)) |
|
|
|
if persistent { |
|
|
|
peer.makePersistent() |
|
|
|
} |
|
|
|
err = sw.addPeer(peer) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Error("Failed to add peer", "address", addr, "err", err) |
|
|
|
peer.CloseConn() |
|
|
|
return nil, err |
|
|
|
} |
|
|
|
sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer) |
|
|
|
return peer, nil |
|
|
|
} |
|
|
|
|
|
|
|
// IsDialing returns true if the switch is currently dialing the given address.
|
|
|
|
func (sw *Switch) IsDialing(addr *NetAddress) bool { |
|
|
|
return sw.dialing.Has(addr.IP.String()) |
|
|
|
dialing = sw.dialing.Size() |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
// Broadcast runs a go routine for each attempted send, which will block
|
|
|
|
@ -411,25 +233,6 @@ func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool { |
|
|
|
return successChan |
|
|
|
} |
|
|
|
|
|
|
|
// NumPeers returns the count of outbound/inbound and outbound-dialing peers.
|
|
|
|
func (sw *Switch) NumPeers() (outbound, inbound, dialing int) { |
|
|
|
peers := sw.peers.List() |
|
|
|
for _, peer := range peers { |
|
|
|
if peer.IsOutbound() { |
|
|
|
outbound++ |
|
|
|
} else { |
|
|
|
inbound++ |
|
|
|
} |
|
|
|
} |
|
|
|
dialing = sw.dialing.Size() |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
// Peers returns the set of peers that are connected to the switch.
|
|
|
|
func (sw *Switch) Peers() IPeerSet { |
|
|
|
return sw.peers |
|
|
|
} |
|
|
|
|
|
|
|
// StopPeerForError disconnects from a peer due to external error.
|
|
|
|
// If the peer is persistent, it will attempt to reconnect.
|
|
|
|
// TODO: make record depending on reason.
|
|
|
|
@ -442,12 +245,27 @@ func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// StopPeerGracefully disconnects from a peer gracefully.
|
|
|
|
// TODO: handle graceful disconnects.
|
|
|
|
func (sw *Switch) StopPeerGracefully(peer Peer) { |
|
|
|
sw.Logger.Info("Stopping peer gracefully") |
|
|
|
sw.stopAndRemovePeer(peer, nil) |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) stopAndRemovePeer(peer Peer, reason interface{}) { |
|
|
|
sw.peers.Remove(peer) |
|
|
|
peer.Stop() |
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
reactor.RemovePeer(peer, reason) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// reconnectToPeer tries to reconnect to the peer, first repeatedly
|
|
|
|
// with a fixed interval, then with exponential backoff.
|
|
|
|
// If no success after all that, it stops trying, and leaves it
|
|
|
|
// to the PEX/Addrbook to find the peer again
|
|
|
|
func (sw *Switch) reconnectToPeer(peer Peer) { |
|
|
|
addr, _ := NewNetAddressString(peer.NodeInfo().RemoteAddr) |
|
|
|
netAddr := peer.NodeInfo().NetAddress() |
|
|
|
start := time.Now() |
|
|
|
sw.Logger.Info("Reconnecting to peer", "peer", peer) |
|
|
|
for i := 0; i < reconnectAttempts; i++ { |
|
|
|
@ -455,7 +273,7 @@ func (sw *Switch) reconnectToPeer(peer Peer) { |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
peer, err := sw.DialPeerWithAddress(addr, true) |
|
|
|
peer, err := sw.DialPeerWithAddress(netAddr, true) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err, "peer", peer) |
|
|
|
// sleep a set amount
|
|
|
|
@ -477,7 +295,7 @@ func (sw *Switch) reconnectToPeer(peer Peer) { |
|
|
|
// sleep an exponentially increasing amount
|
|
|
|
sleepIntervalSeconds := math.Pow(reconnectBackOffBaseSeconds, float64(i)) |
|
|
|
sw.randomSleep(time.Duration(sleepIntervalSeconds) * time.Second) |
|
|
|
peer, err := sw.DialPeerWithAddress(addr, true) |
|
|
|
peer, err := sw.DialPeerWithAddress(netAddr, true) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err, "peer", peer) |
|
|
|
continue |
|
|
|
@ -489,21 +307,97 @@ func (sw *Switch) reconnectToPeer(peer Peer) { |
|
|
|
sw.Logger.Error("Failed to reconnect to peer. Giving up", "peer", peer, "elapsed", time.Since(start)) |
|
|
|
} |
|
|
|
|
|
|
|
// StopPeerGracefully disconnects from a peer gracefully.
|
|
|
|
// TODO: handle graceful disconnects.
|
|
|
|
func (sw *Switch) StopPeerGracefully(peer Peer) { |
|
|
|
sw.Logger.Info("Stopping peer gracefully") |
|
|
|
sw.stopAndRemovePeer(peer, nil) |
|
|
|
//---------------------------------------------------------------------
|
|
|
|
// Dialing
|
|
|
|
|
|
|
|
// IsDialing returns true if the switch is currently dialing the given ID.
|
|
|
|
func (sw *Switch) IsDialing(id ID) bool { |
|
|
|
return sw.dialing.Has(string(id)) |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) stopAndRemovePeer(peer Peer, reason interface{}) { |
|
|
|
sw.peers.Remove(peer) |
|
|
|
peer.Stop() |
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
reactor.RemovePeer(peer, reason) |
|
|
|
// DialPeersAsync dials a list of peers asynchronously in random order (optionally, making them persistent).
|
|
|
|
func (sw *Switch) DialPeersAsync(addrBook *AddrBook, peers []string, persistent bool) error { |
|
|
|
netAddrs, errs := NewNetAddressStrings(peers) |
|
|
|
for _, err := range errs { |
|
|
|
sw.Logger.Error("Error in peer's address", "err", err) |
|
|
|
} |
|
|
|
|
|
|
|
if addrBook != nil { |
|
|
|
// add peers to `addrBook`
|
|
|
|
ourAddr := sw.nodeInfo.NetAddress() |
|
|
|
for _, netAddr := range netAddrs { |
|
|
|
// do not add our address or ID
|
|
|
|
if netAddr.Same(ourAddr) { |
|
|
|
continue |
|
|
|
} |
|
|
|
addrBook.AddAddress(netAddr, ourAddr) |
|
|
|
} |
|
|
|
addrBook.Save() |
|
|
|
} |
|
|
|
|
|
|
|
// permute the list, dial them in random order.
|
|
|
|
perm := sw.rng.Perm(len(netAddrs)) |
|
|
|
for i := 0; i < len(perm); i++ { |
|
|
|
go func(i int) { |
|
|
|
sw.randomSleep(0) |
|
|
|
j := perm[i] |
|
|
|
peer, err := sw.DialPeerWithAddress(netAddrs[j], persistent) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Error("Error dialing peer", "err", err) |
|
|
|
} else { |
|
|
|
sw.Logger.Info("Connected to peer", "peer", peer) |
|
|
|
} |
|
|
|
}(i) |
|
|
|
} |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
// DialPeerWithAddress dials the given peer and runs sw.addPeer if it connects and authenticates successfully.
|
|
|
|
// If `persistent == true`, the switch will always try to reconnect to this peer if the connection ever fails.
|
|
|
|
func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (Peer, error) { |
|
|
|
sw.dialing.Set(string(addr.ID), addr) |
|
|
|
defer sw.dialing.Delete(string(addr.ID)) |
|
|
|
return sw.addOutboundPeerWithConfig(addr, sw.peerConfig, persistent) |
|
|
|
} |
|
|
|
|
|
|
|
// sleep for interval plus some random amount of ms on [0, dialRandomizerIntervalMilliseconds]
|
|
|
|
func (sw *Switch) randomSleep(interval time.Duration) { |
|
|
|
r := time.Duration(sw.rng.Int63n(dialRandomizerIntervalMilliseconds)) * time.Millisecond |
|
|
|
time.Sleep(r + interval) |
|
|
|
} |
|
|
|
|
|
|
|
//------------------------------------------------------------------------------------
|
|
|
|
// Connection filtering
|
|
|
|
|
|
|
|
// FilterConnByAddr returns an error if connecting to the given address is forbidden.
|
|
|
|
func (sw *Switch) FilterConnByAddr(addr net.Addr) error { |
|
|
|
if sw.filterConnByAddr != nil { |
|
|
|
return sw.filterConnByAddr(addr) |
|
|
|
} |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
// FilterConnByPubKey returns an error if connecting to the given public key is forbidden.
|
|
|
|
func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKey) error { |
|
|
|
if sw.filterConnByPubKey != nil { |
|
|
|
return sw.filterConnByPubKey(pubkey) |
|
|
|
} |
|
|
|
return nil |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
// SetAddrFilter sets the function for filtering connections by address.
|
|
|
|
func (sw *Switch) SetAddrFilter(f func(net.Addr) error) { |
|
|
|
sw.filterConnByAddr = f |
|
|
|
} |
|
|
|
|
|
|
|
// SetPubKeyFilter sets the function for filtering connections by public key.
|
|
|
|
func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKey) error) { |
|
|
|
sw.filterConnByPubKey = f |
|
|
|
} |
|
|
|
|
|
|
|
//------------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
func (sw *Switch) listenerRoutine(l Listener) { |
|
|
|
for { |
|
|
|
inConn, ok := <-l.Connections() |
|
|
|
@ -519,131 +413,124 @@ func (sw *Switch) listenerRoutine(l Listener) { |
|
|
|
} |
|
|
|
|
|
|
|
// New inbound connection!
|
|
|
|
err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig) |
|
|
|
err := sw.addInboundPeerWithConfig(inConn, sw.peerConfig) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Info("Ignoring inbound connection: error while adding peer", "address", inConn.RemoteAddr().String(), "err", err) |
|
|
|
continue |
|
|
|
} |
|
|
|
|
|
|
|
// NOTE: We don't yet have the listening port of the
|
|
|
|
// remote (if they have a listener at all).
|
|
|
|
// The peerHandshake will handle that.
|
|
|
|
} |
|
|
|
|
|
|
|
// cleanup
|
|
|
|
} |
|
|
|
|
|
|
|
//------------------------------------------------------------------
|
|
|
|
// Connects switches via arbitrary net.Conn. Used for testing.
|
|
|
|
|
|
|
|
// MakeConnectedSwitches returns n switches, connected according to the connect func.
|
|
|
|
// If connect==Connect2Switches, the switches will be fully connected.
|
|
|
|
// initSwitch defines how the i'th switch should be initialized (ie. with what reactors).
|
|
|
|
// NOTE: panics if any switch fails to start.
|
|
|
|
func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch { |
|
|
|
switches := make([]*Switch, n) |
|
|
|
for i := 0; i < n; i++ { |
|
|
|
switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch) |
|
|
|
func (sw *Switch) addInboundPeerWithConfig(conn net.Conn, config *PeerConfig) error { |
|
|
|
peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodeKey.PrivKey, config) |
|
|
|
if err != nil { |
|
|
|
peer.CloseConn() |
|
|
|
return err |
|
|
|
} |
|
|
|
peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr())) |
|
|
|
if err = sw.addPeer(peer); err != nil { |
|
|
|
peer.CloseConn() |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
if err := StartSwitches(switches); err != nil { |
|
|
|
panic(err) |
|
|
|
// dial the peer; make secret connection; authenticate against the dialed ID;
|
|
|
|
// add the peer.
|
|
|
|
func (sw *Switch) addOutboundPeerWithConfig(addr *NetAddress, config *PeerConfig, persistent bool) (Peer, error) { |
|
|
|
sw.Logger.Info("Dialing peer", "address", addr) |
|
|
|
peer, err := newOutboundPeer(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodeKey.PrivKey, config, persistent) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Error("Failed to dial peer", "address", addr, "err", err) |
|
|
|
return nil, err |
|
|
|
} |
|
|
|
peer.SetLogger(sw.Logger.With("peer", addr)) |
|
|
|
|
|
|
|
for i := 0; i < n; i++ { |
|
|
|
for j := i + 1; j < n; j++ { |
|
|
|
connect(switches, i, j) |
|
|
|
} |
|
|
|
// authenticate peer
|
|
|
|
if addr.ID == "" { |
|
|
|
peer.Logger.Info("Dialed peer with unknown ID - unable to authenticate", "addr", addr) |
|
|
|
} else if addr.ID != peer.ID() { |
|
|
|
peer.CloseConn() |
|
|
|
return nil, fmt.Errorf("Failed to authenticate peer %v. Connected to peer with ID %s", addr, peer.ID()) |
|
|
|
} |
|
|
|
|
|
|
|
return switches |
|
|
|
err = sw.addPeer(peer) |
|
|
|
if err != nil { |
|
|
|
sw.Logger.Error("Failed to add peer", "address", addr, "err", err) |
|
|
|
peer.CloseConn() |
|
|
|
return nil, err |
|
|
|
} |
|
|
|
sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer) |
|
|
|
return peer, nil |
|
|
|
} |
|
|
|
|
|
|
|
// Connect2Switches will connect switches i and j via net.Pipe().
|
|
|
|
// Blocks until a connection is established.
|
|
|
|
// NOTE: caller ensures i and j are within bounds.
|
|
|
|
func Connect2Switches(switches []*Switch, i, j int) { |
|
|
|
switchI := switches[i] |
|
|
|
switchJ := switches[j] |
|
|
|
c1, c2 := netPipe() |
|
|
|
doneCh := make(chan struct{}) |
|
|
|
go func() { |
|
|
|
err := switchI.addPeerWithConnection(c1) |
|
|
|
if err != nil { |
|
|
|
panic(err) |
|
|
|
} |
|
|
|
doneCh <- struct{}{} |
|
|
|
}() |
|
|
|
go func() { |
|
|
|
err := switchJ.addPeerWithConnection(c2) |
|
|
|
if err != nil { |
|
|
|
panic(err) |
|
|
|
} |
|
|
|
doneCh <- struct{}{} |
|
|
|
}() |
|
|
|
<-doneCh |
|
|
|
<-doneCh |
|
|
|
} |
|
|
|
// addPeer performs the Tendermint P2P handshake with a peer
|
|
|
|
// that already has a SecretConnection. If all goes well,
|
|
|
|
// it starts the peer and adds it to the switch.
|
|
|
|
// NOTE: This performs a blocking handshake before the peer is added.
|
|
|
|
// NOTE: If error is returned, caller is responsible for calling peer.CloseConn()
|
|
|
|
func (sw *Switch) addPeer(peer *peer) error { |
|
|
|
// Avoid self
|
|
|
|
if sw.nodeKey.ID() == peer.ID() { |
|
|
|
return ErrSwitchConnectToSelf |
|
|
|
} |
|
|
|
|
|
|
|
// Avoid duplicate
|
|
|
|
if sw.peers.Has(peer.ID()) { |
|
|
|
return ErrSwitchDuplicatePeer |
|
|
|
|
|
|
|
// StartSwitches calls sw.Start() for each given switch.
|
|
|
|
// It returns the first encountered error.
|
|
|
|
func StartSwitches(switches []*Switch) error { |
|
|
|
for _, s := range switches { |
|
|
|
err := s.Start() // start switch and reactors
|
|
|
|
if err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
} |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch { |
|
|
|
privKey := crypto.GenPrivKeyEd25519() |
|
|
|
// new switch, add reactors
|
|
|
|
// TODO: let the config be passed in?
|
|
|
|
s := initSwitch(i, NewSwitch(cfg)) |
|
|
|
s.SetNodeInfo(&NodeInfo{ |
|
|
|
PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519), |
|
|
|
Moniker: cmn.Fmt("switch%d", i), |
|
|
|
Network: network, |
|
|
|
Version: version, |
|
|
|
RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023), |
|
|
|
ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023), |
|
|
|
}) |
|
|
|
s.SetNodePrivKey(privKey) |
|
|
|
return s |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) addPeerWithConnection(conn net.Conn) error { |
|
|
|
peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig) |
|
|
|
if err != nil { |
|
|
|
if err := conn.Close(); err != nil { |
|
|
|
sw.Logger.Error("Error closing connection", "err", err) |
|
|
|
} |
|
|
|
// Filter peer against white list
|
|
|
|
if err := sw.FilterConnByAddr(peer.Addr()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr())) |
|
|
|
if err = sw.addPeer(peer); err != nil { |
|
|
|
peer.CloseConn() |
|
|
|
if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
return nil |
|
|
|
} |
|
|
|
// Exchange NodeInfo with the peer
|
|
|
|
if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error { |
|
|
|
peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config) |
|
|
|
if err != nil { |
|
|
|
if err := conn.Close(); err != nil { |
|
|
|
sw.Logger.Error("Error closing connection", "err", err) |
|
|
|
} |
|
|
|
// Validate the peers nodeInfo against the pubkey
|
|
|
|
if err := peer.NodeInfo().Validate(peer.PubKey()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr())) |
|
|
|
if err = sw.addPeer(peer); err != nil { |
|
|
|
peer.CloseConn() |
|
|
|
|
|
|
|
// Check version, chain id
|
|
|
|
if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo()); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
// All good. Start peer
|
|
|
|
if sw.IsRunning() { |
|
|
|
sw.startInitPeer(peer) |
|
|
|
} |
|
|
|
|
|
|
|
// Add the peer to .peers.
|
|
|
|
// We start it first so that a peer in the list is safe to Stop.
|
|
|
|
// It should not err since we already checked peers.Has().
|
|
|
|
if err := sw.peers.Add(peer); err != nil { |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
sw.Logger.Info("Added peer", "peer", peer) |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
func (sw *Switch) startInitPeer(peer *peer) { |
|
|
|
err := peer.Start() // spawn send/recv routines
|
|
|
|
if err != nil { |
|
|
|
// Should never happen
|
|
|
|
sw.Logger.Error("Error starting peer", "peer", peer, "err", err) |
|
|
|
} |
|
|
|
|
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
reactor.AddPeer(peer) |
|
|
|
} |
|
|
|
} |
Ethan Buchman
7 years ago
GitHub