swarm/network, swarm/storage, p2p/similations: fix stream tests add delivery benchmarks

- memstore garbage collect does not delete open requests
 - dbstore batch write errors handling  somewhat impoved but should be checkable after dbStored fires
 - inproc adapter does not allow message events in p2p server
 - network pkg allows loglevel flag
 - streamer registry gets a defaultSkipCheck option which the RequestFromPeers uses
 - waitForPeers gets a parameter how many peers it should wait for
 - number of peers to wait for is given by the global peerCount NodeID -> int function
 - make delivery and received chunk channels buffered with const deliveryCap
 - change all sendpriority calls to use context to avoid disconnects due to buffer contention (but potential memory leak!)
 - check error for all sends
 - abstract out trigger function for pivot
 - abstract out client calls for IDs
 - introduce channel to control check function is only called if previous one finished even if triggered (dubious)
 - implement benchmarks for delivery through a chain of requests through multiple hops
 - abstract out batchDone function on client
 - rename peer locks to client/serverMu
 - testing CheckResult only gives averages if there are more than one node to passed
 - testing implememts WatchDisconnections which aborts the simulation
 - testing implements PivorTrigger and ClientCall

p2p/simulations/adapters/inproc.go

@@ -112,7 +112,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 			MaxPeers:        math.MaxInt32,
 			NoDiscovery:     true,
 			Dialer:          s,
-			EnableMsgEvents: true,
+			EnableMsgEvents: false,
 		},
 		NoUSB:  true,
 		Logger: log.New("node.id", id.String()),

swarm/network/protocol_test.go

@@ -17,16 +17,29 @@
 package network
 
 import (
+	"flag"
 	"fmt"
+	"os"
 	"sync"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
 )
 
+var (
+	adapter  = flag.String("adapter", "sim", "type of simulation: sim|socket|exec|docker")
+	loglevel = flag.Int("loglevel", 2, "verbosity of logs")
+)
+
+func init() {
+	flag.Parse()
+	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(true))))
+}
+
 type testStore struct {
 	sync.Mutex
 

swarm/network/stream/common_test.go

@@ -39,7 +39,9 @@ var (
 )
 
 var (
-	waitPeerErrC chan error
+	defaultSkipCheck bool
+	waitPeerErrC     chan error
+	chunkSize        = 4096
 )
 
 var services = adapters.Services{
@@ -56,7 +58,7 @@ func init() {
 
 }
 
-// newService
+// NewStreamerService
 func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	id := ctx.Config.ID
 	addr := toAddr(id)
@@ -65,12 +67,11 @@ 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 nil })
+	r := NewRegistry(addr, delivery, store, defaultSkipCheck)
-	r := NewRegistry(addr, delivery, netStore)
 	RegisterSwarmSyncerServer(r, db)
 	RegisterSwarmSyncerClient(r, db)
 	go func() {
-		waitPeerErrC <- waitForPeers(r, 1*time.Second, 1)
+		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
 	}()
 	return r, nil
 }
@@ -96,7 +97,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
-	streamer := NewRegistry(addr, delivery, localStore)
+	streamer := NewRegistry(addr, delivery, localStore, defaultSkipCheck)
 	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
 
 	err = waitForPeers(streamer, 1*time.Second, 1)

swarm/network/stream/delivery.go

@@ -20,12 +20,16 @@ import (
 	"errors"
 	"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"
 )
 
