aboutsummaryrefslogtreecommitdiffstats
path: root/swarm/network/stream/syncer_test.go
diff options
context:
space:
mode:
Diffstat (limited to 'swarm/network/stream/syncer_test.go')
-rw-r--r--swarm/network/stream/syncer_test.go161
1 files changed, 54 insertions, 107 deletions
diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go
index 014ec9a98..be0752a9d 100644
--- a/swarm/network/stream/syncer_test.go
+++ b/swarm/network/stream/syncer_test.go
@@ -24,7 +24,6 @@ import (
"math"
"os"
"sync"
- "sync/atomic"
"testing"
"time"
@@ -38,7 +37,6 @@ import (
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
- mockmem "github.com/ethereum/go-ethereum/swarm/storage/mock/mem"
"github.com/ethereum/go-ethereum/swarm/testutil"
)
@@ -46,9 +44,15 @@ const dataChunkCount = 200
func TestSyncerSimulation(t *testing.T) {
testSyncBetweenNodes(t, 2, dataChunkCount, true, 1)
- testSyncBetweenNodes(t, 4, dataChunkCount, true, 1)
- testSyncBetweenNodes(t, 8, dataChunkCount, true, 1)
- testSyncBetweenNodes(t, 16, dataChunkCount, true, 1)
+ // This test uses much more memory when running with
+ // race detector. Allow it to finish successfully by
+ // reducing its scope, and still check for data races
+ // with the smallest number of nodes.
+ if !raceTest {
+ testSyncBetweenNodes(t, 4, dataChunkCount, true, 1)
+ testSyncBetweenNodes(t, 8, dataChunkCount, true, 1)
+ testSyncBetweenNodes(t, 16, dataChunkCount, true, 1)
+ }
}
func createMockStore(globalStore mock.GlobalStorer, id enode.ID, addr *network.BzzAddr) (lstore storage.ChunkStore, datadir string, err error) {
@@ -73,50 +77,46 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- var store storage.ChunkStore
- var datadir string
-
- node := ctx.Config.Node()
- addr := network.NewAddr(node)
+ addr := network.NewAddr(ctx.Config.Node())
//hack to put addresses in same space
addr.OAddr[0] = byte(0)
- if *useMockStore {
- store, datadir, err = createMockStore(mockmem.NewGlobalStore(), node.ID(), addr)
- } else {
- store, datadir, err = createTestLocalStorageForID(node.ID(), addr)
- }
+ netStore, delivery, clean, err := newNetStoreAndDeliveryWithBzzAddr(ctx, bucket, addr)
if err != nil {
return nil, nil, err
}
- bucket.Store(bucketKeyStore, store)
- cleanup = func() {
- store.Close()
- os.RemoveAll(datadir)
- }
- localStore := store.(*storage.LocalStore)
- netStore, err := storage.NewNetStore(localStore, nil)
- if err != nil {
- return nil, nil, err
- }
- bucket.Store(bucketKeyDB, netStore)
- kad := network.NewKademlia(addr.Over(), network.NewKadParams())
- delivery := NewDelivery(kad, netStore)
- netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
- bucket.Store(bucketKeyDelivery, delivery)
+ var dir string
+ var store *state.DBStore
+ if raceTest {
+ // Use on-disk DBStore to reduce memory consumption in race tests.
+ dir, err = ioutil.TempDir("", "swarm-stream-")
+ if err != nil {
+ return nil, nil, err
+ }
+ store, err = state.NewDBStore(dir)
+ if err != nil {
+ return nil, nil, err
+ }
+ } else {
+ store = state.NewInmemoryStore()
+ }
- r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
+ r := NewRegistry(addr.ID(), delivery, netStore, store, &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
SkipCheck: skipCheck,
}, nil)
- fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
- bucket.Store(bucketKeyFileStore, fileStore)
+ cleanup = func() {
+ r.Close()
+ clean()
+ if dir != "" {
+ os.RemoveAll(dir)
+ }
+ }
return r, cleanup, nil
-
},
})
defer sim.Close()
@@ -139,26 +139,10 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
nodeIndex[id] = i
}
- disconnections := sim.PeerEvents(
- context.Background(),
- sim.NodeIDs(),
- simulation.NewPeerEventsFilter().Drop(),
- )
-
- var disconnected atomic.Value
- go func() {
- for d := range disconnections {
- if d.Error != nil {
- log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
- disconnected.Store(true)
- }
- }
- }()
+ disconnected := watchDisconnections(ctx, sim)
defer func() {
- if err != nil {
- if yes, ok := disconnected.Load().(bool); ok && yes {
- err = errors.New("disconnect events received")
- }
+ if err != nil && disconnected.bool() {
+ err = errors.New("disconnect events received")
}
}()
@@ -167,7 +151,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
id := nodeIDs[j]
client, err := sim.Net.GetNode(id).Client()
if err != nil {
- t.Fatal(err)
+ return fmt.Errorf("node %s client: %v", id, err)
}
sid := nodeIDs[j+1]
client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", FormatSyncBinKey(1), false), NewRange(0, 0), Top)
@@ -183,7 +167,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
size := chunkCount * chunkSize
_, wait, err := fileStore.Store(ctx, testutil.RandomReader(j, size), int64(size), false)
if err != nil {
- t.Fatal(err.Error())
+ return fmt.Errorf("fileStore.Store: %v", err)
}
wait(ctx)
}
@@ -251,44 +235,26 @@ func TestSameVersionID(t *testing.T) {
v := uint(1)
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- var store storage.ChunkStore
- var datadir string
-
- node := ctx.Config.Node()
- addr := network.NewAddr(node)
-
- store, datadir, err = createTestLocalStorageForID(node.ID(), addr)
+ addr, netStore, delivery, clean, err := newNetStoreAndDelivery(ctx, bucket)
if err != nil {
return nil, nil, err
}
- bucket.Store(bucketKeyStore, store)
- cleanup = func() {
- store.Close()
- os.RemoveAll(datadir)
- }
- localStore := store.(*storage.LocalStore)
- netStore, err := storage.NewNetStore(localStore, nil)
- if err != nil {
- return nil, nil, err
- }
- bucket.Store(bucketKeyDB, netStore)
- kad := network.NewKademlia(addr.Over(), network.NewKadParams())
- delivery := NewDelivery(kad, netStore)
- netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
-
- bucket.Store(bucketKeyDelivery, delivery)
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
}, nil)
+ bucket.Store(bucketKeyRegistry, r)
+
//assign to each node the same version ID
r.spec.Version = v
- bucket.Store(bucketKeyRegistry, r)
+ cleanup = func() {
+ r.Close()
+ clean()
+ }
return r, cleanup, nil
-
},
})
defer sim.Close()
@@ -316,7 +282,7 @@ func TestSameVersionID(t *testing.T) {
//the peers should connect, thus getting the peer should not return nil
if registry.getPeer(nodes[1]) == nil {
- t.Fatal("Expected the peer to not be nil, but it is")
+ return errors.New("Expected the peer to not be nil, but it is")
}
return nil
})
@@ -333,46 +299,27 @@ func TestDifferentVersionID(t *testing.T) {
v := uint(0)
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- var store storage.ChunkStore
- var datadir string
-
- 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)
- cleanup = func() {
- store.Close()
- os.RemoveAll(datadir)
- }
- localStore := store.(*storage.LocalStore)
- netStore, err := storage.NewNetStore(localStore, nil)
+ addr, netStore, delivery, clean, err := newNetStoreAndDelivery(ctx, bucket)
if err != nil {
return nil, nil, err
}
- bucket.Store(bucketKeyDB, netStore)
- kad := network.NewKademlia(addr.Over(), network.NewKadParams())
- delivery := NewDelivery(kad, netStore)
- netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
-
- bucket.Store(bucketKeyDelivery, delivery)
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
}, nil)
+ bucket.Store(bucketKeyRegistry, r)
//increase the version ID for each node
v++
r.spec.Version = v
- bucket.Store(bucketKeyRegistry, r)
+ cleanup = func() {
+ r.Close()
+ clean()
+ }
return r, cleanup, nil
-
},
})
defer sim.Close()
@@ -400,7 +347,7 @@ func TestDifferentVersionID(t *testing.T) {
//getting the other peer should fail due to the different version numbers
if registry.getPeer(nodes[1]) != nil {
- t.Fatal("Expected the peer to be nil, but it is not")
+ return errors.New("Expected the peer to be nil, but it is not")
}
return nil
})
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-04 16:09:57 +0800