cmd/swarm/swarm-smoke: pss aync test, send multiple msgs

cmd/swarm/swarm-smoke/main.go

@@ -37,17 +37,18 @@ var (
 )
 
 var (
-	allhosts     string
+	allhosts        string
-	hosts        []string
+	hosts           []string
-	filesize     int
+	filesize        int
-	inputSeed    int
+	inputSeed       int
-	syncDelay    int
+	syncDelay       int
-	httpPort     int
+	httpPort        int
-	wsPort       int
+	wsPort          int
-	verbosity    int
+	verbosity       int
-	timeout      int
+	timeout         int
-	single       bool
+	single          bool
-	trackTimeout int
+	trackTimeout    int
+	pssMessageCount int
 )
 
 func main() {
@@ -116,6 +117,12 @@ func main() {
 			Usage:       "timeout in seconds to wait for GetAllReferences to return",
 			Destination: &trackTimeout,
 		},
+		cli.IntFlag{
+			Name:        "pss-messages",
+			Value:       10,
+			Usage:       "number of pss messages that should be send in the pss smoke test",
+			Destination: &pssMessageCount,
+		},
 	}
 
 	app.Flags = append(app.Flags, []cli.Flag{
@@ -155,10 +162,10 @@ func main() {
 			Action:  wrapCliCommand("sliding-window", slidingWindowCmd),
 		},
 		{
-			Name:    "pss",
+			Name:    "pss-asym",
-			Aliases: []string{"p"},
+			Aliases: []string{"pa"},
-			Usage:   "pss send and receive checks",
+			Usage:   "PSS: send and receive multiple messages across random nodes using asymmetric encryption",
-			Action:  wrapCliCommand("pss", pssChecks),
+			Action:  wrapCliCommand("pss-asym", pssAsymCheck),
 		},
 	}
 

cmd/swarm/swarm-smoke/pss.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/sha1"
+	"errors"
 	"fmt"
 	"math/rand"
 	"time"
@@ -27,7 +28,6 @@ const (
 type pssJob struct {
 	msg  []byte
 	mode int
-	prox bool
 }
 
 type pssNode struct {
@@ -39,24 +39,33 @@ type pssNode struct {
 	msgC           chan pss.APIMsg
 }
 
-// keyed by hex representation of overlay address
 type pssSession struct {
 	topic        pss.Topic
 	msgC         chan pss.APIMsg
 	nodes        []*pssNode
 	requiredJobs map[string]*pssJob
-	allowedJobs  map[string]*pssJob
 }
 
-func pssChecks(ctx *cli.Context, tuid string) error {
+// pssAsymCheck sends one or more messages (depending on pssMessageCount)
-	return pssAsym(ctx, tuid)
+// using asymmetric encryption across random nodes.
-}
+func pssAsymCheck(ctx *cli.Context, tuid string) error {
 
-func pssAsym(ctx *cli.Context, tuid string) error {
+	// use input seed if it has been set
+	if inputSeed != 0 {
+		seed = inputSeed
+	}
+	rand.Seed(int64(seed))
+	if pssMessageCount <= 0 {
+		pssMessageCount = 1
+		log.Warn(fmt.Sprintf("message count should be a positive number. Defaulting to %d", pssMessageCount))
+	}
+	log.Info("pss-asym test started", "msgCount", pssMessageCount)
 
 	errc := make(chan error)
-
 	session := pssSetup()
+	if len(session.nodes) <= 1 {
+		return errors.New("at least 2 nodes are required to be working")
+	}
 	defer func() {
 		for _, n := range session.nodes {
 			n.deregisterFunc()
@@ -64,37 +73,54 @@ func pssAsym(ctx *cli.Context, tuid string) error {
 	}()
 
 	go func() {
-		errc <- pssAsymDo(ctx, session, tuid)
+		var failCount, successCount int64
+		for i := 0; i < pssMessageCount; i++ {
+			err := pssAsymDo(ctx, session, tuid)
+			if err != nil {
+				failCount++
+				log.Error("error sending pss msg", "err", err)
+			} else {
+				successCount++
+			}
+		}
+
+		metrics.GetOrRegisterCounter("pss.asym.failMsg", nil).Inc(failCount)
+		metrics.GetOrRegisterCounter("pss.asym.successMsg", nil).Inc(successCount)
+
+		log.Info("pss-asym test ended", "success", successCount, "failures", failCount)
+
+		if failCount > 0 {
+			errc <- errors.New("some messages were not delivered")
+		} else {
+			errc <- nil
+		}
 	}()
 
 	select {
 	case err := <-errc:
 		if err != nil {
-			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
+			metrics.GetOrRegisterCounter("pss.asym.fail", nil).Inc(1)
 		}
 		return err
 	case <-time.After(time.Duration(timeout) * time.Second):
-		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)
+		metrics.GetOrRegisterCounter("pss.asym.timeout", nil).Inc(1)
-
+		return fmt.Errorf("timeout after %v sec", timeout)
-		e := fmt.Errorf("timeout after %v sec", timeout)
-		return e
 	}
 
 }
 
 func pssSetup() *pssSession {
-
 	// random topic, one per session,  same for all msgs
 	topic := pss.BytesToTopic(testutil.RandomBytes(seed, 4))
 
+	log.Trace("pss random topic", "topic", topic.String())
+
 	session := &pssSession{
 		msgC:         make(chan pss.APIMsg),
 		requiredJobs: make(map[string]*pssJob),
-		allowedJobs:  make(map[string]*pssJob),
 		topic:        topic,
 	}
 
-	// hosts is global :/
 	// set up the necessary info for each pss node
 	for i, host := range hosts {
 		httpHost := fmt.Sprintf("ws://%s:%d", host, 8546)
@@ -105,7 +131,6 @@ func pssSetup() *pssSession {
 		}
 
 		// get overlay address
-		// TODO this should be done automatically for all nodes anyway in all smokes perhaps?
 		var addr hexutil.Bytes
 		err = rpcClient.Call(&addr, "pss_baseAddr")
 		if err != nil {
@@ -124,7 +149,6 @@ func pssSetup() *pssSession {
 		// subscribe to the topic for the session
 		// this creates the incoming message handler automatically
 		// any message received with this topic comes on the channel
-		//msgC := make(chan pss.APIMsg)
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 		defer cancel()
 		sub, err := rpcClient.Subscribe(ctx, "pss", session.msgC, "receive", session.topic, true, false)
@@ -134,8 +158,7 @@ func pssSetup() *pssSession {
 		}
 		session.nodes = append(session.nodes,
 			&pssNode{
-				hostIdx: i,
+				hostIdx:        i,
-				//msgC:           msgC,
 				addr:           []byte(addr),
 				client:         rpcClient,
 				pubkey:         pubkey,
@@ -147,14 +170,20 @@ func pssSetup() *pssSession {
 	return session
 }
 
+// pssAsymDo sends a single PSS message between two random nodes using asymetric encryption
 func pssAsymDo(ctx *cli.Context, session *pssSession, tuid string) error {
 
-	// can we choose? if so, change
+	senderNodeIdx := rand.Intn(len(session.nodes))
-	senderNode := session.nodes[0]
+	senderNode := session.nodes[senderNodeIdx]
+	log.Trace("sender node", "pss_baseAddr", hexutil.Encode(senderNode.addr), "host", hosts[senderNodeIdx])
 
-	// add if single blah blah...
+	// receiving node has to be different than sender
-	recvNodeIdx := rand.Intn(len(session.nodes)-1) + 1
+	recvNodeIdx := senderNodeIdx
+	for recvNodeIdx == senderNodeIdx {
+		recvNodeIdx = rand.Intn(len(session.nodes))
+	}
 	recvNode := session.nodes[recvNodeIdx]
+	log.Trace("recv node", "pss_baseAddr", hexutil.Encode(recvNode.addr), "host", hosts[recvNodeIdx])
 
 	// set recipient in pivot node
 	err := senderNode.client.Call(nil, "pss_setPeerPublicKey", recvNode.pubkey, session.topic, "0x")
@@ -162,46 +191,39 @@ func pssAsymDo(ctx *cli.Context, session *pssSession, tuid string) error {
 		return err
 	}
 
-	// for asym it's not necessary to set recipient, but for sym we must
-	//	err = recvNode.client.Call(nil, "pss_setPeerPublicKey", senderNode.pubkey, session.topic, "0x")
-	//	if err != nil {
-	//		return err
-	//	}
-
 	// create new message and add it to job index to check for receives
 	randomMsg := testutil.RandomBytes(seed, 128)
+	// change seed so that the next random message is different
+	seed = seed + 1
+
 	msgIdx := toMsgIdx(randomMsg)
 	session.requiredJobs[msgIdx] = &pssJob{
 		msg:  randomMsg,
 		mode: pssModeAsym,
 	}
 
-	// send the msg
+	// send the message
 	hostIdx := session.nodes[recvNodeIdx].hostIdx
-	log.Info("sending pss", "sender", hosts[0], "recv", hosts[hostIdx])
+	log.Debug("sending msg", "job", hexutil.Encode([]byte(msgIdx)), "sender", hosts[senderNodeIdx], "recv", hosts[hostIdx])
 	err = senderNode.client.Call(nil, "pss_sendAsym", recvNode.pubkey, session.topic, hexutil.Encode(randomMsg))
 	if err != nil {
 		return err
 	}
 
-	// receive the msg or timeout and fail
+	// receive the message and check for its content
-	// and check its validity
-	pctx, cancel := context.WithTimeout(context.TODO(), time.Second)
-	defer cancel()
 	select {
-	//case res := <-(recvNode.msgC):
+	case res := <-session.msgC:
-	case res := <-(session.msgC):
 		resIdx := toMsgIdx(res.Msg)
 		resJob, ok := session.requiredJobs[resIdx]
 		if !ok {
-			return fmt.Errorf("corrupt message", "job", resIdx)
+			return fmt.Errorf("corrupt message for job %s", resIdx)
 		}
 		if !bytes.Equal(res.Msg, resJob.msg) {
-			return fmt.Errorf("message mismatch", "job", resIdx)
+			return fmt.Errorf("message mismatch. expected: %s got: %s", resJob.msg, res.Msg)
 		}
-		log.Info("got msg", "job", hexutil.Encode([]byte(resIdx)))
+		log.Debug("got msg", "job", hexutil.Encode([]byte(resIdx)))
-	case <-pctx.Done():
+	case <-time.After(1 * time.Second):
-		return pctx.Err()
+		return errors.New("msg timeout after 1 sec")
 	}
 	return nil
 }