-const swarmChunkServerStreamName = "RETRIEVE_REQUEST"
+const (
+	swarmChunkServerStreamName = "RETRIEVE_REQUEST"
+	deliveryCap                = 32
+)
 
 type Delivery struct {
 	db       *storage.DBAPI
@@ -39,7 +43,7 @@ func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery {
 	d := &Delivery{
 		db:       db,
 		overlay:  overlay,
-		receiveC: make(chan *ChunkDeliveryMsg, 10),
+		receiveC: make(chan *ChunkDeliveryMsg, deliveryCap),
 	}
 
 	go d.processReceivedChunks()
@@ -57,7 +61,7 @@ type SwarmChunkServer struct {
 // NewSwarmChunkServer is SwarmChunkServer constructor
 func NewSwarmChunkServer(db *storage.DBAPI) *SwarmChunkServer {
 	s := &SwarmChunkServer{
-		deliveryC: make(chan []byte),
+		deliveryC: make(chan []byte, deliveryCap),
 		batchC:    make(chan []byte),
 		db:        db,
 	}
@@ -103,6 +107,7 @@ type RetrieveRequestMsg struct {
 }
 
 func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) error {
+	log.Debug("received request", "peer", sp.ID(), "hash", req.Key)
 	s, err := sp.getServer(swarmChunkServerStreamName)
 	if err != nil {
 		return err
@@ -112,6 +117,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e
 	if chunk.ReqC != nil {
 		if created {
 			if err := d.RequestFromPeers(chunk.Key[:], false, sp.ID()); err != nil {
+				log.Warn("unable to forward chunk request", "peer", sp.ID(), "key", chunk.Key, "err", err)
 				return nil
 			}
 		}
@@ -128,8 +134,10 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e
 			}
 
 			if req.SkipCheck {
-				sp.Deliver(chunk, s.priority)
+				err := sp.Deliver(chunk, s.priority)
-				return
+				if err != nil {
+					sp.Drop(err)
+				}
 			}
 			streamer.deliveryC <- chunk.Key[:]
 		}()
@@ -137,6 +145,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e
 	}
 	// TODO: call the retrieve function of the outgoing syncer
 	if req.SkipCheck {
+		log.Trace("deliver", "peer", sp.ID(), "hash", chunk.Key)
 		return sp.Deliver(chunk, s.priority)
 	}
 	streamer.deliveryC <- chunk.Key[:]
@@ -154,45 +163,60 @@ func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
 }
 
 func (d *Delivery) processReceivedChunks() {
+R:
 	for req := range d.receiveC {
 		// this should be has locally
 		chunk, err := d.db.Get(req.Key)
-		if err == nil && chunk.ReqC == nil {
+		if err != nil {
-			continue
+			log.Error("not in db? ", "key", req.Key, "chunk", chunk)
+			continue R
+		}
+		if chunk.ReqC == nil {
+			continue R
 		}
 		select {
 		case <-chunk.ReqC:
+			continue R
 		default:
-			chunk.SData = req.SData
-			d.db.Put(chunk)
-			close(chunk.ReqC)
 		}
+		chunk.SData = req.SData
+		d.db.Put(chunk)
+		log.Warn("reecived delivery", "hash", chunk.Key)
+		chunk.WaitToStore()
+		log.Warn("received delivery stored", "hash", chunk.Key)
+		close(chunk.ReqC)
+		log.Warn("received delivery requesters notified", "hash", chunk.Key)
 	}
 }
 
 // RequestFromPeers sends a chunk retrieve request to
 func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...discover.NodeID) error {
 	var success bool
+	var err error
+	log.Warn("request", "hash", hash)
 	d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool {
 		spId := p.(*network.BzzPeer).ID()
 		for _, p := range peersToSkip {
 			if p == spId {
+				log.Warn("skip peer", "peer", spId)
 				return true
 			}
 		}
 		sp := d.getPeer(spId)
+		if sp == nil {
+			log.Warn("peer not found", "id", spId)
+			return true
+		}
 		// TODO: skip light nodes that do not accept retrieve requests
-		err := sp.SendPriority(&RetrieveRequestMsg{
+		err = sp.SendPriority(&RetrieveRequestMsg{
 			Key:       hash,
 			SkipCheck: skipCheck,
 		}, Top)
-		if err == nil {
+		success = true
-			success = true
-		}
 		return false
 	})
 	if success {
-		return nil
+		return err
 	}
 	return errors.New("no peer found")
 }

swarm/network/stream/delivery_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 	"github.com/ethereum/go-ethereum/swarm/storage"
@@ -39,6 +40,7 @@ var (
 	deliveries map[discover.NodeID]*Delivery
 	stores     map[discover.NodeID]storage.ChunkStore
 	toAddr     func(discover.NodeID) *network.BzzAddr
+	peerCount  func(discover.NodeID) int
 )
 
 func TestStreamerRetrieveRequest(t *testing.T) {
@@ -305,13 +307,18 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 }
 
 func TestDeliveryFromNodes(t *testing.T) {
-	testDeliveryFromNodes(t, 2, 1, 8100, true)
+	testDeliveryFromNodes(t, 2, 1, dataChunkCount, true)
-	testDeliveryFromNodes(t, 2, 1, 8100, false)
+	testDeliveryFromNodes(t, 2, 1, dataChunkCount, false)
-	testDeliveryFromNodes(t, 3, 1, 8100, true)
+	testDeliveryFromNodes(t, 4, 1, dataChunkCount, true)
-	testDeliveryFromNodes(t, 3, 1, 8100, false)
+	testDeliveryFromNodes(t, 4, 1, dataChunkCount, false)
+	testDeliveryFromNodes(t, 8, 1, dataChunkCount, true)
+	testDeliveryFromNodes(t, 8, 1, dataChunkCount, false)
+	testDeliveryFromNodes(t, 16, 1, dataChunkCount, true)
+	testDeliveryFromNodes(t, 16, 1, dataChunkCount, false)
 }
 
-func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) {
+func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck bool) {
+	defaultSkipCheck = skipCheck
 	toAddr = network.NewAddrFromNodeID
 	conf := &streamTesting.RunConfig{
 		Adapter:   *adapter,
@@ -331,24 +338,33 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 	for i, id := range sim.IDs {
 		stores[id] = sim.Stores[i]
 	}
+	peerCount = func(id discover.NodeID) int {
+		if sim.IDs[0] == id || sim.IDs[nodes-1] == id {
+			return 1
+		}
+		return 2
+	}
 
 	// here we distribute chunks of a random file into Stores of nodes 1 to nodes
 	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
 	rrdpa.Start()
+	size := chunkCount * chunkSize
 	fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
 	// wait until all chunks stored
 	wait()
-	rrdpa.Stop()
+	defer rrdpa.Stop()
 	if err != nil {
 		t.Fatal(err.Error())
 	}
+	errc := make(chan error, 1)
+	waitPeerErrC = make(chan error)
+	quitC := make(chan struct{})
 
-	action := func(context.Context) error {
+	action := func(ctx 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 {
@@ -362,23 +378,20 @@ 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++ {
+		var j int
-			id := sim.IDs[i]
+		err := sim.CallClient(func(client *rpc.Client) error {
-			node := sim.Net.GetNode(id)
+			err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC)
-			if node == nil {
-				return fmt.Errorf("unknown node: %s", id)
-			}
-			client, err := node.Client()
 			if err != nil {
-				return fmt.Errorf("error getting node client: %s", err)
+				return 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)
 			}
+			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+			defer cancel()
+			j++
+			sid := sim.IDs[j]
+			return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+		}, sim.IDs[0:nodes-1]...)
+		if err != nil {
+			return err
 		}
 
 		// create a retriever dpa for the pivot node
@@ -386,8 +399,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 		retrieveFunc := func(chunk *storage.Chunk) error {
 			return delivery.RequestFromPeers(chunk.Key[:], skipCheck)
 		}
-		dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), retrieveFunc)
+		netStore := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), retrieveFunc)
-		dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
+		dpa := storage.NewDPA(netStore, storage.NewChunkerParams())
 		dpa.Start()
 
 		go func() {
@@ -395,51 +408,40 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 			// 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 := readAll(dpa, fileHash)
-			log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
+			log.Info(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
+			if err != nil {
+				errc <- fmt.Errorf("requesting chunks action error: %v", err)
+			}
 		}()
 		return nil
 	}
-
+	checkC := make(chan struct{})
 	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
+		defer func() { checkC <- struct{}{} }()
 		select {
+		case err := <-errc:
+			return false, err
 		case <-ctx.Done():
 			return false, ctx.Err()
 		default:
 		}
-		// try to locally retrieve the file to check if retrieve requests have been successful
-		node := sim.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 total int64
-		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		err := sim.CallClient(func(client *rpc.Client) error {
-		defer cancel()
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-		// call RPC method to streamer API readAll method to check local availability
+			defer cancel()
-		err = client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash))
+			return 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))
+		}, id)
+		log.Info(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
 		}
