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

@@ -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
 			}
 		}
 	})

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
 }

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...))
 }
 

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 {

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
 }