swarm/network/stream, swarm/storage: simplify testing code, add more debug

- add Close to server - fixes closed leveldb issue
 - Trigger func simplified
 - ClientCall simplified
 - syncer simulation check now call on each node
 - syncer simulation move defer cancel context before teardown
 - added timeout and logging process deliveries
 - improve debug log and comments

swarm/network/stream/delivery.go

@@ -60,6 +60,7 @@ type SwarmChunkServer struct {
 	batchC     chan []byte
 	db         *storage.DBAPI
 	currentLen uint64
+	quit       chan struct{}
 }
 
 // NewSwarmChunkServer is SwarmChunkServer constructor
@@ -68,6 +69,7 @@ func NewSwarmChunkServer(db *storage.DBAPI) *SwarmChunkServer {
 		deliveryC: make(chan []byte, deliveryCap),
 		batchC:    make(chan []byte),
 		db:        db,
+		quit:      make(chan struct{}),
 	}
 	go s.processDeliveries()
 	return s
@@ -79,6 +81,8 @@ func (s *SwarmChunkServer) processDeliveries() {
 	var batchC chan []byte
 	for {
 		select {
+		case <-s.quit:
+			return
 		case hash := <-s.deliveryC:
 			hashes = append(hashes, hash...)
 			batchC = s.batchC
@@ -98,6 +102,11 @@ func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64
 	return
 }
 
+// Close needs to be called on a stream server
+func (s *SwarmChunkServer) Close() {
+	close(s.quit)
+}
+
 // GetData retrives chunk data from db store
 func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) {
 	chunk, err := s.db.Get(storage.Key(key))
@@ -168,30 +177,40 @@ type ChunkDeliveryMsg struct {
 
 func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error {
 	d.counterIn++
+	log.Error("push to receiveC", "hash", storage.Key(req.Key).Hex())
 	d.receiveC <- req
 	return nil
 }
 
 func (d *Delivery) processReceivedChunks() {
-R:
+	done := make(chan struct{})
+	timer := time.NewTimer(2 * time.Second)
+	defer timer.Stop()
+	// R:
 	for req := range d.receiveC {
-		// this should be has locally
+		log.Error("pop from receiveC", "hash", storage.Key(req.Key).Hex())
-		chunk, err := d.db.Get(req.Key)
+		timer.Reset(1 * time.Second)
-		log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
+		go func(req *ChunkDeliveryMsg) {
-		if err == nil {
+			defer func() { done <- struct{}{} }()
-			log.Error("found existing?", "hash", chunk.Key.Hex())
+			// this should be has locally
-			continue R
+			chunk, err := d.db.Get(req.Key)
-		}
+			log.Error("after db.Get", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
-		if err != storage.ErrFetching {
+			if err == nil {
-			panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk))
+				log.Error("found existing?", "hash", chunk.Key.Hex())
-		}
+				// continue R
-		select {
+				return
-		case <-chunk.ReqC:
+			}
-			log.Error("someone else delivered?", "hash", chunk.Key.Hex())
+			if err != storage.ErrFetching {
-			continue R
+				panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk))
-		default:
+			}
-		}
+			select {
-		go func() {
+			case <-chunk.ReqC:
+				log.Error("someone else delivered?", "hash", chunk.Key.Hex())
+				// continue R
+				return
+			default:
+			}
+			// go func() {
 			chunk.SData = req.SData
 			log.Error("received delivery", "hash", chunk.Key.Hex())
 			d.db.Put(chunk)
@@ -201,7 +220,13 @@ R:
 			//log.Warn("received delivery stored", "hash", chunk.Key)
 			log.Error("requesters notified", "hash", chunk.Key.Hex())
 			d.counterDone++
-		}()
+			// }()
+		}(req)
+		select {
+		case <-timer.C:
+			log.Error("!!!unable to process", "hash", req.Key.Hex())
+		case <-done:
+		}
 	}
 }
 

swarm/network/stream/delivery_test.go

@@ -378,22 +378,23 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 
 		// 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
