swarm: replace streamer updateSyncing with peer-based syncing

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

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
+	*protocols.Peer // represents the connection for online peers
-	*BzzAddr                  // remote address -> implements Addr interface = protocols.Peer
+	*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.

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 {

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)
 

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
+	addr            enode.ID
-	api            *API
+	api             *API
-	skipCheck      bool
+	skipCheck       bool
-	clientMu       sync.RWMutex
+	clientMu        sync.RWMutex
-	serverMu       sync.RWMutex
+	serverMu        sync.RWMutex
-	peersMu        sync.RWMutex
+	peersMu         sync.RWMutex
-	serverFuncs    map[string]func(*Peer, string, bool) (Server, error)
+	serverFuncs     map[string]func(*Peer, string, bool) (Server, error)
-	clientFuncs    map[string]func(*Peer, string, bool) (Client, error)
+	clientFuncs     map[string]func(*Peer, string, bool) (Client, error)
-	peers          map[enode.ID]*Peer
+	peers           map[enode.ID]*Peer
-	delivery       *Delivery
+	delivery        *Delivery
-	intervalsStore state.Store
+	intervalsStore  state.Store
-	autoRetrieval  bool // automatically subscribe to retrieve request stream
+	autoRetrieval   bool // automatically subscribe to retrieve request stream
-	maxPeerServers int
+	maxPeerServers  int
-	spec           *protocols.Spec   //this protocol's spec
+	spec            *protocols.Spec   //this protocol's spec
-	balance        protocols.Balance //implements protocols.Balance, for accounting
+	balance         protocols.Balance //implements protocols.Balance, for accounting
-	prices         protocols.Prices  //implements protocols.Prices, provides prices to accounting
+	prices          protocols.Prices  //implements protocols.Prices, provides prices to accounting
-	quit           chan struct{}     // terminates registry goroutines
+	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,
+		addr:            localID,
-		skipCheck:      options.SkipCheck,
+		skipCheck:       options.SkipCheck,
-		serverFuncs:    make(map[string]func(*Peer, string, bool) (Server, error)),
+		serverFuncs:     make(map[string]func(*Peer, string, bool) (Server, error)),
-		clientFuncs:    make(map[string]func(*Peer, string, bool) (Client, error)),
+		clientFuncs:     make(map[string]func(*Peer, string, bool) (Client, error)),
-		peers:          make(map[enode.ID]*Peer),
+		peers:           make(map[enode.ID]*Peer),
-		delivery:       delivery,
+		delivery:        delivery,
-		intervalsStore: intervalsStore,
+		intervalsStore:  intervalsStore,
-		autoRetrieval:  retrieval,
+		autoRetrieval:   retrieval,
-		maxPeerServers: options.MaxPeerServers,
+		maxPeerServers:  options.MaxPeerServers,
-		balance:        balance,
+		balance:         balance,
-		quit:           quit,
+		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
+	streamer.syncMode = options.Syncing
-	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()
-			}
-		}()
-	}
 
 	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 {

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 := &Registry{}
-	r.requestPeerSubscriptions(k, nil)
+	//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 {
+		//if len(subs) == 0 {
-			expectedMsgCount.inc()
+		expectedMsgCount.inc()
-		}
+		//}
-		doRequestSubscription(r, p, bin, subs)
+		doRequestSubscription(r, p, bin)
-		return true
+		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,
+		if realCount != emc {
-		// thus the real count should be half of the actual request subscriptions sent
-		if realCount/2 != emc {
 			return fmt.Errorf("Real subscriptions and expected amount don't match; real: %d, expected: %d", realCount/2, emc)
 		}
 		return nil