swarm/network/stream: fixed retrieval tests; new backend for visualization

swarm/network/stream/snapshot_retrieval_test.go

@@ -129,6 +129,7 @@ func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s no
 	r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
 		DoSync:          true,
 		SyncUpdateDelay: 3 * time.Second,
+		DoRetrieve:      true,
 	})
 
 	fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
@@ -206,36 +207,22 @@ func runFileRetrievalTest(nodeCount int) error {
 		for !allSuccess {
 			for _, id := range nodeIDs {
 				//for each expected chunk, check if it is in the local store
-				localChunks := conf.idToChunksMap[id]
 				localSuccess := true
-				for _, ch := range localChunks {
+				item, ok := sim.NodeItem(id, bucketKeyFileStore)
-					//get the real chunk by the index in the index array
+				if !ok {
-					chunk := conf.hashes[ch]
+					return fmt.Errorf("No registry")
-					log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
+				}
-					//check if the expected chunk is indeed in the localstore
+				fileStore := item.(*storage.FileStore)
-					var err error
+				//check all chunks
-					//check on the node's FileStore (netstore)
+				for i, hash := range conf.hashes {
-					item, ok := sim.NodeItem(id, bucketKeyFileStore)
+					reader, _ := fileStore.Retrieve(context.TODO(), hash)
-					if !ok {
+					//check that we can read the file size and that it corresponds to the generated file size
-						return fmt.Errorf("No registry")
+					if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
-					}
+						//allSuccess = false
-					fileStore := item.(*storage.FileStore)
-					//check all chunks
-					for i, hash := range conf.hashes {
-						reader, _ := fileStore.Retrieve(context.TODO(), hash)
-						//check that we can read the file size and that it corresponds to the generated file size
-						if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
-							allSuccess = false
-							log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
-						} else {
-							log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
-						}
-					}
-					if err != nil {
-						log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
 						localSuccess = false
+						log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
 					} else {
-						log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+						log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
 					}
 				}
 				allSuccess = localSuccess
@@ -295,8 +282,6 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 			conf.addrToIDMap[string(a)] = n
 		}
 
-		//an array for the random files
-		var randomFiles []string
 		//this is the node selected for upload
 		node := sim.RandomUpNode()
 		item, ok := sim.NodeItem(node.ID, bucketKeyStore)
@@ -318,36 +303,22 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 		for !allSuccess {
 			for _, id := range nodeIDs {
 				//for each expected chunk, check if it is in the local store
-				localChunks := conf.idToChunksMap[id]
 				localSuccess := true
-				for _, ch := range localChunks {
+				//check on the node's FileStore (netstore)
-					//get the real chunk by the index in the index array
+				item, ok := sim.NodeItem(id, bucketKeyFileStore)
-					chunk := conf.hashes[ch]
+				if !ok {
-					log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
+					return fmt.Errorf("No registry")
-					//check if the expected chunk is indeed in the localstore
+				}
-					var err error
+				fileStore := item.(*storage.FileStore)
-					//check on the node's FileStore (netstore)
+				//check all chunks
-					item, ok := sim.NodeItem(id, bucketKeyFileStore)
+				for _, hash := range conf.hashes {
-					if !ok {
+					reader, _ := fileStore.Retrieve(context.TODO(), hash)
-						return fmt.Errorf("No registry")
+					//check that we can read the file size and that it corresponds to the generated file size
-					}
+					if s, err := reader.Size(ctx, nil); err != nil || s != int64(chunkSize) {
-					fileStore := item.(*storage.FileStore)
-					//check all chunks
-					for i, hash := range conf.hashes {
-						reader, _ := fileStore.Retrieve(context.TODO(), hash)
-						//check that we can read the file size and that it corresponds to the generated file size
-						if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
-							allSuccess = false
-							log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
-						} else {
-							log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
-						}
-					}
-					if err != nil {
-						log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
 						localSuccess = false
+						log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
 					} else {
-						log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+						log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
 					}
 				}
 				allSuccess = localSuccess

swarm/network/stream/visualized_snapshot_sync_sim_test.go

@@ -53,7 +53,7 @@ func setupSim(serviceMap map[string]simulation.ServiceFunc) (int, int, *simulati
 }
 
 func watchSim(sim *simulation.Simulation) (context.Context, context.CancelFunc) {
-	ctx, cancelSimRun := context.WithTimeout(context.Background(), 2*time.Minute)
+	ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute)
 
 	if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
 		panic(err)
@@ -98,10 +98,10 @@ func TestNonExistingHashesWithServer(t *testing.T) {
 		}
 		fileStore := item.(*storage.FileStore)
 		fakeHash := storage.GenerateRandomChunk(1000).Address()
-		fmt.Println(fakeHash)
+		//fmt.Println(fakeHash)
 		reader, _ := fileStore.Retrieve(context.TODO(), fakeHash)
 		if _, err := reader.Size(ctx, nil); err != nil {
-			fmt.Println("expected error for non-existing chunk")
+			log.Debug("expected error for non-existing chunk")
 		}
 
 		time.Sleep(testDuration)
@@ -109,14 +109,20 @@ func TestNonExistingHashesWithServer(t *testing.T) {
 		return nil
 	})
 	if result.Error != nil {
+		sendSimTerminatedEvent(sim)
 		t.Fatal(result.Error)
 	}
 
+	sendSimTerminatedEvent(sim)
+
+}
+
+func sendSimTerminatedEvent(sim *simulation.Simulation) {
 	evt := &simulations.Event{
-		Type: EventTypeSimTerminated,
+		Type:    EventTypeSimTerminated,
+		Control: false,
 	}
 	sim.Net.Events().Send(evt)
-
 }
 
 func TestSnapshotSyncWithServer(t *testing.T) {
@@ -163,22 +169,25 @@ func TestSnapshotSyncWithServer(t *testing.T) {
 			}
 			if *e.Event.MsgCode == uint64(1) {
 				evt := &simulations.Event{
-					Type: EventTypeChunkOffered,
+					Type:    EventTypeChunkOffered,
-					Node: sim.Net.GetNode(e.NodeID),
+					Node:    sim.Net.GetNode(e.NodeID),
+					Control: false,
 					//Data: fmt.Sprintf("%s", h),
 				}
 				sim.Net.Events().Send(evt)
 			} else if *e.Event.MsgCode == uint64(2) {
 				evt := &simulations.Event{
-					Type: EventTypeChunkWanted,
+					Type:    EventTypeChunkWanted,
-					Node: sim.Net.GetNode(e.NodeID),
+					Node:    sim.Net.GetNode(e.NodeID),
+					Control: false,
 					//Data: fmt.Sprintf("%s", h),
 				}
 				sim.Net.Events().Send(evt)
 			} else if *e.Event.MsgCode == uint64(6) {
 				evt := &simulations.Event{
-					Type: EventTypeChunkDelivered,
+					Type:    EventTypeChunkDelivered,
-					Node: sim.Net.GetNode(e.NodeID),
+					Node:    sim.Net.GetNode(e.NodeID),
+					Control: false,
 					//Data: fmt.Sprintf("%s", h),
 				}
 				sim.Net.Events().Send(evt)
@@ -188,7 +197,8 @@ func TestSnapshotSyncWithServer(t *testing.T) {
 	result := runSim(conf, ctx, sim, chunkCount)
 
 	evt := &simulations.Event{
-		Type: EventTypeSimTerminated,
+		Type:    EventTypeSimTerminated,
+		Control: false,
 	}
 	sim.Net.Events().Send(evt)