aboutsummaryrefslogtreecommitdiffstats
path: root/swarm/network/stream/snapshot_retrieval_test.go
diff options
context:
space:
mode:
Diffstat (limited to 'swarm/network/stream/snapshot_retrieval_test.go')
-rw-r--r--swarm/network/stream/snapshot_retrieval_test.go232
1 files changed, 79 insertions, 153 deletions
diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go
index 6498f599d..09d915d48 100644
--- a/swarm/network/stream/snapshot_retrieval_test.go
+++ b/swarm/network/stream/snapshot_retrieval_test.go
@@ -104,8 +104,47 @@ func TestRetrieval(t *testing.T) {
}
}
-/*
+var retrievalSimServiceMap = map[string]simulation.ServiceFunc{
+ "streamer": retrievalStreamerFunc,
+}
+
+func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
+ n := ctx.Config.Node()
+ addr := network.NewAddr(n)
+ store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
+ if err != nil {
+ return nil, nil, err
+ }
+ bucket.Store(bucketKeyStore, store)
+
+ localStore := store.(*storage.LocalStore)
+ netStore, err := storage.NewNetStore(localStore, nil)
+ if err != nil {
+ return nil, nil, err
+ }
+ kad := network.NewKademlia(addr.Over(), network.NewKadParams())
+ delivery := NewDelivery(kad, netStore)
+ netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
+
+ r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
+ DoSync: true,
+ SyncUpdateDelay: 3 * time.Second,
+ DoRetrieve: true,
+ })
+
+ fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
+ bucket.Store(bucketKeyFileStore, fileStore)
+
+ cleanup = func() {
+ os.RemoveAll(datadir)
+ netStore.Close()
+ r.Close()
+ }
+
+ return r, cleanup, nil
+}
+/*
The test loads a snapshot file to construct the swarm network,
assuming that the snapshot file identifies a healthy
kademlia network. Nevertheless a health check runs in the
@@ -114,43 +153,7 @@ simulation's `action` function.
The snapshot should have 'streamer' in its service list.
*/
func runFileRetrievalTest(nodeCount int) error {
- sim := simulation.New(map[string]simulation.ServiceFunc{
- "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- node := ctx.Config.Node()
- addr := network.NewAddr(node)
- store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
- if err != nil {
- return nil, nil, err
- }
- bucket.Store(bucketKeyStore, store)
-
- localStore := store.(*storage.LocalStore)
- netStore, err := storage.NewNetStore(localStore, nil)
- if err != nil {
- return nil, nil, err
- }
- kad := network.NewKademlia(addr.Over(), network.NewKadParams())
- delivery := NewDelivery(kad, netStore)
- netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
-
- r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
- DoSync: true,
- SyncUpdateDelay: 3 * time.Second,
- })
-
- fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
- bucket.Store(bucketKeyFileStore, fileStore)
-
- cleanup = func() {
- os.RemoveAll(datadir)
- netStore.Close()
- r.Close()
- }
-
- return r, cleanup, nil
-
- },
- })
+ sim := simulation.New(retrievalSimServiceMap)
defer sim.Close()
log.Info("Initializing test config")
@@ -200,49 +203,29 @@ func runFileRetrievalTest(nodeCount int) error {
// File retrieval check is repeated until all uploaded files are retrieved from all nodes
// or until the timeout is reached.
- allSuccess := false
- for !allSuccess {
+ REPEAT:
+ for {
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 {
- //get the real chunk by the index in the index array
- chunk := conf.hashes[ch]
- log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
- //check if the expected chunk is indeed in the localstore
- var err error
- //check on the node's FileStore (netstore)
- item, ok := sim.NodeItem(id, bucketKeyFileStore)
- if !ok {
- return fmt.Errorf("No registry")
- }
- 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
- } else {
- log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+ //for each expected file, check if it is in the local store
+ item, ok := sim.NodeItem(id, bucketKeyFileStore)
+ if !ok {
+ return fmt.Errorf("No filestore")
+ }
+ 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])) {
+ log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
+ time.Sleep(500 * time.Millisecond)
+ continue REPEAT
}
+ log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
}
- allSuccess = localSuccess
}
+ return nil
}
- if !allSuccess {
- return fmt.Errorf("Not all chunks succeeded!")
- }
- return nil
})
if result.Error != nil {
@@ -263,44 +246,7 @@ simulation's `action` function.
The snapshot should have 'streamer' in its service list.
*/
func runRetrievalTest(chunkCount int, nodeCount int) error {
- sim := simulation.New(map[string]simulation.ServiceFunc{
- "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- node := ctx.Config.Node()
- addr := network.NewAddr(node)
- store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
- if err != nil {
- return nil, nil, err
- }
- bucket.Store(bucketKeyStore, store)
-
- localStore := store.(*storage.LocalStore)
- netStore, err := storage.NewNetStore(localStore, nil)
- if err != nil {
- return nil, nil, err
- }
- kad := network.NewKademlia(addr.Over(), network.NewKadParams())
- delivery := NewDelivery(kad, netStore)
- netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
-
- r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
- DoSync: true,
- SyncUpdateDelay: 0,
- })
-
- fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
- bucketKeyFileStore = simulation.BucketKey("filestore")
- bucket.Store(bucketKeyFileStore, fileStore)
-
- cleanup = func() {
- os.RemoveAll(datadir)
- netStore.Close()
- r.Close()
- }
-
- return r, cleanup, nil
-
- },
- })
+ sim := simulation.New(retrievalSimServiceMap)
defer sim.Close()
conf := &synctestConfig{}
@@ -330,8 +276,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)
@@ -349,49 +293,31 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
// File retrieval check is repeated until all uploaded files are retrieved from all nodes
// or until the timeout is reached.
- allSuccess := false
- for !allSuccess {
+ REPEAT:
+ for {
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 {
- //get the real chunk by the index in the index array
- chunk := conf.hashes[ch]
- log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
- //check if the expected chunk is indeed in the localstore
- var err error
- //check on the node's FileStore (netstore)
- item, ok := sim.NodeItem(id, bucketKeyFileStore)
- if !ok {
- return fmt.Errorf("No registry")
- }
- 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
- } else {
- log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+ //check on the node's FileStore (netstore)
+ item, ok := sim.NodeItem(id, bucketKeyFileStore)
+ if !ok {
+ return fmt.Errorf("No filestore")
+ }
+ fileStore := item.(*storage.FileStore)
+ //check all chunks
+ for _, hash := range conf.hashes {
+ reader, _ := fileStore.Retrieve(context.TODO(), hash)
+ //check that we can read the chunk size and that it corresponds to the generated chunk size
+ if s, err := reader.Size(ctx, nil); err != nil || s != int64(chunkSize) {
+ log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s)
+ time.Sleep(500 * time.Millisecond)
+ continue REPEAT
}
+ log.Debug(fmt.Sprintf("Chunk with root hash %x successfully retrieved", hash))
}
- allSuccess = localSuccess
}
+ // all nodes and files found, exit loop and return without error
+ return nil
}
- if !allSuccess {
- return fmt.Errorf("Not all chunks succeeded!")
- }
- return nil
})
if result.Error != nil {
generated by cgit v1.2.3 (git 2.25.1) at 2024-09-13 04:31:27 +0800