From c6d7d0d0fa4fbf976f019ba21572f4dfe2a01e74 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Wed, 28 Feb 2018 15:22:58 -0500 Subject: [PATCH] swarm: RequestSubscription for empty bins --- swarm/network/stream/messages.go | 9 ++ swarm/network/stream/snapshot_sync_test.go | 141 +++++---------------- swarm/network/stream/streamer_test.go | 14 ++ 3 files changed, 53 insertions(+), 111 deletions(-) diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 8fabe2c3bc..b58df047b5 100644 --- a/swarm/network/stream/messages.go +++ b/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 { diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index a7160c539c..164270a2d2 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/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++ { - //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) + err = r.Subscribe(peerId, NewStream("SYNC", []byte{byte(i)}, false), &Range{From: 0, To: 0}, Top) 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++ { - err = r.RequestSubscription(peerId, "SYNC", []byte{byte(i)}, 0, 0, Top, true) + //for c := po + 1; c < po+skip; c++ { + 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]]))) } diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index dae4dbfdc4..2961a7e896 100644 --- a/swarm/network/stream/streamer_test.go +++ b/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})