cmd/swarm/swarm-smoke: sliding window test v2

cmd/swarm/swarm-smoke/sliding_window.go

@@ -20,8 +20,6 @@ import (
 	"bytes"
 	"fmt"
 	"math/rand"
-	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -53,13 +51,16 @@ func slidingWindow(c *cli.Context) error {
 	generateEndpoints(scheme, cluster, appName, from, to)
 	storeSize = storeSize * 4096 //store size is in chunks - transform to bytes
 	hashes := []uploadResult{}   //swarm hashes of the uploads
-	filesize := storeSize / 7    //each file to upload, bytes
 	nodes := to - from
 	networkCapacity := float64(storeSize) * float64(nodes)
 	const iterationTimeout = 30 * time.Second
 	log.Info("sliding window test started", "store size(kb)", int(storeSize/1000), "nodes", nodes, "filesize(kb)", int(filesize/1000), "network capacity(kb)", int(networkCapacity/1000), "timeout", timeout)
 	uploadedBytes := 0
-	for uploadedBytes = 0; uploadedBytes <= int(networkCapacity); uploadedBytes += filesize {
+	networkDepth := 0
+	errored := false
+
+outer:
+	for {
 		seed := int(time.Now().UnixNano() / 1e6)
 		log.Info("uploading to "+endpoints[0]+" and syncing", "seed", seed)
 
@@ -79,103 +80,33 @@ func slidingWindow(c *cli.Context) error {
 			return err
 		}
 
-		log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash))
+		log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash), "sleeping", syncDelay)
 		hashes = append(hashes, uploadResult{hash: hash, digest: fhash})
-	}
+		time.Sleep(time.Duration(syncDelay) * time.Second)
+		uploadedBytes += filesize
 
-	log.Info("done uploading files", "len(hashes)", len(hashes), "sleep for", syncDelay)
+		for i, v := range hashes {
-
-	time.Sleep(time.Duration(syncDelay) * time.Second)
-
-	networkDepth := 0
-	var errored int32 = 0
-
-	timedOut := false
-
-LOOP:
-	for i := len(hashes) - 1; i >= 0; i-- {
-		wg := sync.WaitGroup{}
-		done := time.After(iterationTimeout)
-		if single {
 			rand.Seed(time.Now().UTC().UnixNano())
 			randIndex := 1 + rand.Intn(len(endpoints)-1)
 			ruid := uuid.New()[:8]
-			wg.Add(1)
+			start := time.Now()
-			go func(endpoint string, ruid string) {
+			err := fetch(v.hash, endpoints[randIndex], v.digest, ruid)
-				// points to address:
+			fetchTime := time.Since(start)
-				// need to measure min/max/mean for the results when not in single mode (not all nodes would necessarily give the same result, though they should)
+			if err != nil {
-				defer wg.Done()
+				errored = true
-
+				log.Error("error retrieving hash", "hash idx", i, "err", err)
-			inner:
+				metrics.GetOrRegisterCounter("sliding-window.single.error", nil).Inc(1)
-				for {
+				networkDepth = i
-					select {
+				break outer
-					case <-done:
-						metrics.GetOrRegisterCounter("sliding-window.single.timeout", nil).Inc(1)
-						atomic.AddInt32(&errored, 1)
-						break inner
-					default:
-					}
-
-					start := time.Now()
-					err := fetch(hashes[i].hash, endpoint, hashes[i].digest, ruid)
-					fetchTime := time.Since(start)
-					if err != nil {
-						continue
-					}
-
-					metrics.GetOrRegisterMeter("sliding-window.single.fetch-time", nil).Mark(int64(fetchTime))
-					return
-				}
-
-			}(endpoints[randIndex], ruid)
-		} else {
-			for _, endpoint := range endpoints {
-				ruid := uuid.New()[:8]
-				wg.Add(1)
-				go func(endpoint string, ruid string) {
-					defer wg.Done()
-
-				inner:
-					for {
-						select {
-						case <-done:
-							metrics.GetOrRegisterCounter("sliding-window.multi.timeout", nil).Inc(1)
-							atomic.AddInt32(&errored, 1)
-							break inner
-						default:
-						}
-
-						start := time.Now()
-						err := fetch(hashes[i].hash, endpoint, hashes[i].digest, ruid)
-						fetchTime := time.Since(start)
-						if err != nil {
-							continue
-						}
-
-						metrics.GetOrRegisterMeter("sliding-window.each.fetch-time", nil).Mark(int64(fetchTime))
-						return
-					}
-				}(endpoint, ruid)
 			}
-		}
 
-		wg.Wait()
+			metrics.GetOrRegisterMeter("sliding-window.single.fetch-time", nil).Mark(int64(fetchTime))
-		networkDepth = len(hashes) - i
-		if errored > 0 {
-			break LOOP
-		}
-		select {
-		case <-done:
-			timedOut = true
-			break LOOP
-		default:
 		}
 	}
 
-	log.Info("sliding window test finished", "timed out?", timedOut, "errored?", errored > 0, "networkDepth", networkDepth, "networkDepth(kb)", int(networkDepth*filesize/1000))
+	log.Info("sliding window test finished", "errored?", errored, "networkDepth", networkDepth, "networkDepth(kb)", int(networkDepth*filesize/1000))
 	log.Info("stats", "uploadedFiles", len(hashes), "uploadedKb", uploadedBytes/1000, "filesizeKb", filesize/1000, "networkCapacityKb", int(networkCapacity/1000), "networkCapacityMb", int(networkCapacity/1000000))
 
 	metrics.GetOrRegisterMeter("sliding-window.network-depth", nil).Mark(int64(networkDepth))
-	metrics.GetOrRegisterMeter("sliding-window.uploaded-bytes", nil).Mark(int64(uploadedBytes))
 	return nil
 }