cmd/swarm: remove timeout from wrapCli; revert changes to randomBytes…

… to be uploaded

cmd/swarm/swarm-smoke/feed_upload_and_sync.go

@@ -3,7 +3,6 @@ package main
 import (
 	"bytes"
 	"crypto/md5"
-	crand "crypto/rand"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -16,7 +15,9 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/swarm/storage/feed"
+	"github.com/ethereum/go-ethereum/swarm/testutil"
 	"github.com/pborman/uuid"
 	cli "gopkg.in/urfave/cli.v1"
 )
@@ -25,10 +26,27 @@ const (
 	feedRandomDataLength = 8
 )
 
-// TODO: retrieve with manifest + extract repeating code
-func feedUploadAndSync(c *cli.Context) error {
-	defer func(now time.Time) { log.Info("total time", "time", time.Since(now), "size (kb)", filesize) }(time.Now())
+func feedUploadAndSync(ctx *cli.Context, tuid string) error {
+	errc := make(chan error)
 
+	go func() {
+		errc <- fuas(ctx, tuid)
+	}()
+
+	select {
+	case err := <-errc:
+		if err != nil {
+			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)
+
+		return fmt.Errorf("timeout after %v sec", timeout)
+	}
+}
+
+func fuas(c *cli.Context, tuid string) error {
 	log.Info("generating and uploading feeds to " + httpEndpoint(hosts[0]) + " and syncing")
 
 	// create a random private key to sign updates with and derive the address
@@ -197,13 +215,11 @@ func feedUploadAndSync(c *cli.Context) error {
 	log.Info("all endpoints synced random data successfully")
 
 	// upload test file
-	seed := int(time.Now().UnixNano() / 1e6)
 	log.Info("feed uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)
 
-	h = md5.New()
-	r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
+	randomBytes := testutil.RandomBytes(seed, filesize*1000)
 
-	hash, err := upload(r, filesize*1000, httpEndpoint(hosts[0]))
+	hash, err := upload(randomBytes, httpEndpoint(hosts[0]))
 	if err != nil {
 		return err
 	}

cmd/swarm/swarm-smoke/main.go

@@ -120,25 +120,25 @@ func main() {
 			Name:    "upload_and_sync",
 			Aliases: []string{"c"},
 			Usage:   "upload and sync",
-			Action:  wrapCliCommand("upload-and-sync", true, uploadAndSync),
+			Action:  wrapCliCommand("upload-and-sync", uploadAndSync),
 		},
 		{
 			Name:    "feed_sync",
 			Aliases: []string{"f"},
 			Usage:   "feed update generate, upload and sync",
-			Action:  wrapCliCommand("feed-and-sync", true, feedUploadAndSync),
+			Action:  wrapCliCommand("feed-and-sync", feedUploadAndSync),
 		},
 		{
 			Name:    "upload_speed",
 			Aliases: []string{"u"},
 			Usage:   "measure upload speed",
-			Action:  wrapCliCommand("upload-speed", true, uploadSpeed),
+			Action:  wrapCliCommand("upload-speed", uploadSpeed),
 		},
 		{
 			Name:    "sliding_window",
 			Aliases: []string{"s"},
 			Usage:   "measure network aggregate capacity",
-			Action:  wrapCliCommand("sliding-window", false, slidingWindow),
+			Action:  wrapCliCommand("sliding-window", slidingWindow),
 		},
 	}
 

cmd/swarm/swarm-smoke/sliding_window.go

@@ -17,36 +17,49 @@
 package main
 
 import (
-	"crypto/md5"
-	crand "crypto/rand"
+	"bytes"
 	"fmt"
-	"io"
 	"math/rand"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/swarm/testutil"
 	"github.com/pborman/uuid"
 
 	cli "gopkg.in/urfave/cli.v1"
 )
 
-var seed = time.Now().UTC().UnixNano()
-
-func init() {
-	rand.Seed(seed)
-}
-
 type uploadResult struct {
 	hash   string
 	digest []byte
 }
 
-func slidingWindow(c *cli.Context) error {
+func slidingWindow(ctx *cli.Context, tuid string) error {
+	errc := make(chan error)
+
+	go func() {
+		errc <- sl(ctx, tuid)
+	}()
+
+	select {
+	case err := <-errc:
+		if err != nil {
+			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)
+
+		return fmt.Errorf("timeout after %v sec", timeout)
+	}
+}
+
+func sl(ctx *cli.Context, tuid string) error {
 	hashes := []uploadResult{} //swarm hashes of the uploads
 	nodes := len(hosts)
 	const iterationTimeout = 30 * time.Second
-	log.Info("sliding window test started", "nodes", nodes, "filesize(kb)", filesize, "timeout", timeout)
+	log.Info("sliding window test started", "tuid", tuid, "nodes", nodes, "filesize(kb)", filesize, "timeout", timeout)
 	uploadedBytes := 0
 	networkDepth := 0
 	errored := false
@@ -55,19 +68,23 @@ outer:
 	for {
 		log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)
 
-		h := md5.New()
-		r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
 		t1 := time.Now()
 
-		hash, err := upload(r, filesize*1000, httpEndpoint(hosts[0]))
+		randomBytes := testutil.RandomBytes(seed, filesize*1000)
+
+		hash, err := upload(randomBytes, httpEndpoint(hosts[0]))
 		if err != nil {
 			log.Error(err.Error())
 			return err
 		}
 
 		metrics.GetOrRegisterResettingTimer("sliding-window.upload-time", nil).UpdateSince(t1)
 
-		fhash := h.Sum(nil)
+		fhash, err := digest(bytes.NewReader(randomBytes))
+		if err != nil {
+			log.Error(err.Error())
+			return err
+		}
 
 		log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash), "sleeping", syncDelay)
 		hashes = append(hashes, uploadResult{hash: hash, digest: fhash})

cmd/swarm/swarm-smoke/upload_and_sync.go

@@ -17,50 +17,68 @@
 package main
 
 import (
-	"crypto/md5"
-	crand "crypto/rand"
+	"bytes"
 	"fmt"
-	"io"
 	"math/rand"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/swarm/testutil"
 	"github.com/pborman/uuid"
 
 	cli "gopkg.in/urfave/cli.v1"
 )
 
-func uploadAndSync(c *cli.Context) error {
-	seed := int(time.Now().UnixNano() / 1e6)
+func uploadAndSync(ctx *cli.Context, tuid string) error {
+	randomBytes := testutil.RandomBytes(seed, filesize*1000)
 
-	// test uuid
-	tuid := uuid.New()[:8]
+	errc := make(chan error)
 
-	log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "tuid", tuid, "seed", seed)
+	go func() {
+		errc <- uas(ctx, randomBytes, tuid)
+	}()
+
+	select {
+	case err := <-errc:
+		if err != nil {
+			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)
 
-	h := md5.New()
-	r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
+		// trigger debug functionality on randomBytes
+
+		return fmt.Errorf("timeout after %v sec", timeout)
+	}
+}
+
+func uas(c *cli.Context, randomBytes []byte, tuid string) error {
+	log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "tuid", tuid, "seed", seed)
 
 	t1 := time.Now()
-	hash, err := upload(r, filesize*1000, httpEndpoint(hosts[0]))
+	hash, err := upload(randomBytes, httpEndpoint(hosts[0]))
 	if err != nil {
 		log.Error(err.Error())
 		return err
 	}
 	t2 := time.Since(t1)
 	metrics.GetOrRegisterResettingTimer("upload-and-sync.upload-time", nil).Update(t2)
 
-	fhash := h.Sum(nil)
+	fhash, err := digest(bytes.NewReader(randomBytes))
+	if err != nil {
+		log.Error(err.Error())
+		return err
+	}
 
 	log.Info("uploaded successfully", "tuid", tuid, "hash", hash, "took", t2, "digest", fmt.Sprintf("%x", fhash))
 
 	time.Sleep(time.Duration(syncDelay) * time.Second)
 
 	wg := sync.WaitGroup{}
 	if single {
-		rand.Seed(time.Now().UTC().UnixNano())
 		randIndex := 1 + rand.Intn(len(hosts)-1)
 		ruid := uuid.New()[:8]
 		wg.Add(1)

cmd/swarm/swarm-smoke/upload_speed.go

@@ -17,38 +17,56 @@
 package main
 
 import (
-	"crypto/md5"
-	crand "crypto/rand"
+	"bytes"
 	"fmt"
-	"io"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/swarm/testutil"
 
 	cli "gopkg.in/urfave/cli.v1"
 )
 
-func uploadSpeed(c *cli.Context) error {
-	if len(hosts) < 2 {
-		log.Crit("less than 2 hosts")
-	}
+func uploadSpeed(ctx *cli.Context, tuid string) error {
+	log.Info("uploading to "+hosts[0], "tuid", tuid, "seed", seed)
+	randomBytes := testutil.RandomBytes(seed, filesize*1000)
+
+	errc := make(chan error)
+
+	go func() {
+		errc <- us(ctx, tuid, randomBytes)
+	}()
 
-	seed := int(time.Now().UnixNano() / 1e6)
-	log.Info("uploading to "+hosts[0], "seed", seed)
+	select {
+	case err := <-errc:
+		if err != nil {
+			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)
 
-	h := md5.New()
-	r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
+		// trigger debug functionality on randomBytes
 
+		return fmt.Errorf("timeout after %v sec", timeout)
+	}
+}
+
+func us(c *cli.Context, tuid string, data []byte) error {
 	t1 := time.Now()
-	hash, err := upload(r, filesize*1000, hosts[0])
+	hash, err := upload(data, hosts[0])
 	if err != nil {
 		log.Error(err.Error())
 		return err
 	}
 	metrics.GetOrRegisterCounter("upload-speed.upload-time", nil).Inc(int64(time.Since(t1)))
 
-	fhash := h.Sum(nil)
+	fhash, err := digest(bytes.NewReader(data))
+	if err != nil {
+		log.Error(err.Error())
+		return err
+	}
 
 	log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash))
 	return nil