From 0f4db1c632f9813d02a09f5a956d0af5b59fe565 Mon Sep 17 00:00:00 2001 From: Vlad Date: Fri, 1 Mar 2019 16:36:20 +0400 Subject: [PATCH] swarm/pss: fixed bug in pss.process, test added --- swarm/pss/pss.go | 15 +- swarm/pss/snapshot_test.go | 450 +++++++++++++++++++++++++++++++++++++ 2 files changed, 456 insertions(+), 9 deletions(-) create mode 100644 swarm/pss/snapshot_test.go diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 0a8d757d8a..440eb9cadd 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", label(p.Kademlia.BaseAddr()), "topic", label(pssmsg.Payload.Topic[:])) + log.Trace("handler", "self, topic", fmt.Sprintf("%s %s", label(p.Kademlia.BaseAddr()), 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)) @@ -415,11 +415,11 @@ func (p *Pss) handlePssMsg(ctx context.Context, msg interface{}) error { } isRecipient := p.isSelfPossibleRecipient(pssmsg, isProx) if !isRecipient { - log.Trace("pss was for someone else :'( ... forwarding", "pss", common.ToHex(p.BaseAddr()), "prox", isProx) + log.Trace("pss msg forwarding ===>", "pss", common.ToHex(p.BaseAddr()), "prox", isProx) return p.enqueue(pssmsg) } - log.Trace("pss for us, yay! ... let's process!", "pss", common.ToHex(p.BaseAddr()), "prox", isProx, "raw", isRaw, "topic", label(pssmsg.Payload.Topic[:])) + log.Trace("pss msg processing <===", "pss", common.ToHex(p.BaseAddr()), "prox", isProx, "raw", isRaw, "topic", label(pssmsg.Payload.Topic[:])) if err := p.process(pssmsg, isRaw, isProx); err != nil { qerr := p.enqueue(pssmsg) if qerr != nil { @@ -463,14 +463,11 @@ func (p *Pss) process(pssmsg *PssMsg, raw bool, prox bool) error { payload = recvmsg.Payload } - if len(pssmsg.To) < addressLength { - if err := p.enqueue(pssmsg); err != nil { - return err - } + if len(pssmsg.To) < addressLength || prox { + err = p.enqueue(pssmsg) } p.executeHandlers(psstopic, payload, from, raw, prox, asymmetric, keyid) - - return nil + return err } // copy all registered handlers for respective topic in order to avoid data race or deadlock diff --git a/swarm/pss/snapshot_test.go b/swarm/pss/snapshot_test.go new file mode 100644 index 0000000000..4d44113abe --- /dev/null +++ b/swarm/pss/snapshot_test.go @@ -0,0 +1,450 @@ +package pss + +import ( + "context" + "encoding/binary" + "encoding/json" + "errors" + "flag" + "fmt" + "io/ioutil" + "os" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/p2p/simulations" + "github.com/ethereum/go-ethereum/p2p/simulations/adapters" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/network/simulation" + "github.com/ethereum/go-ethereum/swarm/pot" + "github.com/ethereum/go-ethereum/swarm/state" +) + +var ( + runNodes = flag.Int("nodes", 0, "nodes to start in the network") + runMessages = flag.Int("messages", 0, "messages to send during test") + + topic = BytesToTopic([]byte{0x00, 0x00, 0x06, 0x82}) + mu sync.Mutex // keeps handlerDonc in sync + kademlias = make(map[enode.ID]*network.Kademlia) + nodeAddrs = make(map[enode.ID][]byte) // make predictable overlay addresses from the generated random enode ids + msgsToReceive int // total count of messages to receive, used for terminating the simulation run + recipients = make(map[int][]enode.ID) // for logging output only + msgs [][]byte // recipient addresses of messages + expectedMsgs = make(map[enode.ID][]uint64) // message serials we expect respective nodes to receive + senders = make(map[int]enode.ID) // originating nodes of the messages (intention is to choose as far as possible from the receiving neighborhood) + pof = pot.DefaultPof(256) // generate messages and index them + sim *simulation.Simulation + handlerDone bool // set to true on termination of the simulation run + handlerC = make(chan handlerNotification) // passes message from pss message handler to simulation driver + doneC = make(chan struct{}) // terminates the handler channel listener + errC = make(chan error) // error to pass to main sim thread + msgC = make(chan handlerNotification) // message receipt notification to main sim thread + debugCnt int +) + +// needed to make the enode id of the receiving node available to the handler for triggers +type handlerContextFunc func(*adapters.NodeConfig) *handler + +// struct to notify reception of messages to simulation driver +// TODO To make code cleaner: +// - consider a separate pss unwrap to message event in sim framework (this will make eventual message propagation analysis with pss easier/possible in the future) +// - consider also test api calls to inspect handling results of messages +type handlerNotification struct { + id enode.ID + serial uint64 +} + +func init() { + log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))) +} + +func isDone() bool { + mu.Lock() + defer mu.Unlock() + return handlerDone +} + +func setDone() { + mu.Lock() + defer mu.Unlock() + handlerDone = true +} + +func getCmdParams(t *testing.T) (int, int) { + args := strings.Split(t.Name(), "/") + msgCount, err := strconv.ParseInt(args[1], 10, 16) + if err != nil { + t.Fatal(err) + } + nodeCount, err := strconv.ParseInt(args[2], 10, 16) + if err != nil { + t.Fatal(err) + } + return int(msgCount), int(nodeCount) +} + +func readSnapshot(t *testing.T, nodeCount int) simulations.Snapshot { + f, err := os.Open(fmt.Sprintf("testdata/snapshot_%d.json", nodeCount)) + if err != nil { + t.Fatal(err) + } + jsonbyte, err := ioutil.ReadAll(f) + if err != nil { + t.Fatal(err) + } + var snap simulations.Snapshot + err = json.Unmarshal(jsonbyte, &snap) + if err != nil { + t.Fatal(err) + } + return snap +} + +func initTestVariables(sim *simulation.Simulation, msgCount int) { + log.Debug("-------------------------------------------------------------------------") + var targets string + for _, nodeId := range sim.NodeIDs() { + nodeAddrs[nodeId] = nodeIDToAddr(nodeId) + } + + for i := 0; i < int(msgCount); i++ { + msgAddr := pot.RandomAddress() // we choose message addresses randomly + + msgs = append(msgs, msgAddr.Bytes()) + smallestPo := 256 + + // loop through all nodes and add the message to recipient indices + for _, nod := range sim.Net.GetNodes() { + po, _ := pof(msgs[i], nodeAddrs[nod.ID()], 0) + depth := kademlias[nod.ID()].NeighbourhoodDepth() + + // node has message address within nearest neighborhood depth, that means it is a recipient + if po >= depth { + recipients[i] = append(recipients[i], nod.ID()) + expectedMsgs[nod.ID()] = append(expectedMsgs[nod.ID()], uint64(i)) + msgsToReceive++ + id := nod.ID() + targets += fmt.Sprintf("%x ", id[:4]) + } + + // keep track of the smallest po value in the iteration + // the first node in the smallest value bin will be the sender + if po < smallestPo { + smallestPo = po + senders[i] = nod.ID() + } + } + log.Debug("nn for msg", "targets", len(recipients[i]), "msgidx", i, "msg", common.Bytes2Hex(msgAddr[:8]), "sender", senders[i], "senderpo", smallestPo) + //log.Debug("nn for msg", "rcptcount", len(recipients[i]), "msgidx", i, "msg", common.Bytes2Hex(msgs[i][:8]), "sender", senders[i], "senderpo", smallestPo) + } + log.Debug("msgs to receive", "count", msgsToReceive) + log.Debug("targets", "nodes", targets) +} + +func TestProxNetwork(t *testing.T) { + if (*runNodes > 0 && *runMessages == 0) || (*runMessages > 0 && *runNodes == 0) { + t.Fatal("cannot specify only one of flags --nodes and --messages") + } else if *runNodes > 0 { + t.Run(fmt.Sprintf("%d/%d", *runMessages, *runNodes), testProxNetwork) + } else { + t.Run("1/4", testProxNetwork) + } +} + +// This tests generates a sequenced number of messages with random addresses. +// It then calculates which nodes in the network have the address of each message +// within their nearest neighborhood depth, and stores them as recipients. +// Upon sending the messages, it verifies that the respective message is passed to the message handlers of these recipients. +// It will fail if a recipient handles a message it should not, or if after propagation not all expected messages are handled (timeout) +func testProxNetwork(t *testing.T) { + msgCount, nodeCount := getCmdParams(t) + handlerContextFuncs := make(map[Topic]handlerContextFunc) + handlerContextFuncs[topic] = nodeMsgHandler + services := newProxServices(true, handlerContextFuncs, kademlias) + snap := readSnapshot(t, nodeCount) + sim = simulation.New(services) + defer sim.Close() + err := sim.Net.Load(&snap) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second*16) + defer cancel() + waitTillSerenity(t, snap, sim, 4000) + initTestVariables(sim, msgCount) + result := sim.Run(ctx, runFunc) + if result.Error != nil { + log.Debug("--------------------------------------------------------------------------------", "rcv", debugCnt) + t.Fatal(result.Error) + } + t.Logf("completed %d", result.Duration) +} + +func waitTillSerenity(t *testing.T, snap simulations.Snapshot, sim *simulation.Simulation, timeout int) { + interval := 16 + expected := listSnapConnections(snap.Conns) + for ms := 0; ms < timeout; ms += interval { + actual := listSimConnections(sim.Net.Conns) + if isSerenity(expected, actual) { + return + } else { + time.Sleep(time.Millisecond * time.Duration(interval)) + } + } +} + +func listSnapConnections(conns []simulations.Conn) (res []uint64) { + for _, c := range conns { + res = append(res, getConnectionHash(c.One, c.Other)) + } + return res +} + +func listSimConnections(conns []*simulations.Conn) (res []uint64) { + for _, c := range conns { + res = append(res, getConnectionHash(c.One, c.Other)) + } + return res +} + +// returns an integer connection identifier (similar to 8-byte hash) +func getConnectionHash(a, b enode.ID) uint64 { + var h [8]byte + for i := 0; i < 8; i++ { + h[i] = a[i] ^ b[i] + } + res := binary.LittleEndian.Uint64(h[:]) + return res +} + +// returns true if all connections in expected are listed in actual +func isSerenity(expected []uint64, actual []uint64) bool { + 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 { + last := len(exp) - 1 + if last == 0 { + return true + } + exp[i] = exp[last] + exp = exp[:last] + } + } + } + } + return len(exp) == 0 +} + +func sendAllMsgs(sim *simulation.Simulation, msgs [][]byte, senders map[int]enode.ID) { + for i, msg := range msgs { + log.Debug("sending msg", "idx", i, "from", senders[i]) + nodeClient, err := sim.Net.GetNode(senders[i]).Client() + if err != nil { + log.Crit(err.Error()) + } + var uvarByte [8]byte + binary.PutUvarint(uvarByte[:], uint64(i)) + nodeClient.Call(nil, "pss_sendRaw", hexutil.Encode(msg), hexutil.Encode(topic[:]), hexutil.Encode(uvarByte[:])) + } + log.Debug("all messages sent") +} + +func runFunc(ctx context.Context, sim *simulation.Simulation) error { + go handlerChannelListener(ctx) + time.Sleep(128 * time.Millisecond) + go sendAllMsgs(sim, msgs, senders) + // collect incoming messages and terminate with corresponding status when message handler listener ends + received := 0 + for { + select { + case err := <-errC: + return err + case hn := <-msgC: + received++ + log.Debug("msg received", "msgs_received", received, "total_expected", msgsToReceive, "id", hn.id, "serial", hn.serial) + if received >= msgsToReceive { + close(doneC) + return nil + } + } + } + return nil +} + +func handlerChannelListener(ctx context.Context) { + for { + select { + case <-doneC: // graceful exit + setDone() + errC <- nil + return + + case <-ctx.Done(): // timeout or cancel + setDone() + errC <- ctx.Err() + return + + // incoming message from pss message handler + case handlerNotification := <-handlerC: + // check if recipient has already received all its messages and notify to fail the test if so + xMsgs := expectedMsgs[handlerNotification.id] + if len(xMsgs) == 0 { + setDone() + errC <- fmt.Errorf("too many messages received by recipient %x", handlerNotification.id) + return + } + + // check if message serial is in expected messages for this recipient and notify to fail the test if not + idx := -1 + for i, msg := range xMsgs { + if handlerNotification.serial == msg { + idx = i + break + } + } + if idx == -1 { + setDone() + errC <- fmt.Errorf("message %d received by wrong recipient %v", handlerNotification.serial, handlerNotification.id) + return + } + + // message is ok, so remove that message serial from the recipient expectation array and notify the main sim thread + xMsgs[idx] = xMsgs[len(xMsgs)-1] + xMsgs = xMsgs[:len(xMsgs)-1] + msgC <- handlerNotification + } + } +} + +func nodeMsgHandler(config *adapters.NodeConfig) *handler { + return &handler{ + f: func(msg []byte, p *p2p.Peer, asymmetric bool, keyid string) error { + debugCnt++ + log.Debug("nodeMsgHandler rcv", "cnt", debugCnt) + + // using simple serial in message body, makes it easy to keep track of who's getting what + serial, c := binary.Uvarint(msg) + if c <= 0 { + log.Crit(fmt.Sprintf("corrupt message received by %x (uvarint parse returned %d)", config.ID, c)) + } + + if isDone() { + return errors.New("handlers aborted") // terminate if simulation is over + } + + // pass message context to the listener in the simulation + handlerC <- handlerNotification{ + id: config.ID, + serial: serial, + } + return nil + }, + caps: &handlerCaps{ + raw: true, // we use raw messages for simplicity + prox: true, + }, + } +} + +// an adaptation of the same services setup as in pss_test.go +// replaces pss_test.go when those tests are rewritten to the new swarm/network/simulation package +func newProxServices(allowRaw bool, handlerContextFuncs map[Topic]handlerContextFunc, kademlias map[enode.ID]*network.Kademlia) map[string]simulation.ServiceFunc { + stateStore := state.NewInmemoryStore() + kademlia := func(id enode.ID) *network.Kademlia { + if k, ok := kademlias[id]; ok { + return k + } + params := network.NewKadParams() + params.MaxBinSize = 3 + params.MinBinSize = 1 + params.MaxRetries = 1000 + params.RetryExponent = 2 + params.RetryInterval = 1000000 + kademlias[id] = network.NewKademlia(id[:], params) + return kademlias[id] + } + return map[string]simulation.ServiceFunc{ + "pss": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) { + // execadapter does not exec init() + initTest() + + // create keys in whisper and set up the pss object + ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + keys, err := wapi.NewKeyPair(ctxlocal) + privkey, err := w.GetPrivateKey(keys) + pssp := NewPssParams().WithPrivateKey(privkey) + pssp.AllowRaw = allowRaw + pskad := kademlia(ctx.Config.ID) + ps, err := NewPss(pskad, pssp) + if err != nil { + return nil, nil, err + } + + // register the handlers we've been passed + var deregisters []func() + for tpc, hndlrFunc := range handlerContextFuncs { + deregisters = append(deregisters, ps.Register(&tpc, hndlrFunc(ctx.Config))) + } + + // if handshake mode is set, add the controller + // TODO: This should be hooked to the handshake test file + if useHandshake { + SetHandshakeController(ps, NewHandshakeParams()) + } + + // we expose some api calls for cheating + ps.addAPI(rpc.API{ + Namespace: "psstest", + Version: "0.3", + Service: NewAPITest(ps), + Public: false, + }) + + // return Pss and cleanups + return ps, func() { + // run the handler deregister functions in reverse order + for i := len(deregisters); i > 0; i-- { + deregisters[i-1]() + } + }, nil + }, + "bzz": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) { + // normally translation of enode id to swarm address is concealed by the network package + // however, we need to keep track of it in the test driver aswell. + // if the translation in the network package changes, that can cause thiese tests to unpredictably fail + // therefore we keep a local copy of the translation here + addr := network.NewAddr(ctx.Config.Node()) + addr.OAddr = nodeIDToAddr(ctx.Config.Node().ID()) + + hp := network.NewHiveParams() + hp.Discovery = false + config := &network.BzzConfig{ + OverlayAddr: addr.Over(), + UnderlayAddr: addr.Under(), + HiveParams: hp, + } + return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore, nil, nil), nil, nil + }, + } +} + +// makes sure we create the addresses the same way in driver and service setup +func nodeIDToAddr(id enode.ID) []byte { + return id.Bytes() +}