-		var j int
+		for j := 0; j < nodes-1; j++ {
-		err := sim.CallClient(func(client *rpc.Client) error {
+			id := sim.IDs[j]
-			err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC)
+			err := sim.CallClient(id, func(client *rpc.Client) error {
+				err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC)
+				if err != nil {
+					return 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)
+			})
 			if err != nil {
 				return 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
 		delivery := deliveries[sim.IDs[0]]
 		retrieveFunc := func(chunk *storage.Chunk) error {
@@ -426,22 +427,21 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		default:
 		}
 		var total int64
-		err := sim.CallClient(func(client *rpc.Client) error {
+		err := sim.CallClient(id, func(client *rpc.Client) error {
 			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 			defer cancel()
 			return client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash))
-		}, 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
 	}
 
 	conf.Step = &simulations.Step{
 		Action:  action,
-		Trigger: streamTesting.PivotTrigger(10*time.Millisecond, checkC, sim.IDs[0]),
+		Trigger: streamTesting.Trigger(10*time.Millisecond, quitC, sim.IDs[0]),
 		// we are only testing the pivot node (net.Nodes[0])
 		Expect: &simulations.Expectation{
 			Nodes: sim.IDs[0:1],
@@ -490,7 +490,12 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) {
 }
 
 func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) {
+	defaultSkipCheck = skipCheck
 	toAddr = network.NewAddrFromNodeID
+	timeout := 300 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
 	conf := &streamTesting.RunConfig{
 		Adapter:   *adapter,
 		NodeCount: nodes,
@@ -498,12 +503,12 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 		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 {
@@ -515,17 +520,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 		}
 		return 2
 	}
+	// wait channel for all nodes all peer connections to set up
+	waitPeerErrC = make(chan error)
+
 	// 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{})
+	defer close(quitC)
 
 	action := func(ctx context.Context) error {
 		// each node Subscribes to each other's swarmChunkServerStreamName
@@ -545,18 +552,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 
 		// each node except the last one subscribes to the upstream swarm chunk server stream
 		// which responds to chunk retrieve requests
-		var j int
+		for j := 0; j < nodes-1; j++ {
-		simErrC <- sim.CallClient(func(client *rpc.Client) error {
+			id := sim.IDs[j]
-			err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), simErrC, quitC)
+			simErrC <- sim.CallClient(id, func(client *rpc.Client) error {
-			if err != nil {
+				err := streamTesting.WatchDisconnections(id, client, peerCount(id), simErrC, quitC)
-				return err
+				if err != nil {
-			}
+					return err
-			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+				}
-			defer cancel()
+				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
-			j++
+				defer cancel()
-			sid := sim.IDs[j] // the upstream peer's id
+				sid := sim.IDs[j+1] // the upstream peer's id
-			return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+				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
 	}
@@ -589,10 +597,6 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 	// run the simulation in the background
 	errc := make(chan error)
 	go func() {
-		timeout := 300 * time.Second
-		ctx, cancel := context.WithTimeout(context.Background(), timeout)
-		defer cancel()
-
 		_, err := sim.Run(ctx, conf)
 		errc <- err
 	}()
@@ -608,6 +612,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 			select {
 			case err = <-simErrC:
 			case <-quitC:
+				return
 			}
 			trigger <- sim.IDs[0]
 			checkC <- err
@@ -669,7 +674,6 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 		}
 	}
 	// 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

@@ -179,8 +179,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 				p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err))
 				return
 			}
-		case <-s.quit:
-			return
 		}
 		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
 		err := p.SendPriority(msg, s.priority)

swarm/network/stream/peer.go

@@ -83,6 +83,10 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 	if err != nil {
 		return err
 	}
+	// true only when quiting
+	if len(hashes) == 0 {
+		return nil
+	}
 	if proof == nil {
 		proof = &HandoverProof{
 			Handover: &Handover{},
@@ -171,3 +175,9 @@ func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bo
 	next <- nil // this is to allow wantedKeysMsg before first batch arrives
 	return nil
 }
+
+func (p *Peer) close() {
+	for _, s := range p.servers {
+		s.Close()
+	}
+}

swarm/network/stream/stream.go

@@ -198,7 +198,8 @@ func (r *Registry) run(p *protocols.Peer) error {
 	sp := NewPeer(p, r)
 	r.setPeer(sp)
 	defer r.deletePeer(sp)
-	defer close(sp.quit)
+	// defer close(sp.quit
+	defer sp.close()
 	return sp.Run(sp.HandleMsg)
 }
 
@@ -257,6 +258,7 @@ type server struct {
 type Server interface {
 	SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
 	GetData([]byte) ([]byte, error)
+	Close()
 }
 
 type client struct {
@@ -266,8 +268,8 @@ type client struct {
 	live      bool
 	stream    string
 	key       []byte
-	quit      chan struct{}
+	// quit      chan struct{}
-	next      chan error
+	next chan error
 }
 
 // Client interface for incoming peer Streamer

swarm/network/stream/streamer_test.go

@@ -85,6 +85,9 @@ func (self *testServer) GetData([]byte) ([]byte, error) {
 	return nil, nil
 }
 
+func (self *testServer) Close() {
+}
+
 func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {
 	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()

swarm/network/stream/syncer.go

@@ -42,6 +42,7 @@ type SwarmSyncerServer struct {
 	db        *storage.DBAPI
 	sessionAt uint64
 	start     uint64
+	quit      chan struct{}
 }
 
 // NewSwarmSyncerServer is contructor for SwarmSyncerServer
@@ -56,6 +57,7 @@ func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerS
 		db:        db,
 		sessionAt: sessionAt,
 		start:     start,
+		quit:      make(chan struct{}),
 	}, nil
 }
 
@@ -72,6 +74,11 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) {
 	// })
 }
 
+// Close needs to be called on a stream server
+func (s *SwarmSyncerServer) Close() {
+	close(s.quit)
+}
+
 // GetSection retrieves the actual chunk from localstore
 func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) {
 	chunk, err := s.db.Get(storage.Key(key))
@@ -95,7 +102,12 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 	}
 	ticker := time.NewTicker(10 * time.Millisecond)
 	defer ticker.Stop()
