Add RemoteAddr and ListenAddr to NodeInfo; Refactor IPRange logic
This commit is contained in:
parent
c37e25e76b
commit
432a37857d
|
@ -0,0 +1,29 @@
|
|||
package p2p
|
||||
|
||||
import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// TODO Test
|
||||
func AddToIPRangeCounts(counts map[string]int, ip string) map[string]int {
|
||||
changes := make(map[string]int)
|
||||
ipParts := strings.Split(ip, ":")
|
||||
for i := 1; i < len(ipParts); i++ {
|
||||
prefix := strings.Join(ipParts[:i], ":")
|
||||
counts[prefix] += 1
|
||||
changes[prefix] = counts[prefix]
|
||||
}
|
||||
return changes
|
||||
}
|
||||
|
||||
// TODO Test
|
||||
func CheckIPRangeCounts(counts map[string]int, limits []int) bool {
|
||||
for prefix, count := range counts {
|
||||
ipParts := strings.Split(prefix, ":")
|
||||
numParts := len(ipParts)
|
||||
if limits[numParts] < count {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
1
peer.go
1
peer.go
|
@ -42,6 +42,7 @@ func peerHandshake(conn net.Conn, ourNodeInfo *NodeInfo) (*NodeInfo, error) {
|
|||
if err2 != nil {
|
||||
return nil, err2
|
||||
}
|
||||
peerNodeInfo.RemoteAddr = conn.RemoteAddr().String()
|
||||
return peerNodeInfo, nil
|
||||
}
|
||||
|
||||
|
|
122
peer_set.go
122
peer_set.go
|
@ -1,8 +1,6 @@
|
|||
package p2p
|
||||
|
||||
import (
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
|
@ -16,18 +14,13 @@ type IPeerSet interface {
|
|||
|
||||
//-----------------------------------------------------------------------------
|
||||
|
||||
var (
|
||||
maxPeersPerIPRange = [4]int{11, 7, 5, 3} // ...
|
||||
)
|
||||
|
||||
// PeerSet is a special structure for keeping a table of peers.
|
||||
// Iteration over the peers is super fast and thread-safe.
|
||||
// We also track how many peers per IP range and avoid too many
|
||||
type PeerSet struct {
|
||||
mtx sync.Mutex
|
||||
lookup map[string]*peerSetItem
|
||||
list []*Peer
|
||||
connectedIPs *nestedCounter
|
||||
mtx sync.Mutex
|
||||
lookup map[string]*peerSetItem
|
||||
list []*Peer
|
||||
}
|
||||
|
||||
type peerSetItem struct {
|
||||
|
@ -37,9 +30,8 @@ type peerSetItem struct {
|
|||
|
||||
func NewPeerSet() *PeerSet {
|
||||
return &PeerSet{
|
||||
lookup: make(map[string]*peerSetItem),
|
||||
list: make([]*Peer, 0, 256),
|
||||
connectedIPs: NewNestedCounter(),
|
||||
lookup: make(map[string]*peerSetItem),
|
||||
list: make([]*Peer, 0, 256),
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -52,12 +44,6 @@ func (ps *PeerSet) Add(peer *Peer) error {
|
|||
return ErrSwitchDuplicatePeer
|
||||
}
|
||||
|
||||
// ensure we havent maxed out connections for the peer's IP range yet
|
||||
// and update the IP range counters
|
||||
if !ps.incrIPRangeCounts(peer.Host()) {
|
||||
return ErrSwitchMaxPeersPerIPRange
|
||||
}
|
||||
|
||||
index := len(ps.list)
|
||||
// Appending is safe even with other goroutines
|
||||
// iterating over the ps.list slice.
|
||||
|
@ -92,9 +78,6 @@ func (ps *PeerSet) Remove(peer *Peer) {
|
|||
return
|
||||
}
|
||||
|
||||
// update the IP range counters
|
||||
ps.decrIPRangeCounts(peer.Host())
|
||||
|
||||
index := item.index
|
||||
// Copy the list but without the last element.
|
||||
// (we must copy because we're mutating the list)
|
||||
|
@ -130,98 +113,3 @@ func (ps *PeerSet) List() []*Peer {
|
|||
defer ps.mtx.Unlock()
|
||||
return ps.list
|
||||
}
|
||||
|
||||
//-----------------------------------------------------------------------------
|
||||
// track the number of IPs we're connected to for each IP address range
|
||||
|
||||
// forms an IP address hierarchy tree with counts
|
||||
// the struct itself is not thread safe and should always only be accessed with the ps.mtx locked
|
||||
type nestedCounter struct {
|
||||
count int
|
||||
children map[string]*nestedCounter
|
||||
}
|
||||
|
||||
func NewNestedCounter() *nestedCounter {
|
||||
nc := new(nestedCounter)
|
||||
nc.children = make(map[string]*nestedCounter)
|
||||
return nc
|
||||
}
|
||||
|
||||
// Check if we have too many IPs in the IP range of the incoming connection
|
||||
// Thread safe
|
||||
func (ps *PeerSet) HasMaxForIPRange(conn net.Conn) (ok bool) {
|
||||
ps.mtx.Lock()
|
||||
defer ps.mtx.Unlock()
|
||||
ip, _, _ := net.SplitHostPort(conn.RemoteAddr().String())
|
||||
ipBytes := strings.Split(ip, ".")
|
||||
|
||||
c := ps.connectedIPs
|
||||
for i, ipByte := range ipBytes {
|
||||
if c, ok = c.children[ipByte]; !ok {
|
||||
return false
|
||||
}
|
||||
if maxPeersPerIPRange[i] <= c.count {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Increments counts for this address' IP range
|
||||
// Returns false if we already have enough connections
|
||||
// Not thread safe (only called by ps.Add())
|
||||
func (ps *PeerSet) incrIPRangeCounts(address string) bool {
|
||||
addrParts := strings.Split(address, ".")
|
||||
|
||||
c := ps.connectedIPs
|
||||
return incrNestedCounters(c, addrParts, 0)
|
||||
}
|
||||
|
||||
// Recursively descend the IP hierarchy, checking if we have
|
||||
// max peers for each range and incrementing if not.
|
||||
// Returns false if incr failed because max peers reached for some range counter.
|
||||
func incrNestedCounters(c *nestedCounter, ipBytes []string, index int) bool {
|
||||
ipByte := ipBytes[index]
|
||||
child := c.children[ipByte]
|
||||
if child == nil {
|
||||
child = NewNestedCounter()
|
||||
c.children[ipByte] = child
|
||||
}
|
||||
if index+1 < len(ipBytes) {
|
||||
if !incrNestedCounters(child, ipBytes, index+1) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
if maxPeersPerIPRange[index] <= child.count {
|
||||
return false
|
||||
} else {
|
||||
child.count += 1
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// Decrement counts for this address' IP range
|
||||
func (ps *PeerSet) decrIPRangeCounts(address string) {
|
||||
addrParts := strings.Split(address, ".")
|
||||
|
||||
c := ps.connectedIPs
|
||||
decrNestedCounters(c, addrParts, 0)
|
||||
}
|
||||
|
||||
// Recursively descend the IP hierarchy, decrementing by one.
|
||||
// If the counter is zero, deletes the child.
|
||||
func decrNestedCounters(c *nestedCounter, ipBytes []string, index int) {
|
||||
ipByte := ipBytes[index]
|
||||
child := c.children[ipByte]
|
||||
if child == nil {
|
||||
log.Error("p2p/peer_set decrNestedCounters encountered a missing child counter")
|
||||
return
|
||||
}
|
||||
if index+1 < len(ipBytes) {
|
||||
decrNestedCounters(child, ipBytes, index+1)
|
||||
}
|
||||
child.count -= 1
|
||||
if child.count <= 0 {
|
||||
delete(c.children, ipByte)
|
||||
}
|
||||
}
|
||||
|
|
105
peer_set_test.go
105
peer_set_test.go
|
@ -2,7 +2,6 @@ package p2p
|
|||
|
||||
import (
|
||||
"math/rand"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
. "github.com/tendermint/go-common"
|
||||
|
@ -13,7 +12,8 @@ func randPeer() *Peer {
|
|||
return &Peer{
|
||||
Key: RandStr(12),
|
||||
NodeInfo: &NodeInfo{
|
||||
Address: Fmt("%v.%v.%v.%v:46656", rand.Int()%256, rand.Int()%256, rand.Int()%256, rand.Int()%256),
|
||||
RemoteAddr: Fmt("%v.%v.%v.%v:46656", rand.Int()%256, rand.Int()%256, rand.Int()%256, rand.Int()%256),
|
||||
ListenAddr: Fmt("%v.%v.%v.%v:46656", rand.Int()%256, rand.Int()%256, rand.Int()%256, rand.Int()%256),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
@ -44,7 +44,6 @@ func TestAddRemoveMany(t *testing.T) {
|
|||
|
||||
peers := []*Peer{}
|
||||
N := 100
|
||||
maxPeersPerIPRange = [4]int{N, N, N, N}
|
||||
for i := 0; i < N; i++ {
|
||||
peer := randPeer()
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
|
@ -66,103 +65,3 @@ func TestAddRemoveMany(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
func newPeerInIPRange(ipBytes ...string) *Peer {
|
||||
ips := make([]string, 4)
|
||||
for i, ipByte := range ipBytes {
|
||||
ips[i] = ipByte
|
||||
}
|
||||
for i := len(ipBytes); i < 4; i++ {
|
||||
ips[i] = Fmt("%v", rand.Int()%256)
|
||||
}
|
||||
ipS := strings.Join(ips, ".")
|
||||
return &Peer{
|
||||
Key: RandStr(12),
|
||||
NodeInfo: &NodeInfo{
|
||||
Address: ipS + ":46656",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestIPRanges(t *testing.T) {
|
||||
peerSet := NewPeerSet()
|
||||
|
||||
// test /8
|
||||
maxPeersPerIPRange = [4]int{2, 2, 2, 2}
|
||||
peer := newPeerInIPRange("54", "1")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "2")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "3")
|
||||
if err := peerSet.Add(peer); err == nil {
|
||||
t.Errorf("Added peer when we shouldn't have")
|
||||
}
|
||||
peer = newPeerInIPRange("55", "1")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
|
||||
// test /16
|
||||
peerSet = NewPeerSet()
|
||||
maxPeersPerIPRange = [4]int{3, 2, 1, 1}
|
||||
peer = newPeerInIPRange("54", "112", "1")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "2")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "3")
|
||||
if err := peerSet.Add(peer); err == nil {
|
||||
t.Errorf("Added peer when we shouldn't have")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "113", "1")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
|
||||
// test /24
|
||||
peerSet = NewPeerSet()
|
||||
maxPeersPerIPRange = [4]int{5, 3, 2, 1}
|
||||
peer = newPeerInIPRange("54", "112", "11", "1")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "11", "2")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "11", "3")
|
||||
if err := peerSet.Add(peer); err == nil {
|
||||
t.Errorf("Added peer when we shouldn't have")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "12", "1")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
|
||||
// test /32
|
||||
peerSet = NewPeerSet()
|
||||
maxPeersPerIPRange = [4]int{11, 7, 5, 2}
|
||||
peer = newPeerInIPRange("54", "112", "11", "10")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "11", "10")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "11", "10")
|
||||
if err := peerSet.Add(peer); err == nil {
|
||||
t.Errorf("Added peer when we shouldn't have")
|
||||
}
|
||||
peer = newPeerInIPRange("54", "112", "11", "11")
|
||||
if err := peerSet.Add(peer); err != nil {
|
||||
t.Errorf("Failed to add new peer")
|
||||
}
|
||||
}
|
||||
|
|
|
@ -63,8 +63,9 @@ func (pexR *PEXReactor) GetChannels() []*ChannelDescriptor {
|
|||
|
||||
// Implements Reactor
|
||||
func (pexR *PEXReactor) AddPeer(peer *Peer) {
|
||||
// XXX Move this into Switch
|
||||
// Add the peer to the address book
|
||||
netAddr := NewNetAddressString(peer.Address)
|
||||
netAddr := NewNetAddressString(peer.ListenAddr)
|
||||
if peer.IsOutbound() {
|
||||
if pexR.book.NeedMoreAddrs() {
|
||||
pexR.RequestPEX(peer)
|
||||
|
|
10
switch.go
10
switch.go
|
@ -224,10 +224,6 @@ func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, er
|
|||
return nil, err
|
||||
}
|
||||
|
||||
// The peerNodeInfo is not verified, so overwrite
|
||||
// the IP, and the port too if we dialed out
|
||||
// Everything else we just have to trust
|
||||
peerNodeInfo.Address = sconn.RemoteAddr().String()
|
||||
peer := newPeer(sconn, peerNodeInfo, outbound, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError)
|
||||
|
||||
// Add the peer to .peers
|
||||
|
@ -351,12 +347,6 @@ func (sw *Switch) listenerRoutine(l Listener) {
|
|||
continue
|
||||
}
|
||||
|
||||
// Ignore connections from IP ranges for which we have too many
|
||||
if sw.peers.HasMaxForIPRange(inConn) {
|
||||
log.Info("Ignoring inbound connection: already have enough peers for that IP range", "address", inConn.RemoteAddr().String())
|
||||
continue
|
||||
}
|
||||
|
||||
// New inbound connection!
|
||||
_, err := sw.AddPeerWithConnection(inConn, false)
|
||||
if err != nil {
|
||||
|
|
21
types.go
21
types.go
|
@ -12,12 +12,13 @@ import (
|
|||
const maxNodeInfoSize = 10240 // 10Kb
|
||||
|
||||
type NodeInfo struct {
|
||||
PubKey crypto.PubKeyEd25519 `json:"pub_key"`
|
||||
Moniker string `json:"moniker"`
|
||||
Network string `json:"network"`
|
||||
Address string `json:"address"`
|
||||
Version string `json:"version"` // major.minor.revision
|
||||
Other []string `json:"other"` // other application specific data
|
||||
PubKey crypto.PubKeyEd25519 `json:"pub_key"`
|
||||
Moniker string `json:"moniker"`
|
||||
Network string `json:"network"`
|
||||
RemoteAddr string `json:"remote_addr"`
|
||||
ListenAddr string `json:"listen_addr"`
|
||||
Version string `json:"version"` // major.minor.revision
|
||||
Other []string `json:"other"` // other application specific data
|
||||
}
|
||||
|
||||
// CONTRACT: two nodes are compactible if the major/minor versions match and network match
|
||||
|
@ -53,13 +54,13 @@ func (info *NodeInfo) CompatibleWith(other *NodeInfo) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (info *NodeInfo) Host() string {
|
||||
host, _, _ := net.SplitHostPort(info.Address)
|
||||
func (info *NodeInfo) ListenHost() string {
|
||||
host, _, _ := net.SplitHostPort(info.ListenAddr)
|
||||
return host
|
||||
}
|
||||
|
||||
func (info *NodeInfo) Port() int {
|
||||
_, port, _ := net.SplitHostPort(info.Address)
|
||||
func (info *NodeInfo) ListenPort() int {
|
||||
_, port, _ := net.SplitHostPort(info.ListenAddr)
|
||||
port_i, err := strconv.Atoi(port)
|
||||
if err != nil {
|
||||
return -1
|
||||
|
|
Loading…
Reference in New Issue