client: add bzz client, update smoke tests

client/bzz.go

@@ -0,0 +1,71 @@
+package client
+
+import (
+	"strings"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethersphere/swarm/log"
+	"github.com/ethersphere/swarm/storage"
+)
+
+type Bzz struct {
+	client *rpc.Client
+}
+
+func NewBzz(client *rpc.Client) *Bzz {
+	return &Bzz{
+		client: client,
+	}
+}
+
+// GetChunksBitVectorFromHost returns a bit vector of presence for a given slice of chunks from a given host
+func (b *Bzz) GetChunksBitVectorFromHost(addrs []storage.Address) (string, error) {
+	var hostChunks string
+	const trackChunksPageSize = 7500
+
+	for len(addrs) > 0 {
+		var pageChunks string
+		// get current page size, so that we avoid a slice out of bounds on the last page
+		pagesize := trackChunksPageSize
+		if len(addrs) < trackChunksPageSize {
+			pagesize = len(addrs)
+		}
+
+		err := b.client.Call(&pageChunks, "bzz_has", addrs[:pagesize])
+		if err != nil {
+			return "", err
+		}
+		hostChunks += pageChunks
+		addrs = addrs[pagesize:]
+	}
+
+	return hostChunks, nil
+}
+
+// GetBzzAddrFromHost returns the bzzAddr for a given host
+func (b *Bzz) GetBzzAddrFromHost() (string, error) {
+	var hive string
+
+	err := b.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
+}
+
+// IsSyncing is checking if the node is still receiving chunk deliveries due to pull syncing
+func (b *Bzz) IsSyncing() (bool, error) {
+	var isSyncing bool
+	err := b.client.Call(&isSyncing, "bzz_isSyncing")
+	if err != nil {
+		log.Error("error calling host for isSyncing", "err", err)
+		return false, err
+	}
+
+	return isSyncing, nil
+}

cmd/swarm-smoke/upload_and_sync.go

@@ -24,7 +24,6 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"os"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -33,8 +32,10 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethersphere/swarm/chunk"
+	"github.com/ethersphere/swarm/client"
 	"github.com/ethersphere/swarm/storage"
 	"github.com/ethersphere/swarm/testutil"
+	"golang.org/x/sync/errgroup"
 
 	cli "gopkg.in/urfave/cli.v1"
 )
@@ -116,14 +117,16 @@ func trackChunks(testData []byte, submitMetrics bool) error {
 				return
 			}
 
-			hostChunks, err := getChunksBitVectorFromHost(rpcClient, addrs)
+			bzzClient := client.NewBzz(rpcClient)
+
+			hostChunks, err := bzzClient.GetChunksBitVectorFromHost(addrs)
 			if err != nil {
 				log.Error("error getting chunks bit vector from host", "err", err, "host", httpHost)
 				hasErr = true
 				return
 			}
 
-			bzzAddr, err := getBzzAddrFromHost(rpcClient)
+			bzzAddr, err := bzzClient.GetBzzAddrFromHost()
 			if err != nil {
 				log.Error("error getting bzz addrs from host", "err", err, "host", httpHost)
 				hasErr = true
@@ -175,46 +178,6 @@ func trackChunks(testData []byte, submitMetrics bool) error {
 	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
-	const trackChunksPageSize = 7500
-
-	for len(addrs) > 0 {
-		var pageChunks string
-		// get current page size, so that we avoid a slice out of bounds on the last page
-		pagesize := trackChunksPageSize
-		if len(addrs) < trackChunksPageSize {
-			pagesize = len(addrs)
-		}
-
-		err := client.Call(&pageChunks, "bzz_has", addrs[:pagesize])
-		if err != nil {
-			return "", err
-		}
-		hostChunks += pageChunks
-		addrs = addrs[pagesize:]
-	}
-
-	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.
@@ -338,29 +301,6 @@ func uploadAndSync(c *cli.Context, randomBytes []byte) error {
 	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()
 
@@ -370,22 +310,39 @@ func waitToSync() {
 		time.Sleep(3 * time.Second)
 
 		notSynced := uint64(0)
-		var wg sync.WaitGroup
-		wg.Add(len(hosts))
+
+		var g errgroup.Group
 		for i := 0; i < len(hosts); i++ {
 			i := i
-			go func(idx int) {
-				stillSyncing, err := isSyncing(wsEndpoint(hosts[idx]))
+			g.Go(func() error {
+				rpcClient, err := rpc.Dial(wsEndpoint(hosts[i]))
+				if rpcClient != nil {
+					defer rpcClient.Close()
+				}
+				if err != nil {
+					log.Error("error dialing host", "err", err)
+					return err
+				}
+
+				bzzClient := client.NewBzz(rpcClient)
 
-				if stillSyncing || err != nil {
+				stillSyncing, err := bzzClient.IsSyncing()
+				if err != nil {
+					return err
+				}
+
+				if stillSyncing {
 					atomic.AddUint64(&notSynced, 1)
 				}
-				wg.Done()
-			}(i)
+
+				return nil
+			})
 		}
-		wg.Wait()
 
-		ns = atomic.LoadUint64(&notSynced)
+		// Wait for all RPC calls to complete.
+		if err := g.Wait(); err == nil {
+			ns = atomic.LoadUint64(&notSynced)
+		}
 	}
 
 	t2 := time.Since(t1)