From e631f66b76b5e8a584ad2b67baae2412e5c811ee Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Fri, 14 Dec 2018 09:20:06 -0500 Subject: [PATCH] swarm/network/stream: PR comments --- swarm/network/stream/common_test.go | 1 - swarm/network/stream/stream.go | 2 +- swarm/network/stream/streamer_test.go | 82 +++++++++++++++++++++------ 3 files changed, 65 insertions(+), 20 deletions(-) diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 52a30cb285..7b29626084 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -44,7 +44,6 @@ var ( loglevel = flag.Int("loglevel", 2, "verbosity of logs") nodes = flag.Int("nodes", 0, "number of nodes") chunks = flag.Int("chunks", 0, "number of chunks") - printstats = flag.Bool("printstats", false, "print results to STDOUT") useMockStore = flag.Bool("mockstore", false, "disabled mock store (default: enabled)") longrunning = flag.Bool("longrunning", false, "do run long-running tests") diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index bd4dc2127f..485a69ea28 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -940,7 +940,7 @@ func (api *API) GetPeerSubscriptions() map[string][]string { pstreams := make(map[string][]string) //iterate all streamer peers for id, p := range api.streamer.peers { - streams := make([]string, 0) + var streams []string //every peer has a map of stream servers //every stream server represents a subscription for s := range p.servers { diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index b938f3852c..5468051474 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -1128,6 +1128,8 @@ func TestGetSubscriptionsRPC(t *testing.T) { nodeCount := 16 //set the syncUpdateDelay for sync registrations to start syncUpdateDelay := 500 * time.Millisecond + //we will later need the kad table for each node + bucketKeyKad := simulation.BucketKey("kademlia") //create a standard sim sim := simulation.New(map[string]simulation.ServiceFunc{ "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { @@ -1143,6 +1145,8 @@ func TestGetSubscriptionsRPC(t *testing.T) { return nil, nil, err } kad := network.NewKademlia(addr.Over(), network.NewKadParams()) + //store the kad table + bucket.Store(bucketKeyKad, kad) delivery := NewDelivery(kad, netStore) netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New //configure so that sync registrations actually happen @@ -1174,17 +1178,62 @@ func TestGetSubscriptionsRPC(t *testing.T) { t.Fatal(err) } - //wait till healthy - if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { - t.Fatal(err) - } - + //run the simulation result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { - //we need to wait for some time until registrations are finished... - time.Sleep(syncUpdateDelay + 1*time.Second) + log.Info("Simulation running") nodes := sim.Net.Nodes + //setup the filter for SubscribeMsg + msgs := sim.PeerEvents( + context.Background(), + sim.NodeIDs(), + simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(4), //4 is SubscribeMsg + ) - //iterate all nodes + //setup the vars we need + msgCount := 0 + expectedMsgCount := 0 + allSubscriptionsDone := make(chan struct{}) + + //in a loop, catch all SubscribeMsg and just add up + go func() { + for m := range msgs { + if m.Error != nil { + log.Error("stream message", "err", m.Error) + continue + } + log.Trace("stream message", "node", m.NodeID, "peer", m.PeerID) + //add one + msgCount += 1 + if msgCount == expectedMsgCount { + //the expected amount is reached + allSubscriptionsDone <- struct{}{} + return + } + } + }() + + //first iterate all nodes to get the expected number of subscriptions from the kad table + for _, node := range nodes { + item, ok := sim.NodeItem(node.ID(), bucketKeyKad) + if !ok { + return fmt.Errorf("No kademlia") + } + kad := item.(*network.Kademlia) + //define the function which should run for each connection - just count subscriptions + //this is not actually subscribing but iterating the same way as the subscriptions do, + //as we need just the number + eachBinFunc := func(p *network.Peer, bin int) bool { + expectedMsgCount += 1 + return true + } + //call the actual kademlia for the count + kad.EachBin(kad.BaseAddr(), pot.DefaultPof(kad.MaxProxDisplay), 0, eachBinFunc) + } + log.Debug("Expected message count: ", "expectedMsgCount", expectedMsgCount) + //wait until all subscriptions are done + <-allSubscriptionsDone + log.Info("All subscriptions received") + //now iterate again, this time we call each node via RPC to get its subscriptions for _, node := range nodes { //create rpc client client, err := node.Client() @@ -1203,18 +1252,15 @@ func TestGetSubscriptionsRPC(t *testing.T) { if err != nil { t.Fatal(err) } - //lenght of the subscriptions can not be smaller than number of peers + //length of the subscriptions can not be smaller than number of peers if len(pstreams) < len(registry.peers) { - t.Fatal("No subscriptions have been made") + t.Fatal("Subscription count is smaller than expected") } - //if enabled, print stats to STDOUT - if *printstats { - fmt.Println(fmt.Sprintf("node %s subscriptions:", node.String())) - for p, ps := range pstreams { - fmt.Println(fmt.Sprintf("...with node %s: ", p)) - for _, s := range ps { - fmt.Println(fmt.Sprintf("......%s", s)) - } + log.Debug(fmt.Sprintf("node %s subscriptions:", node.String())) + for p, ps := range pstreams { + log.Debug(fmt.Sprintf("...with node %s: ", p)) + for _, s := range ps { + log.Debug(fmt.Sprintf("......%s", s)) } } }