From 4976fcc91a43b5c7047c51a03985887b694f0fbb Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Thu, 24 Jan 2019 12:02:18 +0100 Subject: swarm: bootnode-mode, new bootnodes and no p2p package discovery (#18498) (cherry picked from commit bbd120354a8d226b446591eeda9f9462cb9b690a) --- swarm/api/config.go | 1 + swarm/network/kademlia.go | 2 +- swarm/network/protocol.go | 10 +++++++++- swarm/network/stream/delivery.go | 9 ++++++++- swarm/network/stream/delivery_test.go | 2 +- swarm/network/stream/stream.go | 5 +++++ swarm/storage/pyramid.go | 7 ------- swarm/swarm.go | 26 +++++++++++++++----------- 8 files changed, 40 insertions(+), 22 deletions(-) (limited to 'swarm') diff --git a/swarm/api/config.go b/swarm/api/config.go index be7385408..54dd67ba8 100644 --- a/swarm/api/config.go +++ b/swarm/api/config.go @@ -66,6 +66,7 @@ type Config struct { DeliverySkipCheck bool MaxStreamPeerServers int LightNodeEnabled bool + BootnodeMode bool SyncUpdateDelay time.Duration SwapAPI string Cors string diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index f9b38fc48..2c2276251 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -279,7 +279,7 @@ func (k *Kademlia) SuggestPeer() (suggestedPeer *BzzAddr, saturationDepth int, c return suggestedPeer, 0, false } -// On inserts the peer as a kademlia peer into the live peers +// On inserts the peer as a kademlia peer into the live peers func (k *Kademlia) On(p *Peer) (uint8, bool) { k.lock.Lock() defer k.lock.Unlock() diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 74e7de126..6f8eadad2 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -67,6 +67,7 @@ type BzzConfig struct { HiveParams *HiveParams NetworkID uint64 LightNode bool + BootnodeMode bool } // Bzz is the swarm protocol bundle @@ -87,7 +88,7 @@ type Bzz struct { // * overlay driver // * peer store func NewBzz(config *BzzConfig, kad *Kademlia, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz { - return &Bzz{ + bzz := &Bzz{ Hive: NewHive(config.HiveParams, kad, store), NetworkID: config.NetworkID, LightNode: config.LightNode, @@ -96,6 +97,13 @@ func NewBzz(config *BzzConfig, kad *Kademlia, store state.Store, streamerSpec *p streamerRun: streamerRun, streamerSpec: streamerSpec, } + + if config.BootnodeMode { + bzz.streamerRun = nil + bzz.streamerSpec = nil + } + + return bzz } // UpdateLocalAddr updates underlayaddress of the running node diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index e1a13fe8d..c9a8dc57a 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -255,8 +255,15 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) ( return true } sp = d.getPeer(id) + // sp is nil, when we encounter a peer that is not registered for delivery, i.e. doesn't support the `stream` protocol if sp == nil { - //log.Warn("Delivery.RequestFromPeers: peer not found", "id", id) + return true + } + // nodes that do not provide stream protocol + // should not be requested, e.g. bootnodes + if !p.HasCap("stream") { + // TODO: if we have no errors, delete this if + log.Error("Delivery.RequestFromPeers: peer doesn't have stream cap. we should have returned at sp == nil") return true } spID = &id diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 70d3829b3..13e13c0f5 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -285,7 +285,7 @@ func TestRequestFromPeers(t *testing.T) { addr := network.RandomAddr() to := network.NewKademlia(addr.OAddr, network.NewKadParams()) delivery := NewDelivery(to, nil) - protocolsPeer := protocols.NewPeer(p2p.NewPeer(dummyPeerID, "dummy", nil), nil, nil) + protocolsPeer := protocols.NewPeer(p2p.NewPeer(dummyPeerID, "dummy", []p2p.Cap{{Name: "stream"}}), nil, nil) peer := network.NewPeer(&network.BzzPeer{ BzzAddr: network.RandomAddr(), LightNode: false, diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index fb571c856..e06048053 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -516,6 +516,11 @@ func (r *Registry) requestPeerSubscriptions(kad *network.Kademlia, subs map[enod // nil as base takes the node's base; we need to pass 255 as `EachConn` runs // from deepest bins backwards kad.EachConn(nil, 255, func(p *network.Peer, po int) bool { + // nodes that do not provide stream protocol + // should not be subscribed, e.g. bootnodes + if !p.HasCap("stream") { + return true + } //if the peer's bin is shallower than the kademlia depth, //only the peer's bin should be subscribed if po < kadDepth { diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go index e5bd7a76a..ed0f843b9 100644 --- a/swarm/storage/pyramid.go +++ b/swarm/storage/pyramid.go @@ -201,8 +201,6 @@ func (pc *PyramidChunker) decrementWorkerCount() { } func (pc *PyramidChunker) Split(ctx context.Context) (k Address, wait func(context.Context) error, err error) { - log.Debug("pyramid.chunker: Split()") - pc.wg.Add(1) pc.prepareChunks(ctx, false) @@ -235,7 +233,6 @@ func (pc *PyramidChunker) Split(ctx context.Context) (k Address, wait func(conte } func (pc *PyramidChunker) Append(ctx context.Context) (k Address, wait func(context.Context) error, err error) { - log.Debug("pyramid.chunker: Append()") // Load the right most unfinished tree chunks in every level pc.loadTree(ctx) @@ -283,8 +280,6 @@ func (pc *PyramidChunker) processor(ctx context.Context, id int64) { } func (pc *PyramidChunker) processChunk(ctx context.Context, id int64, job *chunkJob) { - log.Debug("pyramid.chunker: processChunk()", "id", id) - ref, err := pc.putter.Put(ctx, job.chunk) if err != nil { select { @@ -301,7 +296,6 @@ func (pc *PyramidChunker) processChunk(ctx context.Context, id int64, job *chunk } func (pc *PyramidChunker) loadTree(ctx context.Context) error { - log.Debug("pyramid.chunker: loadTree()") // Get the root chunk to get the total size chunkData, err := pc.getter.Get(ctx, Reference(pc.key)) if err != nil { @@ -386,7 +380,6 @@ func (pc *PyramidChunker) loadTree(ctx context.Context) error { } func (pc *PyramidChunker) prepareChunks(ctx context.Context, isAppend bool) { - log.Debug("pyramid.chunker: prepareChunks", "isAppend", isAppend) defer pc.wg.Done() chunkWG := &sync.WaitGroup{} diff --git a/swarm/swarm.go b/swarm/swarm.go index db52675fd..d17d81320 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -84,12 +84,11 @@ type Swarm struct { tracerClose io.Closer } -// creates a new swarm service instance +// NewSwarm creates a new swarm service instance // implements node.Service // If mockStore is not nil, it will be used as the storage for chunk data. // MockStore should be used only for testing. func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err error) { - if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroAddr) { return nil, fmt.Errorf("empty public key") } @@ -116,10 +115,11 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e config.HiveParams.Discovery = true bzzconfig := &network.BzzConfig{ - NetworkID: config.NetworkID, - OverlayAddr: common.FromHex(config.BzzKey), - HiveParams: config.HiveParams, - LightNode: config.LightNodeEnabled, + NetworkID: config.NetworkID, + OverlayAddr: common.FromHex(config.BzzKey), + HiveParams: config.HiveParams, + LightNode: config.LightNodeEnabled, + BootnodeMode: config.BootnodeMode, } self.stateStore, err = state.NewDBStore(filepath.Join(config.Path, "state-store.db")) @@ -455,12 +455,16 @@ func (self *Swarm) Stop() error { return err } -// implements the node.Service interface -func (self *Swarm) Protocols() (protos []p2p.Protocol) { - protos = append(protos, self.bzz.Protocols()...) +// Protocols implements the node.Service interface +func (s *Swarm) Protocols() (protos []p2p.Protocol) { + if s.config.BootnodeMode { + protos = append(protos, s.bzz.Protocols()...) + } else { + protos = append(protos, s.bzz.Protocols()...) - if self.ps != nil { - protos = append(protos, self.ps.Protocols()...) + if s.ps != nil { + protos = append(protos, s.ps.Protocols()...) + } } return } -- cgit v1.2.3