From 726ab4893b24b7b5e8cbe59b2ff6de0ba0a18cab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jano=C5=A1=20Gulja=C5=A1?= Date: Fri, 20 Apr 2018 12:22:13 +0200 Subject: [PATCH] swarm: high level swarm network simulation This commit includes TestSwarmNetwork test and changes to the swarm that are required for that test to pass. TestSwarmNetwork contains tests for static network simulation with 10 and 100 nodes, and also tests where nodes are added or removed while the availability of files is checked between changes. On every test step, a number of nodes is created and the same number of files are uploaded on each of nodes (one file per node). The check consists of trying to retrieve all files from all nodes. --- cmd/swarm/db.go | 1 + swarm/api/config.go | 4 + swarm/network/kademlia.go | 26 +- .../simulations/discovery/discovery_test.go | 4 +- swarm/network/stream/delivery.go | 23 +- swarm/network/stream/delivery_test.go | 10 +- swarm/network/stream/messages.go | 12 +- swarm/network/stream/peer.go | 1 + swarm/network/stream/stream.go | 17 +- swarm/network/stream/syncer.go | 42 +- swarm/network_test.go | 582 ++++++++++++++++++ swarm/storage/ldbstore.go | 6 +- swarm/storage/localstore.go | 58 +- swarm/storage/resource.go | 5 +- swarm/storage/types.go | 5 +- swarm/swarm.go | 25 +- 16 files changed, 753 insertions(+), 68 deletions(-) create mode 100644 swarm/network_test.go diff --git a/cmd/swarm/db.go b/cmd/swarm/db.go index a080be7f57..fe03f2d160 100644 --- a/cmd/swarm/db.go +++ b/cmd/swarm/db.go @@ -115,5 +115,6 @@ func openLDBStore(path string, basekey []byte) (*storage.LDBStore, error) { storeparams := storage.NewDefaultStoreParams() ldbparams := storage.NewLDBStoreParams(storeparams, path) + ldbparams.BaseKey = basekey return storage.NewLDBStore(ldbparams) } diff --git a/swarm/api/config.go b/swarm/api/config.go index 8f608d6e44..a4bf4c96d0 100644 --- a/swarm/api/config.go +++ b/swarm/api/config.go @@ -28,6 +28,7 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/services/swap" "github.com/ethereum/go-ethereum/swarm/storage" @@ -55,6 +56,7 @@ type Config struct { Port string PublicKey string BzzKey string + NodeID string NetworkId uint64 SwapEnabled bool SyncEnabled bool @@ -113,6 +115,7 @@ func (self *Config) Init(prvKey *ecdsa.PrivateKey) { self.PublicKey = pubkeyhex self.BzzKey = keyhex + self.NodeID = discover.PubkeyID(&prvKey.PublicKey).String() if self.SwapEnabled { self.Swap.Init(self.Contract, prvKey) @@ -120,6 +123,7 @@ func (self *Config) Init(prvKey *ecdsa.PrivateKey) { self.privateKey = prvKey self.LocalStoreParams.Init(self.Path) + self.LocalStoreParams.BaseKey = common.FromHex(keyhex) } func (self *Config) ShiftPrivateKey() (privKey *ecdsa.PrivateKey) { diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 8eba4ad52d..aebed4752d 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -204,6 +204,8 @@ func (k *Kademlia) Register(peers []OverlayAddr) error { k.addrCountC <- k.addrs.Size() } // log.Trace(fmt.Sprintf("%x registered %v peers, %v known, total: %v", k.BaseAddr()[:4], size, known, k.addrs.Size())) + + k.sendNeighbourhoodDepthChange() return nil } @@ -313,13 +315,7 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) { changed = true k.depth = depth } - if k.nDepthC != nil { - nDepth := k.neighbourhoodDepth() - if nDepth != k.nDepth { - k.nDepth = nDepth - k.nDepthC <- nDepth - } - } + k.sendNeighbourhoodDepthChange() return k.depth, changed } @@ -334,6 +330,21 @@ func (k *Kademlia) NeighbourhoodDepthC() <-chan int { return k.nDepthC } +// sendNeighbourhoodDepthChange sends new neighbourhood depth to k.nDepth channel +// if it is initialized. +func (k *Kademlia) sendNeighbourhoodDepthChange() { + // nDepthC is initialized when NeighbourhoodDepthC is called and returned by it. + // It provides signaling of neighbourhood depth change. + // This part of the code is sending new neighbourhood depth to nDepthC if that condition is met. + if k.nDepthC != nil { + nDepth := k.neighbourhoodDepth() + if nDepth != k.nDepth { + k.nDepth = nDepth + k.nDepthC <- nDepth + } + } +} + // AddrCountC returns the channel that sends a new // address count value on each change. // Not receiving from the returned channel will block Register function @@ -367,6 +378,7 @@ func (k *Kademlia) Off(p OverlayConn) { if k.addrCountC != nil { k.addrCountC <- k.addrs.Size() } + k.sendNeighbourhoodDepthChange() } } diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 5d0eb1ae7c..0490903329 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -416,7 +416,7 @@ func discoveryPersistenceSimulation(nodes, conns int, adapter adapters.NodeAdapt return fmt.Errorf("error stopping node %s: %s", node.ID().TerminalString(), err) } } - log.Info(fmt.Sprintf("shutting down nodes took: %s", time.Now().Sub(shutdownStarted))) + log.Info(fmt.Sprintf("shutting down nodes took: %s", time.Since(shutdownStarted))) persistenceEnabled = true discoveryEnabled = false restartTime = time.Now() @@ -430,7 +430,7 @@ func discoveryPersistenceSimulation(nodes, conns int, adapter adapters.NodeAdapt } } - log.Info(fmt.Sprintf("restarting nodes took: %s", time.Now().Sub(restartTime))) + log.Info(fmt.Sprintf("restarting nodes took: %s", time.Since(restartTime))) return nil } diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 21a2609485..28303fe327 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -22,6 +22,7 @@ import ( "fmt" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/swarm/network" @@ -136,19 +137,23 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e chunk, created := d.db.GetOrCreateRequest(req.Key) if chunk.ReqC != nil { if created { - if err := d.RequestFromPeers(chunk.Key[:], false, sp.ID()); err != nil { + if err := d.RequestFromPeers(chunk.Key[:], true, sp.ID()); err != nil { log.Warn("unable to forward chunk request", "peer", sp.ID(), "key", chunk.Key, "err", err) chunk.SetErrored(storage.ErrChunkForward) return nil } } go func() { - t := time.NewTimer(3 * time.Minute) + t := time.NewTimer(10 * time.Minute) defer t.Stop() + log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Key, "node", common.Bytes2Hex(d.overlay.BaseAddr()), "created", created) + start := time.Now() select { case <-chunk.ReqC: + log.Debug("retrieve request ReqC closed", "peer", sp.ID(), "hash", req.Key, "time", time.Since(start)) case <-t.C: + log.Debug("retrieve request timeout", "peer", sp.ID(), "hash", req.Key) chunk.SetErrored(storage.ErrChunkTimeout) return } @@ -208,14 +213,13 @@ R: } chunk.SData = req.SData d.db.Put(chunk) - chunk.WaitToStore() - err = chunk.GetErrored() - if err != nil { + + go func(req *ChunkDeliveryMsg) { + err := chunk.WaitToStore() if err == storage.ErrChunkInvalid { req.peer.Drop(err) } - } - close(chunk.ReqC) + }(req) } } @@ -241,11 +245,14 @@ func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ... Key: hash, SkipCheck: skipCheck, }, Top) + if err != nil { + return true + } success = true return false }) if success { - return err + return nil } return errors.New("no peer found") } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index d4e31343c5..be49ad8a40 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -82,7 +82,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { peer.handleSubscribeMsg(&SubscribeMsg{ Stream: NewStream(swarmChunkServerStreamName, "", false), - History: NewRange(0, 0), + History: nil, Priority: Top, }) @@ -129,9 +129,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { peerID := tester.IDs[0] peer := streamer.getPeer(peerID) + stream := NewStream(swarmChunkServerStreamName, "", false) + peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: NewStream(swarmChunkServerStreamName, "", false), - History: NewRange(0, 0), + Stream: stream, + History: nil, Priority: Top, }) @@ -163,7 +165,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { From: 0, // TODO: why is this 32??? To: 32, - Stream: NewStream(swarmChunkServerStreamName, "", false), + Stream: stream, }, Peer: peerID, }, diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 3e5c03d4d0..d2d2d99d75 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -17,6 +17,7 @@ package stream import ( + "errors" "fmt" "sync" "time" @@ -85,7 +86,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { } }() - log.Debug("%s received subscription", "from", p.streamer.addr.ID(), "peer", p.ID(), "stream", req.Stream, "history", req.History) + log.Debug("received subscription", "from", p.streamer.addr.ID(), "peer", p.ID(), "stream", req.Stream, "history", req.History) f, err := p.streamer.GetServerFunc(req.Stream.Name) if err != nil { @@ -216,7 +217,10 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { // }() go func() { wg.Wait() - c.next <- c.batchDone(p, req, hashes) + select { + case c.next <- c.batchDone(p, req, hashes): + case <-c.quit: + } }() // only send wantedKeysMsg if all missing chunks of the previous batch arrived // except @@ -239,7 +243,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { select { case <-time.After(120 * time.Second): log.Warn("ERROR in handleOfferedHashesMsg, DROPPING peer!", "err", "TIMEOUT") - p.Drop(err) + p.Drop(errors.New("handle offered hashes timeout")) return case err := <-c.next: if err != nil { @@ -247,6 +251,8 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { p.Drop(err) return } + case <-c.quit: + return } log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To) err := p.SendPriority(msg, c.priority) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 324b1577f3..79afd5d425 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -265,6 +265,7 @@ func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created boo priority: cp.priority, to: cp.to, next: next, + quit: make(chan struct{}), intervalsStore: p.streamer.intervalsStore, intervalsKey: intervalsKey, } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index b36ad27138..9ffa6205b4 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -92,8 +92,8 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) { return NewSwarmChunkServer(delivery.db), nil }) - streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ string, _ bool) (Client, error) { - return NewSwarmSyncerClient(p, delivery.db, false) + streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t string, live bool) (Client, error) { + return NewSwarmSyncerClient(p, delivery.db, false, NewStream(swarmChunkServerStreamName, t, live)) }) RegisterSwarmSyncerServer(streamer, db) RegisterSwarmSyncerClient(streamer, db) @@ -127,13 +127,13 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i // wait for kademlia table to be healthy time.Sleep(options.SyncUpdateDelay) - // initial requests for syncing subscription to peers - streamer.updateSyncing() - kad := streamer.delivery.overlay.(*network.Kademlia) depthC := latestIntC(kad.NeighbourhoodDepthC()) addressBookSizeC := latestIntC(kad.AddrCountC()) + // initial requests for syncing subscription to peers + streamer.updateSyncing() + for depth := range depthC { log.Debug("Kademlia neighbourhood depth change", "depth", depth) @@ -516,6 +516,7 @@ type client struct { sessionAt uint64 to uint64 next chan error + quit chan struct{} intervalsKey string intervalsStore state.Store @@ -600,7 +601,11 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error } func (c *client) close() { - close(c.next) + select { + case <-c.quit: + default: + close(c.quit) + } c.Close() } diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 799d097af4..9ecd2fe25a 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -99,13 +99,18 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 if to <= from || from >= s.sessionAt { to = math.MaxUint64 } - ticker := time.NewTicker(1 * time.Second) - defer ticker.Stop() + var ticker *time.Ticker + var wait bool for { - select { - case <-ticker.C: - case <-s.quit: - return nil, 0, 0, nil, nil + if wait { + if ticker == nil { + ticker = time.NewTicker(1000 * time.Millisecond) + } + select { + case <-ticker.C: + case <-s.quit: + return nil, 0, 0, nil, nil + } } err := s.db.Iterator(from, to, s.po, func(key storage.Key, idx uint64) bool { batch = append(batch, key[:]...) @@ -119,6 +124,10 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 if len(batch) > 0 { break } + wait = true + } + if wait { + ticker.Stop() } log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) @@ -134,18 +143,22 @@ type SwarmSyncerClient struct { retrieveC chan *storage.Chunk storeC chan *storage.Chunk db *storage.DBAPI - // chunker storage.Chunker + // chunker storage.Chunker currentRoot storage.Key requestFunc func(chunk *storage.Chunk) end, start uint64 + peer *Peer ignoreExistingRequest bool + stream Stream } // NewSwarmSyncerClient is a contructor for provable data exchange syncer -func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, ignoreExistingRequest bool) (*SwarmSyncerClient, error) { +func NewSwarmSyncerClient(p *Peer, db *storage.DBAPI, ignoreExistingRequest bool, stream Stream) (*SwarmSyncerClient, error) { return &SwarmSyncerClient{ - db: db, + db: db, + peer: p, ignoreExistingRequest: ignoreExistingRequest, + stream: stream, }, nil } @@ -188,16 +201,19 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, ignoreExistingRequest bool // RegisterSwarmSyncerClient registers the client constructor function for // to handle incoming sync streams func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) { - streamer.RegisterClientFunc("SYNC", func(p *Peer, _ string, love bool) (Client, error) { - return NewSwarmSyncerClient(p, db, true) + streamer.RegisterClientFunc("SYNC", func(p *Peer, t string, live bool) (Client, error) { + return NewSwarmSyncerClient(p, db, true, NewStream("SYNC", t, live)) }) } // NeedData func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) { - chunk, created := s.db.GetOrCreateRequest(key) + chunk, _ := s.db.GetOrCreateRequest(key) // TODO: we may want to request from this peer anyway even if the request exists - if chunk.ReqC == nil || (s.ignoreExistingRequest && !created) { + + // ignoreExistingRequest is temporary commented out until its functionality is verified. + // For now, this optimization can be disabled. + if chunk.ReqC == nil { //|| (s.ignoreExistingRequest && !created) { return nil } // create request and wait until the chunk data arrives and is stored diff --git a/swarm/network_test.go b/swarm/network_test.go new file mode 100644 index 0000000000..122d583a3f --- /dev/null +++ b/swarm/network_test.go @@ -0,0 +1,582 @@ +// 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 swarm + +import ( + "context" + "flag" + "fmt" + "io/ioutil" + "math/rand" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/node" + "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/api" + "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/storage" + colorable "github.com/mattn/go-colorable" +) + +var ( + loglevel = flag.Int("loglevel", 4, "verbosity of logs") + longrunning = flag.Bool("longrunning", false, "do run long-running tests") + waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability") +) + +func init() { + rand.Seed(time.Now().UnixNano()) + + flag.Parse() + + log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))) +} + +// TestSwarmNetwork runs a series of test simulations with +// static and dynamic Swarm nodes in network simulation, by +// uploading files to every node and retrieving them. +func TestSwarmNetwork(t *testing.T) { + for _, tc := range []struct { + name string + steps []testSwarmNetworkStep + timeout time.Duration + disabled bool + }{ + { + name: "10_nodes", + steps: []testSwarmNetworkStep{ + { + nodeCount: 10, + }, + }, + timeout: 45 * time.Second, + }, + { + name: "100_nodes", + steps: []testSwarmNetworkStep{ + { + nodeCount: 100, + }, + }, + timeout: 3 * time.Minute, + disabled: !*longrunning, + }, + { + name: "inc_node_count", + steps: []testSwarmNetworkStep{ + { + nodeCount: 2, + }, + { + nodeCount: 5, + }, + { + nodeCount: 10, + }, + }, + timeout: 90 * time.Second, + disabled: !*longrunning, + }, + { + name: "dec_node_count", + steps: []testSwarmNetworkStep{ + { + nodeCount: 10, + }, + { + nodeCount: 6, + }, + { + nodeCount: 3, + }, + }, + timeout: 90 * time.Second, + disabled: !*longrunning, + }, + { + name: "dec_inc_node_count", + steps: []testSwarmNetworkStep{ + { + nodeCount: 5, + }, + { + nodeCount: 3, + }, + { + nodeCount: 10, + }, + }, + timeout: 90 * time.Second, + }, + { + name: "inc_dec_node_count", + steps: []testSwarmNetworkStep{ + { + nodeCount: 3, + }, + { + nodeCount: 5, + }, + { + nodeCount: 25, + }, + { + nodeCount: 10, + }, + { + nodeCount: 4, + }, + }, + timeout: 5 * time.Minute, + disabled: !*longrunning, + }, + } { + if tc.disabled { + continue + } + t.Run(tc.name, func(t *testing.T) { + testSwarmNetwork(t, tc.timeout, tc.steps...) + }) + } +} + +// testSwarmNetworkStep is the configuration +// for the state of the simulation network. +type testSwarmNetworkStep struct { + // number of swarm nodes that must be in the Up state + nodeCount int +} + +// file represents the file uploaded on a particular node. +type file struct { + key storage.Key + data string + nodeID discover.NodeID +} + +// check represents a reference to a file that is retrieved +// from a particular node. +type check struct { + key string + nodeID discover.NodeID +} + +// testSwarmNetwork is a helper function used for testing different +// static and dynamic Swarm network simulations. +// It is responsible for: +// - Setting up a Swarm network simulation, and updates the number of nodes within the network on every step according to steps. +// - Uploading a unique file to every node on every step. +// - May wait for Kademlia on every node to be healthy. +// - Checking if a file is retrievable from all nodes. +func testSwarmNetwork(t *testing.T, timeout time.Duration, steps ...testSwarmNetworkStep) { + dir, err := ioutil.TempDir("", "swarm-network-test") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + + ctx := context.Background() + if timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + + swarms := make(map[discover.NodeID]*Swarm) + files := make([]file, 0) + + services := map[string]adapters.ServiceFunc{ + "swarm": func(ctx *adapters.ServiceContext) (node.Service, error) { + config := api.NewConfig() + config.PssEnabled = false + + dir, err := ioutil.TempDir(dir, "node") + if err != nil { + return nil, err + } + + config.Path = dir + + privkey, err := crypto.GenerateKey() + if err != nil { + return nil, err + } + + config.Init(privkey) + s, err := NewSwarm(nil, nil, config, nil) + if err != nil { + return nil, err + } + log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", s.bzz.BaseAddr())) + swarms[ctx.Config.ID] = s + return s, nil + }, + } + + a := adapters.NewSimAdapter(services) + net := simulations.NewNetwork(a, &simulations.NetworkConfig{ + ID: "0", + DefaultService: "swarm", + }) + defer net.Shutdown() + + trigger := make(chan discover.NodeID) + + sim := simulations.NewSimulation(net) + + for i, step := range steps { + log.Debug("test sync step", "n", i+1, "nodes", step.nodeCount) + + change := step.nodeCount - len(allNodeIDs(net)) + + if change > 0 { + _, err := addNodes(change, net) + if err != nil { + t.Fatal(err) + } + } else if change < 0 { + err := removeNodes(-change, net) + if err != nil { + t.Fatal(err) + } + } else { + t.Logf("step %v: no change in nodes", i) + continue + } + + nodeIDs := allNodeIDs(net) + shuffle(len(nodeIDs), func(i, j int) { + nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i] + }) + for _, id := range nodeIDs { + key, data, err := uploadFile(swarms[id]) + if err != nil { + t.Fatal(err) + } + log.Trace("file uploaded", "node", id, "key", key.String()) + files = append(files, file{ + key: key, + data: data, + nodeID: id, + }) + } + + // Prepare PeerPot map for checking Kademlia health + var ppmap map[string]*network.PeerPot + nIDs := allNodeIDs(net) + addrs := make([][]byte, len(nIDs)) + if *waitKademlia { + for i, id := range nIDs { + addrs[i] = swarms[id].bzz.BaseAddr() + } + ppmap = network.NewPeerPotMap(2, addrs) + } + + var checkStatusM sync.Map + var nodeStatusM sync.Map + var totalFoundCount uint64 + + result := sim.Run(ctx, &simulations.Step{ + Action: func(ctx context.Context) error { + if *waitKademlia { + // Wait for healthy Kademlia on every node before checking files + ticker := time.NewTicker(200 * time.Millisecond) + defer ticker.Stop() + + for range ticker.C { + healthy := true + log.Debug("kademlia health check", "node count", len(nIDs), "addr count", len(addrs)) + for i, id := range nIDs { + swarm := swarms[id] + //PeerPot for this node + addr := common.Bytes2Hex(swarm.bzz.BaseAddr()) + pp := ppmap[addr] + //call Healthy RPC + h := swarm.bzz.Healthy(pp) + //print info + log.Debug(swarm.bzz.String()) + log.Debug("kademlia", "empty bins", pp.EmptyBins, "gotNN", h.GotNN, "knowNN", h.KnowNN, "full", h.Full) + log.Debug("kademlia", "health", h.GotNN && h.KnowNN && h.Full, "addr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()), "id", id, "i", i) + log.Debug("kademlia", "ill condition", !h.GotNN || !h.Full, "addr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()), "id", id, "i", i) + if !h.GotNN || !h.Full { + healthy = false + break + } + } + if healthy { + break + } + } + } + + go func() { + // File retrieval check is repeated until all uploaded files are retrieved from all nodes + // or until the timeout is reached. + for { + if retrieve(net, files, swarms, trigger, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 { + return + } + } + }() + return nil + }, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: allNodeIDs(net), + Check: func(ctx context.Context, id discover.NodeID) (bool, error) { + // The check is done by a goroutine in the action function. + return true, nil + }, + }, + }) + if result.Error != nil { + t.Fatal(result.Error) + } + log.Debug("done: test sync step", "n", i+1, "nodes", step.nodeCount) + } +} + +// allNodeIDs is returning NodeID for every node that is Up. +func allNodeIDs(net *simulations.Network) (nodes []discover.NodeID) { + for _, n := range net.GetNodes() { + if n.Up { + nodes = append(nodes, n.ID()) + } + } + return +} + +// addNodes adds a number of nodes to the network. +func addNodes(count int, net *simulations.Network) (ids []discover.NodeID, err error) { + for i := 0; i < count; i++ { + nodeIDs := allNodeIDs(net) + l := len(nodeIDs) + nodeconf := adapters.RandomNodeConfig() + node, err := net.NewNodeWithConfig(nodeconf) + if err != nil { + return nil, fmt.Errorf("create node: %v", err) + } + err = net.Start(node.ID()) + if err != nil { + return nil, fmt.Errorf("start node: %v", err) + } + + log.Debug("created node", "id", node.ID()) + + // connect nodes in a chain + if l > 0 { + var otherNodeID discover.NodeID + for i := l - 1; i >= 0; i-- { + n := net.GetNode(nodeIDs[i]) + if n.Up { + otherNodeID = n.ID() + break + } + } + log.Debug("connect nodes", "one", node.ID(), "other", otherNodeID) + if err := net.Connect(node.ID(), otherNodeID); err != nil { + return nil, err + } + } + ids = append(ids, node.ID()) + } + return ids, nil +} + +// removeNodes stops a random nodes in the network. +func removeNodes(count int, net *simulations.Network) error { + for i := 0; i < count; i++ { + // allNodeIDs are returning only the Up nodes. + nodeIDs := allNodeIDs(net) + if len(nodeIDs) == 0 { + break + } + node := net.GetNode(nodeIDs[rand.Intn(len(nodeIDs))]) + if err := node.Stop(); err != nil { + return err + } + log.Debug("removed node", "id", node.ID()) + } + return nil +} + +// uploadFile, uploads a short file to the swarm instance +// using the api.Put method. +func uploadFile(swarm *Swarm) (storage.Key, string, error) { + b := make([]byte, 8) + _, err := rand.Read(b) + if err != nil { + return nil, "", err + } + // File data is very short, but it is ensured that its + // uniqueness is very certain. + data := fmt.Sprintf("test content %s %x", time.Now().Round(0), b) + k, wait, err := swarm.api.Put(data, "text/plain", false) + if err != nil { + return nil, "", err + } + if wait != nil { + wait() + } + return k, data, nil +} + +// retrieve is the function that is used for checking the availability of +// uploaded files in testSwarmNetwork test helper function. +func retrieve( + net *simulations.Network, + files []file, + swarms map[discover.NodeID]*Swarm, + trigger chan discover.NodeID, + checkStatusM *sync.Map, + nodeStatusM *sync.Map, + totalFoundCount *uint64, +) (missing uint64) { + shuffle(len(files), func(i, j int) { + files[i], files[j] = files[j], files[i] + }) + + var totalWg sync.WaitGroup + errc := make(chan error) + + nodeIDs := allNodeIDs(net) + + totalCheckCount := len(nodeIDs) * len(files) + + for _, id := range nodeIDs { + if _, ok := nodeStatusM.Load(id); ok { + continue + } + start := time.Now() + var checkCount uint64 + var foundCount uint64 + + totalWg.Add(1) + + var wg sync.WaitGroup + + for _, f := range files { + swarm := swarms[id] + + checkKey := check{ + key: f.key.String(), + nodeID: id, + } + if n, ok := checkStatusM.Load(checkKey); ok && n.(int) == 0 { + continue + } + + checkCount++ + wg.Add(1) + go func(f file, id discover.NodeID) { + defer wg.Done() + + log.Debug("api get: check file", "node", id.String(), "key", f.key.String(), "total files found", atomic.LoadUint64(totalFoundCount)) + + r, _, _, err := swarm.api.Get(f.key, "/") + if err != nil { + errc <- fmt.Errorf("api get: node %s, key %s, kademlia %s: %v", id, f.key, swarm.bzz.Hive, err) + return + } + d, err := ioutil.ReadAll(r) + if err != nil { + errc <- fmt.Errorf("api get: read response: node %s, key %s: kademlia %s: %v", id, f.key, swarm.bzz.Hive, err) + return + } + data := string(d) + if data != f.data { + errc <- fmt.Errorf("file contend missmatch: node %s, key %s, expected %q, got %q", id, f.key, f.data, data) + return + } + checkStatusM.Store(checkKey, 0) + atomic.AddUint64(&foundCount, 1) + log.Info("api get: file found", "node", id.String(), "key", f.key.String(), "content", data, "files found", atomic.LoadUint64(&foundCount)) + }(f, id) + } + + go func(id discover.NodeID) { + defer totalWg.Done() + wg.Wait() + + atomic.AddUint64(totalFoundCount, foundCount) + + if foundCount == checkCount { + log.Info("all files are found for node", "id", id.String(), "duration", time.Since(start)) + nodeStatusM.Store(id, 0) + trigger <- id + return + } + log.Debug("files missing for node", "id", id.String(), "check", checkCount, "found", foundCount) + }(id) + + } + + go func() { + totalWg.Wait() + close(errc) + }() + + var errCount int + for err := range errc { + if err != nil { + errCount++ + } + log.Warn(err.Error()) + } + + log.Info("check stats", "total check count", totalCheckCount, "total files found", atomic.LoadUint64(totalFoundCount), "total errors", errCount) + + return uint64(totalCheckCount) - atomic.LoadUint64(totalFoundCount) +} + +// Backported from stdlib https://golang.org/src/math/rand/rand.go?s=11175:11215#L333 +// +// Replace with rand.Shuffle from go 1.10 when go 1.9 support is dropped. +// +// shuffle pseudo-randomizes the order of elements. +// n is the number of elements. Shuffle panics if n < 0. +// swap swaps the elements with indexes i and j. +func shuffle(n int, swap func(i, j int)) { + if n < 0 { + panic("invalid argument to Shuffle") + } + + // Fisher-Yates shuffle: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle + // Shuffle really ought not be called with n that doesn't fit in 32 bits. + // Not only will it take a very long time, but with 2³¹! possible permutations, + // there's no way that any PRNG can have a big enough internal state to + // generate even a minuscule percentage of the possible permutations. + // Nevertheless, the right API signature accepts an int n, so handle it as best we can. + i := n - 1 + for ; i > 1<<31-1-1; i-- { + j := int(rand.Int63n(int64(i + 1))) + swap(i, j) + } + for ; i > 0; i-- { + j := int(rand.Int31n(int32(i + 1))) + swap(i, j) + } +} diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 42fa731aaf..a4aba77928 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -227,7 +227,11 @@ func encodeIndex(index *dpaDBIndex) []byte { } func encodeData(chunk *Chunk) []byte { - return append(chunk.Key[:], chunk.SData...) + // Always create a new underlying array for the returned byte slice. + // The chunk.Key array may be used in the returned slice which + // may be changed later in the code or by the LevelDB, resulting + // that the Key is changed as well. + return append(append([]byte{}, chunk.Key[:]...), chunk.SData...) } func decodeIndex(data []byte, index *dpaDBIndex) error { diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 4ab87c3e93..2cbb269aa7 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -88,6 +88,18 @@ func NewTestLocalStoreForAddr(params *LocalStoreParams) (*LocalStore, error) { return localStore, nil } +// Put is responsible for doing validation and storage of the chunk +// by using configured ChunkValidators, MemStore and LDBStore. +// If the chunk is not valid, its GetErrored function will +// return ErrChunkInvalid. +// This method will check if the chunk is already in the MemStore +// and it will return it if it is. If there is an error from +// the MemStore.Get, it will be returned by calling GetErrored +// on the chunk. +// This method is responsible for closing Chunk.ReqC channel +// when the chunk is stored in memstore. +// After the LDBStore.Put, it is ensured that the MemStore +// contains the chunk with the same data, but nil ReqC channel. func (self *LocalStore) Put(chunk *Chunk) { valid := true for _, v := range self.Validators { @@ -97,26 +109,52 @@ func (self *LocalStore) Put(chunk *Chunk) { } if !valid { chunk.SetErrored(ErrChunkInvalid) - chunk.dbStoredC <- false + chunk.markAsStored() return } + log.Trace("localstore.put", "key", chunk.Key) self.mu.Lock() defer self.mu.Unlock() chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - c := &Chunk{ - Key: Key(append([]byte{}, chunk.Key...)), - SData: append([]byte{}, chunk.SData...), - Size: chunk.Size, - dbStored: chunk.dbStored, - dbStoredC: chunk.dbStoredC, - dbStoredMu: chunk.dbStoredMu, + + memChunk, err := self.memStore.Get(chunk.Key) + switch err { + case nil: + if memChunk.ReqC == nil { + chunk.markAsStored() + return + } + case ErrChunkNotFound: + default: + chunk.SetErrored(err) + return + } + + self.memStore.Put(chunk) + + if memChunk != nil && memChunk.ReqC != nil { + close(memChunk.ReqC) } dbStorePutCounter.Inc(1) - self.memStore.Put(c) - self.DbStore.Put(c) + self.DbStore.Put(chunk) + + newc := NewChunk(chunk.Key, nil) + newc.SData = chunk.SData + newc.Size = chunk.Size + //newc.dbStored = chunk.dbStored + newc.dbStoredC = chunk.dbStoredC + //newc.dbStoredMu = chunk.dbStoredMu + go func() { + <-chunk.dbStoredC + + self.mu.Lock() + defer self.mu.Unlock() + + self.memStore.Put(newc) + }() } // Get(chunk *Chunk) looks up a chunk in the local stores diff --git a/swarm/storage/resource.go b/swarm/storage/resource.go index 2bc0884bf0..ab9d0eb5ee 100644 --- a/swarm/storage/resource.go +++ b/swarm/storage/resource.go @@ -214,10 +214,7 @@ func (self *ResourceHandler) Validate(key Key, data []byte) bool { } return false } else if signature == nil { - if !bytes.Equal(self.resourceHash(period, version, ens.EnsNode(name)), key) { - return false - } - return true + return bytes.Equal(self.resourceHash(period, version, ens.EnsNode(name)), key) } digest := self.keyDataHash(key, parseddata) addr, err := getAddressFromDataSig(digest, *signature) diff --git a/swarm/storage/types.go b/swarm/storage/types.go index 20f4ce7ddc..16e0536e19 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -32,7 +32,7 @@ import ( "github.com/ethereum/go-ethereum/log" ) -const MaxPO = 7 +const MaxPO = 16 const KeyLength = 32 type Hasher func() hash.Hash @@ -218,8 +218,9 @@ func (c *Chunk) markAsStored() { } } -func (c *Chunk) WaitToStore() { +func (c *Chunk) WaitToStore() error { <-c.dbStoredC + return c.GetErrored() } func GenerateRandomChunk(dataSize int64) *Chunk { diff --git a/swarm/swarm.go b/swarm/swarm.go index 335966f4be..2edab24927 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -32,7 +32,6 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/contracts/chequebook" "github.com/ethereum/go-ethereum/contracts/ens" - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" @@ -129,10 +128,18 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. //self.cloud = &storage.Forwarder{} //self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams) log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store")) - nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey))) - addr := network.NewAddrFromNodeID(nodeid) + + nodeID, err := discover.HexID(config.NodeID) + if err != nil { + return nil, err + } + addr := &network.BzzAddr{ + OAddr: common.FromHex(config.BzzKey), + UAddr: []byte(discover.NewNode(nodeID, net.IP{127, 0, 0, 1}, 30303, 30303).String()), + } + bzzconfig := &network.BzzConfig{ - OverlayAddr: common.FromHex(config.BzzKey), + OverlayAddr: addr.OAddr, UnderlayAddr: addr.UAddr, HiveParams: config.HiveParams, } @@ -220,10 +227,12 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) // Pss = postal service over swarm (devp2p over bzz) - pssparams := pss.NewPssParams(self.privateKey) - self.ps = pss.NewPss(to, pssparams) - if pss.IsActiveHandshake { - pss.SetHandshakeController(self.ps, pss.NewHandshakeParams()) + if config.PssEnabled { + pssparams := pss.NewPssParams(self.privateKey) + self.ps = pss.NewPss(to, pssparams) + if pss.IsActiveHandshake { + pss.SetHandshakeController(self.ps, pss.NewHandshakeParams()) + } } self.api = api.NewApi(self.dpa, self.dns, resourceHandler)