swarm/network/stream: addressed PR comments @janos

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)

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
 			}

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() {}

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)

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

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)
 	}