+		close(quitC)
 		return true, nil
 	}
 
-	trigger := 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])
-		for range ticker.C {
-			trigger <- sim.Net.Nodes[0].ID()
-		}
-	}()
-
 	conf.Step = &simulations.Step{
 		Action:  action,
-		Trigger: trigger,
+		Trigger: streamTesting.PivotTrigger(10*time.Millisecond, checkC, sim.IDs[0]),
 		// we are only testing the pivot node (net.Nodes[0])
 		Expect: &simulations.Expectation{
 			Nodes: sim.IDs[0:1],
@@ -457,3 +459,212 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 	}
 	streamTesting.CheckResult(t, result, startedAt, finishedAt)
 }
+
+func BenchmarkDeliveryFromNodesWithoutCheck(b *testing.B) {
+	for chunks := 32; chunks <= 128; chunks *= 2 {
+		for i := 2; i < 32; i *= 2 {
+			b.Run(
+				fmt.Sprintf("nodes=%v,chunks=%v", i, chunks),
+				func(b *testing.B) {
+					benchmarkDeliveryFromNodes(b, i, 1, chunks, true)
+				},
+			)
+		}
+	}
+}
+
+func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) {
+	for chunks := 32; chunks <= 128; chunks *= 2 {
+		for i := 2; i < 32; i *= 2 {
+			b.Run(
+				fmt.Sprintf("nodes=%v,chunks=%v", i, chunks),
+				func(b *testing.B) {
+					benchmarkDeliveryFromNodes(b, i, 1, chunks, false)
+				},
+			)
+		}
+	}
+}
+
+func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) {
+	toAddr = network.NewAddrFromNodeID
+	conf := &streamTesting.RunConfig{
+		Adapter:   *adapter,
+		NodeCount: nodes,
+		ConnLevel: conns,
+		ToAddr:    toAddr,
+		Services:  services,
+	}
+	defaultSkipCheck = skipCheck
+	sim, teardown, err := streamTesting.NewSimulation(conf)
+	defer teardown()
+	if err != nil {
+		b.Fatal(err.Error())
+	}
+	stores = make(map[discover.NodeID]storage.ChunkStore)
+	deliveries = make(map[discover.NodeID]*Delivery)
+	for i, id := range sim.IDs {
+		stores[id] = sim.Stores[i]
+	}
+	peerCount = func(id discover.NodeID) int {
+		if sim.IDs[0] == id || sim.IDs[nodes-1] == id {
+			return 1
+		}
+		return 2
+	}
+	// create a dpa for the last node in the chain which we are gonna write to
+	remoteDpa := storage.NewDPA(sim.Stores[nodes-1], storage.NewChunkerParams())
+	remoteDpa.Start()
+	defer remoteDpa.Stop()
+
+	// wait channel for all nodes all peer connections to set up
+	waitPeerErrC = make(chan error)
+	// channel to signal simulation initialisation with action call complete
+	// or node disconnections
+	simErrC := make(chan error)
+	quitC := make(chan struct{})
+
+	action := func(ctx 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
+		// waitPeerErrC using a global err channel to share betweem action and node service
+		i := 0
+		for err := range waitPeerErrC {
+			if err != nil {
+				return fmt.Errorf("error waiting for peers: %s", err)
+			}
+			i++
+			if i == nodes {
+				break
+			}
+		}
+
+		// each node except the last one subscribes to the upstream swarm chunk server stream
+		// which responds to chunk retrieve requests
+		var j int
+		simErrC <- sim.CallClient(func(client *rpc.Client) error {
+			err := streamTesting.WatchDisconnections(sim.IDs[j], client, simErrC, quitC)
+			if err != nil {
+				return err
+			}
+			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+			defer cancel()
+			j++
+			sid := sim.IDs[j] // the upstream peer's id
+			return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+		}, sim.IDs[0:nodes-1]...)
+		// signal to the benchmark that setup is complete
+		return err
+	}
+
+	// the check function is only triggered when the benchmark finishes
+	checkC := make(chan error)
+	trigger := make(chan discover.NodeID)
+	check := func(ctx context.Context, id discover.NodeID) (_ bool, err error) {
+		select {
+		case <-ctx.Done():
+			err = ctx.Err()
+		case err = <-checkC:
+		}
+		if err != nil {
+			return false, err
+		}
+		return true, nil
+	}
+
+	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,
+		},
+	}
+
+	// run the simulation in the background
+	errc := make(chan error)
+	go func() {
+		_, err := sim.Run(conf)
+		errc <- err
+	}()
+
+	// wait for simulation action to complete stream subscriptions
+	err = <-simErrC
+	if err != nil {
+		b.Fatalf("simulation failed to initialise. expected no error. got %v", err)
+	}
+	go func() {
+		for {
+			var err error
+			select {
+			case err = <-simErrC:
+			case <-quitC:
+			}
+			trigger <- sim.IDs[0]
+			checkC <- err
+		}
+	}()
+
+	// create a retriever dpa for the pivot node
+	// by now deliveries are set for each node by the streamer service
+	delivery := deliveries[sim.IDs[0]]
+	retrieveFunc := func(chunk *storage.Chunk) error {
+		return delivery.RequestFromPeers(chunk.Key[:], skipCheck)
+	}
+	netStore := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), retrieveFunc)
+
+	// benchmark loop
+	b.ResetTimer()
+	b.StopTimer()
+	for i := 0; i < b.N; i++ {
+		// uploading chunkCount random chunks to the last node
+		hashes := make([]storage.Key, chunkCount)
+		for i := 0; i < chunkCount; i++ {
+			// create actual size real chunks
+			hash, wait, err := remoteDpa.Store(io.LimitReader(crand.Reader, int64(chunkSize)), int64(chunkSize))
+			// wait until all chunks stored
+			wait()
+			if err != nil {
+				b.Fatalf("expected no error. got %v", err)
+			}
+			// collect the hashes
+			hashes[i] = hash
+		}
+		// now benchmark the actual retrieval
+		// netstore.Get is called for each hash in a go routine and errors are collected
+		b.StartTimer()
+		errs := make(chan error)
+		for _, hash := range hashes {
+			go func(h storage.Key) {
+				_, err := netStore.Get(h)
+				log.Warn("test check netstore get", "hash", h, "err", err)
+				errs <- err
+			}(hash)
+		}
+		// count and report retrieval errors
+		// if there are misses then chunk timeout is too low for the distance and volume (?)
+		var total, misses int
+		for err := range errs {
+			if err != nil {
+				log.Warn(err.Error())
+				misses++
+			}
+			total++
+			if total == chunkCount {
+				break
+			}
+		}
+		b.StopTimer()
+		if misses > 0 {
+			simErrC <- fmt.Errorf("%v chunk not found out of %v", misses, total)
+		}
+	}
+	// benchmark over, trigger the check function to conclude the simulation
+	close(quitC)
+	err = <-errc
+	if err != nil {
+		b.Fatalf("expected no error. got %v", err)
+	}
+}

