From 6de0d73449d57318f8d976aae784f64d3b5ecc43 Mon Sep 17 00:00:00 2001 From: zelig Date: Wed, 17 May 2017 14:09:38 -0700 Subject: [PATCH] swarm/network: kademlia health check for discovery test --- swarm/network/hive.go | 9 +-- swarm/network/kademlia.go | 65 +++++++++++++++++++ swarm/network/protocol.go | 8 ++- .../simulations/discovery/discovery_test.go | 6 +- 4 files changed, 77 insertions(+), 11 deletions(-) diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 2feb7d98d6..9fb20d2815 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -55,6 +55,7 @@ type Overlay interface { String() string BaseAddr() []byte + Healthy([][]byte) bool } // HiveParams holds the config options to hive @@ -191,14 +192,6 @@ func (self *Hive) PeerInfo(id discover.NodeID) interface{} { return interface{}(addr) } -// Healthy reports the health state of the kademlia connectivity -// -func (self *Hive) Healthy() bool { - // TODO: determine if we have enough peers to consider the network - // to be healthy - return true -} - func (self *Hive) Register(peers chan OverlayAddr) error { defer self.wake() return self.Overlay.Register(peers) diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 80ecad5962..dc8faa6d0d 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -23,6 +23,8 @@ import ( "time" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/pot" ) @@ -499,3 +501,66 @@ func (self *Kademlia) Prune(c <-chan time.Time) { } }() } + +func NewPeerPot(kadMinProxSize int, ids ...*adapters.NodeId) map[discover.NodeID][][]byte { + // create a table of all nodes for health check + np := pot.NewPot(nil, 0) + for _, id := range ids { + o := ToOverlayAddr(id) + np, _, _ = pot.Add(np, pot.NewBytesVal(o, nil)) + } + nnmap := make(map[discover.NodeID][][]byte) + + for _, id := range ids { + pl := 0 + var nns [][]byte + np.EachNeighbour(pot.NewBytesVal(id.Bytes(), nil), func(val pot.PotVal, po int) bool { + a := val.(pot.BytesAddress).Address() + nns = append(nns, a) + if len(nns) >= kadMinProxSize { + pl = po + } + return pl == 0 || pl == po + }) + nnmap[id.NodeID] = nns + } + return nnmap +} + +func (self *Kademlia) FirstEmptyBin() (i int) { + i = -1 + self.conns.EachBin(pot.NewBytesVal(self.base, nil), 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + if po > i+1 { + i = po + return false + } + i = po + return true + }) + return i +} + +func (self *Kademlia) Full() bool { + return self.FirstEmptyBin() >= self.Depth() +} + +// Healthy reports the health state of the kademlia connectivity +// +func (self *Kademlia) Healthy(peers [][]byte) bool { + return self.gotNearestNeighbours(peers) && self.Full() +} + +func (self *Kademlia) gotNearestNeighbours(peers [][]byte) (got bool) { + pm := make(map[string]bool) + for _, p := range peers { + pm[string(p)] = true + } + self.EachConn(nil, 255, func(p OverlayConn, po int, nn bool) bool { + if !nn { + return false + } + _, got = pm[string(p.Address())] + return got + }) + return got +} diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 1678e2de5a..cb8b4eca2f 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -64,6 +64,7 @@ type Addr interface { Over() []byte Under() []byte String() string + Update(OverlayAddr) OverlayAddr } // Peer interface represents an live peer connection @@ -349,7 +350,12 @@ func NewNodeIdFromAddr(addr Addr) *adapters.NodeId { func NewAddrFromNodeId(n *adapters.NodeId) *bzzAddr { id := n.NodeID return &bzzAddr{ - OAddr: crypto.Keccak256(id[:]), + OAddr: ToOverlayAddr(n), UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()), } } + +// ToOverlayAddr creates an overlayaddress from NodeID +func ToOverlayAddr(id *adapters.NodeId) []byte { + return crypto.Keccak256(id.Bytes()) +} diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 3e1fde34b3..a02e1d67f3 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -19,6 +19,7 @@ import ( // serviceName is used with the exec adapter so the exec'd binary knows which // service to execute const serviceName = "discovery" +const testMinProxBinSize = 2 var services = adapters.Services{ serviceName: newService, @@ -94,6 +95,7 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) { } return nil } + nnmap := network.NewPeerPot(testMinProxBinSize, ids...) check := func(ctx context.Context, id *adapters.NodeId) (bool, error) { select { case <-ctx.Done(): @@ -110,7 +112,7 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) { return false, fmt.Errorf("error getting node client: %s", err) } var healthy bool - if err := client.Call(&healthy, "hive_healthy", nil); err != nil { + if err := client.Call(&healthy, "hive_healthy", nnmap[id.NodeID]); err != nil { return false, fmt.Errorf("error getting node health: %s", err) } return healthy, nil @@ -180,7 +182,7 @@ func newService(id *adapters.NodeId, snapshot []byte) node.Service { addr := network.NewAddrFromNodeId(id) kp := network.NewKadParams() - kp.MinProxBinSize = 2 + kp.MinProxBinSize = testMinProxBinSize kp.MaxBinSize = 3 kp.MinBinSize = 1 kp.MaxRetries = 1000