forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 02:12:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/network/stream: TestIntervals not waiting for chunk to be stored

Browse source
This commit is contained in:
Fabio Barone 2018-07-27 11:53:50 -05:00
parent 0e3a050c87
commit 3d4f288a9b
6 changed files with 41 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
swarm/network/stream/common_test.go
View file

@ -49,7 +49,6 @@ var (
chunks = flag.Int("chunks", 0, "number of chunks") chunks = flag.Int("chunks", 0, "number of chunks")
useMockStore = flag.Bool("mockstore", false, "disabled mock store (default: enabled)") useMockStore = flag.Bool("mockstore", false, "disabled mock store (default: enabled)")
longrunning = flag.Bool("longrunning", false, "do run long-running tests") longrunning = flag.Bool("longrunning", false, "do run long-running tests")
waitKademlia = flag.Bool("waitkademlia", true, "wait for healthy kademlia before checking files availability")
bucketKeyDB = simulation.BucketKey("db") bucketKeyDB = simulation.BucketKey("db")
bucketKeyStore = simulation.BucketKey("store") bucketKeyStore = simulation.BucketKey("store")
@ -236,7 +235,7 @@ func generateRandomFile() (string, error) {
} }
//create a local store for the given node //create a local store for the given node
func createTestLocalStorageForId(id discover.NodeID, addr *network.BzzAddr) (storage.ChunkStore, string, error) { func createTestLocalStorageForID(id discover.NodeID, addr *network.BzzAddr) (storage.ChunkStore, string, error) {
var datadir string var datadir string
var err error var err error
datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString())) datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString()))

4
swarm/network/stream/delivery_test.go
View file

@ -314,7 +314,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -501,7 +501,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }

35
swarm/network/stream/intervals_test.go
View file