swarm/network/stream/messages.go

@@ -79,8 +79,12 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error {
 	if err != nil {
 		return nil
 	}
-	log.Debug("received subscription", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
-	go p.SendOfferedHashes(os, req.From, req.To)
+	go func() {
+		if err := p.SendOfferedHashes(os, req.From, req.To); err != nil {
+			p.Drop(err)
+		}
+	}()
 	return nil
 }
 
@@ -128,14 +132,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	}
 	go func() {
 		wg.Wait()
-		if tf := s.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil {
+		s.next <- s.batchDone(p, req, hashes)
-			tp, err := tf()
-			if err != nil {
-				return
-			}
-			p.SendPriority(tp, s.priority)
-		}
-		s.next <- struct{}{}
 	}()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
@@ -143,7 +140,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 		s.sessionAt = req.From
 	}
 	from, to := s.nextBatch(req.To)
-	log.Debug("received batch", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
 	if from == to {
 		return nil
 	}
@@ -157,12 +154,19 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	}
 	go func() {
 		select {
-		case <-s.next:
+		case err := <-s.next:
+			if err != nil {
+				p.Drop(err)
+				return
+			}
 		case <-s.quit:
 			return
 		}
-		log.Debug("want batch", "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
+		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
-		p.SendPriority(msg, s.priority)
+		err := p.SendPriority(msg, s.priority)
+		if err != nil {
+			p.Drop(err)
+		}
 	}()
 	return nil
 }
