swarm: time for chunk at node in snapshot_sync_test

swarm/network/stream/common_test.go

@@ -52,6 +52,7 @@ var (
 	defaultSkipCheck bool
 	waitPeerErrC     chan error
 	chunkSize        = 4096
+	registries       map[discover.NodeID]*TestRegistry
 )
 
 var services = adapters.Services{
@@ -92,7 +93,9 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
 	}()
 	dpa := storage.NewDPA(storage.NewNetStore(store, nil), storage.NewDPAParams())
-	return &TestRegistry{Registry: r, dpa: dpa}, nil
+	testRegistry := &TestRegistry{Registry: r, dpa: dpa}
+	registries[id] = testRegistry
+	return testRegistry, nil
 }
 
 //create a local store for the given node
@@ -112,12 +115,18 @@ func createTestLocalStorageForId(id discover.NodeID, addr *network.BzzAddr) (sto
 	return store, nil
 }
 
+func datadirsCleanup() {
+	for _, id := range ids {
+		os.RemoveAll(datadirs[id])
+	}
+}
+
 //local stores need to be cleaned up after the sim is done
 func localStoreCleanup() {
 	log.Info("Cleaning up...")
-	for i := 0; i < len(ids); i++ {
-		stores[ids[i]].Close()
-		os.RemoveAll(datadirs[ids[i]])
+	for _, id := range ids {
+		registries[id].Close()
+		stores[id].Close()
 	}
 	log.Info("Local store cleanup done")
 }

swarm/network/stream/messages.go

@@ -74,15 +74,6 @@ func (p *Peer) handleRequestSubscription(req *RequestSubscriptionMsg) (err error
 	return p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority)
 }
 
-func (p *Peer) handleRequestSubscription(req *RequestSubscriptionMsg) (err error) {
-	log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr.ID(), p.ID(), req.Stream))
-	err = p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
 func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 	defer func() {
 		if err != nil {

swarm/network/stream/snapshot_sync_test.go

@@ -33,7 +33,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	"github.com/ethereum/go-ethereum/pot"
 	"github.com/ethereum/go-ethereum/swarm/network"
-	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
+	//streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -47,6 +47,8 @@ var (
 	datadirs  map[discover.NodeID]string
 	conf      *synctestConfig
 	ppmap     map[discover.NodeID]*network.PeerPot
+
+	printed bool
 )
 
 type synctestConfig struct {
@@ -84,6 +86,8 @@ func initSyncTest() {
 	datadirs = make(map[discover.NodeID]string)
 	//deliveries for each node
 	deliveries = make(map[discover.NodeID]*Delivery)
+	//registries, map of discover.NodeID to its streamer
+	registries = make(map[discover.NodeID]*TestRegistry)
 
 	//channel to wait for peers connected
 	waitPeerErrC = make(chan error)
@@ -194,12 +198,32 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 	//map of overlay address to discover ID
 	conf.addrToIdMap = make(map[string]discover.NodeID)
 	//First load the snapshot from the file
+	var actionTicker *time.Ticker
+	var timingTicker *time.Ticker
+	trigger := make(chan discover.NodeID)
+	// channel to signal simulation initialisation with action call complete
+	// or node disconnections
+	disconnectC := make(chan error)
+	quitC := make(chan struct{})
+
 	net, err := initNetWithSnapshot(nodeCount)
 	if err != nil {
 		return err
 	}
-	defer net.Shutdown()
-
+	cleanup := func() {
+		timingTicker.Stop()
+		actionTicker.Stop()
+		//close(trigger)
+		close(quitC)
+		close(disconnectC)
+		close(waitPeerErrC)
+		//after the test, clean up local stores initialized with createLocalStoreForId
+		localStoreCleanup()
+		//shutdown the snapshot network
+		net.Shutdown()
+		//datadirsCleanup()
+	}
+	defer cleanup()
 	//get the nodes of the network
 	nodes := net.GetNodes()
 	//select one index at random...
@@ -226,13 +250,6 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 	log.Info("Test config successfully initialized")
 
 	ppmap = network.NewPeerPot(testMinProxBinSize, ids, conf.addrs)
-	// channel to signal simulation initialisation with action call complete
-	// or node disconnections
-	disconnectC := make(chan error)
-	quitC := make(chan struct{})
-
-	//after the test, clean up local stores initialized with createLocalStoreForId
-	defer localStoreCleanup()
 
 	//define the action to be performed before the test checks: start syncing
 	action := func(ctx context.Context) error {
@@ -286,10 +303,10 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 					log.Debug(kt)
 				}
 			}
-			err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
-			if err != nil {
-				return err
-			}
+			//err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
+			//if err != nil {
+			//	return err
+			//}
 			err = client.CallContext(ctx, nil, "stream_startSyncing")
 			if err != nil {
 				return err
@@ -309,10 +326,19 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 		//finally map chunks to the closest addresses
 		mapKeysToNodes(conf)
 
+		//periodically check if chunks have arrived at nodes
+		actionTicker = time.NewTicker(time.Second / 100)
+		go func() {
+			startTime = time.Now()
+			for range actionTicker.C {
+				checkChunkIsAtNode(conf)
+			}
+		}()
+		log.Info("Action terminated")
+
 		return nil
 	}
 
-	trigger := make(chan discover.NodeID)
 	//check defines what will be checked during the test
 	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
 		select {
@@ -323,7 +349,6 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 			return false, ctx.Err()
 		default:
 		}
-
 		log.Trace(fmt.Sprintf("Checking node: %s", id))
 		//select the local store for the given node
 		lstore := stores[id]
@@ -353,25 +378,16 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
+	timingTicker = time.NewTicker(time.Second * 1)
 	//for each tick, run the checks on all nodes
 	go func() {
-		ticker := time.NewTicker(time.Second * 1)
-		for range ticker.C {
+		for range timingTicker.C {
 			for i := 0; i < len(ids); i++ {
 				log.Trace(fmt.Sprintf("triggering step %d, id %s", i, ids[i]))
 				trigger <- ids[i]
 			}
 		}
 	}()
-	/*
-		go func() {
-			startTime = time.Now()
-			ticker := time.NewTicker(time.Second / 10)
-			for range ticker.C {
-				checkChunkIsAtNode(conf)
-			}
-		}()
-	*/
 
 	log.Info("Starting simulation run...")
 	//run the simulation
@@ -383,11 +399,9 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 			Check: check,
 		},
 	})
