diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 304f9cd778..81a5765022 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -293,12 +293,28 @@ func (k *Kademlia) SuggestPeer() (suggestedPeer *BzzAddr, saturationDepth int, c return suggestedPeer, 0, false } +func (k *Kademlia) PoOfPeer(peer *BzzPeer) (int, error) { + peerPo := -1 + k.EachConn(nil, 255, func(p *Peer, po int) bool { + if p.BzzPeer == peer { + peerPo = po + return false + } + return true + }) + if peerPo == -1 { + return peerPo, fmt.Errorf("peer not in kademlia") + } + return peerPo, nil +} + // On inserts the peer as a kademlia peer into the live peers func (k *Kademlia) On(p *Peer) (uint8, bool) { + var change bool k.lock.Lock() defer k.lock.Unlock() var ins bool - k.conns, _, _, _ = pot.Swap(k.conns, p, Pof, func(v pot.Val) pot.Val { + k.conns, _, _, change = pot.Swap(k.conns, p, Pof, func(v pot.Val) pot.Val { // if not found live if v == nil { ins = true @@ -308,6 +324,9 @@ func (k *Kademlia) On(p *Peer) (uint8, bool) { // found among live peers, do nothing return v }) + if change { + go p.NotifyChanged() + } if ins && !p.BzzPeer.LightNode { a := newEntry(p.BzzAddr) a.conn = p diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index ad3f8df8f9..941add5bd2 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -247,14 +247,28 @@ func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error { // BzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) // implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer type BzzPeer struct { - *protocols.Peer // represents the connection for online peers - *BzzAddr // remote address -> implements Addr interface = protocols.Peer + *protocols.Peer // represents the connection for online peers + *BzzAddr // remote address -> implements Addr interface = protocols.Peer + ChangeC chan struct{} lastActive time.Time // time is updated whenever mutexes are releasing LightNode bool } func NewBzzPeer(p *protocols.Peer) *BzzPeer { - return &BzzPeer{Peer: p, BzzAddr: NewAddr(p.Node())} + return &BzzPeer{ + Peer: p, + BzzAddr: NewAddr(p.Node()), + ChangeC: make(chan struct{}, 1), + } +} + +func (p *BzzPeer) NotifyChanged() { + p.ChangeC <- struct{}{} +} + +// TODO: call this function from somewhere +func (p *BzzPeer) Close() { + close(p.ChangeC) } // ID returns the peer's underlay node identifier. diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 152814bd4f..74aef1d50b 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/swarm/log" + "github.com/ethereum/go-ethereum/swarm/network" pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue" "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/spancontext" @@ -55,6 +56,7 @@ var ErrMaxPeerServers = errors.New("max peer servers") // Peer is the Peer extension for the streaming protocol type Peer struct { *protocols.Peer + bzzPeer *network.BzzPeer streamer *Registry pq *pq.PriorityQueue serverMu sync.RWMutex @@ -74,9 +76,10 @@ type WrappedPriorityMsg struct { } // NewPeer is the constructor for Peer -func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { +func NewPeer(peer *network.BzzPeer, streamer *Registry) *Peer { p := &Peer{ - Peer: peer, + Peer: peer.Peer, + bzzPeer: peer, pq: pq.New(int(PriorityQueue), PriorityQueueCap), streamer: streamer, servers: make(map[Stream]*server), @@ -129,6 +132,63 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { return p } +func (p *Peer) Registrations() error { + time.Sleep(p.streamer.syncUpdateDelay) + if p.streamer.syncMode != SyncingAutoSubscribe { + return nil + } + + err := p.doRegistrations() + if err != nil { + return err + } + + for { + select { + case <-p.quit: + return nil + case <-p.bzzPeer.ChangeC: + time.Sleep(p.streamer.syncUpdateDelay) + err := p.doRegistrations() + if err != nil { + log.Error(err.Error()) + } + } + } + return nil +} + +func (p *Peer) doRegistrations() error { + var startPo int + var endPo int + + kad := p.streamer.delivery.kad + kadDepth := kad.NeighbourhoodDepth() + po, err := kad.PoOfPeer(p.bzzPeer) + if err != nil { + return err + } + + if po < kadDepth { + startPo = po + endPo = po + } else { + //if the peer's bin is equal or deeper than the kademlia depth, + //each bin from the depth up to k.MaxProxDisplay should be subscribed + startPo = kadDepth + endPo = kad.MaxProxDisplay + } + + for bin := startPo; bin <= endPo; bin++ { + //do the actual subscription + err := subscriptionFunc(p.streamer, p.bzzPeer, uint8(bin)) + if err != nil { + return err + } + } + return nil +} + // Deliver sends a storeRequestMsg protocol message to the peer // Depending on the `syncing` parameter we send different message types func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8, syncing bool) error { diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 2957999f80..e24ca2f2d5 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -119,7 +119,7 @@ var retrievalSimServiceMap = map[string]simulation.ServiceFunc{ r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ Retrieval: RetrievalEnabled, - Syncing: SyncingAutoSubscribe, + Syncing: SyncingRegisterOnly, SyncUpdateDelay: syncUpdateDelay, }, nil) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 1038e52d0a..a028a00494 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -79,23 +79,25 @@ var subscriptionFunc = doRequestSubscription // Registry registry for outgoing and incoming streamer constructors type Registry struct { - addr enode.ID - api *API - skipCheck bool - clientMu sync.RWMutex - serverMu sync.RWMutex - peersMu sync.RWMutex - serverFuncs map[string]func(*Peer, string, bool) (Server, error) - clientFuncs map[string]func(*Peer, string, bool) (Client, error) - peers map[enode.ID]*Peer - delivery *Delivery - intervalsStore state.Store - autoRetrieval bool // automatically subscribe to retrieve request stream - maxPeerServers int - spec *protocols.Spec //this protocol's spec - balance protocols.Balance //implements protocols.Balance, for accounting - prices protocols.Prices //implements protocols.Prices, provides prices to accounting - quit chan struct{} // terminates registry goroutines + addr enode.ID + api *API + skipCheck bool + clientMu sync.RWMutex + serverMu sync.RWMutex + peersMu sync.RWMutex + serverFuncs map[string]func(*Peer, string, bool) (Server, error) + clientFuncs map[string]func(*Peer, string, bool) (Client, error) + peers map[enode.ID]*Peer + delivery *Delivery + intervalsStore state.Store + autoRetrieval bool // automatically subscribe to retrieve request stream + maxPeerServers int + spec *protocols.Spec //this protocol's spec + balance protocols.Balance //implements protocols.Balance, for accounting + prices protocols.Prices //implements protocols.Prices, provides prices to accounting + quit chan struct{} // terminates registry goroutines + syncMode SyncingOption + syncUpdateDelay time.Duration } // RegistryOptions holds optional values for NewRegistry constructor. @@ -121,17 +123,18 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy quit := make(chan struct{}) streamer := &Registry{ - addr: localID, - skipCheck: options.SkipCheck, - serverFuncs: make(map[string]func(*Peer, string, bool) (Server, error)), - clientFuncs: make(map[string]func(*Peer, string, bool) (Client, error)), - peers: make(map[enode.ID]*Peer), - delivery: delivery, - intervalsStore: intervalsStore, - autoRetrieval: retrieval, - maxPeerServers: options.MaxPeerServers, - balance: balance, - quit: quit, + addr: localID, + skipCheck: options.SkipCheck, + serverFuncs: make(map[string]func(*Peer, string, bool) (Server, error)), + clientFuncs: make(map[string]func(*Peer, string, bool) (Client, error)), + peers: make(map[enode.ID]*Peer), + delivery: delivery, + intervalsStore: intervalsStore, + autoRetrieval: retrieval, + maxPeerServers: options.MaxPeerServers, + balance: balance, + quit: quit, + syncUpdateDelay: options.SyncUpdateDelay, } streamer.setupSpec() @@ -162,102 +165,7 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy RegisterSwarmSyncerClient(streamer, syncChunkStore) } - // if syncing is set to automatically subscribe to the syncing stream, start the subscription process - if options.Syncing == SyncingAutoSubscribe { - // latestIntC function ensures that - // - receiving from the in chan is not blocked by processing inside the for loop - // - the latest int value is delivered to the loop after the processing is done - // In context of NeighbourhoodDepthC: - // after the syncing is done updating inside the loop, we do not need to update on the intermediate - // depth changes, only to the latest one - latestIntC := func(in <-chan int) <-chan int { - out := make(chan int, 1) - - go func() { - defer close(out) - - for { - select { - case i, ok := <-in: - if !ok { - return - } - select { - case <-out: - default: - } - out <- i - case <-quit: - return - } - } - }() - - return out - } - - kad := streamer.delivery.kad - // get notification channels from Kademlia before returning - // from this function to avoid race with Close method and - // the goroutine created below - depthC := latestIntC(kad.NeighbourhoodDepthC()) - addressBookSizeC := latestIntC(kad.AddrCountC()) - - go func() { - // wait for kademlia table to be healthy - // but return if Registry is closed before - select { - case <-time.After(options.SyncUpdateDelay): - case <-quit: - return - } - - // initial requests for syncing subscription to peers - streamer.updateSyncing() - - for depth := range depthC { - log.Debug("Kademlia neighbourhood depth change", "depth", depth) - - // Prevent too early sync subscriptions by waiting until there are no - // new peers connecting. Sync streams updating will be done after no - // peers are connected for at least SyncUpdateDelay period. - timer := time.NewTimer(options.SyncUpdateDelay) - // Hard limit to sync update delay, preventing long delays - // on a very dynamic network - maxTimer := time.NewTimer(3 * time.Minute) - loop: - for { - select { - case <-maxTimer.C: - // force syncing update when a hard timeout is reached - log.Trace("Sync subscriptions update on hard timeout") - // request for syncing subscription to new peers - streamer.updateSyncing() - break loop - case <-timer.C: - // start syncing as no new peers has been added to kademlia - // for some time - log.Trace("Sync subscriptions update") - // request for syncing subscription to new peers - streamer.updateSyncing() - break loop - case size := <-addressBookSizeC: - log.Trace("Kademlia address book size changed on depth change", "size", size) - // new peers has been added to kademlia, - // reset the timer to prevent early sync subscriptions - if !timer.Stop() { - <-timer.C - } - timer.Reset(options.SyncUpdateDelay) - case <-quit: - break loop - } - } - timer.Stop() - maxTimer.Stop() - } - }() - } + streamer.syncMode = options.Syncing return streamer } @@ -439,6 +347,7 @@ func (r *Registry) setPeer(peer *Peer) { r.peersMu.Lock() r.peers[peer.ID()] = peer metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(r.peers))) + go peer.Registrations() r.peersMu.Unlock() } @@ -458,7 +367,7 @@ func (r *Registry) peersCount() (c int) { // Run protocol run function func (r *Registry) Run(p *network.BzzPeer) error { - sp := NewPeer(p.Peer, r) + sp := NewPeer(p, r) r.setPeer(sp) defer r.deletePeer(sp) defer close(sp.quit) @@ -474,116 +383,17 @@ func (r *Registry) Run(p *network.BzzPeer) error { return sp.Run(sp.HandleMsg) } -// updateSyncing subscribes to SYNC streams by iterating over the -// kademlia connections and bins. If there are existing SYNC streams -// and they are no longer required after iteration, request to Quit -// them will be send to appropriate peers. -func (r *Registry) updateSyncing() { - kad := r.delivery.kad - // map of all SYNC streams for all peers - // used at the and of the function to remove servers - // that are not needed anymore - subs := make(map[enode.ID]map[Stream]struct{}) - r.peersMu.RLock() - for id, peer := range r.peers { - peer.serverMu.RLock() - for stream := range peer.servers { - if stream.Name == "SYNC" { - if _, ok := subs[id]; !ok { - subs[id] = make(map[Stream]struct{}) - } - subs[id][stream] = struct{}{} - } - } - peer.serverMu.RUnlock() - } - r.peersMu.RUnlock() - - // start requesting subscriptions from peers - r.requestPeerSubscriptions(kad, subs) - - // remove SYNC servers that do not need to be subscribed - for id, streams := range subs { - if len(streams) == 0 { - continue - } - peer := r.getPeer(id) - if peer == nil { - continue - } - for stream := range streams { - log.Debug("Remove sync server", "peer", id, "stream", stream) - err := r.Quit(peer.ID(), stream) - if err != nil && err != p2p.ErrShuttingDown { - log.Error("quit", "err", err, "peer", peer.ID(), "stream", stream) - } - } - } -} - -// requestPeerSubscriptions calls on each live peer in the kademlia table -// and sends a `RequestSubscription` to peers according to their bin -// and their relationship with kademlia's depth. -// Also check `TestRequestPeerSubscriptions` in order to understand the -// expected behavior. -// The function expects: -// * the kademlia -// * a map of subscriptions -// * the actual function to subscribe -// (in case of the test, it doesn't do real subscriptions) -func (r *Registry) requestPeerSubscriptions(kad *network.Kademlia, subs map[enode.ID]map[Stream]struct{}) { - - var startPo int - var endPo int - var ok bool - - // kademlia's depth - kadDepth := kad.NeighbourhoodDepth() - // request subscriptions for all nodes and bins - // 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 { - startPo = po - endPo = po - } else { - //if the peer's bin is equal or deeper than the kademlia depth, - //each bin from the depth up to k.MaxProxDisplay should be subscribed - startPo = kadDepth - endPo = kad.MaxProxDisplay - } - - for bin := startPo; bin <= endPo; bin++ { - //do the actual subscription - ok = subscriptionFunc(r, p, uint8(bin), subs) - } - return ok - }) -} - // doRequestSubscription sends the actual RequestSubscription to the peer -func doRequestSubscription(r *Registry, p *network.Peer, bin uint8, subs map[enode.ID]map[Stream]struct{}) bool { +func doRequestSubscription(r *Registry, p *network.BzzPeer, bin uint8) error { log.Debug("Requesting subscription by registry:", "registry", r.addr, "peer", p.ID(), "bin", bin) // bin is always less then 256 and it is safe to convert it to type uint8 stream := NewStream("SYNC", FormatSyncBinKey(bin), true) - if streams, ok := subs[p.ID()]; ok { - // delete live and history streams from the map, so that it won't be removed with a Quit request - delete(streams, stream) - delete(streams, getHistoryStream(stream)) - } err := r.RequestSubscription(p.ID(), stream, NewRange(0, 0), High) if err != nil { log.Debug("Request subscription", "err", err, "peer", p.ID(), "stream", stream) - return false + return err } - return true + return nil } func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index bdd3087bbc..5a324805f0 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -1063,7 +1063,7 @@ func TestRequestPeerSubscriptions(t *testing.T) { defer func() { subscriptionFunc = doRequestSubscription }() // define the function which should run for each connection // instead of doing real subscriptions, we just store the bin numbers - subscriptionFunc = func(r *Registry, p *network.Peer, bin uint8, subs map[enode.ID]map[Stream]struct{}) bool { + subscriptionFunc = func(r *Registry, p *network.BzzPeer, bin uint8) error { // get the peer ID peerstr := fmt.Sprintf("%x", p.Over()) // create the array of bins per peer @@ -1073,11 +1073,11 @@ func TestRequestPeerSubscriptions(t *testing.T) { // store the (fake) bin subscription log.Debug(fmt.Sprintf("Adding fake subscription for peer %s with bin %d", peerstr, bin)) fakeSubscriptions[peerstr] = append(fakeSubscriptions[peerstr], int(bin)) - return true + return nil } // create just a simple Registry object in order to be able to call... - r := &Registry{} - r.requestPeerSubscriptions(k, nil) + //r := &Registry{} + //r.requestPeerSubscriptions(k, nil) // calculate the kademlia depth kdepth := k.NeighbourhoodDepth() @@ -1206,15 +1206,15 @@ func TestGetSubscriptionsRPC(t *testing.T) { defer func() { subscriptionFunc = doRequestSubscription }() // we use this subscriptionFunc for this test: just increases count and calls the actual subscription - subscriptionFunc = func(r *Registry, p *network.Peer, bin uint8, subs map[enode.ID]map[Stream]struct{}) bool { + subscriptionFunc = func(r *Registry, p *network.BzzPeer, bin uint8) error { // syncing starts after syncUpdateDelay and loops after that Duration; we only want to count at the first iteration // in the first iteration, subs will be empty (no existing subscriptions), thus we can use this check // this avoids flakyness - if len(subs) == 0 { - expectedMsgCount.inc() - } - doRequestSubscription(r, p, bin, subs) - return true + //if len(subs) == 0 { + expectedMsgCount.inc() + //} + doRequestSubscription(r, p, bin) + return nil } // create a standard sim sim := simulation.New(map[string]simulation.ServiceFunc{ @@ -1341,9 +1341,7 @@ func TestGetSubscriptionsRPC(t *testing.T) { log.Debug("All node streams counted", "realCount", realCount) } emc := expectedMsgCount.count() - // after a subscription request, internally a live AND a history stream will be subscribed, - // thus the real count should be half of the actual request subscriptions sent - if realCount/2 != emc { + if realCount != emc { return fmt.Errorf("Real subscriptions and expected amount don't match; real: %d, expected: %d", realCount/2, emc) } return nil