From 9f1b6852c99700c515a2ad8f49e94d178fe30de4 Mon Sep 17 00:00:00 2001 From: Vlad Date: Fri, 8 Mar 2019 19:30:17 +0400 Subject: [PATCH] swarm/pss: refactoring, file renamed --- swarm/network/simulation/kademlia.go | 35 +++++++++------ swarm/pss/{snapshot_test.go => prox_test.go} | 46 ++++++++++---------- swarm/pss/pss.go | 2 +- 3 files changed, 45 insertions(+), 38 deletions(-) rename swarm/pss/{snapshot_test.go => prox_test.go} (93%) diff --git a/swarm/network/simulation/kademlia.go b/swarm/network/simulation/kademlia.go index 16752c4586..06329a02bd 100644 --- a/swarm/network/simulation/kademlia.go +++ b/swarm/network/simulation/kademlia.go @@ -99,9 +99,13 @@ func (s *Simulation) kademlias() (ks map[enode.ID]*network.Kademlia) { return ks } +// WaitTillSnapshotRecreated is blocking until all the connections specified +// in the snapshot are actually up and running. +// It differs from WaitTillHealthy, which waits only until all the kademlias are +// healthy (it might happen even before all the connections are established). func (s *Simulation) WaitTillSnapshotRecreated(ctx context.Context, snap simulations.Snapshot) error { - expected := listSnapshotConnections(snap.Conns) - ticker := time.NewTicker(16 * time.Millisecond) + expected := getSnapshotConnections(snap.Conns) + ticker := time.NewTicker(150 * time.Millisecond) defer ticker.Stop() for { @@ -109,7 +113,7 @@ func (s *Simulation) WaitTillSnapshotRecreated(ctx context.Context, snap simulat case <-ctx.Done(): return ctx.Err() case <-ticker.C: - actual := s.listActualConnections() + actual := s.getActualConnections() if isAllDeployed(expected, actual) { return nil } @@ -117,7 +121,7 @@ func (s *Simulation) WaitTillSnapshotRecreated(ctx context.Context, snap simulat } } -func (s *Simulation) listActualConnections() (res []uint64) { +func (s *Simulation) getActualConnections() (res []uint64) { kademlias := s.kademlias() for base, k := range kademlias { k.EachConn(base[:], 256, func(p *network.Peer, _ int) bool { @@ -131,9 +135,10 @@ func (s *Simulation) listActualConnections() (res []uint64) { return res } -func listSnapshotConnections(conns []simulations.Conn) (res []uint64) { +func getSnapshotConnections(conns []simulations.Conn) (res []uint64) { for _, c := range conns { res = append(res, getConnectionHash(c.One, c.Other)) + c.String() } return res } @@ -150,17 +155,19 @@ func getConnectionHash(a, b enode.ID) uint64 { // returns true if all connections in expected are listed in actual func isAllDeployed(expected []uint64, actual []uint64) bool { + if len(expected) == 0 { + return true + } + exp := make([]uint64, len(expected)) copy(exp, expected) - if len(exp) > 0 { - for _, c := range actual { - // remove value c from exp - for i := 0; i < len(exp); i++ { - if exp[i] == c { - exp = removeListElement(exp, i) - if len(exp) == 0 { - return true - } + for _, c := range actual { + // remove value c from exp + for i := 0; i < len(exp); i++ { + if exp[i] == c { + exp = removeListElement(exp, i) + if len(exp) == 0 { + return true } } } diff --git a/swarm/pss/snapshot_test.go b/swarm/pss/prox_test.go similarity index 93% rename from swarm/pss/snapshot_test.go rename to swarm/pss/prox_test.go index efca60b60c..b90896b1a7 100644 --- a/swarm/pss/snapshot_test.go +++ b/swarm/pss/prox_test.go @@ -51,10 +51,10 @@ var ( mu sync.Mutex // keeps handlerDonc in sync sim *simulation.Simulation - handlerDone bool // set to true on termination of the simulation run - msgsToReceive int // total count of messages to receive, used for terminating the simulation run - maxMessages int - msgCnt int + handlerDone bool // set to true on termination of the simulation run + requiredMessages int + allowedMessages int + messageCount int kademlias map[enode.ID]*network.Kademlia nodeAddrs map[enode.ID][]byte // make predictable overlay addresses from the generated random enode ids @@ -72,9 +72,9 @@ var ( func resetTestVariables() { handlerDone = false - msgsToReceive = 0 - maxMessages = 0 - msgCnt = 0 + requiredMessages = 0 + allowedMessages = 0 + messageCount = 0 msgs = nil sim = nil @@ -133,8 +133,8 @@ func readSnapshot(t *testing.T, nodeCount int) simulations.Snapshot { return snap } -func assingTestVariables(sim *simulation.Simulation, msgCount int) { - log.Debug("-------------------------------------------------------------------------") +func assignTestVariables(sim *simulation.Simulation, msgCount int) { + log.Debug("TestProxNetwork start") for _, nodeId := range sim.NodeIDs() { nodeAddrs[nodeId] = nodeIDToAddr(nodeId) } @@ -144,7 +144,7 @@ func assingTestVariables(sim *simulation.Simulation, msgCount int) { msgs = append(msgs, msgAddr.Bytes()) smallestPo := 256 var targets []enode.ID - var prev int + var closestPO int // loop through all nodes and add the message to recipient indices for _, nod := range sim.Net.GetNodes() { @@ -152,16 +152,16 @@ func assingTestVariables(sim *simulation.Simulation, msgCount int) { depth := kademlias[nod.ID()].NeighbourhoodDepth() // only nodes with closest IDs (wrt msg) will receive the msg - if po > prev { - prev = po + if po > closestPO { + closestPO = po targets = nil targets = append(targets, nod.ID()) - } else if po == prev { + } else if po == closestPO { targets = append(targets, nod.ID()) } if po >= depth { - maxMessages++ + allowedMessages++ allowed[i] = append(allowed[i], nod.ID()) allowedMsgs[nod.ID()] = append(allowedMsgs[nod.ID()], uint64(i)) } @@ -173,7 +173,7 @@ func assingTestVariables(sim *simulation.Simulation, msgCount int) { } } - msgsToReceive += len(targets) + requiredMessages += len(targets) for _, id := range targets { recipients[i] = append(recipients[i], id) expectedMsgs[id] = append(expectedMsgs[id], uint64(i)) @@ -181,7 +181,7 @@ func assingTestVariables(sim *simulation.Simulation, msgCount int) { log.Debug("nn for msg", "targets", len(recipients[i]), "msgidx", i, "msg", common.Bytes2Hex(msgAddr[:8]), "sender", senders[i], "senderpo", smallestPo) } - log.Debug("msgs to receive", "count", msgsToReceive) + log.Debug("msgs to receive", "count", requiredMessages) } func TestProxNetwork(t *testing.T) { @@ -218,14 +218,14 @@ func testProxNetwork(t *testing.T) { if err != nil { t.Fatalf("failed to recreate snapshot: %s", err) } - assingTestVariables(sim, msgCount) + assignTestVariables(sim, msgCount) result := sim.Run(ctx, runFunc) if result.Error != nil { // context deadline exceeded // however, it might just mean that not all possible messages are received // now we must check if all required messages are received - log.Debug("--------------------------------------------------------------------------------", "rcv", msgCnt) - if msgCnt < msgsToReceive { + log.Debug("TestProxNetwork finnished", "rcv", messageCount) + if messageCount < requiredMessages { t.Fatal(result.Error) } } @@ -258,8 +258,8 @@ func runFunc(ctx context.Context, sim *simulation.Simulation) error { return err case hn := <-msgC: received++ - log.Debug("msg received", "msgs_received", received, "total_expected", msgsToReceive, "id", hn.id, "serial", hn.serial) - if received >= maxMessages { + log.Debug("msg received", "msgs_received", received, "total_expected", requiredMessages, "id", hn.id, "serial", hn.serial) + if received == allowedMessages { close(doneC) return nil } @@ -316,8 +316,8 @@ func handlerChannelListener(ctx context.Context) { func nodeMsgHandler(config *adapters.NodeConfig) *handler { return &handler{ f: func(msg []byte, p *p2p.Peer, asymmetric bool, keyid string) error { - msgCnt++ - log.Debug("nodeMsgHandler rcv", "cnt", msgCnt) + messageCount++ + log.Debug("nodeMsgHandler rcv", "cnt", messageCount) // using simple serial in message body, makes it easy to keep track of who's getting what serial, c := binary.Uvarint(msg) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 440eb9cadd..0d02c9b8de 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -379,7 +379,7 @@ func (p *Pss) handlePssMsg(ctx context.Context, msg interface{}) error { if !ok { return fmt.Errorf("invalid message type. Expected *PssMsg, got %T ", msg) } - log.Trace("handler", "self, topic", fmt.Sprintf("%s %s", label(p.Kademlia.BaseAddr()), label(pssmsg.Payload.Topic[:]))) + log.Trace("handler", "self", label(p.Kademlia.BaseAddr()), "topic", label(pssmsg.Payload.Topic[:])) if int64(pssmsg.Expire) < time.Now().Unix() { metrics.GetOrRegisterCounter("pss.expire", nil).Inc(1) log.Warn("pss filtered expired message", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", common.ToHex(pssmsg.To))