@@ -185,10 +189,9 @@ func (m WantedHashesMsg) String() string {
 // * sends the next batch of unsynced keys
 // * sends the actual data chunks as per WantedHashesMsg
 func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
-	log.Debug("received wanted batch", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
 	s, err := p.getServer(req.Stream + keyToString(req.Key))
 	if err != nil {
-		log.Debug(err.Error())
 		return err
 	}
 	hashes := s.currentBatch

swarm/network/stream/peer.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
@@ -27,16 +28,18 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
+var sendTimeout = 5 * time.Second
+
 // Peer is the Peer extention for the streaming protocol
 type Peer struct {
 	*protocols.Peer
-	streamer   *Registry
+	streamer *Registry
-	pq         *pq.PriorityQueue
+	pq       *pq.PriorityQueue
-	outgoingMu sync.RWMutex
+	serverMu sync.RWMutex
-	incomingMu sync.RWMutex
+	clientMu sync.RWMutex
-	servers    map[string]*server
+	servers  map[string]*server
-	clients    map[string]*client
+	clients  map[string]*client
-	quit       chan struct{}
+	quit     chan struct{}
 }
 
 // NewPeer is the constructor for Peer
@@ -64,12 +67,14 @@ func (p *Peer) Deliver(chunk *storage.Chunk, priority uint8) error {
 		Key:   chunk.Key,
 		SData: chunk.SData,
 	}
-	return p.pq.Push(nil, msg, int(priority))
+	return p.SendPriority(msg, priority)
 }
 
-// Deliver sends a storeRequestMsg protocol message to the peer
+// SendPriority sends message to the peer using the outgoing priority queue
 func (p *Peer) SendPriority(msg interface{}, priority uint8) error {
-	return p.pq.Push(nil, msg, int(priority))
+	ctx, cancel := context.WithTimeout(context.Background(), sendTimeout)
+	defer cancel()
+	return p.pq.Push(ctx, msg, int(priority))
 }
 
 // SendOfferedHashes sends OfferedHashesMsg protocol msg
@@ -92,13 +97,13 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 		Stream:        s.stream,
 		Key:           s.key,
 	}
