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(ctx *cli.Context, tuid string) error {
-func feedUploadAndSync(c *cli.Context) error {
+	errc := make(chan error)
-	defer func(now time.Time) { log.Info("total time", "time", time.Since(now), "size (kb)", filesize) }(time.Now())
 
+	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()
+	randomBytes := testutil.RandomBytes(seed, filesize*1000)
-	r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
 
-	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"
+	"bytes"
-	crand "crypto/rand"
 	"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,11 +68,11 @@ 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
@@ -67,7 +80,11 @@ outer:
 
 		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,34 +17,49 @@
 package main
 
 import (
-	"crypto/md5"
+	"bytes"
-	crand "crypto/rand"
 	"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 {
+func uploadAndSync(ctx *cli.Context, tuid string) error {
-	seed := int(time.Now().UnixNano() / 1e6)
+	randomBytes := testutil.RandomBytes(seed, filesize*1000)
 
-	// test uuid
+	errc := make(chan error)
-	tuid := uuid.New()[:8]
 
+	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)
+
+		// 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)
 
-	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]))
+	hash, err := upload(randomBytes, httpEndpoint(hosts[0]))
 	if err != nil {
 		log.Error(err.Error())
 		return err
@@ -52,7 +67,11 @@ func uploadAndSync(c *cli.Context) error {
 	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))
 
@@ -60,7 +79,6 @@ func uploadAndSync(c *cli.Context) error {
 
 	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"
+	"bytes"
-	crand "crypto/rand"
 	"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 {
+func uploadSpeed(ctx *cli.Context, tuid string) error {
-	if len(hosts) < 2 {
+	log.Info("uploading to "+hosts[0], "tuid", tuid, "seed", seed)
-		log.Crit("less than 2 hosts")
+	randomBytes := testutil.RandomBytes(seed, filesize*1000)
+
+	errc := make(chan error)
+
+	go func() {
+		errc <- us(ctx, tuid, randomBytes)
+	}()
+
+	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)
+
+		// trigger debug functionality on randomBytes
+
+		return fmt.Errorf("timeout after %v sec", timeout)
 	}
+}
 
-	seed := int(time.Now().UnixNano() / 1e6)
+func us(c *cli.Context, tuid string, data []byte) error {
-	log.Info("uploading to "+hosts[0], "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, 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

cmd/swarm/swarm-smoke/util.go

@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math/rand"
 	"net/http"
 	"net/http/httptrace"
 	"os"
@@ -37,13 +38,19 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/api/client"
 	"github.com/ethereum/go-ethereum/swarm/spancontext"
 	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/pborman/uuid"
 	cli "gopkg.in/urfave/cli.v1"
 )
 
 var (
-	commandName = ""
+	commandName       = ""
+	seed        int64 = time.Now().UTC().UnixNano()
 )
 
+func init() {
+	rand.Seed(seed)
+}
+
 func httpEndpoint(host string) string {
 	return fmt.Sprintf("http://%s:%d", host, httpPort)
 }
@@ -52,47 +59,28 @@ func wsEndpoint(host string) string {
 	return fmt.Sprintf("ws://%s:%d", host, wsPort)
 }
 
-func wrapCliCommand(name string, killOnTimeout bool, command func(*cli.Context) error) func(*cli.Context) error {
+func wrapCliCommand(name string, command func(*cli.Context, string) error) func(*cli.Context) error {
 	return func(ctx *cli.Context) error {
 		log.PrintOrigins(true)
 		log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(verbosity), log.StreamHandler(os.Stdout, log.TerminalFormat(false))))
 
+		// test uuid
+		tuid := uuid.New()[:8]
+
+		commandName = name
+
 		hosts = strings.Split(allhosts, ",")
 
 		defer func(now time.Time) {
 			totalTime := time.Since(now)
-			log.Info("total time", "time", totalTime, "kb", filesize)
+			log.Info("total time", "tuid", tuid, "time", totalTime, "kb", filesize)
 			metrics.GetOrRegisterResettingTimer(name+".total-time", nil).Update(totalTime)
 		}(time.Now())
 
-		log.Info("smoke test starting", "task", name, "timeout", timeout)
+		log.Info("smoke test starting", "tuid", tuid, "task", name, "timeout", timeout)
-		commandName = name
 		metrics.GetOrRegisterCounter(name, nil).Inc(1)
 
-		errc := make(chan error)
+		return command(ctx, tuid)
-		done := make(chan struct{})
-
-		if killOnTimeout {
-			go func() {
-				<-time.After(time.Duration(timeout) * time.Second)
-				close(done)
-			}()
-		}
-
-		go func() {
-			errc <- command(ctx)
-		}()
-
-		select {
-		case err := <-errc:
-			if err != nil {
-				metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", name), nil).Inc(1)
-			}
-			return err
-		case <-done:
-			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", name), nil).Inc(1)
-			return fmt.Errorf("timeout after %v sec", timeout)
-		}
 	}
 }
 
@@ -210,14 +198,14 @@ func fetch(hash string, endpoint string, original []byte, ruid string, tuid stri
 }
 
 // upload an arbitrary byte as a plaintext file  to `endpoint` using the api client
-func upload(r io.Reader, size int, endpoint string) (string, error) {
+func upload(data []byte, endpoint string) (string, error) {
 	swarm := client.NewClient(endpoint)
 	f := &client.File{
-		ReadCloser: ioutil.NopCloser(r),
+		ReadCloser: ioutil.NopCloser(bytes.NewReader(data)),
 		ManifestEntry: api.ManifestEntry{
 			ContentType: "text/plain",
 			Mode:        0660,
-			Size:        int64(size),
+			Size:        int64(len(data)),
 		},
 	}
 

swarm/testutil/file.go

@@ -47,9 +47,9 @@ func TempFileWithContent(t *testing.T, content string) string {
 
 // RandomBytes returns pseudo-random deterministic result
 // because test fails must be reproducible
-func RandomBytes(seed, length int) []byte {
+func RandomBytes(seed int64, length int) []byte {
 	b := make([]byte, length)
-	reader := rand.New(rand.NewSource(int64(seed)))
+	reader := rand.New(rand.NewSource(seed))
 	for n := 0; n < length; {
 		read, err := reader.Read(b[n:])
 		if err != nil {
@@ -60,6 +60,6 @@ func RandomBytes(seed, length int) []byte {
 	return b
 }
 
-func RandomReader(seed, length int) *bytes.Reader {
+func RandomReader(seed int64, length int) *bytes.Reader {
 	return bytes.NewReader(RandomBytes(seed, length))
 }