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: parametrized sync tests for live and history

Browse source
This commit is contained in:
Fabio Barone 2018-03-08 17:03:48 -05:00
parent acccab01b3
commit 9c28a5f196
9 changed files with 196 additions and 491 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
p2p/simulations/mocker.go
View file

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

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

@ -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")
)

8
swarm/network/stream/messages.go
View file

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

8
swarm/network/stream/peer.go
View file

@ -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

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

@ -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 <http://www.gnu.org/licenses/>.
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
}

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

@ -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 {

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

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

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

@ -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

13
swarm/network/stream/testing/testing.go
View file

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