swarm/network: Draft test for syncer

swarm/network/bitvector/bitvector.go

@@ -1,6 +1,8 @@
 package bitvector
 
-import "errors"
+import (
+	"errors"
+)
 
 var errInvalidLength = errors.New("invalid length")
 

swarm/network/request_test.go

@@ -305,12 +305,9 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 
 }
 
-// serviceName is used with the exec adapter so the exec'd binary knows which
-// service to execute
-const serviceName = "delivery"
-
 var services = adapters.Services{
-	serviceName: newDeliveryService,
+	"delivery": newDeliveryService,
+	"syncer":   newSyncerService,
 }
 
 var (
@@ -498,6 +495,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter
 					return err
 				}
 				// wait until all chunks stored
+				// TODO: is wait() necessary?
 				wait()
 				// assign the fileHash to a global so that it is available for the check function
 				fileHash = hash
@@ -549,7 +547,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter
 			}
 		}
 
-		result, err := runSimulation(nodes, conns, action, trigger, check, adapter)
+		result, err := runSimulation(nodes, conns, "delivery", action, trigger, check, adapter)
 		if err != nil {
 			return nil, fmt.Errorf("Setting up simulation failed: %v", err)
 		}
@@ -560,7 +558,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter
 	}
 }
 
-func runSimulation(nodes, conns int, action func(*simulations.Network) func(context.Context) error, trigger func(*simulations.Network) chan discover.NodeID, check func(*simulations.Network, *storage.DPA) func(context.Context, discover.NodeID) (bool, error), adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+func runSimulation(nodes, conns int, serviceName string, action func(*simulations.Network) func(context.Context) error, trigger func(*simulations.Network) chan discover.NodeID, check func(*simulations.Network, *storage.DPA) func(context.Context, discover.NodeID) (bool, error), adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
 	// create network
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
 		ID:             "0",
@@ -654,18 +652,21 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) {
 	nodeCount++
 
 	log.Warn("new service created")
-	return &testDeliveryService{
+	self := &testStreamerService{
 		addr:     addr,
 		streamer: streamer,
-	}, nil
+	}
+	self.run = self.runDelivery
+	return self, nil
 }
 
-type testDeliveryService struct {
+type testStreamerService struct {
 	addr     *BzzAddr
 	streamer *Streamer
+	run      func(p *p2p.Peer, rw p2p.MsgReadWriter) error
 }
 
-func (tds *testDeliveryService) Protocols() []p2p.Protocol {
+func (tds *testStreamerService) Protocols() []p2p.Protocol {
 	log.Warn("Protocols function", "run", tds.run)
 	return []p2p.Protocol{
 		{
@@ -679,19 +680,19 @@ func (tds *testDeliveryService) Protocols() []p2p.Protocol {
 	}
 }
 
-func (b *testDeliveryService) APIs() []rpc.API {
+func (b *testStreamerService) APIs() []rpc.API {
 	return []rpc.API{}
 }
 
-func (b *testDeliveryService) Start(server *p2p.Server) error {
+func (b *testStreamerService) Start(server *p2p.Server) error {
 	return nil
 }
 
-func (b *testDeliveryService) Stop() error {
+func (b *testStreamerService) Stop() error {
 	return nil
 }
 
-func (b *testDeliveryService) run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 	bzzPeer := &bzzPeer{
 		Peer:      protocols.NewPeer(p, rw, StreamerSpec),
 		localAddr: b.addr,

swarm/network/streamer.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
@@ -264,7 +265,7 @@ func (self *StreamerPeer) getOutgoingStreamer(s string) (*outgoingStreamer, erro
 	defer self.outgoingLock.RUnlock()
 	streamer := self.outgoing[s]
 	if streamer == nil {
-		return nil, fmt.Errorf("stream '%v' not provided", s)
+		return nil, fmt.Errorf("outgoing stream '%v' not provided to peer %v", s, self.ID())
 	}
 	return streamer, nil
 }
@@ -274,7 +275,7 @@ func (self *StreamerPeer) getIncomingStreamer(s string) (*incomingStreamer, erro
 	defer self.incomingLock.RUnlock()
 	streamer := self.incoming[s]
 	if streamer == nil {
-		return nil, fmt.Errorf("stream '%v' not provided", s)
+		return nil, fmt.Errorf("incoming stream '%v' not provided to peer %v", s, self.ID())
 	}
 	return streamer, nil
 }
@@ -348,6 +349,7 @@ func (self *incomingStreamer) nextBatch(from uint64) (nextFrom uint64, nextTo ui
 
 // Subscribe initiates the streamer
 func (self *Streamer) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
+	log.Warn("!!!!!! Subscribe ", "peer", peerId)
 	f, err := self.GetIncomingStreamer(s)
 	if err != nil {
 		return err
@@ -362,7 +364,7 @@ func (self *Streamer) Subscribe(peerId discover.NodeID, s string, t []byte, from
 	if err != nil {
 		return err
 	}
-	err = peer.setIncomingStreamer(s, is, priority, live)
+	err = peer.setIncomingStreamer(s+string(t), is, priority, live)
 	if err != nil {
 		return err
 	}

swarm/network/syncer.go

@@ -21,6 +21,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -58,6 +60,7 @@ func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage.
 
 // to obtain the chunks from key or request db entry only
 func (self *DbAccess) getOrCreateRequest(key storage.Key) (*storage.Chunk, bool) {
+	log.Warn("getOrCreateRequest", "self", self)
 	return self.loc.GetOrCreateRequest(key)
 }
 
@@ -95,9 +98,11 @@ func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSy
 
 const maxPO = 32
 
-func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) {
+func RegisterOutgoingSyncer(streamer *Streamer, db *DbAccess) {
 	streamer.RegisterOutgoingStreamer("SYNC", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
 		syncType, po := parseSyncLabel(t)
+		// TODO: make this work for HISTORY too
+		syncType = "LIVE"
 		switch syncType {
 		case "LIVE":
 			return NewOutgoingSwarmSyncer(true, po, db)
@@ -128,17 +133,29 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64,
 	if from == 0 {
 		from = self.start
 	}
-	err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool {
+	if to <= from {
-		batch = append(batch, key[:]...)
+		to = math.MaxUint64
-		i++
-		to = idx
-		return i < batchSize
-	})
-	if err != nil {
-		return nil, 0, 0, nil, err
 	}
+	log.Warn("!!!!!!!!!!!!! setNextBatch", "from", from, "to", to, "currentStoreCount", self.db.currentBucketStorageIndex(1))
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+	for range ticker.C {
+		err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool {
+			batch = append(batch, key[:]...)
+			i++
+			to = idx
+			return i < batchSize
+		})
+		if err != nil {
+			return nil, 0, 0, nil, err
+		}
+		if len(batch) > 0 {
+			break
+		}
+	}
+
 	log.Debug("Swarm batch", "po", self.po, "len", i, "from", from, "to", to)
-	return batch, from, to, nil, nil
+	return batch, from, to + 1, nil, nil
 }
 
 // IncomingSwarmSyncer
@@ -212,16 +229,9 @@ func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) {
 	}
 }
 
-func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
+func RegisterIncomingSyncer(streamer *Streamer, db *DbAccess) {
 	streamer.RegisterIncomingStreamer("SYNC", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-		syncType, _ := parseSyncLabel(t)
+		return NewIncomingSwarmSyncer(p, db, nil)
-		switch syncType {
-		case "LIVE":
-			return NewIncomingSwarmSyncer(p, nil, nil)
-		case "HISTORY":
-			return NewIncomingSwarmSyncer(p, nil, nil)
-		}
-		return nil, fmt.Errorf("unknown sync type %q", syncType)
 	})
 	// stream = fmt.Sprintf("SYNC-%02d-delete", po)
 	// streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {

swarm/network/syncer_test.go

@@ -0,0 +1,220 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library 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 Lesser General Public License for more details.
+//
+// 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 network
+
+import (
+	"context"
+	crand "crypto/rand"
+	"fmt"
+	"io"
+	"math"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/p2p/simulations"
+	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+var nodeAddrById map[discover.NodeID]*BzzAddr
+
+func TestSyncerSimulation(t *testing.T) {
+	testSimulation(t, testSyncBetweenNodes(2, 1, 81000, true, 1))
+}
+
+func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+	return func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+		nodeAddrById = make(map[discover.NodeID]*BzzAddr)
+		trigger := func(net *simulations.Network) chan discover.NodeID {
+			triggerC := make(chan discover.NodeID)
+			ticker := time.NewTicker(500 * time.Millisecond)
+			go func() {
+				defer ticker.Stop()
+				// we are only testing the pivot node (net.Nodes[0]) but simulation needs
+				// all nodes to pass the check so we trigger each and the check function
+				// will trivially return true
+				for i := 1; i < nodes; i++ {
+					triggerC <- net.Nodes[i].ID()
+				}
+				for range ticker.C {
+					triggerC <- net.Nodes[0].ID()
+				}
+			}()
+			return triggerC
+		}
+
+		action := func(net *simulations.Network) func(context.Context) error {
+			// here we distribute chunks of a random file into localstores of nodes 1 to nodes
+			rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams())
+			rrdpa.Start()
+			// create a retriever dpa for the pivot node
+			dpacs := storage.NewDpaChunkStore(localStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
+			dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
+			dpa.Start()
+			return func(context.Context) error {
+				defer rrdpa.Stop()
+				// upload an actual random file of size size
+				_, _, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+				if err != nil {
+					return err
+				}
+				// // wait until all chunks stored
+				// wait()
+				// // assign the fileHash to a global so that it is available for the check function
+				// fileHash = hash
+				// 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
+				// 	time.Sleep(2 * time.Second)
+				// 	n, err := mustReadAll(dpa, fileHash)
+				// 	log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
+				// }()
+				return nil
+			}
+		}
+
+		check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) {
+			dbAccesses := make([]*DbAccess, nodes)
+
+			for i := 0; i < nodes; i++ {
+				dbAccesses[i] = NewDbAccess(localStores[i].(*storage.LocalStore))
+			}
+			return func(ctx context.Context, id discover.NodeID) (bool, error) {
+				var found, total int
+				dbAccesses[1].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
+					_, err := dbAccesses[0].get(key)
+					if err == nil {
+						found++
+					}
+					total++
+					return true
+				})
+
+				//
+				// if id != net.Nodes[0].ID() {
+				// 	return true, nil
+				// }
+				select {
+				case <-ctx.Done():
+					return false, ctx.Err()
+				default:
+				}
+				return found == total, nil
+				// // try to locally retrieve the file to check if retrieve requests have been successful
+				// log.Warn(fmt.Sprintf("try to locally retrieve %v", fileHash))
+				// total, err := mustReadAll(dpa, fileHash)
+				// if err != nil || total != size {
+				// 	log.Warn(fmt.Sprintf("number of bytes read %v/%v (error: %v)", total, size, err))
+				// 	return false, nil
+				// }
+				// return true, nil
+				// node := net.GetNode(id)
+				// if node == nil {
+				// 	return false, fmt.Errorf("unknown node: %s", id)
+				// }
+				// client, err := node.Client()
+				// if err != nil {
+				// 	return false, fmt.Errorf("error getting node client: %s", err)
+				// }
+				// var response int
+				// if err := client.Call(&response, "test_haslocal", hash); err != nil {
+				// 	return false, fmt.Errorf("error getting bzz_has response: %s", err)
+				// }
+				// log.Debug(fmt.Sprintf("node has: %v\n%v", id, response))
+				// return response == 0, nil
+			}
+		}
+
+		result, err := runSimulation(nodes, conns, "syncer", action, trigger, check, adapter)
+		if err != nil {
+			return nil, fmt.Errorf("Setting up simulation failed: %v", err)
+		}
+		if result.Error != nil {
+			return nil, fmt.Errorf("Simulation failed: %s", result.Error)
+		}
+		return result, err
+	}
+}
+
+func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) {
+	id := ctx.Config.ID
+	addr := NewAddrFromNodeID(id)
+	kad := NewKademlia(addr.Over(), NewKadParams())
+	localStore := localStores[nodeCount]
+	dbAccess := NewDbAccess(localStore.(*storage.LocalStore))
+	streamer := NewStreamer(NewDelivery(kad, dbAccess))
+	log.Warn("!!!!!!!! Registering syncers")
+	RegisterIncomingSyncer(streamer, dbAccess)
+	RegisterOutgoingSyncer(streamer, dbAccess)
+	addrBytes := addr.Address()
+	if nodeCount == 0 {
+		// the delivery service for the pivot node is assigned globally
+		// so that the simulation action call can use it for the
+		// swarm enabled dpa
+		delivery = streamer.delivery
+		addrBytes[0] = 0x0
+	} else {
+		addrBytes[0] = 0xF0
+	}
+	addr = &BzzAddr{
+		OAddr: addrBytes,
+		UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()),
+	}
+	nodeAddrById[id] = addr
+
+	//else {
+	// 	RegisterOutgoingSyncer(streamer, dbAccess)
+	// }
+	nodeCount++
+
+	log.Warn("new service created")
+	self := &testStreamerService{
+		addr:     addr,
+		streamer: streamer,
+	}
+	self.run = self.runSyncer
+	return self, nil
+}
+
+func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+	bzzPeer := &bzzPeer{
+		Peer:      protocols.NewPeer(p, rw, StreamerSpec),
+		localAddr: b.addr,
+		BzzAddr:   nodeAddrById[p.ID()],
+	}
+	b.streamer.delivery.overlay.On(bzzPeer)
+	defer b.streamer.delivery.overlay.Off(bzzPeer)
+	go func() {
+		// each node Subscribes to each other's retrieveRequestStream
+		// need to wait till an aynchronous process registers the peers in streamer.peers
+		// that is used by Subscribe
+		time.Sleep(1 * time.Second)
+		err := b.streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, true)
+		if err != nil {
+			log.Warn("error in subscribe", "err", err)
+		}
+	}()
+	return b.streamer.Run(bzzPeer)
+}

swarm/swarm.go

@@ -128,8 +128,8 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 
 	dbAccess := network.NewDbAccess(self.lstore)
 	self.streamer = network.NewStreamer(to, dbAccess)
-	network.RegisterOutgoingSyncers(self.streamer, dbAccess)
+	network.RegisterOutgoingSyncer(self.streamer, dbAccess)
-	network.RegisterIncomingSyncers(self.streamer, dbAccess)
+	network.RegisterIncomingSyncer(self.streamer, dbAccess)
 
 	self.bzz = network.NewBzz(bzzconfig, to, nil, self.streamer)