-	log.Debug("Swarm syncer offer batch", "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
+	log.Warn("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
 	return p.SendPriority(msg, s.priority)
 }
 
 func (p *Peer) getServer(s string) (*server, error) {
-	p.outgoingMu.RLock()
+	p.serverMu.RLock()
-	defer p.outgoingMu.RUnlock()
+	defer p.serverMu.RUnlock()
 
 	server := p.servers[s]
 	if server == nil {
@@ -108,8 +113,8 @@ func (p *Peer) getServer(s string) (*server, error) {
 }
 
 func (p *Peer) getClient(s string) (*client, error) {
-	p.incomingMu.RLock()
+	p.clientMu.RLock()
-	defer p.incomingMu.RUnlock()
+	defer p.clientMu.RUnlock()
 
 	client := p.clients[s]
 	if client == nil {
@@ -119,8 +124,8 @@ func (p *Peer) getClient(s string) (*client, error) {
 }
 
 func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*server, error) {
-	p.outgoingMu.Lock()
+	p.serverMu.Lock()
-	defer p.outgoingMu.Unlock()
+	defer p.serverMu.Unlock()
 
 	sk := s + keyToString(key)
 	if p.servers[sk] != nil {
@@ -137,14 +142,14 @@ func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*serve
 }
 
 func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error {
-	p.incomingMu.Lock()
+	p.clientMu.Lock()
-	defer p.incomingMu.Unlock()
+	defer p.clientMu.Unlock()
 
 	sk := s + keyToString(key)
 	if p.clients[sk] != nil {
 		return fmt.Errorf("client %v already registered", sk)
 	}
-	next := make(chan struct{}, 1)
+	next := make(chan error, 1)
 	// var intervals *Intervals
 	// if !live {
 	// key := s + p.ID().String()
@@ -159,6 +164,6 @@ func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bo
 		stream:   s,
 		key:      key,
 	}
-	next <- struct{}{} // this is to allow wantedKeysMsg before first batch arrives
+	next <- nil // this is to allow wantedKeysMsg before first batch arrives
 	return nil
 }

swarm/network/stream/stream.go

@@ -38,8 +38,8 @@ const (
 	Mid
 	High
 	Top
-	PriorityQueue        // number of queues
+	PriorityQueue         // number of queues
-	PriorityQueueCap = 3 // queue capacity
+	PriorityQueueCap = 32 // queue capacity
 	HashSize         = 32
 )
 
@@ -47,6 +47,7 @@ const (
 type Registry struct {
 	api         *API
 	addr        *network.BzzAddr
+	skipCheck   bool
 	clientMu    sync.RWMutex
 	serverMu    sync.RWMutex
 	peersMu     sync.RWMutex
@@ -58,9 +59,10 @@ type Registry struct {
 }
 
 // NewRegistry is Streamer constructor
-func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore) *Registry {
+func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, skipCheck bool) *Registry {
 	streamer := &Registry{
 		addr:        addr,
+		skipCheck:   skipCheck,
 		store:       store,
 		serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)),
 		clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)),
@@ -154,7 +156,7 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, t
 }
 
 func (r *Registry) Retrieve(chunk *storage.Chunk) error {
-	return r.delivery.RequestFromPeers(chunk.Key[:], false)
+	return r.delivery.RequestFromPeers(chunk.Key[:], r.skipCheck)
 }
 
 func (r *Registry) NodeInfo() interface{} {
@@ -265,7 +267,7 @@ type client struct {
 	stream    string
 	key       []byte
 	quit      chan struct{}
-	next      chan struct{}
+	next      chan error
 }
 
 // Client interface for incoming peer Streamer
@@ -305,6 +307,17 @@ func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
 	return nextFrom, nextTo
 }
 
+func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error {
+	if tf := c.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil {
+		tp, err := tf()
+		if err != nil {
+			return err
+		}
+		return p.SendPriority(tp, c.priority)
+	}
+	return nil
+}
+
 // Spec is the spec of the streamer protocol
 var Spec = &protocols.Spec{
 	Name:       "stream",

swarm/network/stream/syncer.go

@@ -29,8 +29,8 @@ import (
 )
 
 const (
-	BatchSize = 2
+	// BatchSize = 2
-	// BatchSize = 128
+	BatchSize = 128
 )
 
 // SwarmSyncerServer implements an OutgoingStreamer for history syncing on bins
@@ -171,6 +171,8 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (
 // 	}
 // }
 
+// RegisterSwarmSyncerClient registers the client constructor function for
+// to handle incoming sync streams
 func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
 	streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte) (Client, error) {
 		return NewSwarmSyncerClient(p, db, nil)
@@ -180,12 +182,16 @@ func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
 // NeedData
 func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
 	chunk, _ := s.db.GetOrCreateRequest(key)
+	log.Warn("created request", "key", chunk.Key)
 	// TODO: we may want to request from this peer anyway even if the request exists
 	if chunk.ReqC == nil {
 		return nil
 	}
 	// create request and wait until the chunk data arrives and is stored
-	return chunk.WaitToStore
+	return func() {
+		chunk.WaitToStore()
+		log.Warn("stored", "key", chunk.Key)
+	}
 }
 
 // BatchDone

swarm/network/stream/syncer_test.go

@@ -28,19 +28,27 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
+const dataChunkCount = 500
+
 func TestSyncerSimulation(t *testing.T) {
-	testSyncBetweenNodes(t, 2, 1, 81000, true, 1)
+	testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1)
-	testSyncBetweenNodes(t, 2, 1, 81000, false, 1)
+	// testSyncBetweenNodes(t, 2, 1, dataChunkCount, false, 1)
-	testSyncBetweenNodes(t, 3, 1, 81000, true, 1)
+	testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1)
-	testSyncBetweenNodes(t, 3, 1, 81000, false, 1)
+	// testSyncBetweenNodes(t, 4, 1, dataChunkCount, false, 1)
+	testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1)
+	// testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1)
+	testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1)
+	// testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1)
 }
 
