diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index e3f06ad487..69862cc831 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -19,7 +19,6 @@ package stream import ( "context" crand "crypto/rand" - "encoding/binary" "errors" "flag" "fmt" @@ -35,7 +34,6 @@ import ( "github.com/ethereum/go-ethereum/log" "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/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/pot" @@ -185,118 +183,6 @@ func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) { return total, nil } -//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) - if err != nil { - return nil, err - } - - c := client.Client.(*testExternalClient) - - return c.hashes, nil -} - -func enableNotifications(r *Registry, peerId discover.NodeID, s Stream) error { - peer := r.getPeer(peerId) - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - client, err := peer.getClient(ctx, s) - if err != nil { - return err - } - - close(client.Client.(*testExternalClient).enableNotificationsC) - - return nil -} - -// TODO: merge functionalities of testExternalClient and testExternalServer -// with testClient and testServer. - -type testExternalClient struct { - hashes chan []byte - db *storage.DBAPI - enableNotificationsC chan struct{} -} - -func newTestExternalClient(db *storage.DBAPI) *testExternalClient { - return &testExternalClient{ - hashes: make(chan []byte), - db: db, - enableNotificationsC: make(chan struct{}), - } -} - -func (c *testExternalClient) NeedData(ctx context.Context, hash []byte) func() { - chunk, _ := c.db.GetOrCreateRequest(ctx, hash) - if chunk.ReqC == nil { - return nil - } - c.hashes <- hash - return func() { - chunk.WaitToStore() - } -} - -func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { - return nil -} - -func (c *testExternalClient) Close() {} - -const testExternalServerBatchSize = 10 - -type testExternalServer struct { - t string - keyFunc func(key []byte, index uint64) - sessionAt uint64 - maxKeys uint64 -} - -func newTestExternalServer(t string, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer { - if keyFunc == nil { - keyFunc = binary.BigEndian.PutUint64 - } - return &testExternalServer{ - t: t, - keyFunc: keyFunc, - sessionAt: sessionAt, - maxKeys: maxKeys, - } -} - -func (s *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { - if from == 0 && to == 0 { - from = s.sessionAt - to = s.sessionAt + testExternalServerBatchSize - } - if to-from > testExternalServerBatchSize { - to = from + testExternalServerBatchSize - 1 - } - if from >= s.maxKeys && to > s.maxKeys { - return nil, 0, 0, nil, io.EOF - } - if to > s.maxKeys { - to = s.maxKeys - } - b := make([]byte, HashSize*(to-from+1)) - for i := from; i <= to; i++ { - s.keyFunc(b[(i-from)*HashSize:(i-from+1)*HashSize], i) - } - return b, from, to, nil, nil -} - -func (s *testExternalServer) GetData(context.Context, []byte) ([]byte, error) { - return make([]byte, 4096), nil -} - -func (s *testExternalServer) Close() {} - func uploadFilesToNodes(sim *simulation.Simulation) ([]storage.Address, []string, error) { nodes := sim.UpNodeIDs() nodeCnt := len(nodes) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index ce9ee0a684..0dbada133c 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -424,8 +424,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } }() - log.Debug("Waiting for kademlia") if *waitKademlia { + log.Debug("Waiting for kademlia") if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { return err } diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index 7e973a1d74..70714b35ef 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -25,10 +25,12 @@ import ( "os" "sync" "testing" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" @@ -36,12 +38,6 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -var ( - externalStreamName = "externalStream" - externalStreamSessionAt uint64 = 50 - externalStreamMaxKeys uint64 = 100 -) - func TestIntervals(t *testing.T) { testIntervals(t, true, nil, false) testIntervals(t, false, NewRange(9, 26), false) @@ -55,6 +51,9 @@ func TestIntervals(t *testing.T) { func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { nodes := 2 chunkCount := dataChunkCount + externalStreamName := "externalStream" + externalStreamSessionAt := uint64(50) + externalStreamMaxKeys := uint64(100) sim := simulation.New(map[string]simulation.ServiceFunc{ "intervalsStreamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { @@ -201,8 +200,6 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { if i > externalStreamMaxKeys { return } - //case err = <-liveSubscription.Err(): - // return case <-ctx.Done(): return } @@ -254,8 +251,6 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { if i > historyTo { return } - //case err = <-historySubscription.Err(): - // return case <-ctx.Done(): return } @@ -276,3 +271,111 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { t.Fatal(result.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) + if err != nil { + return nil, err + } + + c := client.Client.(*testExternalClient) + + return c.hashes, nil +} + +func enableNotifications(r *Registry, peerId discover.NodeID, s Stream) error { + peer := r.getPeer(peerId) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + client, err := peer.getClient(ctx, s) + if err != nil { + return err + } + + close(client.Client.(*testExternalClient).enableNotificationsC) + + return nil +} + +type testExternalClient struct { + hashes chan []byte + db *storage.DBAPI + enableNotificationsC chan struct{} +} + +func newTestExternalClient(db *storage.DBAPI) *testExternalClient { + return &testExternalClient{ + hashes: make(chan []byte), + db: db, + enableNotificationsC: make(chan struct{}), + } +} + +func (c *testExternalClient) NeedData(ctx context.Context, hash []byte) func() { + chunk, _ := c.db.GetOrCreateRequest(ctx, hash) + if chunk.ReqC == nil { + return nil + } + c.hashes <- hash + return func() { + chunk.WaitToStore() + } +} + +func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { + return nil +} + +func (c *testExternalClient) Close() {} + +const testExternalServerBatchSize = 10 + +type testExternalServer struct { + t string + keyFunc func(key []byte, index uint64) + sessionAt uint64 + maxKeys uint64 +} + +func newTestExternalServer(t string, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer { + if keyFunc == nil { + keyFunc = binary.BigEndian.PutUint64 + } + return &testExternalServer{ + t: t, + keyFunc: keyFunc, + sessionAt: sessionAt, + maxKeys: maxKeys, + } +} + +func (s *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { + if from == 0 && to == 0 { + from = s.sessionAt + to = s.sessionAt + testExternalServerBatchSize + } + if to-from > testExternalServerBatchSize { + to = from + testExternalServerBatchSize - 1 + } + if from >= s.maxKeys && to > s.maxKeys { + return nil, 0, 0, nil, io.EOF + } + if to > s.maxKeys { + to = s.maxKeys + } + b := make([]byte, HashSize*(to-from+1)) + for i := from; i <= to; i++ { + s.keyFunc(b[(i-from)*HashSize:(i-from+1)*HashSize], i) + } + return b, from, to, nil, nil +} + +func (s *testExternalServer) GetData(context.Context, []byte) ([]byte, error) { + return make([]byte, 4096), nil +} + +func (s *testExternalServer) Close() {} diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index e48fc8fd15..27107b6b8b 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -135,7 +135,7 @@ func runFileRetrievalTest(nodeCount int) error { r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ DoSync: true, - SyncUpdateDelay: 0, + SyncUpdateDelay: 3 * time.Second, }) fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) @@ -320,11 +320,6 @@ func runRetrievalTest(chunkCount int, nodeCount int) error { //an array for the random files var randomFiles []string - //channel to signal when the upload has finished - //uploadFinished := make(chan struct{}) - //channel to trigger new node checks - - //get the the node at that index //this is the node selected for upload node := sim.RandomUpNode() item, ok := sim.NodeItem(node.ID, bucketKeyStore) diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index ba85952f1d..056390560d 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -42,11 +42,6 @@ import ( const testMinProxBinSize = 2 const MaxTimeout = 600 -var ( - //this global is needed to run the RPC test - globConf *synctestConfig -) - type synctestConfig struct { addrs [][]byte hashes []storage.Address @@ -214,7 +209,6 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { if *useMockStore { gDir, globalStore, err = createGlobalStore() if err != nil { - //return false, fmt.Errorf("Something went wrong; using mockStore enabled but globalStore is nil") return fmt.Errorf("Something went wrong; using mockStore enabled but globalStore is nil") } defer func() { @@ -227,7 +221,6 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { } for !allSuccess { for _, id := range nodeIDs { - //log.Trace("file uploaded", "node", id, "key", key.String()) //for each expected chunk, check if it is in the local store localChunks := conf.idToChunksMap[id] localSuccess := true @@ -281,7 +274,6 @@ chunk addresses actually do have the chunks in their local stores. The test loads a snapshot file to construct the swarm network, assuming that the snapshot file identifies a healthy kademlia network. The snapshot should have 'streamer' in its service list. - */ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { sim := simulation.New(map[string]simulation.ServiceFunc{ @@ -345,7 +337,6 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { } var subscriptionCount int - globConf = conf filter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(4) eventC := sim.PeerEvents(ctx, nodeIDs, filter) @@ -360,7 +351,7 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { registry := item.(*Registry) var cnt int - cnt, err = startSyncing(registry) + cnt, err = startSyncing(registry, conf) if err != nil { return err } @@ -410,7 +401,6 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { allSuccess := false for !allSuccess { for _, id := range nodeIDs { - //log.Trace("file uploaded", "node", id, "key", key.String()) //for each expected chunk, check if it is in the local store localChunks := conf.idToChunksMap[id] localSuccess := true @@ -461,7 +451,7 @@ func testSyncingViaDirectSubscribe(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 startSyncing(r *Registry) (int, error) { +func startSyncing(r *Registry, conf *synctestConfig) (int, error) { var err error kad, ok := r.delivery.overlay.(*network.Kademlia) @@ -476,7 +466,7 @@ func startSyncing(r *Registry) (int, error) { histRange := &Range{} subCnt++ - err = r.RequestSubscription(globConf.addrToIdMap[string(conn.Address())], NewStream("SYNC", FormatSyncBinKey(uint8(po)), true), histRange, Top) + err = r.RequestSubscription(conf.addrToIdMap[string(conn.Address())], NewStream("SYNC", FormatSyncBinKey(uint8(po)), true), histRange, Top) if err != nil { log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err)) return false diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index c75cf7bea1..8e5d937d9b 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -127,7 +127,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // defer cancel should come before defer simulation teardown defer cancel() - _, err := sim.AddNodesAndConnectFull(nodes) + _, err := sim.AddNodesAndConnectChain(nodes) if err != nil { t.Fatal(err) }