-	for range ticker.C {
+	for {
+		select {
+		case <-ticker.C:
+		case <-s.quit:
+			return nil, 0, 0, nil, nil
+		}
 		err := s.db.Iterator(from, to, s.po, func(key storage.Key, idx uint64) bool {
 			batch = append(batch, key[:]...)
 			i++

swarm/network/stream/syncer_test.go

@@ -44,6 +44,7 @@ func TestSyncerSimulation(t *testing.T) {
 	// testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1)
 	// // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1)
 	testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1)
+	// testSyncBetweenNodes(t, 32, 1, dataChunkCount, true, 1)
 	// testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1)
 }
 
@@ -61,34 +62,49 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		ToAddr:    toAddr,
 		Services:  services,
 	}
+	// create context for simulation run
+	timeout := 30 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	// defer cancel should come before defer simulation teardown
+	defer cancel()
 
+	// create simulation network with the config
 	sim, teardown, err := streamTesting.NewSimulation(conf)
 	defer teardown()
 	if err != nil {
 		t.Fatal(err.Error())
 	}
 
+	// DEBUG:
 	defer func() {
 		for _, id := range sim.IDs {
 			deliveries[id].PrintCounters(id)
 		}
-		// for id, delivery := range deliveries {
-		// 	delivery.PrintCounters(id)
-		// }
 	}()
 
+	// HACK: these are global variables in the test so that they are available for
+	// the service constructor function
+	// TODO: will this work with exec/docker adapter?
+	// localstore of nodes made available for action and check calls
 	stores = make(map[discover.NodeID]storage.ChunkStore)
-	deliveries = make(map[discover.NodeID]*Delivery)
+	nodeIndex := make(map[discover.NodeID]int)
 	for i, id := range sim.IDs {
+		nodeIndex[id] = i
 		stores[id] = sim.Stores[i]
 	}
+	deliveries = make(map[discover.NodeID]*Delivery)
+	// peerCount function gives the number of peer connections for a nodeID
+	// this is needed for the service run function to wait until
+	// each protocol  instance runs and the streamer peers are available
 	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
+	waitPeerErrC = make(chan error)
+
+	// here we distribute chunks of a random file into stores 1...nodes
 	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
 	rrdpa.Start()
 	size := chunkCount * chunkSize
@@ -100,12 +116,14 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		t.Fatal(err.Error())
 	}
 
-	// collect hashes in po 1 from all nodes
+	// create DBAPI-s for all nodes
-	hashes := make([][]storage.Key, nodes)
 	dbs := make([]*storage.DBAPI, nodes)
 	for i := 0; i < nodes; i++ {
 		dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore))
 	}
+
+	// collect hashes in po 1 bin for each node
+	hashes := make([][]storage.Key, nodes)
 	totalHashes := 0
 	hashCounts := make([]int, nodes)
 	for i := nodes - 1; i >= 0; i-- {
@@ -120,9 +138,12 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		})
 	}
 
+	// errc is error channel for simulation
 	errc := make(chan error, 1)
-	waitPeerErrC = make(chan error)
 	quitC := make(chan struct{})
+	defer close(quitC)
+
+	// action is subscribe
 	action := func(ctx context.Context) error {
 		// need to wait till an aynchronous process registers the peers in streamer.peers
 		// that is used by Subscribe
@@ -139,24 +160,29 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			}
 		}
 		// each node Subscribes to each other's swarmChunkServerStreamName
-		j := 0
+		for j := 0; j < nodes-1; j++ {
-		return sim.CallClient(func(client *rpc.Client) error {
+			id := sim.IDs[j]
-			err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC)
+			err := sim.CallClient(id, func(client *rpc.Client) error {
+				// report disconnect events to the error channel cos peers should not disconnect
+				err := streamTesting.WatchDisconnections(id, client, peerCount(id), errc, quitC)
+				if err != nil {
+					return err
+				}
+				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+				defer cancel()
+				// start syncing, i.e., subscribe to upstream peers po 1 bin
+				sid := sim.IDs[j+1]
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{1}, 0, 0, Top, false)
+			})
 			if err != nil {
 				return err
 			}
-			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+		}
-			defer cancel()
+		return nil
-			j++
-			return client.CallContext(ctx, nil, "stream_subscribeStream", sim.IDs[j], "SYNC", []byte{1}, 0, 0, Top, false)
-		}, sim.IDs[0:nodes-1]...)
 	}
 
 	// 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) {
