From 9c28a5f1967bea0f81acf537fdd2397f54542d2b Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 8 Mar 2018 17:03:48 -0500 Subject: [PATCH] swarm: parametrized sync tests for live and history --- p2p/simulations/mocker.go | 18 +- swarm/network/stream/common_test.go | 2 +- swarm/network/stream/messages.go | 8 +- swarm/network/stream/peer.go | 8 +- .../network/stream/snapshot_retrieval_test.go | 270 ------------- swarm/network/stream/snapshot_sync_test.go | 362 ++++++++---------- swarm/network/stream/streamer_test.go | 2 +- swarm/network/stream/syncer.go | 4 +- swarm/network/stream/testing/testing.go | 13 +- 9 files changed, 196 insertions(+), 491 deletions(-) delete mode 100644 swarm/network/stream/snapshot_retrieval_test.go diff --git a/p2p/simulations/mocker.go b/p2p/simulations/mocker.go index 389b1e3ec3..b370fe2cd2 100644 --- a/p2p/simulations/mocker.go +++ b/p2p/simulations/mocker.go @@ -103,13 +103,7 @@ func startStop(net *Network, quit chan struct{}, nodeCount int) { func probabilistic(net *Network, quit chan struct{}, nodeCount int) { nodes, err := connectNodesInRing(net, nodeCount) if err != nil { - select { - case <-quit: - //error may be due to abortion of mocking; so the quit channel is closed - return - default: - panic("Could not startup node network for mocker") - } + panic("Could not startup node network for mocker") } for { select { @@ -150,7 +144,7 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) { log.Debug(fmt.Sprintf("node %v shutting down", nodes[i])) err := net.Stop(nodes[i]) if err != nil { - log.Error("Error stopping node", "node", nodes[i]) + log.Error(fmt.Sprintf("Error stopping node %s", nodes[i])) wg.Done() continue } @@ -158,7 +152,7 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) { time.Sleep(randWait) err := net.Start(id) if err != nil { - log.Error("Error starting node", "node", id) + log.Error(fmt.Sprintf("Error starting node %s", id)) } wg.Done() }(nodes[i]) @@ -175,7 +169,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error) conf := adapters.RandomNodeConfig() node, err := net.NewNodeWithConfig(conf) if err != nil { - log.Error("Error creating a node!", "err", err) + log.Error("Error creating a node! %s", err) return nil, err } ids[i] = node.ID() @@ -183,7 +177,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error) for _, id := range ids { if err := net.Start(id); err != nil { - log.Error("Error starting a node!", "err", err) + log.Error("Error starting a node! %s", err) return nil, err } log.Debug(fmt.Sprintf("node %v starting up", id)) @@ -191,7 +185,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error) for i, id := range ids { peerID := ids[(i+1)%len(ids)] if err := net.Connect(id, peerID); err != nil { - log.Error("Error connecting a node to a peer!", "err", err) + log.Error("Error connecting a node to a peer! %s", err) return nil, err } } diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index b5de857b48..e384c0f707 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -44,7 +44,7 @@ import ( ) var ( - adapter = flag.String("adapter", "sim", "type of simulation: sim|socket|exec|docker") + adapter = flag.String("adapter", "socket", "type of simulation: sim|socket|exec|docker") loglevel = flag.Int("loglevel", 4, "verbosity of logs") ) diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 02a09d1e6a..3d18a321e7 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -110,6 +110,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { go func() { if err := p.SendOfferedHashes(os, from, to); err != nil { + log.Error("ERROR in SendOfferedHashes, DROPPING peer!", "err", err) p.Drop(err) } }() @@ -127,6 +128,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { } go func() { if err := p.SendOfferedHashes(os, req.History.From, req.History.To); err != nil { + log.Error("ERROR in SendOfferedHashes, DROPPING peer!", "err", err) p.Drop(err) } }() @@ -235,11 +237,13 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { } go func() { select { - case <-time.After(30 * time.Second): + case <-time.After(120 * time.Second): + log.Error("ERROR in handleOfferedHashesMsg, DROPPING peer!", "err", "TIMEOUT") p.Drop(err) return case err := <-c.next: if err != nil { + log.Error("ERROR in handleOfferedHashesMsg, DROPPING peer!", "err", err) p.Drop(err) return } @@ -247,6 +251,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To) err := p.SendPriority(msg, c.priority) if err != nil { + log.Error("ERROR in handleOfferedHashesMsg, DROPPING peer!", "err", err) p.Drop(err) } }() @@ -279,6 +284,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { // launch in go routine since GetBatch blocks until new hashes arrive go func() { if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { + log.Error("ERROR in handleWantedHashesMsg, DROPPING peer!", "err", err) p.Drop(err) } }() diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 3b0a90b0f9..324b1577f3 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -30,7 +30,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -var sendTimeout = 5 * time.Second +var sendTimeout = 30 * time.Second type notFoundError struct { t string @@ -83,7 +83,6 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { // Deliver sends a storeRequestMsg protocol message to the peer func (p *Peer) Deliver(chunk *storage.Chunk, priority uint8) error { - //fmt.Println(fmt.Sprintf("DELIVER from %s: to %s : chunk: %s", string(p.streamer.addr.Under()), p.ID(), chunk.String())) msg := &ChunkDeliveryMsg{ Key: chunk.Key, SData: chunk.SData, @@ -294,13 +293,8 @@ func (p *Peer) setClientParams(s Stream, params *clientParams) error { if p.clients[s] != nil { return fmt.Errorf("client %s already exists", s) } -<<<<<<< HEAD if p.clientParams[s] != nil { return fmt.Errorf("client params %s already set", s) -======= - if p.clientParams[sk] != nil { - return fmt.Errorf("client params %v already set, %s to %s", sk, p.streamer.addr.ID(), p.ID()) ->>>>>>> e13194f15... swarm: Subscribe to bins with RequestSubscription } p.clientParams[s] = params return nil diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go deleted file mode 100644 index 709756462d..0000000000 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ /dev/null @@ -1,270 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . -package stream - -import ( - //"context" - crand "crypto/rand" - "flag" - "fmt" - "io" - "math/rand" - // "os" - "github.com/ethereum/go-ethereum/log" - "testing" - "time" - // "github.com/ethereum/go-ethereum/node" - // "github.com/ethereum/go-ethereum/pot" - "github.com/ethereum/go-ethereum/p2p/discover" - //"github.com/ethereum/go-ethereum/p2p/simulations" - // "github.com/ethereum/go-ethereum/p2p/simulations/adapters" - //"github.com/ethereum/go-ethereum/swarm/network" - "github.com/ethereum/go-ethereum/swarm/storage" - //streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing" -) - -var rootHash storage.Key - -func init() { - flag.Parse() - rand.Seed(time.Now().Unix()) - - initRetrievalTest() -} - -func initRetrievalTest() { -} - -func TestRetrieval_4(t *testing.T) { retrievalTest(t, 4) } - -/* -func TestRetrieval_1(t *testing.T) { retrievalTest(t, 1) } -func TestSyncing_4(t *testing.T) { testSyncing(t, 4) } -func TestSyncing_8(t *testing.T) { testSyncing(t, 8) } -func TestSyncing_32(t *testing.T) { testSyncing(t, 32) } -func TestSyncing_128(t *testing.T) { testSyncing(t, 128) } -func TestSyncing_256(t *testing.T) { testSyncing(t, 256) } -func TestSyncing_1024(t *testing.T) { testSyncing(t,1024) } - -// Benchmarks to test the average time it takes for an N-node ring -// to full a healthy kademlia topology -func BenchmarkSyncing_1(b *testing.B) { benchmarkSyncing(b, 1) } -func BenchmarkSyncing_4(b *testing.B) { benchmarkSyncing(b, 4) } -func BenchmarkSyncing_8(b *testing.B) { benchmarkSyncing(b, 8) } -func BenchmarkSyncing_32(b *testing.B) { benchmarkSyncing(b, 32) } -func BenchmarkSyncing_128(b *testing.B) { benchmarkSyncing(b, 128) } -func BenchmarkSyncing_256(b *testing.B) { benchmarkSyncing(b, 256) } -func BenchmarkSyncing_1024(b *testing.B) { benchmarkSyncing(b, 1024) } - -func benchmarkSyncing(b *testing.B, chunkCount int) { - for i := 0; i < b.N; i++ { - result, err := testSyncing(b.T, chunkCount) - if err != nil { - b.Fatalf("setting up simulation failed", result) - } - if result.Error != nil { - b.Logf("simulation failed: %s", result.Error) - } - } -} - -*/ -func retrievalTest(t *testing.T, chunkCount int) { - err := runRetrievalTest(chunkCount) - if err != nil { - t.Fatal(err) - } -} - -/* -The test generates the given number of chunks, -then uploads these to a random node. -Afterwards for every chunk generated, the nearest node addresses -are identified, syncing is started, and finally we verify -that the nodes closer to the 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 runRetrievalTest(chunkCount int) error { - /* - //First load the snapshot from the file - net, err := initNetWithSnapshot() - if err != nil { - return err - } - defer net.Shutdown() - - //get the nodes of the network - nodes := net.GetNodes() - //select one index at random... - idx := rand.Intn(len(nodes)) - //...and get the the node at that index - //this is the node selected for upload - uploadNode := nodes[idx] - //now select a node at random which will be used to retrieve - ridx := rand.Intn(len(nodes)) - //make sure uploadNode nad retrieveNode are not the same - if ridx == idx { - if ridx == len(nodes)-1 { - ridx = 0 - } else { - ridx += 1 - } - } - retrieveNode := nodes[ridx] - //iterate over all nodes... - for c := 0; c < len(nodes); c++ { - //create an array of discovery nodeIDS - ids[c] = nodes[c].ID() - } - - // channel to signal simulation initialisation with action call complete - // or node disconnections - //disconnectC := make(chan error) - //quitC := make(chan struct{}) - - //after the test, clean up local stores initialized with createLocalStoreForId - defer localStoreCleanup() - - trigger := make(chan discover.NodeID) - //triggerCheck defines what will be checked during the test - triggerCheck := func(ctx context.Context, id discover.NodeID) (bool, error) { - select { - case <-ctx.Done(): - return false, ctx.Err() - //case <-disconnectC: - // log.Error("Disconnect event detected") - // return false, ctx.Err() - default: - } - - log.Warn(fmt.Sprintf("Checking node: %s", id)) - //select the !!!!NETstore!!! for the given node - /* - lstore := stores[id] - if _,err := lstore.Get(rootHash); err !=nil { - log.Warn("File Not Found") - return false, nil - } - log.Warn("File Found") - */ - /* - return true, nil - } - - //for each tick, select a new node to be checked - ticker := time.NewTicker(time.Second * 1) - go func() { - for i := 0; i < len(ids); i++ { - <-ticker.C - trigger <- ids[i] - log.Debug(fmt.Sprintf("triggering step %d, id %s", i, ids[i])) - } - }() - - timeout := 300 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - //define the action to be performed before the test checks: start syncing - action := func(ctx context.Context) error { - // need to wait till an aynchronous process registers the peers in streamer.peers - // that is used by Subscribe - // the global peerCount function tells how many connections each node has - // TODO: this is to be reimplemented with peerEvent watcher without global var - i := 0 - for err := range waitPeerErrC { - if err != nil { - return fmt.Errorf("error waiting for peers: %s", err) - } - i++ - if i == len(ids)-1 { - break - } - } - - // each node Subscribes to each other's swarmChunkServerStreamName - for j := 0; j < len(ids); j++ { - log.Debug(fmt.Sprintf("subscribe: %d", j)) - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - client, err := net.GetNode(ids[j]).Client() - if err != nil { - return err - } - //RPC call to subscribe, select bin 0 - //client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false) - // report disconnect events to the error channel cos peers should not disconnect - //err = streamTesting.WatchDisconnections(ids[j], client, disconnectC, quitC) - //if err != nil { - // return err - //} - // start syncing, i.e., subscribe to upstream peers po 1 bin - //each node subscribes to the next index, last subscribes to 0 - idx := j + 1 - if j == len(ids)-1 { - idx = 0 - } - sid := ids[idx] - client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false) - } - //now upload the chunks to the selected random single node - rootHash, err = uploadFileToRandomNodeStore(node.ID(), chunkCount) - if err != nil { - return err - } - //finally map chunks to the closest addresses - //chunksForAddressesMap = mapIdsToKeys(chunks, ids) - log.Debug(fmt.Sprintf("%v", chunksForAddressesMap)) - - return nil - } - //run the simulation - result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{ - Action: action, - Trigger: trigger, - Expect: &simulations.Expectation{ - Nodes: ids, - Check: triggerCheck, - }, - }) - //close(quitC) - if result.Error != nil { - return result.Error - } - */ - return nil -} - -//upload a file(chunks) to a single local node store -func uploadFileToRandomNodeStore(id discover.NodeID, chunkCount int) (storage.Key, error) { - log.Debug(fmt.Sprintf("Uploading to node id: %s", id)) - lstore := stores[id] - size := chunkCount * chunkSize - dpa := storage.NewDPA(lstore, storage.NewChunkerParams()) - dpa.Start() - rootHash, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) - wait() - if err != nil { - return nil, err - } - - defer dpa.Stop() - - return rootHash, nil -} diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index e028b495eb..9a9acc4df1 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -24,6 +24,7 @@ import ( "io/ioutil" "math/rand" "os" + "sync" "testing" "time" @@ -40,6 +41,7 @@ import ( ) const testMinProxBinSize = 2 +const MAX_TIMEOUT = 600 var ( pof = pot.DefaultPof(256) @@ -50,10 +52,8 @@ var ( datadirs map[discover.NodeID]string ppmap map[discover.NodeID]*network.PeerPot - requestedSubscriptions int - receivedSubscriptions int - subscriptionsFinished bool - printed bool + live bool + history bool ) type synctestConfig struct { @@ -67,8 +67,7 @@ type synctestConfig struct { } func init() { - //rand.Seed(time.Now().Unix()) - rand.Seed(100) + rand.Seed(time.Now().Unix()) initSyncTest() } @@ -106,72 +105,77 @@ func initSyncTest() { } } -func TestSyncing_1_16(t *testing.T) { testSyncing(t, 1, 16) } -func TestSyncing_1_32(t *testing.T) { testSyncing(t, 1, 32) } -func TestSyncing_1_64(t *testing.T) { testSyncing(t, 1, 64) } -func TestSyncing_1_128(t *testing.T) { testSyncing(t, 1, 128) } -func TestSyncing_1_256(t *testing.T) { testSyncing(t, 1, 256) } -func TestSyncing_4_16(t *testing.T) { testSyncing(t, 4, 16) } -func TestSyncing_4_32(t *testing.T) { testSyncing(t, 4, 32) } -func TestSyncing_4_64(t *testing.T) { testSyncing(t, 4, 64) } -func TestSyncing_4_128(t *testing.T) { testSyncing(t, 4, 128) } -func TestSyncing_4_256(t *testing.T) { testSyncing(t, 4, 256) } -func TestSyncing_8_16(t *testing.T) { testSyncing(t, 8, 16) } -func TestSyncing_8_32(t *testing.T) { testSyncing(t, 8, 32) } -func TestSyncing_8_64(t *testing.T) { testSyncing(t, 8, 64) } -func TestSyncing_8_128(t *testing.T) { testSyncing(t, 8, 128) } -func TestSyncing_8_256(t *testing.T) { testSyncing(t, 8, 256) } -func TestSyncing_32_16(t *testing.T) { testSyncing(t, 32, 16) } -func TestSyncing_32_32(t *testing.T) { testSyncing(t, 32, 32) } -func TestSyncing_32_64(t *testing.T) { testSyncing(t, 32, 64) } -func TestSyncing_32_128(t *testing.T) { testSyncing(t, 32, 128) } -func TestSyncing_32_256(t *testing.T) { testSyncing(t, 32, 256) } -func TestSyncing_128_16(t *testing.T) { testSyncing(t, 128, 16) } -func TestSyncing_128_32(t *testing.T) { testSyncing(t, 128, 32) } -func TestSyncing_128_64(t *testing.T) { testSyncing(t, 128, 64) } -func TestSyncing_128_128(t *testing.T) { testSyncing(t, 128, 128) } -func TestSyncing_128_256(t *testing.T) { testSyncing(t, 128, 256) } -func TestSyncing_256_16(t *testing.T) { testSyncing(t, 256, 16) } -func TestSyncing_256_32(t *testing.T) { testSyncing(t, 256, 32) } -func TestSyncing_256_64(t *testing.T) { testSyncing(t, 256, 64) } -func TestSyncing_256_128(t *testing.T) { testSyncing(t, 256, 128) } +//This file executes a number of tests with the syntax +//TestSyncing_x_y +//x is the number of chunks which will be uploaded +//y is the number of nodes for the test +func TestSyncing_1_16(t *testing.T) { testSyncing(t, 1, 16) } +func TestSyncing_1_32(t *testing.T) { testSyncing(t, 1, 32) } +func TestSyncing_1_64(t *testing.T) { testSyncing(t, 1, 64) } +func TestSyncing_1_128(t *testing.T) { testSyncing(t, 1, 128) } +func TestSyncing_1_256(t *testing.T) { testSyncing(t, 1, 256) } +func TestSyncing_4_16(t *testing.T) { testSyncing(t, 4, 16) } +func TestSyncing_4_32(t *testing.T) { testSyncing(t, 4, 32) } +func TestSyncing_4_64(t *testing.T) { testSyncing(t, 4, 64) } +func TestSyncing_4_128(t *testing.T) { testSyncing(t, 4, 128) } +func TestSyncing_8_16(t *testing.T) { testSyncing(t, 8, 16) } +func TestSyncing_8_32(t *testing.T) { testSyncing(t, 8, 32) } +func TestSyncing_8_64(t *testing.T) { testSyncing(t, 8, 64) } +func TestSyncing_8_128(t *testing.T) { testSyncing(t, 8, 128) } +func TestSyncing_32_16(t *testing.T) { testSyncing(t, 32, 16) } +func TestSyncing_32_32(t *testing.T) { testSyncing(t, 32, 32) } +func TestSyncing_32_64(t *testing.T) { testSyncing(t, 32, 64) } +func TestSyncing_128_16(t *testing.T) { testSyncing(t, 128, 16) } +func TestSyncing_128_32(t *testing.T) { testSyncing(t, 128, 32) } +func TestSyncing_128_64(t *testing.T) { testSyncing(t, 128, 64) } +func TestSyncing_256_16(t *testing.T) { testSyncing(t, 256, 16) } +func TestSyncing_256_32(t *testing.T) { testSyncing(t, 256, 32) } +func TestSyncing_1024_16(t *testing.T) { testSyncing(t, 1024, 16) } + +//The following tests have been disabled because they seem to hit resource limits +//on developer machines and/or are long running +/* +func TestSyncing_256_64(t *testing.T) { testSyncing(t, 256, 64) } +func TestSyncing_8_256(t *testing.T) { testSyncing(t, 8, 256) } +func TestSyncing_32_256(t *testing.T) { testSyncing(t, 32, 256) } +func TestSyncing_32_128(t *testing.T) { testSyncing(t, 32, 128) } +func TestSyncing_256_128(t *testing.T) { testSyncing(t, 256, 128) } +func TestSyncing_128_128(t *testing.T) { testSyncing(t, 128, 128) } func TestSyncing_256_256(t *testing.T) { testSyncing(t, 256, 256) } -func TestSyncing_1024_16(t *testing.T) { testSyncing(t, 1024, 16) } -func TestSyncing_1024_32(t *testing.T) { testSyncing(t, 1024, 32) } -func TestSyncing_1024_64(t *testing.T) { testSyncing(t, 1024, 64) } +func TestSyncing_128_256(t *testing.T) { testSyncing(t, 128, 256) } +func TestSyncing_1024_32(t *testing.T) { testSyncing(t, 1024, 32) } +func TestSyncing_1024_64(t *testing.T) { testSyncing(t, 1024, 64) } func TestSyncing_1024_128(t *testing.T) { testSyncing(t, 1024, 128) } func TestSyncing_1024_256(t *testing.T) { testSyncing(t, 1024, 256) } - -// Benchmarks to test the average time it takes for an N-node ring -// to full a healthy kademlia topology -/* -func BenchmarkSyncing_1(b *testing.B) { benchmarkSyncing(b, 1) } -func BenchmarkSyncing_4(b *testing.B) { benchmarkSyncing(b, 4) } -func BenchmarkSyncing_8(b *testing.B) { benchmarkSyncing(b, 8) } -func BenchmarkSyncing_32(b *testing.B) { benchmarkSyncing(b, 32) } -func BenchmarkSyncing_128(b *testing.B) { benchmarkSyncing(b, 128) } -func BenchmarkSyncing_256(b *testing.B) { benchmarkSyncing(b, 256) } -func BenchmarkSyncing_1024(b *testing.B) { benchmarkSyncing(b, 1024) } - -func benchmarkSyncing(b *testing.B, chunkCount int) { - for i := 0; i < b.N; i++ { - result, err := testSyncing(b.T, chunkCount) - if err != nil { - b.Fatalf("setting up simulation failed", result) - } - if result.Error != nil { - b.Logf("simulation failed: %s", result.Error) - } - } -} */ func testSyncing(t *testing.T, chunkCount int, nodeCount int) { ids = make([]discover.NodeID, nodeCount) - err := runSyncTest(chunkCount, nodeCount) + + //test live and NO history + log.Info("Testing live and no history") + live = true + history = false + err := runSyncTest(chunkCount, nodeCount, live, history) if err != nil { t.Fatal(err) } + //finaly test history only + log.Info("Testing history only") + live = false + history = true + err = runSyncTest(chunkCount, nodeCount, live, history) + if err != nil { + t.Fatal(err) + } + //test live and history + log.Info("Testing live and history") + live = true + err = runSyncTest(chunkCount, nodeCount, live, history) + if err != nil { + t.Fatal(err) + } + } /* @@ -186,11 +190,15 @@ 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. -This tests LIVE syncing, as the file is uploaded *after* sync streams have been setup. -For HISTORY syncing a different test is needed. +For every test run, a series of three tests will be executed: +- a LIVE test first, where first subscriptions are established, + then a file (random chunks) is uploaded +- a HISTORY test, where the file is uploaded first, and then + the subscriptions are established +- a crude LIVE AND HISTORY test last, where (different) chunks + are uploaded twice, once before and once after subscriptions */ -func runSyncTest(chunkCount int, nodeCount int) error { - +func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error { //initialize the test struct conf = &synctestConfig{} //mapping of nearest node addresses for chunk hashes @@ -201,9 +209,8 @@ func runSyncTest(chunkCount int, nodeCount int) error { conf.idToAddrMap = make(map[discover.NodeID][]byte) //map of overlay address to discover ID conf.addrToIdMap = make(map[string]discover.NodeID) + conf.chunks = make([]storage.Key, 0) //First load the snapshot from the file - var actionTicker *time.Ticker - var timingTicker *time.Ticker trigger := make(chan discover.NodeID) // channel to signal simulation initialisation with action call complete // or node disconnections @@ -215,21 +222,17 @@ func runSyncTest(chunkCount int, nodeCount int) error { if err != nil { return err } - //define the cleanup function - cleanup := func() { - timingTicker.Stop() - actionTicker.Stop() - close(quitC) - close(disconnectC) - //after the test, clean up local stores initialized with createLocalStoreForId - localStoreCleanup() + //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() - } - //do cleanup after test is terminated - defer cleanup() + //close(disconnectC) + close(quitC) + }() //get the nodes of the network nodes := net.GetNodes() //select one index at random... @@ -258,39 +261,53 @@ func runSyncTest(chunkCount int, nodeCount int) error { //only needed for healthy call when debugging ppmap = network.NewPeerPot(testMinProxBinSize, ids, conf.addrs) - //variables needed to wait for all subscriptions established before uploading - subscriptionsDone := make(chan struct{}) - errc := make(chan error) //define the action to be performed before the test checks: start syncing action := func(ctx context.Context) error { + if history { + log.Info("Uploading for history") + //If testing only history, we upload the chunk(s) first + chunks, err := uploadFileToSingleNodeStore(node.ID(), chunkCount) + if err != nil { + return err + } + conf.chunks = append(conf.chunks, chunks...) + //finally map chunks to the closest addresses + mapKeysToNodes(conf) + } + + errc := make(chan error) + //variables needed to wait for all subscriptions established before uploading + subscriptionsDone := make(chan struct{}) + + //now setup and start event watching in order to know when we can upload + ctx, watchCancel := context.WithTimeout(context.Background(), MAX_TIMEOUT*time.Second) + defer watchCancel() + log.Info("Setting up stream subscription") // each node Subscribes to each other's swarmChunkServerStreamName + var wg sync.WaitGroup for j, id := range ids { log.Trace(fmt.Sprintf("subscribe: %d", j)) - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) - defer cancel() client, err := net.GetNode(id).Client() if err != nil { return err } + wg.Add(1) + watchSubscriptionEvents(ctx, id, client, &wg, errc) - //now setup and start event watching in order to know when we can upload - watchCtx, watchCancel := context.WithTimeout(context.Background(), 15*time.Second) - defer watchCancel() - watchSubscriptionEvents(watchCtx, id, client, subscriptionsDone, errc) - - if log.Lvl(*loglevel) == log.LvlDebug { - //print uploading node kademlia - if j == idx { - var kt string - err := client.CallContext(ctx, &kt, "stream_getKad") - if err != nil { - return err - } - - log.Debug("uploading node kad") - log.Debug(kt) + if log.Lvl(*loglevel) >= log.LvlDebug { + //uncomment this if to see only the uploader's node kademlia + //otherwise print all kademlias + //if j == idx { + var kt string + err = client.CallContext(ctx, &kt, "stream_getKad") + if err != nil { + return err } + + log.Debug("kad table " + node.ID().String()) + log.Debug(kt) + //} } //watch for peers disconnecting err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC) @@ -303,34 +320,33 @@ func runSyncTest(chunkCount int, nodeCount int) error { return err } } - //only at this point all subscriptions have been finished and - //and we have a final number of subscriptions we need to wait for - subscriptionsFinished = true + //now wait until the number of expected subscriptions has been finished + + go func() { + wg.Wait() + close(subscriptionsDone) + }() + select { case <-subscriptionsDone: - close(subscriptionsDone) case err := <-errc: return err } log.Info("Stream subscriptions successfully requested") - //now upload the chunks to the selected random single node - conf.chunks, err = uploadFileToSingleNodeStore(node.ID(), chunkCount) - if err != nil { - return err - } - log.Info(fmt.Sprintf("Uploaded %d chunks to random single node", chunkCount)) - //finally map chunks to the closest addresses - mapKeysToNodes(conf) - - //periodically check if chunks have arrived at nodes - actionTicker = time.NewTicker(time.Second / 100) - go func() { - startTime = time.Now() - for range actionTicker.C { - checkChunkIsAtNode(conf) + if live { + //now upload the chunks to the selected random single node + chunks, err := uploadFileToSingleNodeStore(node.ID(), chunkCount) + if err != nil { + return err } - }() + conf.chunks = append(conf.chunks, chunks...) + //finally map chunks to the closest addresses + log.Debug(fmt.Sprintf("Uploaded chunks for live syncing: %v", conf.chunks)) + mapKeysToNodes(conf) + log.Info(fmt.Sprintf("Uploaded %d chunks to random single node", chunkCount)) + } + log.Info("Action terminated") return nil @@ -341,9 +357,9 @@ func runSyncTest(chunkCount int, nodeCount int) error { select { case <-ctx.Done(): return false, ctx.Err() - case <-disconnectC: - log.Error("Disconnect event detected") - 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)) @@ -361,22 +377,23 @@ func runSyncTest(chunkCount int, nodeCount int) error { 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 { - log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) + log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) allSuccess = false } else { - log.Trace(fmt.Sprintf("Chunk %s FOUND for id %s", chunk, id)) + log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) } } return allSuccess, nil } - timeout := 120 * time.Second + timeout := MAX_TIMEOUT * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - timingTicker = time.NewTicker(time.Second * 1) //for each tick, run the checks on all nodes + timingTicker := time.NewTicker(time.Second * 1) + defer timingTicker.Stop() go func() { for range timingTicker.C { for i := 0; i < len(ids); i++ { @@ -396,6 +413,7 @@ func runSyncTest(chunkCount int, nodeCount int) error { Check: check, }, }) + if result.Error != nil { return result.Error } @@ -433,61 +451,24 @@ func (r *TestRegistry) StartSyncing(ctx context.Context) error { kad.EachBin(r.addr.Over(), pof, 0, func(conn network.OverlayConn, po int) bool { //identify begin and start index of the bin(s) we want to subscribe to log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), conf.addrToIdMap[string(conn.Address())], po)) - - err = r.RequestSubscription(conf.addrToIdMap[string(conn.Address())], NewStream("SYNC", []byte{uint8(po)}, true), &Range{}, Top) + //fmt.Printf("rs: %s peer %s bin %d\n", r.addr.ID(), conf.addrToIdMap[string(conn.Address())], po) + var histRange *Range + if !history { + histRange = nil + } else { + histRange = &Range{} + } + err = r.RequestSubscription(conf.addrToIdMap[string(conn.Address())], NewStream("SYNC", FormatSyncBinKey(uint8(po)), live), histRange, Top) if err != nil { log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err)) return false } - requestedSubscriptions += 1 return true }) return nil } -//periodically check if chunks have arrived at nodes -func checkChunkIsAtNode(conf *synctestConfig) { - allOk := true - if printed { - return - } - //for every chunk, get the array of nodes it should have arrived to - for chunk, nodes := range conf.chunksToNodesMap { - //for every of those nodes - for _, node := range nodes { - //check that the chunk is in that localstore - if ok, _ := stores[conf.addrToIdMap[string(conf.addrs[node])]].Get([]byte(chunk)); ok != nil { - //if it is there, write the amount of time passed in the retrievalMap - //first create the map for the chunk if it is not there yet - if len(conf.retrievalMap[chunk]) == 0 { - conf.retrievalMap[chunk] = make(map[string]time.Duration) - } - //if the time value for the chunk at this node has not been recorded yet... - if _, ok := conf.retrievalMap[chunk][string(conf.addrs[node])]; !ok { - //...record it - conf.retrievalMap[chunk][string(conf.addrs[node])] = time.Since(startTime) - } - } else { - allOk = false - } - //if one of the chunks hasn't arrived yet, don't print - if conf.retrievalMap[chunk][string(conf.addrs[node])] == 0 { - allOk = false - } - } - } - if allOk && !printed { - log.Info("All chunks arrived at destination") - for ch, n := range conf.retrievalMap { - for a, t := range n { - log.Info(fmt.Sprintf("Chunk %v at node %s took %v ms", storage.Key([]byte((ch))).String()[:8], conf.addrToIdMap[string(a)].String()[0:8], t.Seconds()*1e3)) - } - } - printed = true - } -} - //map chunk keys to addresses which are responsible func mapKeysToNodes(conf *synctestConfig) { kmap := make(map[string][]int) @@ -569,7 +550,7 @@ func uploadFileToSingleNodeStore(id discover.NodeID, chunkCount int) ([]storage. } //Here we wait until all connections from the snapshot are up -func waitForSnapshotConnsUp(ctx context.Context, net *simulations.Network, done chan struct{}, connCount int, errc chan error) { +func waitForSnapshotConnsUp(ctx context.Context, net *simulations.Network, connCount int, errc chan error) { arrivedConns := 0 events := make(chan *simulations.Event) //subscribe to all events from the network @@ -587,7 +568,7 @@ func waitForSnapshotConnsUp(ctx context.Context, net *simulations.Network, done arrivedConns++ //the amount of expected connections has been reached, so we can stop waiting if arrivedConns == connCount { - done <- struct{}{} + errc <- nil return } } @@ -597,29 +578,25 @@ func waitForSnapshotConnsUp(ctx context.Context, net *simulations.Network, done } } } - return } //initialize a network from a snapshot func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) { - adapter := "sim" - var a adapters.NodeAdapter //add the streamer service to the node adapter - //discovery["streamer"] = NewStreamerService - if adapter == "exec" { + if *adapter == "exec" { dirname, err := ioutil.TempDir(".", "") if err != nil { return nil, err } a = adapters.NewExecAdapter(dirname) - } else if adapter == "sock" { + } else if *adapter == "socket" { a = adapters.NewSocketAdapter(services) - } else if adapter == "tcp" { + } else if *adapter == "tcp" { a = adapters.NewTCPAdapter(services) - } else if adapter == "sim" { + } else if *adapter == "sim" { a = adapters.NewSimAdapter(services) } @@ -655,12 +632,11 @@ func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) { log.Info("Waiting for p2p connections to be established...") //wait until all node connections are established //setup variables - errc := make(chan error) ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() connCount := len(snap.Conns) - done := make(chan struct{}) - go waitForSnapshotConnsUp(ctx, net, done, connCount, errc) + errc := make(chan error) + go waitForSnapshotConnsUp(ctx, net, connCount, errc) //now we can load the snapshot err = net.Load(&snap) @@ -669,10 +645,10 @@ func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) { } //finally wait until connections are established select { - case <-done: - close(done) case err = <-errc: - return nil, err + if err != nil { + return nil, err + } } log.Info("Snapshot loaded and connections established") return net, nil @@ -680,7 +656,7 @@ func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) { //we want to wait for subscriptions to be established before uploading to test //that live syncing is working correctly -func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, done chan struct{}, errc chan error) { +func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, wg *sync.WaitGroup, errc chan error) { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { @@ -698,12 +674,8 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp case e := <-events: //just catch SubscribeMsg if e.Type == p2p.PeerEventTypeMsgRecv && e.Protocol == "stream" && e.MsgCode != nil && *e.MsgCode == 4 { - receivedSubscriptions += 1 - //only check for done if subscription process is finished - if subscriptionsFinished && (receivedSubscriptions == requestedSubscriptions) { - done <- struct{}{} - return - } + wg.Done() + return } case err := <-sub.Err(): if err != nil { diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index 30d005f7f6..44622c9955 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -46,7 +46,7 @@ func TestStreamerRequestSubscription(t *testing.T) { t.Fatal(err) } - stream := NewStream("foo", nil, false) + stream := NewStream("foo", "", false) err = streamer.RequestSubscription(tester.IDs[0], stream, &Range{}, Top) if err == nil || err.Error() != "stream foo not registered" { t.Fatalf("Expected error %v, got %v", "stream foo not registered", err) diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index c3317e734f..154fb89c04 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -193,9 +193,9 @@ func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) { // NeedData func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) { - chunk, need := s.db.GetOrCreateRequest(key) + chunk, created := s.db.GetOrCreateRequest(key) // TODO: we may want to request from this peer anyway even if the request exists - if chunk.ReqC == nil || need == false { + if chunk.ReqC == nil || created == false { return nil } // create request and wait until the chunk data arrives and is stored diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index ad76f4b39a..4367df8e35 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -223,17 +223,26 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error return fmt.Errorf("error getting peer events for node %v: %s", id, err) } go func() { + defer sub.Unsubscribe() for { select { case <-quitC: return case e := <-events: if e.Type == p2p.PeerEventTypeDrop { - errc <- fmt.Errorf("peerEvent for node %v: %v", id, e) + select { + case errc <- fmt.Errorf("peerEvent for node %v: %v", id, e): + case <-quitC: + return + } } case err := <-sub.Err(): if err != nil { - errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err) + select { + case errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err): + case <-quitC: + return + } } } }