From 6f5ed03e15e2e5930bb3b2050b934d72ad4833c8 Mon Sep 17 00:00:00 2001 From: Elad_ Date: Thu, 12 Apr 2018 10:08:36 +0200 Subject: [PATCH] kademlia persistence simulation (#347) * swarm/network: Kademlia updates to full and SuggestPeer methods * swarm/network: add more tests and fixes * fixed a bug in state store that did not close the file handle on shutdown * added test for kademlia state storage across sessions --- p2p/simulations/adapters/inproc.go | 2 +- p2p/simulations/network.go | 11 +- swarm/network/hive.go | 21 +- swarm/network/kademlia.go | 6 +- .../simulations/discovery/discovery_test.go | 252 +++++++++++++++++- 5 files changed, 275 insertions(+), 17 deletions(-) diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 8752d04458..3dbecd92e9 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -274,7 +274,7 @@ func (self *SimNode) Start(snapshots map[string][]byte) error { for _, name := range self.config.Services { if err := self.node.Register(newService(name)); err != nil { regErr = err - return + break } } }) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 6919da1cd5..911e1c8a60 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -87,7 +87,10 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) if conf.Reachable == nil { conf.Reachable = func(otherID discover.NodeID) bool { _, err := self.InitConn(conf.ID, otherID) - return err == nil + if err != nil && bytes.Compare(conf.ID.Bytes(), otherID.Bytes()) < 0 { + return false + } + return true } } @@ -448,9 +451,11 @@ func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn { // this is cheating as the simulation is used as an oracle and know about // remote peers attempt to connect to a node which will then not initiate the connection func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) { + log.Debug(fmt.Sprintf("InitConn(oneID: %v, otherID: %v)", oneID, otherID)) self.lock.Lock() defer self.lock.Unlock() if oneID == otherID { + log.Trace(fmt.Sprintf("refusing to connect to self %v", oneID)) return nil, fmt.Errorf("refusing to connect to self %v", oneID) } conn, err := self.getOrCreateConn(oneID, otherID) @@ -458,15 +463,19 @@ func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) { return nil, err } if time.Since(conn.initiated) < dialBanTimeout { + log.Trace(fmt.Sprintf("connection between %v and %v recently attempted", oneID, otherID)) return nil, fmt.Errorf("connection between %v and %v recently attempted", oneID, otherID) } if conn.Up { + log.Trace(fmt.Sprintf("%v and %v already connected", oneID, otherID)) return nil, fmt.Errorf("%v and %v already connected", oneID, otherID) } err = conn.nodesUp() if err != nil { + log.Trace(fmt.Sprintf("nodes not up: %v", err)) return nil, fmt.Errorf("nodes not up: %v", err) } + log.Debug("InitConn - connection initiated") conn.initiated = time.Now() return conn, nil } diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 172403379d..f6992fcfe1 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -18,7 +18,6 @@ package network import ( "fmt" - "math/rand" "sync" "time" @@ -71,7 +70,7 @@ func NewHiveParams() *HiveParams { Discovery: true, PeersBroadcastSetSize: 3, MaxPeersPerRequest: 5, - KeepAliveInterval: 1000 * time.Millisecond, + KeepAliveInterval: 500 * time.Millisecond, } } @@ -102,10 +101,12 @@ func NewHive(params *HiveParams, overlay Overlay, store state.Store) *Hive { // server is used to connect to a peer based on its NodeID or enode URL // these are called on the p2p.Server which runs on the node func (h *Hive) Start(server *p2p.Server) error { - log.Trace(fmt.Sprintf("%08x hive starting", h.BaseAddr()[:4])) + log.Info(fmt.Sprintf("%08x hive starting", h.BaseAddr()[:4])) // if state store is specified, load peers to prepopulate the overlay address book if h.Store != nil { + log.Info("detected an existing store. trying to load peers") if err := h.loadPeers(); err != nil { + log.Error(fmt.Sprintf("%08x hive encoutered an error trying to load peers", h.BaseAddr()[:4])) return err } } @@ -123,7 +124,12 @@ func (h *Hive) Stop() error { log.Info(fmt.Sprintf("%08x hive stopping, saving peers", h.BaseAddr()[:4])) h.ticker.Stop() if h.Store != nil { - return h.savePeers() + if err := h.savePeers(); err != nil { + return fmt.Errorf("could not save peers to persistence store: %v", err) + } + if err := h.Store.Close(); err != nil { + return fmt.Errorf("could not close file handle to persistence store: %v", err) + } } log.Info(fmt.Sprintf("%08x hive stopped, dropping peers", h.BaseAddr()[:4])) h.EachConn(nil, 255, func(p OverlayConn, _ int, _ bool) bool { @@ -139,8 +145,9 @@ func (h *Hive) Stop() error { // at each iteration, ask the overlay driver to suggest the most preferred peer to connect to // as well as advertises saturation depth if needed func (h *Hive) connect() { - time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) for range h.ticker.C { + log.Trace(fmt.Sprintf("%08x hive connect()", h.BaseAddr()[:4])) + addr, depth, changed := h.SuggestPeer() if h.Discovery && changed { NotifyDepth(uint8(depth), h) @@ -203,14 +210,16 @@ func ToAddr(pa OverlayPeer) *BzzAddr { // loadPeers, savePeer implement persistence callback/ func (h *Hive) loadPeers() error { var as []*BzzAddr - err := h.Store.Get("peers", &as) if err != nil { if err == state.ErrNotFound { + log.Info(fmt.Sprintf("hive %08x: no persisted peers found", h.BaseAddr()[:4])) return nil } return err } + log.Info(fmt.Sprintf("hive %08x: peers loaded", h.BaseAddr()[:4])) + return h.Register(toOverlayAddrs(as...)) } diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index c06e524aea..15e6b4204b 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -263,10 +263,14 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) { if po >= depth { return false } - return f(func(val pot.Val, _ int) bool { + ok := f(func(val pot.Val, _ int) bool { a = k.callable(val) return a == nil }) + if !ok { + return false + } + return true }) // found a candidate if a != nil { diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index fcc8aa38b2..5d0eb1ae7c 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -9,6 +9,8 @@ import ( "io/ioutil" "math/rand" "os" + "path" + "strings" "sync" "testing" "time" @@ -21,6 +23,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/state" colorable "github.com/mattn/go-colorable" ) @@ -28,13 +31,46 @@ import ( // service to execute const serviceName = "discovery" const testMinProxBinSize = 2 +const discoveryPersistenceDatadir = "discovery_persistence_test_store" + +var discoveryPersistencePath = path.Join(os.TempDir(), discoveryPersistenceDatadir) +var discoveryEnabled = true +var persistenceEnabled = false var services = adapters.Services{ serviceName: newService, } +func cleanDbStores() error { + entries, err := ioutil.ReadDir(os.TempDir()) + if err != nil { + return err + } + + for _, f := range entries { + if strings.HasPrefix(f.Name(), discoveryPersistenceDatadir) { + os.RemoveAll(path.Join(os.TempDir(), f.Name())) + } + } + return nil + +} + +func getDbStore(nodeID string) (*state.DBStore, error) { + if _, err := os.Stat(discoveryPersistencePath + "_" + nodeID); os.IsNotExist(err) { + log.Info(fmt.Sprintf("directory for nodeID %s does not exist. creating...", nodeID)) + ioutil.TempDir("", discoveryPersistencePath+"_"+nodeID) + } + log.Info(fmt.Sprintf("opening storage directory for nodeID %s", nodeID)) + store, err := state.NewDBStore(discoveryPersistencePath + "_" + nodeID) + if err != nil { + return nil, err + } + return store, nil +} + var ( - nodeCount = flag.Int("nodes", 16, "number of nodes to create (default 10)") + nodeCount = flag.Int("nodes", 10, "number of nodes to create (default 10)") initCount = flag.Int("conns", 1, "number of originally connected peers (default 1)") snapshotFile = flag.String("snapshot", "", "create snapshot") loglevel = flag.Int("loglevel", 3, "verbosity of logs") @@ -110,6 +146,14 @@ func TestDiscoverySimulationSimAdapter(t *testing.T) { testDiscoverySimulationSimAdapter(t, *nodeCount, *initCount) } +func TestDiscoveryPersistenceSimulationSimAdapter(t *testing.T) { + testDiscoveryPersistenceSimulationSimAdapter(t, *nodeCount, *initCount) +} + +func testDiscoveryPersistenceSimulationSimAdapter(t *testing.T, nodes, conns int) { + testDiscoveryPersistenceSimulation(t, nodes, conns, adapters.NewSimAdapter(services)) +} + func testDiscoverySimulationSimAdapter(t *testing.T, nodes, conns int) { testDiscoverySimulation(t, nodes, conns, adapters.NewSimAdapter(services)) } @@ -145,6 +189,26 @@ func testDiscoverySimulation(t *testing.T, nodes, conns int, adapter adapters.No t.Logf("Setup: %s, shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt)) } +func testDiscoveryPersistenceSimulation(t *testing.T, nodes, conns int, adapter adapters.NodeAdapter) map[int][]byte { + persistenceEnabled = true + discoveryEnabled = true + + result, err := discoveryPersistenceSimulation(nodes, conns, adapter) + + if err != nil { + t.Fatalf("Setting up simulation failed: %v", err) + } + if result.Error != nil { + t.Fatalf("Simulation failed: %s", result.Error) + } + t.Logf("Simulation with %d nodes passed in %s", nodes, result.FinishedAt.Sub(result.StartedAt)) + // set the discovery and persistence flags again to default so other + // tests will not be affected + discoveryEnabled = true + persistenceEnabled = false + return nil +} + func benchmarkDiscovery(b *testing.B, nodes, conns int) { for i := 0; i < b.N; i++ { result, err := discoverySimulation(nodes, conns, adapters.NewSimAdapter(services)) @@ -268,6 +332,172 @@ func discoverySimulation(nodes, conns int, adapter adapters.NodeAdapter) (*simul return result, nil } +func discoveryPersistenceSimulation(nodes, conns int, adapter adapters.NodeAdapter) (*simulations.StepResult, error) { + cleanDbStores() + defer cleanDbStores() + + // create network + net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ + ID: "0", + DefaultService: serviceName, + }) + defer net.Shutdown() + trigger := make(chan discover.NodeID) + ids := make([]discover.NodeID, nodes) + var addrs [][]byte + + for i := 0; i < nodes; i++ { + conf := adapters.RandomNodeConfig() + node, err := net.NewNodeWithConfig(conf) + if err != nil { + panic(err) + } + if err != nil { + return nil, fmt.Errorf("error starting node: %s", err) + } + if err := net.Start(node.ID()); err != nil { + return nil, fmt.Errorf("error starting node %s: %s", node.ID().TerminalString(), err) + } + if err := triggerChecks(trigger, net, node.ID()); err != nil { + return nil, fmt.Errorf("error triggering checks for node %s: %s", node.ID().TerminalString(), err) + } + ids[i] = node.ID() + a := network.ToOverlayAddr(ids[i].Bytes()) + + addrs = append(addrs, a) + } + + // run a simulation which connects the 10 nodes in a ring and waits + // for full peer discovery + ppmap := network.NewPeerPotMap(testMinProxBinSize, addrs) + + var restartTime time.Time + + action := func(ctx context.Context) error { + ticker := time.NewTicker(500 * time.Millisecond) + + for range ticker.C { + isHealthy := true + for _, id := range ids { + //call Healthy RPC + node := net.GetNode(id) + if node == nil { + return fmt.Errorf("unknown node: %s", id) + } + client, err := node.Client() + if err != nil { + return fmt.Errorf("error getting node client: %s", err) + } + healthy := &network.Health{} + addr := common.Bytes2Hex(network.ToOverlayAddr(id.Bytes())) + if err := client.Call(&healthy, "hive_healthy", ppmap[addr]); err != nil { + return fmt.Errorf("error getting node health: %s", err) + } + + log.Info(fmt.Sprintf("NODE: %s, IS HEALTHY: %t", id.String(), healthy.GotNN && healthy.KnowNN && healthy.Full)) + if !healthy.GotNN || !healthy.Full { + isHealthy = false + break + } + } + if isHealthy { + break + } + } + ticker.Stop() + + log.Info("reached healthy kademlia. starting to shutdown nodes.") + shutdownStarted := time.Now() + // stop all ids, then start them again + for _, id := range ids { + node := net.GetNode(id) + + if err := net.Stop(node.ID()); err != nil { + return fmt.Errorf("error stopping node %s: %s", node.ID().TerminalString(), err) + } + } + log.Info(fmt.Sprintf("shutting down nodes took: %s", time.Now().Sub(shutdownStarted))) + persistenceEnabled = true + discoveryEnabled = false + restartTime = time.Now() + for _, id := range ids { + node := net.GetNode(id) + if err := net.Start(node.ID()); err != nil { + return fmt.Errorf("error starting node %s: %s", node.ID().TerminalString(), err) + } + if err := triggerChecks(trigger, net, node.ID()); err != nil { + return fmt.Errorf("error triggering checks for node %s: %s", node.ID().TerminalString(), err) + } + } + + log.Info(fmt.Sprintf("restarting nodes took: %s", time.Now().Sub(restartTime))) + + return nil + } + //connects in a chain + wg := sync.WaitGroup{} + //connects in a ring + for i := range ids { + for j := 1; j <= conns; j++ { + k := (i + j) % len(ids) + if k == i { + k = (k + 1) % len(ids) + } + wg.Add(1) + go func(i, k int) { + defer wg.Done() + net.Connect(ids[i], ids[k]) + }(i, k) + } + } + wg.Wait() + log.Debug(fmt.Sprintf("nodes: %v", len(addrs))) + // construct the peer pot, so that kademlia health can be checked + check := func(ctx context.Context, id discover.NodeID) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + node := net.GetNode(id) + if node == nil { + return false, fmt.Errorf("unknown node: %s", id) + } + client, err := node.Client() + if err != nil { + return false, fmt.Errorf("error getting node client: %s", err) + } + healthy := &network.Health{} + addr := common.Bytes2Hex(network.ToOverlayAddr(id.Bytes())) + if err := client.Call(&healthy, "hive_healthy", ppmap[addr]); err != nil { + return false, fmt.Errorf("error getting node health: %s", err) + } + log.Info(fmt.Sprintf("node %4s healthy: got nearest neighbours: %v, know nearest neighbours: %v, saturated: %v", id, healthy.GotNN, healthy.KnowNN, healthy.Full)) + + return healthy.KnowNN && healthy.GotNN && healthy.Full, nil + } + + // 64 nodes ~ 1min + // 128 nodes ~ + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: ids, + Check: check, + }, + }) + if result.Error != nil { + return result, nil + } + + return result, nil +} + // triggerChecks triggers a simulation step check whenever a peer is added or // removed from the given node, and also every second to avoid a race between // peer events and kademlia becoming healthy @@ -315,11 +545,6 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { kp := network.NewKadParams() kp.MinProxBinSize = testMinProxBinSize - kp.MaxBinSize = 3 - kp.MinBinSize = 1 - kp.MaxRetries = 1000 - kp.RetryExponent = 2 - kp.RetryInterval = 50000000 if ctx.Config.Reachable != nil { kp.Reachable = func(o network.OverlayAddr) bool { @@ -327,9 +552,11 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { } } kad := network.NewKademlia(addr.Over(), kp) - hp := network.NewHiveParams() - hp.KeepAliveInterval = 200 * time.Millisecond + hp.KeepAliveInterval = time.Duration(200) * time.Millisecond + hp.Discovery = discoveryEnabled + + log.Info(fmt.Sprintf("discovery for nodeID %s is %t", ctx.Config.ID.String(), hp.Discovery)) config := &network.BzzConfig{ OverlayAddr: addr.Over(), @@ -337,5 +564,14 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { HiveParams: hp, } + if persistenceEnabled { + log.Info(fmt.Sprintf("persistence enabled for nodeID %s", ctx.Config.ID.String())) + store, err := getDbStore(ctx.Config.ID.String()) + if err != nil { + return nil, err + } + return network.NewBzz(config, kad, store, nil, nil), nil + } + return network.NewBzz(config, kad, nil, nil, nil), nil }