From 9d30da5e372427c9a3fe30bad4fd76678b44d393 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Tue, 11 Dec 2018 11:21:46 -0500 Subject: [PATCH] swarm/network/stream: cleanup visualized snapshot sync test --- .../visualized_snapshot_sync_sim_test.go | 96 ++++++++----------- 1 file changed, 40 insertions(+), 56 deletions(-) diff --git a/swarm/network/stream/visualized_snapshot_sync_sim_test.go b/swarm/network/stream/visualized_snapshot_sync_sim_test.go index ef184f9cb7..36585661a4 100644 --- a/swarm/network/stream/visualized_snapshot_sync_sim_test.go +++ b/swarm/network/stream/visualized_snapshot_sync_sim_test.go @@ -155,10 +155,11 @@ func sendSimTerminatedEvent(sim *simulation.Simulation) { //It also sends some custom events so that the frontend //can visualize messages like SendOfferedMsg, WantedHashesMsg, DeliveryMsg func TestSnapshotSyncWithServer(t *testing.T) { - - feedwrap := &netWrapper{} //t.Skip("temporarily disabled as simulations.WaitTillHealthy cannot be trusted") - //nodeCount, chunkCount, sim := setupSim(simServiceMap) + + //define a wrapper object to be able to pass around data + wrapper := &netWrapper{} + nodeCount := *nodes chunkCount := *chunks @@ -166,6 +167,9 @@ func TestSnapshotSyncWithServer(t *testing.T) { nodeCount = 32 chunkCount = 1 } + + log.Info(fmt.Sprintf("Running the simulation with %d nodes and %d chunks", nodeCount, chunkCount)) + sim := simulation.New(map[string]simulation.ServiceFunc{ "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { n := ctx.Config.Node() @@ -192,7 +196,7 @@ func TestSnapshotSyncWithServer(t *testing.T) { tr := &testRegistry{ Registry: r, - w: feedwrap, + w: wrapper, } bucket.Store(bucketKeyRegistry, tr) @@ -205,7 +209,8 @@ func TestSnapshotSyncWithServer(t *testing.T) { return tr, cleanup, nil }, - }).WithServer(":8888") + }).WithServer(":8888") //start with the HTTP server + defer sim.Close() log.Info("Initializing test config") @@ -217,8 +222,8 @@ func TestSnapshotSyncWithServer(t *testing.T) { conf.addrToIDMap = make(map[string]enode.ID) //array where the generated chunk hashes will be stored conf.hashes = make([]storage.Address, 0) - - feedwrap.setNetwork(sim.Net) + //pass the network to the wrapper object + wrapper.setNetwork(sim.Net) err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount)) if err != nil { panic(err) @@ -227,54 +232,6 @@ func TestSnapshotSyncWithServer(t *testing.T) { ctx, cancelSimRun := watchSim(sim) defer cancelSimRun() - /* - //setup filters in the event feed - offeredHashesFilter := simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(1) - wantedFilter := simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(2) - deliveryFilter := simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(10) - eventC := sim.PeerEvents(ctx, sim.UpNodeIDs(), offeredHashesFilter, wantedFilter, deliveryFilter) - - quit := make(chan struct{}) - - go func() { - for e := range eventC { - select { - case <-quit: - fmt.Println("quitting event loop") - return - default: - } - if e.Error != nil { - t.Fatal(e.Error) - } - - if e.Event.Msg.Code == uint64(1) { - evt := &simulations.Event{ - Type: EventTypeChunkOffered, - Node: sim.Net.GetNode(e.NodeID), - Control: false, - } - go sim.Net.Events().Send(evt) - } else if e.Event.Msg.Code == uint64(2) { - evt := &simulations.Event{ - Type: EventTypeChunkWanted, - Node: sim.Net.GetNode(e.NodeID), - Control: false, - } - go sim.Net.Events().Send(evt) - } else if e.Event.Msg.Code == uint64(10) { - evt := &simulations.Event{ - Type: EventTypeChunkDelivered, - Node: sim.Net.GetNode(e.NodeID), - Control: false, - } - go sim.Net.Events().Send(evt) - } else { - fmt.Println(fmt.Sprintf("msg code: %d", e.Event.Msg.Code)) - } - } - }() - */ //run the sim result := runSim(conf, ctx, sim, chunkCount) @@ -292,27 +249,36 @@ func TestSnapshotSyncWithServer(t *testing.T) { log.Info("Simulation ended") } +//testRegistry embeds registry +//it allows to replace the protocol run function type testRegistry struct { *Registry w *netWrapper } +//Protocols replaces the protocol's run function func (tr *testRegistry) Protocols() []p2p.Protocol { regProto := tr.Registry.Protocols() + //set the `stream` protocol's run function with the testRegistry's one regProto[0].Run = tr.runProto return regProto } +//runProto is the new overwritten protocol's run function for this test func (tr *testRegistry) runProto(p *p2p.Peer, rw p2p.MsgReadWriter) error { + //create a custom rw message ReadWriter testRw := &testMsgReadWriter{ MsgReadWriter: rw, Peer: p, w: tr.w, Registry: tr.Registry, } + //now run the actual upper layer `Registry`'s protocol function return tr.runProtocol(p, testRw) } +//testMsgReadWriter is a custom rw +//it will allow us to re-use the message twice type testMsgReadWriter struct { *Registry p2p.MsgReadWriter @@ -320,26 +286,33 @@ type testMsgReadWriter struct { w *netWrapper } +//netWrapper wrapper object so we can pass data around type netWrapper struct { net *simulations.Network } +//set the network to the wrapper for later use (used inside the custom rw) func (w *netWrapper) setNetwork(n *simulations.Network) { w.net = n } +//get he network from the wrapper (used inside the custom rw) func (w *netWrapper) getNetwork() *simulations.Network { return w.net } // ReadMsg reads a message from the underlying MsgReadWriter and emits a // "message received" event +//we do this because we are interested in the Payload of the message for custom use +//in this test, but messages can only be consumed once (stream io.Reader) func (ev *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { + //read the message from the underlying rw msg, err := ev.MsgReadWriter.ReadMsg() if err != nil { return msg, err } + //don't do anything with message codes we actually are not needing/reading subCodes := []uint64{1, 2, 10} found := false for _, c := range subCodes { @@ -347,10 +320,16 @@ func (ev *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { found = true } } + //just return if not a msg code we are interested in if !found { return msg, nil } + //we use a io.TeeReader so that we can read the message twice + //the Payload is a io.Reader, so if we read from it, the actual protocol handler + //cannot access it anymore. + //But we need that handler to be able to consume the message as normal, + //as if we would not do anything here with that message var buf bytes.Buffer tee := io.TeeReader(msg.Payload, &buf) @@ -360,21 +339,26 @@ func (ev *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { ReceivedAt: msg.ReceivedAt, Payload: tee, } + //assign the copy for later use msg.Payload = &buf + //now let's look into the message var wmsg protocols.WrappedMsg err = mcp.Decode(&wmsg) if err != nil { log.Error(err.Error()) return msg, err } + //create a new message from the code val, ok := ev.Registry.GetSpec().NewMsg(mcp.Code) if !ok { return msg, errors.New(fmt.Sprintf("Invalid message code: %v", msg.Code)) } + //decode it if err := rlp.DecodeBytes(wmsg.Payload, val); err != nil { return msg, errors.New(fmt.Sprintf("Decoding error <= %v: %v", msg, err)) } + //now for every message type we are interested in, create a custom event and send it var evt *simulations.Event switch val := val.(type) { case *OfferedHashesMsg: @@ -389,7 +373,6 @@ func (ev *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { Type: EventTypeChunkWanted, Node: ev.w.getNetwork().GetNode(ev.ID()), Control: false, - Data: val.Want, } case *ChunkDeliveryMsgSyncing: evt = &simulations.Event{ @@ -400,6 +383,7 @@ func (ev *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { } } if evt != nil { + //send custom event to feed; frontend will listen to it and display ev.w.getNetwork().Events().Send(evt) } return msg, nil