diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 76007b6658..e3f06ad487 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -28,15 +28,14 @@ import ( "math/rand" "os" "strings" + "sync/atomic" "testing" "time" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" - "github.com/ethereum/go-ethereum/rpc" + //"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/pot" @@ -67,6 +66,7 @@ var ( func init() { flag.Parse() + rand.Seed(time.Now().UnixNano()) log.PrintOrigins(true) log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))) @@ -142,20 +142,31 @@ func waitForPeers(streamer *Registry, timeout time.Duration, expectedPeers int) } } -type TestRegistry struct { - *Registry - fileStore *storage.FileStore +type roundRobinStore struct { + index uint32 + stores []storage.ChunkStore } -func (r *TestRegistry) APIs() []rpc.API { - a := r.Registry.APIs() - a = append(a, rpc.API{ - Namespace: "stream", - Version: "3.0", - Service: r, - Public: true, - }) - return a +func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore { + return &roundRobinStore{ + stores: stores, + } +} + +func (rrs *roundRobinStore) Get(ctx context.Context, addr storage.Address) (*storage.Chunk, error) { + return nil, errors.New("get not well defined on round robin store") +} + +func (rrs *roundRobinStore) Put(ctx context.Context, chunk *storage.Chunk) { + i := atomic.AddUint32(&rrs.index, 1) + idx := int(i) % len(rrs.stores) + rrs.stores[idx].Put(ctx, chunk) +} + +func (rrs *roundRobinStore) Close() { + for _, store := range rrs.stores { + store.Close() + } } func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) { @@ -174,34 +185,8 @@ func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) { return total, nil } -func (r *TestRegistry) ReadAll(hash common.Hash) (int64, error) { - return readAll(r.fileStore, hash[:]) -} - -func (r *TestRegistry) Start(server *p2p.Server) error { - return r.Registry.Start(server) -} - -func (r *TestRegistry) Stop() error { - return r.Registry.Stop() -} - -type TestExternalRegistry struct { - *Registry -} - -func (r *TestExternalRegistry) APIs() []rpc.API { - a := r.Registry.APIs() - a = append(a, rpc.API{ - Namespace: "stream", - Version: "3.0", - Service: r, - Public: true, - }) - return a -} - -func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) { +//func getHashes(r *Registry, ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) { +func getHashes(r *Registry, ctx context.Context, peerId discover.NodeID, s Stream) (chan []byte, error) { peer := r.getPeer(peerId) client, err := peer.getClient(ctx, s) @@ -211,41 +196,10 @@ func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.No c := client.Client.(*testExternalClient) - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return nil, fmt.Errorf("Subscribe not supported") - } - - sub := notifier.CreateSubscription() - - go func() { - // if we begin sending event immediately some events - // will probably be dropped since the subscription ID might not be send to - // the client. - // ref: rpc/subscription_test.go#L65 - time.Sleep(1 * time.Second) - for { - select { - case h := <-c.hashes: - <-c.enableNotificationsC // wait for notification subscription to complete - if err := notifier.Notify(sub.ID, h); err != nil { - log.Warn(fmt.Sprintf("rpc sub notifier notify stream %s: %v", s, err)) - } - case err := <-sub.Err(): - if err != nil { - log.Warn(fmt.Sprintf("caught subscription error in stream %s: %v", s, err)) - } - case <-notifier.Closed(): - log.Trace(fmt.Sprintf("rpc sub notifier closed")) - return - } - } - }() - - return sub, nil + return c.hashes, nil } -func (r *TestExternalRegistry) EnableNotifications(peerId discover.NodeID, s Stream) error { +func enableNotifications(r *Registry, peerId discover.NodeID, s Stream) error { peer := r.getPeer(peerId) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -302,7 +256,6 @@ type testExternalServer struct { keyFunc func(key []byte, index uint64) sessionAt uint64 maxKeys uint64 - streamer *TestExternalRegistry } func newTestExternalServer(t string, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer { @@ -382,10 +335,6 @@ func uploadFilesToNodes(sim *simulation.Simulation) ([]storage.Address, []string return rootAddrs, rfiles, nil } -func init() { - rand.Seed(time.Now().UnixNano()) -} - //generate a random file (string) func generateRandomFile() (string, error) { //generate a random file size between minFileSize and maxFileSize diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 8f9594da7c..ce9ee0a684 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -27,7 +27,6 @@ import ( "testing" "time" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" @@ -332,6 +331,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ SkipCheck: skipCheck, }) + bucket.Store(bucketKeyRegistry, r) retrieveFunc := func(ctx context.Context, chunk *storage.Chunk) error { return delivery.RequestFromPeers(ctx, chunk.Addr[:], skipCheck) @@ -339,16 +339,15 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck netStore := storage.NewNetStore(localStore, retrieveFunc) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) bucket.Store(bucketKeyFileStore, fileStore) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) defer sim.Close() log.Info("Adding nodes to simulation") - _, err := sim.AddNodesAndConnectFull(nodes) + _, err := sim.AddNodesAndConnectChain(nodes) if err != nil { t.Fatal(err) } @@ -357,19 +356,34 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck ctx := context.Background() result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { nodeIDs := sim.UpNodeIDs() - // create a retriever FileStore for the pivot node - log.Debug("Selecting pivot node") - //select first node - node := nodeIDs[0] - - item, ok := sim.NodeItem(node, bucketKeyFileStore) - if !ok { - return fmt.Errorf("No filestore") + //determine the pivot node to be the first node of the simulation + sim.SetPivotNode(nodeIDs[0]) + //distribute chunks of a random file into Stores of nodes 1 to nodes + //we will do this by creating a file store with an underlying round-robin store: + //the file store will create a hash for the uploaded file, but every chunk will be + //distributed to different nodes via round-robin scheduling + log.Debug("Writing file to round-robin file store") + //to do this, we create an array for chunkstores (length minus one, the pivot node) + stores := make([]storage.ChunkStore, len(nodeIDs)-1) + //we then need to get all stores from the sim.... + lStores := sim.NodesItems(bucketKeyStore) + i := 0 + //...iterate the buckets... + for id, bucketVal := range lStores { + //...and remove the one which is the pivot node + if id == *sim.PivotNodeID() { + continue + } + //the other ones are added to the array... + stores[i] = bucketVal.(storage.ChunkStore) + i++ } - fileStore := item.(*storage.FileStore) + //...which then gets passed to the round-robin file store + roundRobinFileStore := storage.NewFileStore(newRoundRobinStore(stores...), storage.NewFileStoreParams()) + //now we can actually upload a (random) file to the round-robin store size := chunkCount * chunkSize log.Debug("Storing data to file store") - fileHash, wait, err := fileStore.Store(ctx, io.LimitReader(crand.Reader, int64(size)), int64(size), false) + fileHash, wait, err := roundRobinFileStore.Store(ctx, io.LimitReader(crand.Reader, int64(size)), int64(size), false) // wait until all chunks stored if err != nil { return err @@ -379,22 +393,31 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck return err } - for j := 0; j < len(nodeIDs)-1; j++ { - client, err := sim.Net.GetNode(nodeIDs[j]).Client() + //each of the nodes (except pivot node) subscribes to the stream of the next node + for j, node := range nodeIDs[0 : nodes-1] { + sid := nodeIDs[j+1] + item, ok := sim.NodeItem(node, bucketKeyRegistry) + if !ok { + return fmt.Errorf("No registry") + } + registry := item.(*Registry) + err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top) if err != nil { return err } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - sid := nodeIDs[j+1] - client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top) } + //get the pivot node's filestore + item, ok := sim.NodeItem(*sim.PivotNodeID(), bucketKeyFileStore) + if !ok { + return fmt.Errorf("No filestore") + } + pivotFileStore := item.(*storage.FileStore) log.Debug("Starting retrieval routine") go func() { // start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks // we must wait for the peer connections to have started before requesting - n, err := readAll(fileStore, fileHash) + n, err := readAll(pivotFileStore, fileHash) log.Info(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err) if err != nil { t.Fatalf("requesting chunks action error: %v", err) @@ -424,26 +447,20 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } }() + //finally check that the pivot node gets all chunks via the root hash log.Debug("Check retrieval") - allSuccess := true - id := nodeIDs[0] - client, err := sim.Net.GetNode(id).Client() - if err != nil { - return err - } + success := true var total int64 - ctx, cancel := context.WithTimeout(ctx, 5*time.Second) - defer cancel() - err = client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash)) + total, err = readAll(pivotFileStore, fileHash) if err != nil { return err } log.Info(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err)) if err != nil || total != int64(size) { - allSuccess = false + success = false } - if !allSuccess { + if !success { return fmt.Errorf("Test failed, chunks not available on all nodes") } log.Debug("Test terminated successfully") @@ -510,18 +527,17 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip return delivery.RequestFromPeers(ctx, chunk.Addr[:], skipCheck) } netStore := storage.NewNetStore(localStore, retrieveFunc) - bucket.Store(bucketKeyNetStore, netStore) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} + bucket.Store(bucketKeyFileStore, fileStore) - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) defer sim.Close() log.Info("Initializing test config") - _, err := sim.AddNodesAndConnectFull(nodes) + _, err := sim.AddNodesAndConnectChain(nodes) if err != nil { b.Fatal(err) } diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index f8599e568b..7e973a1d74 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -25,7 +25,6 @@ import ( "os" "sync" "testing" - "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" @@ -79,6 +78,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ SkipCheck: skipCheck, }) + bucket.Store(bucketKeyRegistry, r) r.RegisterClientFunc(externalStreamName, func(p *Peer, t string, live bool) (Client, error) { return newTestExternalClient(db), nil @@ -90,9 +90,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { fileStore := storage.NewFileStore(localStore, storage.NewFileStoreParams()) bucket.Store(bucketKeyFileStore, fileStore) - testRegistry := &TestExternalRegistry{r} - - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) @@ -129,16 +127,13 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { t.Fatal(err) } - client, err := sim.Net.GetNode(checker).Client() - if err != nil { - log.Error("Get Client error: %v", "err", err) - return err + item, ok = sim.NodeItem(checker, bucketKeyRegistry) + if !ok { + return fmt.Errorf("No registry") } - ctx, cancel := context.WithTimeout(ctx, 100*time.Second) - defer cancel() - err = client.CallContext(ctx, nil, "stream_subscribeStream", storer, NewStream(externalStreamName, "", live), history, Top) + registry := item.(*Registry) + err = registry.Subscribe(storer, NewStream(externalStreamName, "", live), history, Top) if err != nil { - log.Error("Stream subscribe error: %v", "err", err) return err } @@ -180,17 +175,16 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { }() // live stream - liveHashesChan := make(chan []byte) - liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", storer, NewStream(externalStreamName, "", true)) + var liveHashesChan chan []byte + liveHashesChan, err = getHashes(registry, ctx, storer, NewStream(externalStreamName, "", true)) if err != nil { log.Error("Subscription error: %v", "err", err) return } - defer liveSubscription.Unsubscribe() i := externalStreamSessionAt // we have subscribed, enable notifications - err = client.CallContext(ctx, nil, "stream_enableNotifications", storer, NewStream(externalStreamName, "", true)) + err = enableNotifications(registry, storer, NewStream(externalStreamName, "", true)) if err != nil { return } @@ -207,8 +201,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { if i > externalStreamMaxKeys { return } - case err = <-liveSubscription.Err(): - return + //case err = <-liveSubscription.Err(): + // return case <-ctx.Done(): return } @@ -227,12 +221,11 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { }() // history stream - historyHashesChan := make(chan []byte) - historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", storer, NewStream(externalStreamName, "", false)) + var historyHashesChan chan []byte + historyHashesChan, err = getHashes(registry, ctx, storer, NewStream(externalStreamName, "", false)) if err != nil { return } - defer historySubscription.Unsubscribe() var i uint64 historyTo := externalStreamMaxKeys @@ -244,7 +237,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { } // we have subscribed, enable notifications - err = client.CallContext(ctx, nil, "stream_enableNotifications", storer, NewStream(externalStreamName, "", false)) + err = enableNotifications(registry, storer, NewStream(externalStreamName, "", false)) if err != nil { return } @@ -261,8 +254,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { if i > historyTo { return } - case err = <-historySubscription.Err(): - return + //case err = <-historySubscription.Err(): + // return case <-ctx.Done(): return } diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 3d2b5b32ba..e48fc8fd15 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -21,6 +21,7 @@ import ( "os" "sync" "testing" + "time" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p/discover" @@ -45,7 +46,10 @@ const ( //Number of nodes can be provided via commandline too. func TestFileRetrieval(t *testing.T) { if *nodes != 0 { - fileRetrievalTest(t, *nodes) + err := runFileRetrievalTest(*nodes) + if err != nil { + t.Fatal(err) + } } else { nodeCnt := []int{16} //if the `longrunning` flag has been provided @@ -54,7 +58,10 @@ func TestFileRetrieval(t *testing.T) { nodeCnt = append(nodeCnt, 32, 64, 128) } for _, n := range nodeCnt { - fileRetrievalTest(t, n) + err := runFileRetrievalTest(n) + if err != nil { + t.Fatal(err) + } } } } @@ -69,7 +76,10 @@ func TestRetrieval(t *testing.T) { //if nodes/chunks have been provided via commandline, //run the tests with these values if *nodes != 0 && *chunks != 0 { - retrievalTest(t, *chunks, *nodes) + err := runRetrievalTest(*chunks, *nodes) + if err != nil { + t.Fatal(err) + } } else { var nodeCnt []int var chnkCnt []int @@ -85,28 +95,15 @@ func TestRetrieval(t *testing.T) { } for _, n := range nodeCnt { for _, c := range chnkCnt { - retrievalTest(t, c, n) + err := runRetrievalTest(c, n) + if err != nil { + t.Fatal(err) + } } } } } -//Run the file retrieval test -func fileRetrievalTest(t *testing.T, nodeCount int) { - err := runFileRetrievalTest(nodeCount) - if err != nil { - t.Fatal(err) - } -} - -//Run the chunks retrieval test -func retrievalTest(t *testing.T, chunkCount int, nodeCount int) { - err := runRetrievalTest(chunkCount, nodeCount) - if err != nil { - t.Fatal(err) - } -} - /* The test loads a snapshot file to construct the swarm network, @@ -142,22 +139,15 @@ func runFileRetrievalTest(nodeCount int) error { }) fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) - bucketKeyFileStore = simulation.BucketKey("filestore") bucket.Store(bucketKeyFileStore, fileStore) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) defer sim.Close() log.Info("Initializing test config") - _, err := sim.AddNodesAndConnectFull(3) - if err != nil { - return err - } - ctx := context.Background() conf := &synctestConfig{} //map of discover ID to indexes of chunks expected at that ID @@ -167,11 +157,14 @@ func runFileRetrievalTest(nodeCount int) error { //array where the generated chunk hashes will be stored conf.hashes = make([]storage.Address, 0) - err = sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount)) + err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount)) if err != nil { return err } + ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute) + defer cancelSimRun() + result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { nodeIDs := sim.UpNodeIDs() for _, n := range nodeIDs { @@ -291,9 +284,8 @@ func runRetrievalTest(chunkCount int, nodeCount int) error { fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) bucketKeyFileStore = simulation.BucketKey("filestore") bucket.Store(bucketKeyFileStore, fileStore) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index c7d8e15d73..ba85952f1d 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -62,12 +62,12 @@ type synctestConfig struct { //to the pivot node, and we check that nodes get the chunks //they are expected to store based on the syncing protocol. //Number of chunks and nodes can be provided via commandline too. -func TestSyncingViaRegistry(t *testing.T) { +func TestSyncingViaGlobalSync(t *testing.T) { //if nodes/chunks have been provided via commandline, //run the tests with these values if *nodes != 0 && *chunks != 0 { log.Info(fmt.Sprintf("Running test with %d chunks and %d nodes...", *chunks, *nodes)) - testSyncing(t, *chunks, *nodes) + testSyncingViaGlobalSync(t, *chunks, *nodes) } else { var nodeCnt []int var chnkCnt []int @@ -84,18 +84,21 @@ func TestSyncingViaRegistry(t *testing.T) { for _, chnk := range chnkCnt { for _, n := range nodeCnt { log.Info(fmt.Sprintf("Long running test with %d chunks and %d nodes...", chnk, n)) - testSyncing(t, chnk, n) + testSyncingViaGlobalSync(t, chnk, n) } } } } -func TestSyncingViaRPC(t *testing.T) { +func TestSyncingViaDirectSubscribe(t *testing.T) { //if nodes/chunks have been provided via commandline, //run the tests with these values if *nodes != 0 && *chunks != 0 { log.Info(fmt.Sprintf("Running test with %d chunks and %d nodes...", *chunks, *nodes)) - testSyncingViaRPC(t, *chunks, *nodes) + err := testSyncingViaDirectSubscribe(*chunks, *nodes) + if err != nil { + t.Fatal(err) + } } else { var nodeCnt []int var chnkCnt []int @@ -112,13 +115,16 @@ func TestSyncingViaRPC(t *testing.T) { for _, chnk := range chnkCnt { for _, n := range nodeCnt { log.Info(fmt.Sprintf("Long running test with %d chunks and %d nodes...", chnk, n)) - testSyncingViaRPC(t, chnk, n) + err := testSyncingViaDirectSubscribe(chnk, n) + if err != nil { + t.Fatal(err) + } } } } } -func testSyncing(t *testing.T, chunkCount int, nodeCount int) { +func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { sim := simulation.New(map[string]simulation.ServiceFunc{ "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { @@ -142,11 +148,9 @@ func testSyncing(t *testing.T, chunkCount int, nodeCount int) { DoSync: true, SyncUpdateDelay: 3 * time.Second, }) + bucket.Store(bucketKeyRegistry, r) - fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} - - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) @@ -267,14 +271,6 @@ func testSyncing(t *testing.T, chunkCount int, nodeCount int) { } } -//Do run the tests -func testSyncingViaRPC(t *testing.T, chunkCount int, nodeCount int) { - err := runSyncTest(chunkCount, nodeCount) - if err != nil { - t.Fatal(err) - } -} - /* The test generates the given number of chunks @@ -287,7 +283,7 @@ assuming that the snapshot file identifies a healthy kademlia network. The snapshot should have 'streamer' in its service list. */ -func runSyncTest(chunkCount int, nodeCount int) error { +func testSyncingViaDirectSubscribe(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) { @@ -308,11 +304,12 @@ func runSyncTest(chunkCount int, nodeCount int) error { delivery := NewDelivery(kad, db) r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil) + bucket.Store(bucketKeyRegistry, r) fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} + bucket.Store(bucketKeyFileStore, fileStore) - return testRegistry, cleanup, nil + return r, cleanup, nil }, }) @@ -353,15 +350,17 @@ func runSyncTest(chunkCount int, nodeCount int) error { filter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(4) eventC := sim.PeerEvents(ctx, nodeIDs, filter) - for j, id := range nodeIDs { + for j, node := range nodeIDs { log.Trace(fmt.Sprintf("Start syncing subscriptions: %d", j)) - client, err := sim.Net.GetNode(id).Client() - if err != nil { - return err - } //start syncing! + item, ok := sim.NodeItem(node, bucketKeyRegistry) + if !ok { + return fmt.Errorf("No registry") + } + registry := item.(*Registry) + var cnt int - err = client.CallContext(ctx, &cnt, "stream_startSyncing") + cnt, err = startSyncing(registry) if err != nil { return err } @@ -379,8 +378,7 @@ func runSyncTest(chunkCount int, nodeCount int) error { break } } - //get the the node at that index - //this is the node selected for upload + //select a random node for upload node := sim.RandomUpNode() item, ok := sim.NodeItem(node.ID, bucketKeyStore) if !ok { @@ -463,7 +461,7 @@ func runSyncTest(chunkCount int, nodeCount int) error { //issues `RequestSubscriptionMsg` to peers, based on po, by iterating over //the kademlia's `EachBin` function. //returns the number of subscriptions requested -func (r *TestRegistry) StartSyncing(ctx context.Context) (int, error) { +func startSyncing(r *Registry) (int, error) { var err error kad, ok := r.delivery.overlay.(*network.Kademlia) diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 2265d48de3..c75cf7bea1 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -114,10 +114,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) bucket.Store(bucketKeyFileStore, fileStore) - testRegistry := &TestRegistry{Registry: r, fileStore: fileStore} - bucket.Store(bucketKeyRegistry, testRegistry) - return testRegistry, cleanup, nil + return r, cleanup, nil }, })