forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-18 01:43:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm: added multiple file retrieval test

Browse source
This commit is contained in:
Fabio Barone 2018-04-12 21:47:19 -05:00
parent f530257139
commit e6f01d019d
3 changed files with 387 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -56,13 +56,11 @@ var (
var (
defaultSkipCheck bool
defaultDoRetrieve bool
waitPeerErrC chan error
chunkSize = 4096
registries map[discover.NodeID]*TestRegistry
createStoreFunc func(id discover.NodeID, addr *network.BzzAddr) (storage.ChunkStore, error)
getRetrieveFunc = defaultRetrieveFunc
doRetrieve = defaultDoRetrieve
subscriptionCount = 0
)
@ -97,7 +95,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
deliveries[id] = delivery
r := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{
SkipCheck: defaultSkipCheck,
DoRetrieve: defaultDoRetrieve,
DoRetrieve: false,
})
RegisterSwarmSyncerServer(r, db)
RegisterSwarmSyncerClient(r, db)

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

@ -19,6 +19,7 @@ import (
"context"
"fmt"
"math/rand"
"strings"
"testing"
"time"
@ -30,6 +31,13 @@ import (
"github.com/ethereum/go-ethereum/swarm/storage"
)
//constants for random file generation
const (
minFileSize = 2
maxFileSize = 40
charset = ".,/.!@#$%^&*()-=:;<>abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
)
func initRetrievalTest() {
//global func to get overlay address from discover ID
toAddr = func(id discover.NodeID) *network.BzzAddr {
@ -64,6 +72,30 @@ func initRetrievalTest() {
}
}
//This test is a retrieval test for nodes.
//A configurable number of nodes can be
//provided to the test.
//Files are uploaded to nodes, other nodes try to retrieve the file
//Number of nodes can be provided via commandline too.
func TestFileRetrieval(t *testing.T) {
if *nodes != 0 {
fileRetrievalTest(t, *nodes)
} else {
var nodeCnt []int
//if the `longrunning` flag has been provided
//run more test combinations
if *longrunning {
nodeCnt = []int{16, 32, 128, 256}
} else {
//default test
nodeCnt = []int{16}
}
for _, n := range nodeCnt {
fileRetrievalTest(t, n)
}
}
}
//This test is a retrieval test for nodes.
//One node is randomly selected to be the pivot node.
//A configurable number of chunks and nodes can be
@ -96,6 +128,33 @@ func TestRetrieval(t *testing.T) {
}
}
//Every test runs 3 times, a live, a history, and a live AND history
func fileRetrievalTest(t *testing.T, nodeCount int) {
//test live and NO history
log.Info("Testing live and no history", "nodeCount", nodeCount)
live = true
history = false
err := runFileRetrievalTest(nodeCount)
if err != nil {
t.Fatal(err)
}
//test history only
log.Info("Testing history only", "nodeCount", nodeCount)
live = false
history = true
err = runFileRetrievalTest(nodeCount)
if err != nil {
t.Fatal(err)
}
//finally test live and history
log.Info("Testing live and history", "nodeCount", nodeCount)
live = true
err = runFileRetrievalTest(nodeCount)
if err != nil {
t.Fatal(err)
}
}
//Every test runs 3 times, a live, a history, and a live AND history
func retrievalTest(t *testing.T, chunkCount int, nodeCount int) {
//test live and NO history
@ -123,6 +182,273 @@ func retrievalTest(t *testing.T, chunkCount int, nodeCount int) {
}
}
/*
The upload is done by dependency to the global
`live` and `history` variables;
If `live` is set, first stream subscriptions are established,
then files are uploaded to nodes.
If `history` is enabled, first upload files, then build up subscriptions.
The test loads a snapshot file to construct the swarm network,
assuming that the snapshot file identifies a healthy
kademlia network. Nevertheless a health check runs in the
simulation's `action` function.
The snapshot should have 'streamer' in its service list.
*/
func runFileRetrievalTest(nodeCount int) error {
//for every run (live, history), int the variables
initRetrievalTest()
//the ids of the snapshot nodes, initiate only now as we need nodeCount
ids = make([]discover.NodeID, nodeCount)
//channel to check for disconnection errors
disconnectC := make(chan error)
//channel to close disconnection watcher routine
quitC := make(chan struct{})
//the test conf (using same as in `snapshot_sync_test`
conf = &synctestConfig{}
//map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Key, 0)
//load nodes from the snapshot file
net, err := initNetWithSnapshot(nodeCount)
if err != nil {
return err
}
//do cleanup after test is terminated
defer func() {
//shutdown the snapshot network
net.Shutdown()
//after the test, clean up local stores initialized with createLocalStoreForId
localStoreCleanup()
//finally clear all data directories
datadirsCleanup()
}()
//get the nodes of the network
nodes := net.GetNodes()
//iterate over all nodes...
for c := 0; c < len(nodes); c++ {
//create an array of discovery nodeIDS
ids[c] = nodes[c].ID()
a := network.ToOverlayAddr(ids[c].Bytes())
//append it to the array of all overlay addresses
conf.addrs = append(conf.addrs, a)
conf.addrToIdMap[string(a)] = ids[c]
}
//needed for healthy call
ppmap = network.NewPeerPot(testMinProxBinSize, ids, conf.addrs)
//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
trigger := make(chan discover.NodeID)
//simulation action
action := func(ctx context.Context) error {
//first run the health check on all nodes,
//wait until nodes are all healthy
ticker := time.NewTicker(200 * time.Millisecond)
defer ticker.Stop()
for range ticker.C {
healthy := true
for _, id := range ids {
r := registries[id]
//PeerPot for this node
pp := ppmap[id]
//call Healthy RPC
h := r.delivery.overlay.Healthy(pp)
//print info
log.Debug(r.delivery.overlay.String())
log.Debug(fmt.Sprintf("IS HEALTHY: %t", h.GotNN && h.KnowNN && h.Full))
if !h.GotNN || !h.Full {
healthy = false
break
}
}
if healthy {
break
}
}
if history {
log.Info("Uploading for history")
//If testing only history, we upload the chunk(s) first
conf.hashes, randomFiles, err = uploadFilesToNodes(nodes)
if err != nil {
return err
}
}
//variables needed to wait for all subscriptions established before uploading
errc := make(chan error)
//now setup and start event watching in order to know when we can upload
ctx, watchCancel := context.WithTimeout(context.Background(), MaxTimeout*time.Second)
defer watchCancel()
log.Info("Setting up stream subscription")
//We need two iterations, one to subscribe to the subscription events
//(so we know when setup phase is finished), and one to
//actually run the stream subscriptions. We can't do it in the same iteration,
//because while the first nodes in the loop are setting up subscriptions,
//the latter ones have not subscribed to listen to peer events yet,
//and then we miss events.
//first iteration: setup disconnection watcher and subscribe to peer events
for j, id := range ids {
log.Trace(fmt.Sprintf("Subscribe to subscription events: %d", j))
client, err := net.GetNode(id).Client()
if err != nil {
return err
}
//watch for peers disconnecting
err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
if err != nil {
return err
}
//check for `SubscribeMsg` events to know when setup phase is complete
watchSubscriptionEvents(ctx, id, client, errc)
}
//second iteration: start syncing and setup stream subscriptions
for j, id := range ids {
log.Trace(fmt.Sprintf("Start syncing and stream subscriptions: %d", j))
client, err := net.GetNode(id).Client()
if err != nil {
return err
}
//start syncing!
var cnt int
err = client.CallContext(ctx, &cnt, "stream_startSyncing")
if err != nil {
return err
}
//increment the number of subscriptions we need to wait for
//by the count returned from startSyncing (SYNC subscriptions)
subscriptionCount += cnt
//now also add the number of RETRIEVAL_REQUEST subscriptions
for snid := range registries[id].peers {
subscriptionCount++
err = client.CallContext(ctx, nil, "stream_subscribeStream", snid, NewStream(swarmChunkServerStreamName, "", false), nil, Top)
if err != nil {
return err
}
}
}
//now wait until the number of expected subscriptions has been finished
//`watchSubscriptionEvents` will write with a `nil` value to errc
//every time a `SubscriptionMsg` has been received
for err := range errc {
if err != nil {
return err
}
//`nil` received, decrement count
subscriptionCount--
//all subscriptions received
if subscriptionCount == 0 {
break
}
}
log.Info("Stream subscriptions successfully requested, action terminated")
if live {
//upload generated files to nodes
var hashes []storage.Key
var rfiles []string
hashes, rfiles, err = uploadFilesToNodes(nodes)
if err != nil {
return err
}
conf.hashes = append(conf.hashes, hashes...)
randomFiles = append(randomFiles, rfiles...)
//signal to the trigger loop that the upload has finished
uploadFinished <- struct{}{}
}
return nil
}
//check defines what will be checked during the test
check := func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
case e := <-disconnectC:
log.Error(e.Error())
return false, fmt.Errorf("Disconnect event detected, network unhealthy")
default:
}
log.Trace(fmt.Sprintf("Checking node: %s", id))
//if there are more than one chunk, test only succeeds if all expected chunks are found
allSuccess := true
//check on the node's dpa (netstore)
dpa := registries[id].dpa
//check all chunks
for i, hash := range conf.hashes {
reader := dpa.Retrieve(hash)
//check that we can read the file size and that it corresponds to the generated file size
if s, err := reader.Size(nil); err != nil || s != int64(len(randomFiles[i])) {
allSuccess = false
log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
} else {
log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
}
}
return allSuccess, nil
}
//for each tick, run the checks on all nodes
timingTicker := time.NewTicker(5 * time.Second)
defer timingTicker.Stop()
go func() {
//for live upload, we should wait for uploads to have finished
//before starting to trigger the checks, due to file size
if live {
<-uploadFinished
}
for range timingTicker.C {
for i := 0; i < len(ids); i++ {
log.Trace(fmt.Sprintf("triggering step %d, id %s", i, ids[i]))
trigger <- ids[i]
}
}
}()
log.Info("Starting simulation run...")
timeout := MaxTimeout * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
//run the simulation
result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
Action: action,
Trigger: trigger,
Expect: &simulations.Expectation{
Nodes: ids,
Check: check,
},
})
if result.Error != nil {
return result.Error
}
return nil
}
/*
The test generates the given number of chunks.
@ -152,14 +478,10 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
quitC := make(chan struct{})
//the test conf (using same as in `snapshot_sync_test`
conf = &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int)
//map of discover ID to kademlia overlay address
conf.idToAddrMap = make(map[discover.NodeID][]byte)
//map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored
conf.chunks = make([]storage.Key, 0)
conf.hashes = make([]storage.Key, 0)
//load nodes from the snapshot file
net, err := initNetWithSnapshot(nodeCount)
if err != nil {
@ -167,7 +489,6 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
}
//do cleanup after test is terminated
defer func() {
doRetrieve = defaultDoRetrieve
//shutdown the snapshot network
net.Shutdown()
//after the test, clean up local stores initialized with createLocalStoreForId
@ -189,7 +510,6 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
a := network.ToOverlayAddr(ids[c].Bytes())
//append it to the array of all overlay addresses
conf.addrs = append(conf.addrs, a)
conf.idToAddrMap[ids[c]] = a
conf.addrToIdMap[string(a)] = ids[c]
}
@ -227,7 +547,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
if history {
log.Info("Uploading for history")
//If testing only history, we upload the chunk(s) first
conf.chunks, err = uploadFileToSingleNodeStore(uploadNode.ID(), chunkCount)
conf.hashes, err = uploadFileToSingleNodeStore(uploadNode.ID(), chunkCount)
if err != nil {
return err
}
@ -237,7 +557,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
errc := make(chan error)
//now setup and start event watching in order to know when we can upload
ctx, watchCancel := context.WithTimeout(context.Background(), MAX_TIMEOUT*time.Second)
ctx, watchCancel := context.WithTimeout(context.Background(), MaxTimeout*time.Second)
defer watchCancel()
log.Info("Setting up stream subscription")
@ -314,7 +634,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
if err != nil {
return err
}
conf.chunks = append(conf.chunks, chnks...)
conf.hashes = append(conf.hashes, chnks...)
}
return nil
@ -345,7 +665,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
//check on the node's dpa (netstore)
dpa := registries[id].dpa
//check all chunks
for _, chnk := range conf.chunks {
for _, chnk := range conf.hashes {
reader := dpa.Retrieve(chnk)
//assuming that reading the Size of the chunk is enough to know we found it
if s, err := reader.Size(nil); err != nil || s != chunkSize {
@ -372,7 +692,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
log.Info("Starting simulation run...")
timeout := MAX_TIMEOUT * time.Second
timeout := MaxTimeout * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
@ -392,3 +712,43 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
return nil
}
//upload generated files to nodes
//every node gets one file uploaded
func uploadFilesToNodes(nodes []*simulations.Node) ([]storage.Key, []string, error) {
nodeCnt := len(nodes)
log.Debug(fmt.Sprintf("Uploading %d files to nodes", nodeCnt))
//array holding generated files
rfiles := make([]string, nodeCnt)
//array holding the root hashes of the files
rootkeys := make([]storage.Key, nodeCnt)
//for every node, generate a file and upload
for i, n := range nodes {
id := n.ID()
dpa := registries[id].dpa
//generate a file
rfiles[i] = generateRandomFile()
//store it (upload it) on the dpa
rk, wait, err := dpa.Store(strings.NewReader(rfiles[i]), int64(len(rfiles[i])), false)
log.Debug("Uploaded random string file to node")
wait()
if err != nil {
return nil, nil, err
}
rootkeys[i] = rk
}
return rootkeys, rfiles, nil
}
//generate a random file (string)
func generateRandomFile() string {
//generate a random file size between minFileSize and maxFileSize
fileSize := rand.Intn(maxFileSize-minFileSize) + minFileSize
log.Debug(fmt.Sprintf("Generated file with filesize %d kB", fileSize))
b := make([]byte, fileSize*1024)
for i := range b {
b[i] = charset[rand.Intn(len(charset))]
}
return string(b)
}

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

@ -42,7 +42,7 @@ import (
)
const testMinProxBinSize = 2
const MAX_TIMEOUT = 600
const MaxTimeout = 600
var (
pof = pot.DefaultPof(256)
@ -61,10 +61,9 @@ var (
type synctestConfig struct {
addrs [][]byte
chunks []storage.Key
hashes []storage.Key
idToChunksMap map[discover.NodeID][]int
chunksToNodesMap map[string][]int
idToAddrMap map[discover.NodeID][]byte
addrToIdMap map[string]discover.NodeID
}
@ -201,12 +200,10 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
conf = &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int)
//map of discover ID to kademlia overlay address
conf.idToAddrMap = make(map[discover.NodeID][]byte)
//map of overlay address to discover ID
conf.addrToIdMap = make(map[string]discover.NodeID)
//array where the generated chunk hashes will be stored
conf.chunks = make([]storage.Key, 0)
conf.hashes = make([]storage.Key, 0)
//channel to trigger node checks in the simulation
trigger := make(chan discover.NodeID)
//channel to check for disconnection errors
@ -254,7 +251,6 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
//the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID,
//so we need to know which overlay addr maps to which nodeID
conf.idToAddrMap[ids[c]] = a
conf.addrToIdMap[string(a)] = ids[c]
}
log.Info("Test config successfully initialized")
@ -297,7 +293,7 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
if err != nil {
return err
}
conf.chunks = append(conf.chunks, chunks...)
conf.hashes = append(conf.hashes, chunks...)
//finally map chunks to the closest addresses
mapKeysToNodes(conf)
}
@ -306,7 +302,7 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
errc := make(chan error)
//now setup and start event watching in order to know when we can upload
ctx, watchCancel := context.WithTimeout(context.Background(), MAX_TIMEOUT*time.Second)
ctx, watchCancel := context.WithTimeout(context.Background(), MaxTimeout*time.Second)
defer watchCancel()
log.Info("Setting up stream subscription")
@ -384,13 +380,13 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
log.Info("Stream subscriptions successfully requested")
if live {
//now upload the chunks to the selected random single node
chunks, err := uploadFileToSingleNodeStore(node.ID(), chunkCount)
hashes, err := uploadFileToSingleNodeStore(node.ID(), chunkCount)
if err != nil {
return err
}
conf.chunks = append(conf.chunks, chunks...)
conf.hashes = append(conf.hashes, hashes...)
//finally map chunks to the closest addresses
log.Debug(fmt.Sprintf("Uploaded chunks for live syncing: %v", conf.chunks))
log.Debug(fmt.Sprintf("Uploaded chunks for live syncing: %v", conf.hashes))
mapKeysToNodes(conf)
log.Info(fmt.Sprintf("Uploaded %d chunks to random single node", chunkCount))
}
@ -421,7 +417,7 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
//for each expected chunk, check if it is in the local store
for _, ch := range localChunks {
//get the real chunk by the index in the index array
chunk := conf.chunks[ch]
chunk := conf.hashes[ch]
log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
//check if the expected chunk is indeed in the localstore
if _, err := lstore.Get(chunk); err != nil {
@ -449,7 +445,7 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
log.Info("Starting simulation run...")
timeout := MAX_TIMEOUT * time.Second
timeout := MaxTimeout * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
@ -527,11 +523,11 @@ func mapKeysToNodes(conf *synctestConfig) {
np, _, _ = pot.Add(np, a, pof)
}
//for each address, run EachNeighbour on the chunk hashes pot to identify closest nodes
log.Trace(fmt.Sprintf("Generated hash chunk(s): %v", conf.chunks))
for i := 0; i < len(conf.chunks); i++ {
log.Trace(fmt.Sprintf("Generated hash chunk(s): %v", conf.hashes))
for i := 0; i < len(conf.hashes); i++ {
pl := 256 //highest possible proximity
var nns []int
np.EachNeighbour([]byte(conf.chunks[i]), pof, func(val pot.Val, po int) bool {
np.EachNeighbour([]byte(conf.hashes[i]), pof, func(val pot.Val, po int) bool {
a := val.([]byte)
if pl < 256 && pl != po {
return false
@ -548,7 +544,7 @@ func mapKeysToNodes(conf *synctestConfig) {
}
return true
})
kmap[string(conf.chunks[i])] = nns
kmap[string(conf.hashes[i])] = nns
}
for addr, chunks := range nodemap {
//this selects which chunks are expected to be found with the given node