swarm/network/stream:

- NewStreamerService now uses netStore
 - processReceivedChunks fix negation error; chunks found and no Req should be ignored
 - processReceivedChunks fix logic: no error but ReqC doesnt block
 - rename mustReadAll to ReadAll
 - use common.Hash in RPC hash
 - on streamer api field
 - streamer API ReadAll function takes hexencoded string, byte slice aint cut it
 - streamer service Start/Stop start and stop api.dpa. fixes nil chunk channel issue

swarm/network/stream/common_test.go

@@ -65,8 +65,8 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	db := storage.NewDBAPI(store.(*storage.LocalStore))
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
-	//netStore := storage.NewNetStore(store.(*storage.LocalStore), func(*storage.Chunk) error { return errors.New("not retrieved yet") })
+	netStore := storage.NewNetStore(store.(*storage.LocalStore), func(*storage.Chunk) error { return nil })
-	r := NewRegistry(addr, delivery, store)
+	r := NewRegistry(addr, delivery, netStore)
 	RegisterSwarmSyncerServer(r, db)
 	RegisterSwarmSyncerClient(r, db)
 	go func() {

swarm/network/stream/delivery.go

@@ -18,10 +18,8 @@ package stream
 
 import (
 	"errors"
-	"fmt"
 	"time"
 
-	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
@@ -151,27 +149,21 @@ type ChunkDeliveryMsg struct {
 }
 
 func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
-	chunk, err := d.db.Get(req.Key)
-	if err != nil {
-		return err
-	}
-
 	d.receiveC <- req
-
-	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, d))
 	return nil
 }
 
 func (d *Delivery) processReceivedChunks() {
 	for req := range d.receiveC {
+		// this should be has locally
 		chunk, err := d.db.Get(req.Key)
-		if err != nil {
+		if err == nil && chunk.ReqC == nil {
 			continue
 		}
-		chunk.SData = req.SData
 		select {
 		case <-chunk.ReqC:
 		default:
+			chunk.SData = req.SData
 			d.db.Put(chunk)
 			close(chunk.ReqC)
 		}

swarm/network/stream/delivery_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
@@ -335,26 +336,19 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
 	rrdpa.Start()
 	fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
-	defer rrdpa.Stop()
-	if err != nil {
-		t.Fatal(err.Error())
-	}
 	// wait until all chunks stored
-	// TODO: is wait() necessary?
 	wait()
-	// each node Subscribes to each other's swarmChunkServerStreamName
+	rrdpa.Stop()
-	// need to wait till an aynchronous process registers the peers in streamer.peers
-	// that is used by Subscribe
-	// time.Sleep(1 * time.Second)
-	// err := streamer.Subscribe(p.ID(), swarmChunkServerStreamName, nil, 0, 0, Top, true)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
 
-	waitPeerErrC = make(chan error)
-
 	action := func(context.Context) error {
-
+		// each node Subscribes to each other's swarmChunkServerStreamName
+		// need to wait till an aynchronous process registers the peers in streamer.peers
+		// that is used by Subscribe
+		// using a global err channel to share betweem action and node service
+		waitPeerErrC = make(chan error)
 		i := 0
 		for err := range waitPeerErrC {
 			if err != nil {
@@ -366,6 +360,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 			}
 		}
 
+		// each node subscribes to the upstream swarm chunk server stream
+		// which responds to chunk retrieve requests all but the last node in the chain does not
 		for i := 0; i < len(sim.IDs)-1; i++ {
 			id := sim.IDs[i]
 			node := sim.Net.GetNode(id)
@@ -376,6 +372,9 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 			if err != nil {
 				return fmt.Errorf("error getting node client: %s", err)
 			}
+			// rpc call to streamer API subscribing to chunk Server to their
+			// unique upstream except for the last node in the chain
+			// Note in this test we only test one direction
 			sid := sim.IDs[i+1]
 			if err := client.Call(nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false); err != nil {
 				return fmt.Errorf("error subscribing: %s", err)
@@ -384,16 +383,18 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 
 		// create a retriever dpa for the pivot node
 		delivery := deliveries[sim.IDs[0]]
-		dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
+		retrieveFunc := func(chunk *storage.Chunk) error {
+			return delivery.RequestFromPeers(chunk.Key[:], skipCheck)
+		}
+		dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), retrieveFunc)
 		dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
 		dpa.Start()
 
 		go func() {
 			defer dpa.Stop()
-			log.Debug(fmt.Sprintf("retrieve %v", fileHash))
 			// start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks
 			// we must wait for the peer connections to have started before requesting
-			n, err := mustReadAll(dpa, fileHash)
+			n, err := readAll(dpa, fileHash)
 			log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
 		}()
 		return nil
@@ -417,8 +418,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 		var total int64
 		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 		defer cancel()
-		err = client.CallContext(ctx, &total, "stream_readAll", fileHash)
+		// call RPC method to streamer API readAll method to check local availability
-		// total, err := mustReadAll(dpa, fileHash)
+		err = client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash))
 		log.Debug(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err))
 		if err != nil || total != int64(size) {
 			return false, nil
@@ -439,6 +440,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 	conf.Step = &simulations.Step{
 		Action:  action,
 		Trigger: trigger,
+		// we are only testing the pivot node (net.Nodes[0])
 		Expect: &simulations.Expectation{
 			Nodes: sim.IDs[0:1],
 			Check: check,

swarm/network/stream/stream.go

@@ -22,6 +22,7 @@ import (
 	"math"
 	"sync"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 
@@ -44,6 +45,7 @@ const (
 
 // Registry registry for outgoing and incoming streamer constructors
 type Registry struct {
+	api         *API
 	addr        *network.BzzAddr
 	clientMu    sync.RWMutex
 	serverMu    sync.RWMutex
@@ -65,6 +67,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkS
 		peers:       make(map[discover.NodeID]*Peer),
 		delivery:    delivery,
 	}
+	streamer.api = NewAPI(streamer, streamer.store)
 	delivery.getPeer = streamer.getPeer
 	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, t []byte) (Server, error) {
 		return NewSwarmChunkServer(delivery.db), nil
@@ -271,7 +274,7 @@ type Client interface {
 	BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error)
 }
 
-// NextBatch adjusts the indexes by inspecting the intervals
+// nextBatch adjusts the indexes by inspecting the intervals
 func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
 	var intervals []uint64
 	if c.live {
@@ -302,7 +305,7 @@ func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
 	return nextFrom, nextTo
 }
 
-// Spec is the spec of the streamer protocol.
+// Spec is the spec of the streamer protocol
 var Spec = &protocols.Spec{
 	Name:       "stream",
 	Version:    1,
@@ -336,17 +339,19 @@ func (r *Registry) APIs() []rpc.API {
 		{
 			Namespace: "stream",
 			Version:   "0.1",
-			Service:   NewAPI(r, r.store),
+			Service:   r.api,
 			Public:    true,
 		},
 	}
 }
 
 func (r *Registry) Start(server *p2p.Server) error {
+	r.api.dpa.Start()
 	return nil
 }
 
 func (r *Registry) Stop() error {
+	r.api.dpa.Stop()
 	return nil
 }
 
@@ -363,7 +368,7 @@ func NewAPI(r *Registry, store storage.ChunkStore) *API {
 	}
 }
 
-func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) {
+func readAll(dpa *storage.DPA, hash []byte) (int64, error) {
 	r := dpa.Retrieve(hash)
 	buf := make([]byte, 1024)
 	var n int
@@ -379,21 +384,8 @@ func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) {
 	return total, nil
 }
 
-func (api *API) ReadAll(hash []byte) (int64, error) {
+func (api *API) ReadAll(hash common.Hash) (int64, error) {
-	r := api.dpa.Retrieve(hash)
+	return readAll(api.dpa, hash[:])
-	buf := make([]byte, 1024)
-	var n int
-	var total int64
-	var err error
-	for (total == 0 || n > 0) && err == nil {
-		n, err = r.ReadAt(buf, total)
-		total += int64(n)
-	}
-	if err != nil && err != io.EOF {
-		return total, err
-	}
-	return total, nil
-	//return mustReadAll(api.dpa, hash)
 }
 
 func (api *API) SubscribeStream(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {

swarm/network/stream/syncer_test.go

@@ -35,7 +35,9 @@ import (
 
 func TestSyncerSimulation(t *testing.T) {
 	testSyncBetweenNodes(t, 2, 1, 81000, true, 1)
+	testSyncBetweenNodes(t, 2, 1, 81000, false, 1)
 	testSyncBetweenNodes(t, 3, 1, 81000, true, 1)
+	testSyncBetweenNodes(t, 3, 1, 81000, false, 1)
 }
 
 func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, po uint8) {
@@ -59,7 +61,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool,
 	}
 	stores = make(map[discover.NodeID]storage.ChunkStore)
 	deliveries = make(map[discover.NodeID]*Delivery)
-	log.Warn("Stores", "len", len(sim.Stores))
 	for i, id := range sim.IDs {
 		stores[id] = sim.Stores[i]
 	}
@@ -136,7 +137,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool,
 
 		var found, total int
 		for i := 1; i < nodes; i++ {
-
 			dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
 				_, err := dbs[0].Get(key)
 				if err == nil {

swarm/storage/chunker.go

@@ -13,7 +13,6 @@
 //
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
 package storage
 
 import (
@@ -463,8 +462,7 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
 	case <-chunk.C: // bells are ringing, data have been delivered
 	}
 	if len(chunk.SData) == 0 {
-		return nil // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
+		return nil
-
 	}
 	return chunk
 }

swarm/storage/netstore.go

@@ -18,10 +18,7 @@ package storage
 
 import (
 	"encoding/binary"
-	"fmt"
 	"time"
-
-	"github.com/ethereum/go-ethereum/log"
 )
 
 // NetStore implements the ChunkStore interface,
@@ -43,7 +40,6 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
 	var created bool
 	chunk, created = self.localStore.GetOrCreateRequest(key)
 	if chunk.ReqC == nil {
-		log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData)))
 		return
 	}
 
@@ -57,7 +53,6 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
 
 	select {
 	case <-t.C:
-		log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log()))
 		return nil, notFound
 	case <-chunk.ReqC:
 	}