swarm: RequestSubscription for empty bins

swarm/network/stream/messages.go

@@ -74,6 +74,15 @@ func (p *Peer) handleRequestSubscription(req *RequestSubscriptionMsg) (err error
 	return p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority)
 }
 
+func (p *Peer) handleRequestSubscription(req *RequestSubscriptionMsg) (err error) {
+	log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s", p.streamer.addr.ID(), p.ID()))
+	err = p.streamer.Subscribe(p.ID(), req.Stream, &Range{}, req.Priority)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 	defer func() {
 		if err != nil {

swarm/network/stream/snapshot_sync_test.go

@@ -24,20 +24,16 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"os"
-	//"sort"
 	"testing"
 	"time"
 
 	"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/pot"
-	//	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
-	//streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 )
 
 const testMinProxBinSize = 2
@@ -48,10 +44,8 @@ var (
 	startTime time.Time
 	ids       []discover.NodeID
 	datadirs  map[discover.NodeID]string
-	overlays  map[discover.NodeID]network.Overlay
 	conf      *synctestConfig
 	ppmap     map[discover.NodeID]*network.PeerPot
-	//conf      := &synctestConfig{}
 )
 
 type synctestConfig struct {
@@ -88,8 +82,6 @@ func initSyncTest() {
 	datadirs = make(map[discover.NodeID]string)
 	//deliveries for each node
 	deliveries = make(map[discover.NodeID]*Delivery)
-	//overlays (kademlia)
-	overlays = make(map[discover.NodeID]network.Overlay)
 
 	//channel to wait for peers connected
 	waitPeerErrC = make(chan error)
@@ -292,12 +284,8 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 		}
 
 		time.Sleep(10 * time.Second)
-		//peerPot := network.NewPeerPot(testMinProxBinSize, ids, conf.addrs)
 		// each node Subscribes to each other's swarmChunkServerStreamName
 		for j := 0; j < len(ids); j++ {
-			//overlays[ids[j]].EachAddr(nil, 0, func(addr network.OverlayAddr, po int, nn bool) bool {
-			//overlays[ids[j]].EachAddr(nil, 256, func(addr network.OverlayAddr, po int, nn bool) bool {
-			//overlays[ids[j]].EachConn(nil, 256, func(addr network.OverlayConn, po int, nn bool) bool {
 			log.Debug(fmt.Sprintf("subscribe: %d", j))
 			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 			defer cancel()
@@ -310,36 +298,7 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 				log.Error(fmt.Sprintf("FAILED CallContext %v", err))
 				return nil
 			}
-			//	})
-
 		}
-		/*
-					for j := 0; j < len(ids); j++ {
-						log.Debug(fmt.Sprintf("subscribe: %d", j))
-						ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
-						defer cancel()
-						client, err := net.GetNode(ids[j]).Client()
-						if err != nil {
-							return err
-						}
-						//RPC call to subscribe, select bin 0
-						//client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
-						// report disconnect events to the error channel cos peers should not disconnect
-						//err = streamTesting.WatchDisconnections(ids[j], client, disconnectC, quitC)
-						//if err != nil {
-						//  return err
-						//}
-						// start syncing, i.e., subscribe to upstream peers po 1 bin
-						//each node subscribes to the next index, last subscribes to 0
-						idx := j + 1
-						if j == len(ids)-1 {
-							idx = 0
-						}
-						sid := ids[idx]
-						//client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
-						client.CallContext(ctx, nil, "stream_startSyncing", sid, po, nn)
-			}
-		*/
 		time.Sleep(10 * time.Second)
 		//now upload the chunks to the selected random single node
 		conf.chunks, err = uploadFileToSingleNodeStore(node.ID(), chunkCount)
@@ -390,79 +349,28 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 	return nil
 }
 
-func newSyncingProtocol(ctx *adapters.ServiceContext) (node.Service, error) {
-	var err error
-	id := ctx.Config.ID
-	addr := toAddr(id)
-	kad := network.NewKademlia(addr.Over(), network.NewKadParams())
-	overlays[id] = kad
-	stores[id], err = createTestLocalStorageForId(id, addr)
-	if err != nil {
-		return nil, err
-	}
-	store := stores[id].(*storage.LocalStore)
-	db := storage.NewDBAPI(store)
-	delivery := NewDelivery(kad, db)
-	deliveries[id] = delivery
-	netStore := storage.NewNetStore(store, nil)
-	r := NewRegistry(addr, delivery, netStore, defaultSkipCheck)
-
-	RegisterSwarmSyncerServer(r, db)
-	RegisterSwarmSyncerClient(r, db)
-
-	//externalStreamName := "syncProtocol"
-	//hashesChan := make(chan []byte) // this chanel is only for one client, in need for more clients, create a map
-	/*
-
-		//r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) {
-			r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte) (Client, error) {
-				return newTestExternalClient(t, hashesChan), nil
-			})
-			//r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) {
-			r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte) (Server, error) {
-				return newTestExternalServer(t), nil
-			})
-	*/
-	go func() {
-		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
-	}()
-
-	//return &TestExternalRegistry{r, hashesChan}, nil
-	return &TestRegistry{Registry: r}, nil
-}
-
 func (r *TestRegistry) StartSyncing(ctx context.Context) error {
-	//func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) {
 	var err error
 
-	//fmt.Println(r.delivery.overlay.String())
 	add := r.addr.ID()
-	//fmt.Println(add)
 	pp := ppmap[add]
 	h := r.delivery.overlay.Healthy(pp)
 	fmt.Println("----------------------------------")
 	fmt.Println(r.delivery.overlay.String())
 	fmt.Println(fmt.Sprintf("IS HEALTHY: %t", h.GotNN && h.KnowNN && h.Full))
-	pos := make(map[int]discover.NodeID)
-	r.delivery.overlay.EachConn(nil, 256, func(addr network.OverlayConn, po int, nn bool) bool {
-		//r.delivery.overlay.EachAddr(nil, 256, func(addr network.OverlayAddr, po int, nn bool) bool {
-		//fmt.Println("A")
-		//fmt.Println(po)
 
+	pos := make(map[int]discover.NodeID)
+
+	r.delivery.overlay.EachConn(nil, 256, func(addr network.OverlayConn, po int, nn bool) bool {
 		lastPO := po
 		if nn {
 			lastPO = maxPO
 		}
-		//fmt.Println(lastPO)
-		//fmt.Println("E")
 		peerId := conf.addrToIdMap[string(addr.Address())]
 		fmt.Println(fmt.Sprintf("node %s has conn with %s at po %d and is nn: %t", r.addr.ID(), peerId, po, nn))
 		pos[po] = peerId
 		for i := po; i <= lastPO; i++ {
-			//for ; i <= lastPO; i++ {
+			err = r.Subscribe(peerId, NewStream("SYNC", []byte{byte(i)}, false), &Range{From: 0, To: 0}, Top)
-			//for i := 0; i <= maxPO; i++ {
-			//fmt.Println(fmt.Sprintf("SUBSCRIBING %s TO %s and PO %d", string(r.addr.ID().String()), peerId, i))
-			err = r.Subscribe(peerId, "SYNC", []byte{byte(i)}, 0, 0, Top, true)
 			if err != nil {
 				log.Error(fmt.Sprintf("Error subscribing! %v", err))
 				return false
@@ -471,31 +379,44 @@ func (r *TestRegistry) StartSyncing(ctx context.Context) error {
 		return true
 	})
 	prev := 0
-	r.delivery.overlay.conns.EachBin(nil, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
+	kad, ok := r.delivery.overlay.(*network.Kademlia)
+	if !ok {
+		return fmt.Errorf("Not a Kademlia!")
+	}
+
+	kad.EachBin(r.addr.Over(), pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
 		skip := po - prev
+		/*
+			fmt.Println(prev)
+			fmt.Println(po)
+			fmt.Println(skip)
+		*/
+		remember := make(map[int]bool)
 		if skip > 1 {
 			f(func(val pot.Val, i int) bool {
-				for c := po + 1; c < po+skip; c++ {
+				//for c := po + 1; c < po+skip; c++ {
-					err = r.RequestSubscription(peerId, "SYNC", []byte{byte(i)}, 0, 0, Top, true)
+				for c := po - 1; c > po-skip; c-- {
+					//fmt.Println(c)
+					if exists, _ := remember[c]; exists {
+						continue
+					}
+					a := val.(network.OverlayPeer)
+					log.Warn(fmt.Sprintf("Request subscription for bin: %d", c))
+					log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s", r.addr.ID(), conf.addrToIdMap[string(a.Address())]))
+					err = r.RequestSubscription(conf.addrToIdMap[string(a.Address())], NewStream("SYNC", []byte{byte(uint8(c))}, false), Top)
 					if err != nil {
 						log.Error(fmt.Sprintf("Error subscribing! %v", err))
 						return false
 					}
+					remember[c] = true
 				}
+				return true
 			})
 		}
+		prev = po
+		return true
 	})
 
-	/*
-		for k, pid := range pos {
-			i := k - 1
-			for p := pos[i]; p == (discover.NodeID{}) && i > 0; i-- {
-				fmt.Println(fmt.Sprintf("Subscribe to bin %d", i))
-				r.Subscribe(pid, "SYNC", []byte{byte(i)}, 0, 0, Top, true)
-				p = pos[i-1]
-			}
-		}
-	*/
 	return nil
 }
 
@@ -552,8 +473,6 @@ func mapKeysToNodes(conf *synctestConfig) *synctestConfig {
 			}
 			return true
 		})
-		//kmap[conf.chunks[i].String()] = nns
-		//kmap[string(conf.chunks[i])] = nns
 		kmap[conf.chunks[i].String()] = nns
 		//log.Debug(fmt.Sprintf("Length for id %s: %d",ids[i],len(kmap[ids[i]])))
 	}

swarm/network/stream/streamer_test.go

@@ -39,6 +39,20 @@ func TestStreamerSubscribe(t *testing.T) {
 	}
 }
 
+func TestStreamerRequestSubscription(t *testing.T) {
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stream := NewStream("foo", nil, false)
+	err = streamer.RequestSubscription(tester.IDs[0], stream, Top)
+	if err == nil || err.Error() != "stream foo not registered" {
+		t.Fatalf("Expected error %v, got %v", "stream foo not registered", err)
+	}
+}
+
 var (
 	hash0     = sha3.Sum256([]byte{0})
 	hash1     = sha3.Sum256([]byte{1})