diff options
Diffstat (limited to 'cmd/swarm/swarm-smoke/upload_and_sync.go')
-rw-r--r-- | cmd/swarm/swarm-smoke/upload_and_sync.go | 327 |
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(¬Synced, 1) + } + wg.Done() + }(i) + } + wg.Wait() + + ns = atomic.LoadUint64(¬Synced) + } + + t2 := time.Since(t1) + metrics.GetOrRegisterResettingTimer("upload-and-sync.single.wait-for-sync.deployment", nil).Update(t2) +} |