-	close(quitC)
 	if result.Error != nil {
 		return result.Error
 	}
-
 	log.Info("Simulation terminated")
 	return nil
 }
@@ -432,7 +446,6 @@ func (r *TestRegistry) StartSyncing(ctx context.Context) error {
 
 	//iterate over each bin and solicit needed subscription to bins
 	kad.EachBin(r.addr.Over(), pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
-
 		//identify begin and start index of the bin(s) we want to subscribe to
 		if po < kadDepth {
 			//not nn
@@ -471,28 +484,45 @@ func (r *TestRegistry) StartSyncing(ctx context.Context) error {
 	return nil
 }
 
+//periodically check if chunks have arrived at nodes
 func checkChunkIsAtNode(conf *synctestConfig) {
 	allOk := true
+	if printed {
+		return
+	}
+	//for every chunk, get the array of nodes it should have arrived to
 	for chunk, nodes := range conf.chunksToNodesMap {
+		//for every of those nodes
 		for _, node := range nodes {
+			//check that the chunk is in that localstore
 			if ok, _ := stores[conf.addrToIdMap[string(conf.addrs[node])]].Get([]byte(chunk)); ok != nil {
+				//if it is there, write the amount of time passed in the retrievalMap
+				//first create the map for the chunk if it is not there yet
 				if len(conf.retrievalMap[chunk]) == 0 {
 					conf.retrievalMap[chunk] = make(map[string]time.Duration)
 				}
+				//if the time value for the chunk at this node has not been recorded yet...
+				if _, ok := conf.retrievalMap[chunk][string(conf.addrs[node])]; !ok {
+					//...record it
 					conf.retrievalMap[chunk][string(conf.addrs[node])] = time.Since(startTime)
 				}
+			} else {
+				allOk = false
+			}
+			//if one of the chunks hasn't arrived yet, don't print
 			if conf.retrievalMap[chunk][string(conf.addrs[node])] == 0 {
 				allOk = false
 			}
 		}
 	}
-	if allOk {
+	if allOk && !printed {
 		log.Info("All chunks arrived at destination")
 		for ch, n := range conf.retrievalMap {
 			for a, t := range n {
-				log.Info(fmt.Sprintf("Chunk %s at node %s took %v ms", string(ch), string(a), t.Seconds()*1e3))
+				log.Info(fmt.Sprintf("Chunk %v at node %s took %v ms", storage.Key([]byte((ch))).String()[:8], conf.addrToIdMap[string(a)].String()[0:8], t.Seconds()*1e3))
 			}
 		}
+		printed = true
 	}
 }
 
@@ -529,7 +559,7 @@ func mapKeysToNodes(conf *synctestConfig) {
 			}
 			return true
 		})
-		kmap[conf.chunks[i].String()] = nns
+		kmap[string(conf.chunks[i])] = nns
 		//log.Debug(fmt.Sprintf("Length for id %s: %d",ids[i],len(kmap[ids[i]])))
 	}
 	if log.Lvl(*loglevel) == log.LvlTrace {

swarm/network/stream/testing/testing.go

@@ -228,7 +228,9 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error
 			case <-quitC:
 				return
 			case e := <-events:
+				if e.Type == p2p.PeerEventTypeDrop {
 					errc <- fmt.Errorf("peerEvent for node %v: %v", id, e)
+				}
 			case err := <-sub.Err():
 				if err != nil {
 					errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err)