-func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, po uint8) {
+func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck bool, po uint8) {
+	defaultSkipCheck = skipCheck
 	toAddr = func(id discover.NodeID) *network.BzzAddr {
 		addr := network.NewAddrFromNodeID(id)
 		addr.OAddr[0] = byte(0)
@@ -64,30 +72,43 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool,
 	for i, id := range sim.IDs {
 		stores[id] = sim.Stores[i]
 	}
-
+	peerCount = func(id discover.NodeID) int {
+		if sim.IDs[0] == id || sim.IDs[nodes-1] == id {
+			return 1
+		}
+		return 2
+	}
 	// here we distribute chunks of a random file into Stores of nodes 1 to nodes
 	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
 	rrdpa.Start()
+	size := chunkCount * chunkSize
 	_, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	// need to wait cos we then immediately collect the relevant bin content
+	wait()
 	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
-	// 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)
-	// create a retriever dpa for the pivot node
-	action := func(context.Context) error {
 
+	// collect hashes in po 1 from all nodes
+	var hashes []storage.Key
+	dbs := make([]*storage.DBAPI, nodes)
+	for i := 0; i < nodes; i++ {
+		dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore))
+	}
+	for i := 1; i < nodes; i++ {
+		dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
+			hashes = append(hashes, key)
+			return true
+		})
+	}
+
+	waitPeerErrC = make(chan error)
+	action := func(ctx context.Context) error {
+		// need to wait till an aynchronous process registers the peers in streamer.peers
+		// that is used by Subscribe
+		// the global peerCount function tells how many connections each node has
+		// TODO: this is to be reimplemented with peerEvent watcher without global var
 		i := 0
 		for err := range waitPeerErrC {
 			if err != nil {
@@ -98,71 +119,42 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool,
 				break
 			}
 		}
-
+		// each node Subscribes to each other's swarmChunkServerStreamName
-		for i := 0; i < len(sim.IDs)-1; i++ {
+		j := 0
-			id := sim.IDs[i]
+		return sim.CallClient(func(client *rpc.Client) error {
-			// if err := streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
+			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
-			// 	log.Warn("error in subscribe", "err", err)
+			defer cancel()
-			// }
+			j++
-			node := sim.Net.GetNode(id)
+			return client.CallContext(ctx, nil, "stream_subscribeStream", sim.IDs[j], "SYNC", []byte{1}, 0, 0, Top, false)
-			if node == nil {
+		}, sim.IDs[0:nodes-1]...)
-				return fmt.Errorf("unknown node: %s", id)
-			}
-			client, err := node.Client()
-			if err != nil {
-				return fmt.Errorf("error getting node client: %s", err)
-			}
-			sid := sim.IDs[i+1]
-			if err := client.Call(nil, "stream_subscribeStream", sid, "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
-				return fmt.Errorf("error subscribing: %s", err)
-			}
-		}
-		return nil
-	}
-
-	dbs := make([]*storage.DBAPI, nodes)
-	for i := 0; i < nodes; i++ {
-		dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore))
 	}
 
+	// this makes sure check is not called before the previous call finishes
+	checkC := make(chan struct{})
 	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
-		if id != sim.Net.Nodes[0].ID() {
+		defer func() { checkC <- struct{}{} }()
-			return true, nil
+
-		}
 		select {
 		case <-ctx.Done():
 			return false, ctx.Err()
 		default:
 		}
 
-		var found, total int
+		var found int
-		for i := 1; i < nodes; i++ {
+		total := len(hashes)
-			dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
+		for _, key := range hashes {
-				_, err := dbs[0].Get(key)
+			_, err := dbs[0].Get(key)
-				if err == nil {
+			if err == nil {
-					found++
+				found++
-				}
+			}
-				total++
-				return true
-			})
 		}
 		log.Debug("sync check", "bin", po, "found", found, "total", total)
 		return found == total, nil
 	}
 
