aboutsummaryrefslogtreecommitdiffstats
path: root/cmd/swarm/swarm-smoke/upload_and_sync.go
diff options
context:
space:
mode:
Diffstat (limited to 'cmd/swarm/swarm-smoke/upload_and_sync.go')
-rw-r--r--cmd/swarm/swarm-smoke/upload_and_sync.go327
1 files changed, 248 insertions, 79 deletions
diff --git a/cmd/swarm/swarm-smoke/upload_and_sync.go b/cmd/swarm/swarm-smoke/upload_and_sync.go
index 6c20a4fa6..7338e3473 100644
--- a/cmd/swarm/swarm-smoke/upload_and_sync.go
+++ b/cmd/swarm/swarm-smoke/upload_and_sync.go
@@ -19,26 +19,27 @@ package main
import (
"bytes"
"context"
+ "encoding/hex"
"fmt"
"io/ioutil"
"math/rand"
"os"
"strings"
"sync"
+ "sync/atomic"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/rpc"
- "github.com/ethereum/go-ethereum/swarm/api"
+ "github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/testutil"
- "github.com/pborman/uuid"
cli "gopkg.in/urfave/cli.v1"
)
-func uploadAndSyncCmd(ctx *cli.Context, tuid string) error {
+func uploadAndSyncCmd(ctx *cli.Context) error {
// use input seed if it has been set
if inputSeed != 0 {
seed = inputSeed
@@ -49,7 +50,7 @@ func uploadAndSyncCmd(ctx *cli.Context, tuid string) error {
errc := make(chan error)
go func() {
- errc <- uploadAndSync(ctx, randomBytes, tuid)
+ errc <- uploadAndSync(ctx, randomBytes)
}()
var err error
@@ -65,7 +66,7 @@ func uploadAndSyncCmd(ctx *cli.Context, tuid string) error {
}
// trigger debug functionality on randomBytes
- e := trackChunks(randomBytes[:])
+ e := trackChunks(randomBytes[:], true)
if e != nil {
log.Error(e.Error())
}
@@ -73,51 +74,180 @@ func uploadAndSyncCmd(ctx *cli.Context, tuid string) error {
return err
}
-func trackChunks(testData []byte) error {
+func trackChunks(testData []byte, submitMetrics bool) error {
addrs, err := getAllRefs(testData)
if err != nil {
return err
}
for i, ref := range addrs {
- log.Trace(fmt.Sprintf("ref %d", i), "ref", ref)
+ log.Debug(fmt.Sprintf("ref %d", i), "ref", ref)
}
+ var globalYes, globalNo int
+ var globalMu sync.Mutex
+ var hasErr bool
+
+ var wg sync.WaitGroup
+ wg.Add(len(hosts))
+
+ var mu sync.Mutex // mutex protecting the allHostsChunks and bzzAddrs maps
+ allHostChunks := map[string]string{} // host->bitvector of presence for chunks
+ bzzAddrs := map[string]string{} // host->bzzAddr
+
for _, host := range hosts {
- httpHost := fmt.Sprintf("ws://%s:%d", host, 8546)
+ host := host
+ go func() {
+ defer wg.Done()
+ httpHost := fmt.Sprintf("ws://%s:%d", host, 8546)
- hostChunks := []string{}
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
- rpcClient, err := rpc.Dial(httpHost)
- if err != nil {
- log.Error("error dialing host", "err", err, "host", httpHost)
- continue
- }
+ rpcClient, err := rpc.DialContext(ctx, httpHost)
+ if rpcClient != nil {
+ defer rpcClient.Close()
+ }
+ if err != nil {
+ log.Error("error dialing host", "err", err, "host", httpHost)
+ hasErr = true
+ return
+ }
- var hasInfo []api.HasInfo
- err = rpcClient.Call(&hasInfo, "bzz_has", addrs)
- if err != nil {
- log.Error("error calling rpc client", "err", err, "host", httpHost)
- continue
- }
+ hostChunks, err := getChunksBitVectorFromHost(rpcClient, addrs)
+ if err != nil {
+ log.Error("error getting chunks bit vector from host", "err", err, "host", httpHost)
+ hasErr = true
+ return
+ }
+
+ bzzAddr, err := getBzzAddrFromHost(rpcClient)
+ if err != nil {
+ log.Error("error getting bzz addrs from host", "err", err, "host", httpHost)
+ hasErr = true
+ return
+ }
+
+ mu.Lock()
+ allHostChunks[host] = hostChunks
+ bzzAddrs[host] = bzzAddr
+ mu.Unlock()
+
+ yes, no := 0, 0
+ for _, val := range hostChunks {
+ if val == '1' {
+ yes++
+ } else {
+ no++
+ }
+ }
+
+ if no == 0 {
+ log.Info("host reported to have all chunks", "host", host)
+ }
- count := 0
- for _, info := range hasInfo {
- if info.Has {
- hostChunks = append(hostChunks, "1")
- } else {
- hostChunks = append(hostChunks, "0")
- count++
+ log.Debug("chunks", "chunks", hostChunks, "yes", yes, "no", no, "host", host)
+
+ if submitMetrics {
+ globalMu.Lock()
+ globalYes += yes
+ globalNo += no
+ globalMu.Unlock()
+ }
+ }()
+ }
+
+ wg.Wait()
+
+ checkChunksVsMostProxHosts(addrs, allHostChunks, bzzAddrs)
+
+ if !hasErr && submitMetrics {
+ // remove the chunks stored on the uploader node
+ globalYes -= len(addrs)
+
+ metrics.GetOrRegisterCounter("deployment.chunks.yes", nil).Inc(int64(globalYes))
+ metrics.GetOrRegisterCounter("deployment.chunks.no", nil).Inc(int64(globalNo))
+ metrics.GetOrRegisterCounter("deployment.chunks.refs", nil).Inc(int64(len(addrs)))
+ }
+
+ return nil
+}
+
+// getChunksBitVectorFromHost returns a bit vector of presence for a given slice of chunks from a given host
+func getChunksBitVectorFromHost(client *rpc.Client, addrs []storage.Address) (string, error) {
+ var hostChunks string
+
+ err := client.Call(&hostChunks, "bzz_has", addrs)
+ if err != nil {
+ return "", err
+ }
+
+ return hostChunks, nil
+}
+
+// getBzzAddrFromHost returns the bzzAddr for a given host
+func getBzzAddrFromHost(client *rpc.Client) (string, error) {
+ var hive string
+
+ err := client.Call(&hive, "bzz_hive")
+ if err != nil {
+ return "", err
+ }
+
+ // we make an ugly assumption about the output format of the hive.String() method
+ // ideally we should replace this with an API call that returns the bzz addr for a given host,
+ // but this also works for now (provided we don't change the hive.String() method, which we haven't in some time
+ ss := strings.Split(strings.Split(hive, "\n")[3], " ")
+ return ss[len(ss)-1], nil
+}
+
+// checkChunksVsMostProxHosts is checking:
+// 1. whether a chunk has been found at less than 2 hosts. Considering our NN size, this should not happen.
+// 2. if a chunk is not found at its closest node. This should also not happen.
+// Together with the --only-upload flag, we could run this smoke test and make sure that our syncing
+// functionality is correct (without even trying to retrieve the content).
+//
+// addrs - a slice with all uploaded chunk refs
+// allHostChunks - host->bit vector, showing what chunks are present on what hosts
+// bzzAddrs - host->bzz address, used when determining the most proximate host for a given chunk
+func checkChunksVsMostProxHosts(addrs []storage.Address, allHostChunks map[string]string, bzzAddrs map[string]string) {
+ for k, v := range bzzAddrs {
+ log.Trace("bzzAddr", "bzz", v, "host", k)
+ }
+
+ for i := range addrs {
+ var foundAt int
+ maxProx := -1
+ var maxProxHost string
+ for host := range allHostChunks {
+ if allHostChunks[host][i] == '1' {
+ foundAt++
+ }
+
+ ba, err := hex.DecodeString(bzzAddrs[host])
+ if err != nil {
+ panic(err)
+ }
+
+ // calculate the host closest to any chunk
+ prox := chunk.Proximity(addrs[i], ba)
+ if prox > maxProx {
+ maxProx = prox
+ maxProxHost = host
}
}
- if count == 0 {
- log.Info("host reported to have all chunks", "host", host)
+ if allHostChunks[maxProxHost][i] == '0' {
+ log.Error("chunk not found at max prox host", "ref", addrs[i], "host", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
+ } else {
+ log.Trace("chunk present at max prox host", "ref", addrs[i], "host", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
}
- log.Trace("chunks", "chunks", strings.Join(hostChunks, ""), "host", host)
+ // if chunk found at less than 2 hosts
+ if foundAt < 2 {
+ log.Error("chunk found at less than two hosts", "foundAt", foundAt, "ref", addrs[i])
+ }
}
- return nil
}
func getAllRefs(testData []byte) (storage.AddressCollection, error) {
@@ -126,19 +256,17 @@ func getAllRefs(testData []byte) (storage.AddressCollection, error) {
return nil, fmt.Errorf("unable to create temp dir: %v", err)
}
defer os.RemoveAll(datadir)
- fileStore, err := storage.NewLocalFileStore(datadir, make([]byte, 32))
+ fileStore, err := storage.NewLocalFileStore(datadir, make([]byte, 32), chunk.NewTags())
if err != nil {
return nil, err
}
- ctx, cancel := context.WithTimeout(context.Background(), time.Duration(trackTimeout)*time.Second)
- defer cancel()
reader := bytes.NewReader(testData)
- return fileStore.GetAllReferences(ctx, reader, false)
+ return fileStore.GetAllReferences(context.Background(), reader, false)
}
-func uploadAndSync(c *cli.Context, randomBytes []byte, tuid string) error {
- log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "tuid", tuid, "seed", seed)
+func uploadAndSync(c *cli.Context, randomBytes []byte) error {
+ log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)
t1 := time.Now()
hash, err := upload(randomBytes, httpEndpoint(hosts[0]))
@@ -155,53 +283,94 @@ func uploadAndSync(c *cli.Context, randomBytes []byte, tuid string) error {
return err
}
- log.Info("uploaded successfully", "tuid", tuid, "hash", hash, "took", t2, "digest", fmt.Sprintf("%x", fhash))
+ log.Info("uploaded successfully", "hash", hash, "took", t2, "digest", fmt.Sprintf("%x", fhash))
- time.Sleep(time.Duration(syncDelay) * time.Second)
+ // wait to sync and log chunks before fetch attempt, only if syncDelay is set to true
+ if syncDelay {
+ waitToSync()
- wg := sync.WaitGroup{}
- if single {
- randIndex := 1 + rand.Intn(len(hosts)-1)
- ruid := uuid.New()[:8]
- wg.Add(1)
- go func(endpoint string, ruid string) {
- for {
- start := time.Now()
- err := fetch(hash, endpoint, fhash, ruid, tuid)
- if err != nil {
- continue
- }
- ended := time.Since(start)
+ log.Debug("chunks before fetch attempt", "hash", hash)
- metrics.GetOrRegisterResettingTimer("upload-and-sync.single.fetch-time", nil).Update(ended)
- log.Info("fetch successful", "tuid", tuid, "ruid", ruid, "took", ended, "endpoint", endpoint)
- wg.Done()
- return
- }
- }(httpEndpoint(hosts[randIndex]), ruid)
- } else {
- for _, endpoint := range hosts[1:] {
- ruid := uuid.New()[:8]
- wg.Add(1)
- go func(endpoint string, ruid string) {
- for {
- start := time.Now()
- err := fetch(hash, endpoint, fhash, ruid, tuid)
- if err != nil {
- continue
- }
- ended := time.Since(start)
-
- metrics.GetOrRegisterResettingTimer("upload-and-sync.each.fetch-time", nil).Update(ended)
- log.Info("fetch successful", "tuid", tuid, "ruid", ruid, "took", ended, "endpoint", endpoint)
- wg.Done()
- return
- }
- }(httpEndpoint(endpoint), ruid)
+ err = trackChunks(randomBytes, false)
+ if err != nil {
+ log.Error(err.Error())
}
}
- wg.Wait()
- log.Info("all hosts synced random file successfully")
+
+ if onlyUpload {
+ log.Debug("only-upload is true, stoppping test", "hash", hash)
+ return nil
+ }
+
+ randIndex := 1 + rand.Intn(len(hosts)-1)
+
+ for {
+ start := time.Now()
+ err := fetch(hash, httpEndpoint(hosts[randIndex]), fhash, "")
+ if err != nil {
+ time.Sleep(2 * time.Second)
+ continue
+ }
+ ended := time.Since(start)
+
+ metrics.GetOrRegisterResettingTimer("upload-and-sync.single.fetch-time", nil).Update(ended)
+ log.Info("fetch successful", "took", ended, "endpoint", httpEndpoint(hosts[randIndex]))
+ break
+ }
return nil
}
+
+func isSyncing(wsHost string) (bool, error) {
+ rpcClient, err := rpc.Dial(wsHost)
+ if rpcClient != nil {
+ defer rpcClient.Close()
+ }
+
+ if err != nil {
+ log.Error("error dialing host", "err", err)
+ return false, err
+ }
+
+ var isSyncing bool
+ err = rpcClient.Call(&isSyncing, "bzz_isSyncing")
+ if err != nil {
+ log.Error("error calling host for isSyncing", "err", err)
+ return false, err
+ }
+
+ log.Debug("isSyncing result", "host", wsHost, "isSyncing", isSyncing)
+
+ return isSyncing, nil
+}
+
+func waitToSync() {
+ t1 := time.Now()
+
+ ns := uint64(1)
+
+ for ns > 0 {
+ time.Sleep(3 * time.Second)
+
+ notSynced := uint64(0)
+ var wg sync.WaitGroup
+ wg.Add(len(hosts))
+ for i := 0; i < len(hosts); i++ {
+ i := i
+ go func(idx int) {
+ stillSyncing, err := isSyncing(wsEndpoint(hosts[idx]))
+
+ if stillSyncing || err != nil {
+ atomic.AddUint64(&notSynced, 1)
+ }
+ wg.Done()
+ }(i)
+ }
+ wg.Wait()
+
+ ns = atomic.LoadUint64(&notSynced)
+ }
+
+ t2 := time.Since(t1)
+ metrics.GetOrRegisterResettingTimer("upload-and-sync.single.wait-for-sync.deployment", nil).Update(t2)
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-26 00:21:06 +0800