swarm: code cleanup, move to ethersphere/swarm (#19661)

1 files changed, 0 insertions, 376 deletions

diff --git a/cmd/swarm/swarm-smoke/upload_and_sync.go b/cmd/swarm/swarm-smoke/upload_and_sync.go
deleted file mode 100644
index 7338e3473..000000000
--- a/cmd/swarm/swarm-smoke/upload_and_sync.go
+++ /dev/null
@@ -1,376 +0,0 @@
-// Copyright 2018 The go-ethereum Authors
-// This file is part of go-ethereum.
-//
-// go-ethereum is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// go-ethereum is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU General Public License for more details.
-//
-// You should have received a copy of the GNU General Public License
-// along with go-ethereum. If not, see <http://www.gnu.org/licenses/>.
-
-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/chunk"
-	"github.com/ethereum/go-ethereum/swarm/storage"
-	"github.com/ethereum/go-ethereum/swarm/testutil"
-
-	cli "gopkg.in/urfave/cli.v1"
-)
-
-func uploadAndSyncCmd(ctx *cli.Context) error {
-	// use input seed if it has been set
-	if inputSeed != 0 {
-		seed = inputSeed
-	}
-
-	randomBytes := testutil.RandomBytes(seed, filesize*1000)
-
-	errc := make(chan error)
-
-	go func() {
-		errc <- uploadAndSync(ctx, randomBytes)
-	}()
-
-	var err error
-	select {
-	case err = <-errc:
-		if err != nil {
-			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
-		}
-	case <-time.After(time.Duration(timeout) * time.Second):
-		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)
-
-		err = fmt.Errorf("timeout after %v sec", timeout)
-	}
-
-	// trigger debug functionality on randomBytes
-	e := trackChunks(randomBytes[:], true)
-	if e != nil {
-		log.Error(e.Error())
-	}
-
-	return err
-}
-
-func trackChunks(testData []byte, submitMetrics bool) error {
-	addrs, err := getAllRefs(testData)
-	if err != nil {
-		return err
-	}
-
-	for i, ref := range addrs {
-		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 {
-		host := host
-		go func() {
-			defer wg.Done()
-			httpHost := fmt.Sprintf("ws://%s:%d", host, 8546)
-
-			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-			defer cancel()
-
-			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
-			}
-
-			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)
-			}
-
-			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 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])
-		}
-
-		// 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])
-		}
-	}
-}
-
-func getAllRefs(testData []byte) (storage.AddressCollection, error) {
-	datadir, err := ioutil.TempDir("", "chunk-debug")
-	if err != nil {
-		return nil, fmt.Errorf("unable to create temp dir: %v", err)
-	}
-	defer os.RemoveAll(datadir)
-	fileStore, err := storage.NewLocalFileStore(datadir, make([]byte, 32), chunk.NewTags())
-	if err != nil {
-		return nil, err
-	}
-
-	reader := bytes.NewReader(testData)
-	return fileStore.GetAllReferences(context.Background(), reader, false)
-}
-
-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]))
-	if err != nil {
-		log.Error(err.Error())
-		return err
-	}
-	t2 := time.Since(t1)
-	metrics.GetOrRegisterResettingTimer("upload-and-sync.upload-time", nil).Update(t2)
-
-	fhash, err := digest(bytes.NewReader(randomBytes))
-	if err != nil {
-		log.Error(err.Error())
-		return err
-	}
-
-	log.Info("uploaded successfully", "hash", hash, "took", t2, "digest", fmt.Sprintf("%x", fhash))
-
-	// wait to sync and log chunks before fetch attempt, only if syncDelay is set to true
-	if syncDelay {
-		waitToSync()
-
-		log.Debug("chunks before fetch attempt", "hash", hash)
-
-		err = trackChunks(randomBytes, false)
-		if err != nil {
-			log.Error(err.Error())
-		}
-	}
-
-	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)
-}