-	trigger := 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])
-		for range ticker.C {
-			trigger <- sim.Net.Nodes[0].ID()
-		}
-	}()
-
 	conf.Step = &simulations.Step{
 		Action:  action,
-		Trigger: trigger,
+		Trigger: streamTesting.PivotTrigger(10*time.Millisecond, checkC, sim.IDs[0]),
 		Expect: &simulations.Expectation{
 			Nodes: sim.IDs[0:1],
 			Check: check,

swarm/network/stream/testing/testing.go

@@ -28,9 +28,11 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
@@ -61,7 +63,8 @@ func SetStores(addrs ...network.Addr) ([]storage.ChunkStore, func(), error) {
 		stores[i] = store
 	}
 	teardown := func() {
-		for _, datadir := range datadirs {
+		for i, datadir := range datadirs {
+			stores[i].Close()
 			os.RemoveAll(datadir)
 		}
 	}
@@ -94,20 +97,22 @@ func NewAdapter(adapterType string, services adapters.Services) (adapter adapter
 }
 
 func CheckResult(t *testing.T, result *simulations.StepResult, startedAt, finishedAt time.Time) {
-	t.Logf("Simulation with %d nodes passed in %s", len(result.Passes), result.FinishedAt.Sub(result.StartedAt))
+	t.Logf("Simulation passed in %s", result.FinishedAt.Sub(result.StartedAt))
-	var min, max time.Duration
+	if len(result.Passes) > 1 {
-	var sum int
+		var min, max time.Duration
-	for _, pass := range result.Passes {
+		var sum int
-		duration := pass.Sub(result.StartedAt)
+		for _, pass := range result.Passes {
-		if sum == 0 || duration < min {
+			duration := pass.Sub(result.StartedAt)
-			min = duration
+			if sum == 0 || duration < min {
+				min = duration
+			}
+			if duration > max {
+				max = duration
+			}
+			sum += int(duration.Nanoseconds())
 		}
-		if duration > max {
+		t.Logf("Min: %s, Max: %s, Average: %s", min, max, time.Duration(sum/len(result.Passes))*time.Nanosecond)
-			max = duration
-		}
-		sum += int(duration.Nanoseconds())
 	}
-	t.Logf("Min: %s, Max: %s, Average: %s", min, max, time.Duration(sum/len(result.Passes))*time.Nanosecond)
 	t.Logf("Setup: %s, Shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt))
 }
 
@@ -195,8 +200,7 @@ func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) {
 		}
 	}
 	wg.Wait()
-
+	log.Info(fmt.Sprintf("simulation with %v nodes", len(s.Addrs)))
-	log.Debug(fmt.Sprintf("nodes: %v", len(s.Addrs)))
 
 	// create an only locally retrieving dpa for the pivot node to test
 	// if retriee requests have arrived
@@ -206,3 +210,59 @@ func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) {
 	result := simulations.NewSimulation(s.Net).Run(ctx, conf.Step)
 	return result, nil
 }
+
+func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error {
+	events := make(chan *p2p.PeerEvent)
+	sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
+	if err != nil {
+		return fmt.Errorf("error getting peer events for node %v: %s", id, err)
+	}
+	go func() {
+		defer sub.Unsubscribe()
+		select {
+		case <-quitC:
+			return
+		case e := <-events:
+			errc <- fmt.Errorf("peerEvent for node %v: %v", id, e)
+		case err := <-sub.Err():
+			if err != nil {
+				errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err)
+			}
+		}
+	}()
+	return nil
+}
+
+func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID) chan discover.NodeID {
+	trigger := make(chan discover.NodeID)
+	go func() {
+		ticker := time.NewTicker(d)
+		defer ticker.Stop()
+		// we are only testing the pivot node (net.Nodes[0])
+		for range ticker.C {
+			for _, id := range ids {
+				trigger <- id
+			}
+			<-checkC
+		}
+	}()
+	return trigger
+}
+
+func (sim *Simulation) CallClient(f func(*rpc.Client) error, ids ...discover.NodeID) error {
+	for _, id := range ids {
+		node := sim.Net.GetNode(id)
+		if node == nil {
+			return fmt.Errorf("unknown node: %s", id)
+		}
+		client, err := node.Client()
+		if err != nil {
+			return fmt.Errorf("error getting node client: %s", err)
+		}
+		err = f(client)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

swarm/storage/dbstore.go

@@ -599,8 +599,9 @@ func (s *DbStore) writeBatches() {
 		s.batchC = make(chan bool)
 		s.batch = new(leveldb.Batch)
 		s.lock.Unlock()
-		log.Trace(fmt.Sprintf("DbStore: spawn batch write (%d chunks) ", b.Len()))
+		err := s.writeBatch(b, e, d, a)
-		s.writeBatch(b, e, d, a)
+		// TODO: set this error on the batch, then tell the chunk
+		log.Trace(fmt.Sprintf("DbStore: spawn batch write (%d chunks): %v", b.Len(), err))
 		close(c)
 		if e >= s.capacity {
 			log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e))
@@ -611,15 +612,16 @@ func (s *DbStore) writeBatches() {
 }
 
 // must be called non concurrently
-func (s *DbStore) writeBatch(b *leveldb.Batch, entryCnt, dataIdx, accessCnt uint64) {
+func (s *DbStore) writeBatch(b *leveldb.Batch, entryCnt, dataIdx, accessCnt uint64) error {
 	b.Put(keyEntryCnt, U64ToBytes(entryCnt))
 	b.Put(keyDataIdx, U64ToBytes(dataIdx))
 	b.Put(keyAccessCnt, U64ToBytes(accessCnt))
 	l := s.batch.Len()
 	if err := s.db.Write(b); err != nil {
-		log.Error(fmt.Sprintf("unable to write batch: %v", err))
+		return fmt.Errorf("unable to write batch: %v", err)
 	}
 	log.Trace(fmt.Sprintf("DbStore: batch write (%d chunks) complete", l))
+	return nil
 }
 
 // newMockEncodeDataFunc returns a function that stores the chunk data

swarm/storage/memstore.go

@@ -240,7 +240,7 @@ func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
 
 func (s *MemStore) removeOldest() {
 	node := s.memtree
-
+	log.Warn("purge memstore")
 	for node.entry == nil {
 
 		aidx := uint(0)
@@ -284,9 +284,11 @@ func (s *MemStore) removeOldest() {
 	<-node.entry.dbStored
 	log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v saved to DBStore. Ready to clear from mem.", node.entry.Key.Log()))
 
-	if node.entry.SData != nil {
+	if node.entry.ReqC == nil {
 		node.entry = nil
 		s.entryCnt--
+	} else {
+		return
 	}
 
 	node.access[0] = 0