@ -60,7 +60,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -132,10 +132,6 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
return fmt.Errorf("No registry") return fmt.Errorf("No registry")
} }
registry := item.(*Registry) registry := item.(*Registry)
err = registry.Subscribe(storer, NewStream(externalStreamName, "", live), history, Top)
if err != nil {
return err
}
liveErrC := make(chan error) liveErrC := make(chan error)
historyErrC := make(chan error) historyErrC := make(chan error)
@ -174,7 +170,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
// live stream // live stream
var liveHashesChan chan []byte var liveHashesChan chan []byte
liveHashesChan, err = getHashes(registry, ctx, storer, NewStream(externalStreamName, "", true)) liveHashesChan, err = getHashes(ctx, registry, storer, NewStream(externalStreamName, "", true))
if err != nil { if err != nil {
log.Error("Subscription error: %v", "err", err) log.Error("Subscription error: %v", "err", err)
return return
@ -218,7 +214,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
// history stream // history stream
var historyHashesChan chan []byte var historyHashesChan chan []byte
historyHashesChan, err = getHashes(registry, ctx, storer, NewStream(externalStreamName, "", false)) historyHashesChan, err = getHashes(ctx, registry, storer, NewStream(externalStreamName, "", false))
if err != nil { if err != nil {
return return
} }
@ -256,6 +252,10 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
} }
}() }()
err = registry.Subscribe(storer, NewStream(externalStreamName, "", live), history, Top)
if err != nil {
return err
}
if err := <-liveErrC; err != nil { if err := <-liveErrC; err != nil {
return err return err
} }
@ -271,8 +271,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
} }
} }
func getHashes(r *Registry, ctx context.Context, peerId discover.NodeID, s Stream) (chan []byte, error) { func getHashes(ctx context.Context, r *Registry, peerID discover.NodeID, s Stream) (chan []byte, error) {
peer := r.getPeer(peerId) peer := r.getPeer(peerID)
client, err := peer.getClient(ctx, s) client, err := peer.getClient(ctx, s)
if err != nil { if err != nil {
@ -284,8 +284,8 @@ func getHashes(r *Registry, ctx context.Context, peerId discover.NodeID, s Strea
return c.hashes, nil return c.hashes, nil
} }
func enableNotifications(r *Registry, peerId discover.NodeID, s Stream) error { func enableNotifications(r *Registry, peerID discover.NodeID, s Stream) error {
peer := r.getPeer(peerId) peer := r.getPeer(peerID)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
@ -320,9 +320,16 @@ func (c *testExternalClient) NeedData(ctx context.Context, hash []byte) func() {
return nil return nil
} }
c.hashes <- hash c.hashes <- hash
return func() { //NOTE: This was failing on go1.9.x with a deadlock.
chunk.WaitToStore() //Sometimes this function would just block
} //It is commented now, but it may be well worth after the chunk refactor
//to re-enable this and see if the problem has been addressed
/*
return func() {
return chunk.WaitToStore()
}
*/
return nil
} }
func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {

12
swarm/network/stream/snapshot_retrieval_test.go
View file

@ -119,7 +119,7 @@ func runFileRetrievalTest(nodeCount int) error {
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -153,7 +153,7 @@ func runFileRetrievalTest(nodeCount int) error {
//map of discover ID to indexes of chunks expected at that ID //map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int) conf.idToChunksMap = make(map[discover.NodeID][]int)
//map of overlay address to discover ID //map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID) conf.addrToIDMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored //array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0) conf.hashes = make([]storage.Address, 0)
@ -175,7 +175,7 @@ func runFileRetrievalTest(nodeCount int) error {
//the proximity calculation is on overlay addr, //the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID, //the p2p/simulations check func triggers on discover.NodeID,
//so we need to know which overlay addr maps to which nodeID //so we need to know which overlay addr maps to which nodeID
conf.addrToIdMap[string(a)] = n conf.addrToIDMap[string(a)] = n
} }
//an array for the random files //an array for the random files
@ -262,7 +262,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -295,7 +295,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
//map of discover ID to indexes of chunks expected at that ID //map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int) conf.idToChunksMap = make(map[discover.NodeID][]int)
//map of overlay address to discover ID //map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID) conf.addrToIDMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored //array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0) conf.hashes = make([]storage.Address, 0)
@ -315,7 +315,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
//the proximity calculation is on overlay addr, //the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID, //the p2p/simulations check func triggers on discover.NodeID,
//so we need to know which overlay addr maps to which nodeID //so we need to know which overlay addr maps to which nodeID
conf.addrToIdMap[string(a)] = n conf.addrToIDMap[string(a)] = n
} }
//an array for the random files //an array for the random files

20
swarm/network/stream/snapshot_sync_test.go
View file

@ -47,7 +47,7 @@ type synctestConfig struct {
hashes []storage.Address hashes []storage.Address
idToChunksMap map[discover.NodeID][]int idToChunksMap map[discover.NodeID][]int
chunksToNodesMap map[string][]int chunksToNodesMap map[string][]int
addrToIdMap map[string]discover.NodeID addrToIDMap map[string]discover.NodeID
} }
//This test is a syncing test for nodes. //This test is a syncing test for nodes.
@ -125,7 +125,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -157,7 +157,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
//map of discover ID to indexes of chunks expected at that ID //map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int) conf.idToChunksMap = make(map[discover.NodeID][]int)
//map of overlay address to discover ID //map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID) conf.addrToIDMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored //array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0) conf.hashes = make([]storage.Address, 0)
@ -179,7 +179,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
//the proximity calculation is on overlay addr, //the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID, //the p2p/simulations check func triggers on discover.NodeID,
//so we need to know which overlay addr maps to which nodeID //so we need to know which overlay addr maps to which nodeID
conf.addrToIdMap[string(a)] = n conf.addrToIDMap[string(a)] = n
} }
//get the the node at that index //get the the node at that index
@ -281,7 +281,7 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
id := ctx.Config.ID id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForId(id, addr) store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -314,7 +314,7 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
//map of discover ID to indexes of chunks expected at that ID //map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int) conf.idToChunksMap = make(map[discover.NodeID][]int)
//map of overlay address to discover ID //map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID) conf.addrToIDMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored //array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0) conf.hashes = make([]storage.Address, 0)
@ -333,7 +333,7 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
//the proximity calculation is on overlay addr, //the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID, //the p2p/simulations check func triggers on discover.NodeID,
//so we need to know which overlay addr maps to which nodeID //so we need to know which overlay addr maps to which nodeID
conf.addrToIdMap[string(a)] = n conf.addrToIDMap[string(a)] = n
} }
var subscriptionCount int var subscriptionCount int
@ -466,7 +466,7 @@ func startSyncing(r *Registry, conf *synctestConfig) (int, error) {
histRange := &Range{} histRange := &Range{}
subCnt++ subCnt++
err = r.RequestSubscription(conf.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 { if err != nil {
log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err)) log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err))
return false return false
@ -499,7 +499,7 @@ func mapKeysToNodes(conf *synctestConfig) {
return false return false
} }
if pl == 256 || pl == po { if pl == 256 || pl == po {
log.Trace(fmt.Sprintf("appending %s", conf.addrToIdMap[string(a)])) log.Trace(fmt.Sprintf("appending %s", conf.addrToIDMap[string(a)]))
nns = append(nns, indexmap[string(a)]) nns = append(nns, indexmap[string(a)])
nodemap[string(a)] = append(nodemap[string(a)], i) nodemap[string(a)] = append(nodemap[string(a)], i)
} }
@ -514,7 +514,7 @@ func mapKeysToNodes(conf *synctestConfig) {
} }
for addr, chunks := range nodemap { for addr, chunks := range nodemap {
//this selects which chunks are expected to be found with the given node //this selects which chunks are expected to be found with the given node
conf.idToChunksMap[conf.addrToIdMap[addr]] = chunks conf.idToChunksMap[conf.addrToIDMap[addr]] = chunks
} }
log.Debug(fmt.Sprintf("Map of expected chunks by ID: %v", conf.idToChunksMap)) log.Debug(fmt.Sprintf("Map of expected chunks by ID: %v", conf.idToChunksMap))
conf.chunksToNodesMap = kmap conf.chunksToNodesMap = kmap

2
swarm/network/stream/syncer_test.go
View file

@ -84,7 +84,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
} }
store, datadir, err = createMockStore(globalStore, id, addr) store, datadir, err = createMockStore(globalStore, id, addr)
} else { } else {
store, datadir, err = createTestLocalStorageForId(id, addr) store, datadir, err = createTestLocalStorageForID(id, addr)
} }
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err