-		defer func() { checkC <- struct{}{} }()
-
 		select {
 		case err := <-errc:
 			return false, err
@@ -165,54 +191,37 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		default:
 		}
 
-		var pass bool
+		log.Error("starting dbs check", "node", id)
-		var i int
+		i := nodeIndex[id]
-		log.Error("staring dbs check")
+		var total, found int
-		for i = nodes - 1; i >= 0; i-- {
+		for j := i; j < nodes; j++ {
-			nodeHashCount := hashCounts[i]
+			total += len(hashes[j])
-			nodeHashFound := 0
+			for _, key := range hashes[j] {
-			for j := i; j < nodes; j++ {
+				chunk, err := dbs[i].Get(key)
-				nodeHashes := hashes[j]
+				if err == storage.ErrFetching {
-				for _, key := range nodeHashes {
+					<-chunk.ReqC
-					chunk, err := dbs[i].Get(key)
+				} else if err != nil {
-					if err == storage.ErrFetching {
+					log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err)
-						<-chunk.ReqC
+					continue
-						nodeHashFound++
-					} else if err == nil {
-						nodeHashFound++
-					} else {
-						log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err)
-					}
 				}
-			}
+				// needed for leveldb not to be closed?
-			log.Error("sync check", "node", sim.IDs[i], "index", i, "bin", po, "found", nodeHashFound, "total", nodeHashCount)
+				// chunk.WaitToStore()
-			pass = nodeHashFound == nodeHashCount
+				found++
-			if !pass {
-				break
 			}
 		}
-		// log.Error("sync check", "bin", po, "found", found, "total", totalHashes)
+		log.Error("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total)
-		// pass := found == totalHashes
+		return total == found, nil
-		if !pass {
-			return false, nil
-		}
-		close(quitC)
-		return true, nil
-
 	}
 
 	conf.Step = &simulations.Step{
 		Action:  action,
-		Trigger: streamTesting.PivotTrigger(500*time.Millisecond, checkC, sim.IDs[0]),
+		Trigger: streamTesting.Trigger(500*time.Millisecond, quitC, sim.IDs[0:nodes-1]...),
 		Expect: &simulations.Expectation{
 			Nodes: sim.IDs[0:1],
 			Check: check,
 		},
 	}
 	startedAt := time.Now()
-	timeout := 30 * time.Second
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
 	result, err := sim.Run(ctx, conf)
 	finishedAt := time.Now()
 	if err != nil {

swarm/network/stream/testing/testing.go

@@ -234,7 +234,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCou
 	return nil
 }
 
-func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID) chan discover.NodeID {
+func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID {
 	trigger := make(chan discover.NodeID)
 	go func() {
 		ticker := time.NewTicker(d)
@@ -242,28 +242,24 @@ func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID)
 		// we are only testing the pivot node (net.Nodes[0])
 		for range ticker.C {
 			for _, id := range ids {
-				trigger <- id
+				select {
+				case trigger <- id:
+				case <-quitC:
+				}
 			}
-			<-checkC
 		}
 	}()
 	return trigger
 }
 
-func (sim *Simulation) CallClient(f func(*rpc.Client) error, ids ...discover.NodeID) error {
+func (sim *Simulation) CallClient(id discover.NodeID, f func(*rpc.Client) error) error {
-	for _, id := range ids {
+	node := sim.Net.GetNode(id)
-		node := sim.Net.GetNode(id)
+	if node == nil {
-		if node == nil {
+		return fmt.Errorf("unknown node: %s", id)
-			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
+	client, err := node.Client()
+	if err != nil {
+		return fmt.Errorf("error getting node client: %s", err)
+	}
+	return f(client)
 }

swarm/storage/dbstore.go

@@ -781,27 +781,22 @@ func (s *DbStore) Close() {
 	s.db.Close()
 }
 
-// initialises a sync iterator from a syncToken (passed in with the handshake)
+// SyncIterator(start, stop, po, f) calls f on each hash of a bin po from start to stop
 func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error {
-	// probably, the lock is not needed
+	sincekey := getDataKey(since, po)
-	// s.lock.Lock()
-	// defer s.lock.Unlock()
-
 	untilkey := getDataKey(until, po)
-
 	it := s.db.NewIterator()
-	seek := getDataKey(since, po)
-	it.Seek(seek)
 	defer it.Release()
+	it.Seek(sincekey)
 
 	for it.Next() {
 		dbkey := it.Key()
 		if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 {
 			break
 		}
-
 		key := make([]byte, 32)
-		copy(key, it.Value()[:32])
+		val := it.Value()
+		copy(key, val[:32])
 		if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) {
 			break
 		}