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)