swarm: replace streamer updateSyncing with peer-based syncing by holisticode · Pull Request #1325 · ethersphere/swarm

holisticode · 2019-04-03T15:42:19Z

This PR is an experiment and work in progress.

It is a replacement for udpateSyncing in the Registry to run subscriptions from peers and not globally via the registry.

It is meant as a discussion base to refine the architecture to replace updateSyncing.

Currently there are no removal of subscriptions yet. Some parts may be qualified as bad or even ugly, consequence of quick hacks in order to have first results.

Currently the smoke tests produce a timeout at about 50% rate with this approach. Also, the retrieval tests are very slow (explaining the timeouts)

fixes #1329

zelig

good stuff, lets do it properly ;)

zelig · 2019-04-05T04:53:06Z

 	return suggestedPeer, 0, false
 }

+func (k *Kademlia) PoOfPeer(peer *BzzPeer) int {


do we need this function? do we need exporting?

I can use chunk.Proximity() directly where it's used if you prefer

zelig · 2019-04-05T05:00:39Z

 			return a
 		})
+		k.lock.Unlock()
+		k.notifyKadChange()


all you need is keep the global depth change channel.

Each time that changes, you see it changed from depth=1 to depth 3.

Now you iterate over peers of PO between 2 and 1 and trigger their change channel
the new depth is put on the peers depthC channel

no, like this we introduce the same problem again: if there is a new peer added but there is no depth change, then we don't request registrations to that guy

zelig · 2019-04-05T05:01:57Z

-	*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{}


as I say this could be chan int

zelig · 2019-04-05T05:02:34Z

+}
+
+// TODO: call this function from somewhere
+func (p *BzzPeer) Close() {


this channel does not need closing

zelig · 2019-04-05T05:04:58Z

 	p := &Peer{
-		Peer:         peer,
+		Peer:         peer.Peer,
+		bzzPeer:      peer,


not sure i understand why you need a change here. One peer can just extend the other if we need both

zelig · 2019-04-05T05:06:23Z

+			// which means that new subscription requests were being issued before
+			// a first round finished and the servers were being created
+			//TODO: needs investigation about why that is
+			timer = time.NewTimer(p.streamer.syncUpdateDelay)


why do we need a timer here?

I would keep the logic for initial wait for now. Lets assume it is signalled on startSyncing channel.
then the peer connection would start an update loop like this:

var change chan int po := proximity(p.Over(), p.baseAddr) newdepth := kad.NeighbourhoodDepth nn = po >= newdepth for { depth = newdepth select { case <- startC: change = peer.ChangeC case newdepth := <- change: if changed := nn ^ po >= depth; changed { nn = !nn if nn { //request peer to subscribe to PO bins depth, depth+1, po - 1, po+1, ... MaxProxDisplay continue } //quit all but po continue } // if only depth changed then if nn { if newdepth < depth { // request peer to subscribe to PO bins newdepth, newdepth+1,... depth -1 } else { // quit PO depth, depth+1, ... newdepth-1 case <- p.quit: // quit all subs } }

zelig · 2019-04-10T03:50:31Z

+	depth := k.NeighbourhoodDepth()
 	k.EachConn(nil, 255, func(p *Peer, po int) bool {
-		go p.NotifyChanged()
+		go p.NotifyChanged(depth)


if put in a go routine, changes are not guaranteed to be in proper order. You need syncronous write here.
The peer chance channel should be bufferred.

zelig · 2019-04-10T04:01:02Z

+				nn = !nn
+				if nn {
+					//request peer to subscribe to PO bins depth, depth+1, po - 1, po+1, ... MaxProxDisplay
+					poList := []int{depth, depth + 1, po - 1}


this is not correct, since po can == depth etc.
You need

for i := depth; i <= kad.MaxProxDisplay; i++ { if po != i { p.doSubscribe(i) // with some error handling } }

zelig · 2019-04-10T04:07:44Z

+		case <-timer.C:
+			change = p.bzzPeer.ChangeC
+		case newdepth := <-change:
+			if changed := nn != (po >= depth); changed {


here it is best to drain all changes from the channel to arrive at the last one

for { select { case newdepth = <-change: case default: } }

zelig · 2019-04-10T04:10:46Z

+					continue
+				}
+				//quit all but po
+				poList := make([]int, 1)


for i := depth; i <= kad.MaxProxDisplay; i++ { if po != i { p.doQuit(i) } }

zelig · 2019-04-10T04:13:09Z

-				log.Error(err.Error())
+			// if only depth changed then
+			if nn {
+				if newdepth < depth {


for i := newdepth; i <= depth-1; i++ { p.doSubscribe(i) }

zelig · 2019-04-10T04:15:09Z

+						return err
+					}
+				} else {
+					// quit PO depth, depth+1, ... newdepth-1


for i := depth + 1; i < newdepth; i++ { p.doQuit(i) // with some error handling }

zelig · 2019-04-10T04:16:20Z

-		//do the actual subscription
-		err := subscriptionFunc(p.streamer, p.bzzPeer, uint8(bin))
+//request peer to subscribe to PO bins depth, depth+1, po - 1, po+1, ... MaxProxDisplay
+func (p *Peer) doRegister(poList []int) error {


we call this in an iterator loop, no need to call with a slice and iterate.
just call it with a single po arg

let's not call it doRegister, but doSubscribe or better
subscribeToSyncStream

zelig · 2019-04-10T04:19:24Z

 }

+// quit - TODO: should this return error?
+func (p *Peer) doQuit(poList []int) {


again no iteration needed and best to call the function quitSyncStream

zelig · 2019-04-10T04:21:14Z

+		live := NewStream("SYNC", FormatSyncBinKey(uint8(po)), true)
+		history := getHistoryStream(live)
+		err := p.streamer.Quit(p.ID(), live)
+		if err != nil && err != p2p.ErrShuttingDown {


not sure you need to bother catching these

zelig · 2019-04-10T04:26:25Z

+// then starting a simulation, distribute chunks to nodes
+// and start retrieval.
+// The snapshot should have 'streamer' in its service list.
+func runPureRetrievalTest(nodeCount int, chunkCount int) error {


Yes, such a test is super important. I always though we had written this already...

zelig · 2019-04-13T11:53:17Z

obsolete by #1336

swarm: replace streamer updateSyncing with peer-based syncing

32a5721

holisticode added in progress sync-optimization stability labels Apr 3, 2019

holisticode self-assigned this Apr 3, 2019

holisticode requested review from janos, nonsense and zelig April 3, 2019 15:42

holisticode added 3 commits April 3, 2019 11:06

swarm/network/stream: timers instead of Sleep and comments

8db1e20

swarm/network/stream: don't use depth change but kad change!

4c04127

swarm/network/stream: better concurrency; po by Proximity() not kad

6bb7f52

zelig suggested changes Apr 5, 2019

View reviewed changes

holisticode added 2 commits April 6, 2019 12:36

swarm/network/stream: PoOfPeer() not needed

1516996

swarm/network/stream: add cancellation and optimize; PureRetrievalTest

2a2cbc1

zelig suggested changes Apr 10, 2019

View reviewed changes

zelig closed this Apr 13, 2019

holisticode deleted the subs-by-peer branch September 2, 2019 15:20

Conversation

holisticode commented Apr 3, 2019 • edited by zelig Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

zelig left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

holisticode Apr 9, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zelig commented Apr 13, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

holisticode commented Apr 3, 2019 •

edited by zelig

Loading

holisticode Apr 9, 2019 •

edited

Loading