Merge branch 'swarm-network-rewrite-syncer-test' into swarm-network-rewrite-syncer

swarm/network/stream/common_test.go

@@ -63,11 +63,12 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	id := ctx.Config.ID
 	addr := toAddr(id)
 	kad := network.NewKademlia(addr.Over(), network.NewKadParams())
-	store := stores[id]
+	store := stores[id].(*storage.LocalStore)
-	db := storage.NewDBAPI(store.(*storage.LocalStore))
+	db := storage.NewDBAPI(store)
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
-	r := NewRegistry(addr, delivery, store, defaultSkipCheck)
+	netStore := storage.NewNetStore(store, nil)
+	r := NewRegistry(addr, delivery, netStore, defaultSkipCheck)
 	RegisterSwarmSyncerServer(r, db)
 	RegisterSwarmSyncerClient(r, db)
 	go func() {

swarm/network/stream/delivery.go

@@ -17,6 +17,7 @@
 package stream
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"time"
@@ -57,6 +58,7 @@ type SwarmChunkServer struct {
 	batchC     chan []byte
 	db         *storage.DBAPI
 	currentLen uint64
+	quit       chan struct{}
 }
 
 // NewSwarmChunkServer is SwarmChunkServer constructor
@@ -65,6 +67,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
@@ -76,6 +79,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
@@ -88,17 +93,32 @@ func (s *SwarmChunkServer) processDeliveries() {
 
 // SetNextBatch
 func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) {
-	hashes = <-s.batchC
+	select {
+	case hashes = <-s.batchC:
+	case <-s.quit:
+		return
+	}
+
 	from = s.currentLen
 	s.currentLen += uint64(len(hashes))
 	to = s.currentLen
 	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 {
+func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) {
-	chunk, _ := s.db.Get(storage.Key(key))
+	chunk, err := s.db.Get(storage.Key(key))
-	return chunk.SData
+	if err == storage.ErrFetching {
+		<-chunk.ReqC
+	} else if err != nil {
+		return nil, err
+	}
+	return chunk.SData, nil
 }
 
 // RetrieveRequestMsg is the protocol msg for chunk retrieve requests
@@ -156,9 +176,11 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e
 type ChunkDeliveryMsg struct {
 	Key   storage.Key
 	SData []byte // the stored chunk Data (incl size)
+	peer  *Peer  // set in handleChunkDeliveryMsg
 }
 
-func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
+func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error {
+	req.peer = sp
 	d.receiveC <- req
 	return nil
 }
@@ -168,6 +190,9 @@ R:
 	for req := range d.receiveC {
 		// this should be has locally
 		chunk, err := d.db.Get(req.Key)
+		if !bytes.Equal(chunk.Key, req.Key) {
+			panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID()))
+		}
 		if err == nil {
 			continue R
 		}
@@ -176,16 +201,14 @@ R:
 		}
 		select {
 		case <-chunk.ReqC:
+			log.Error("someone else delivered?", "hash", chunk.Key.Hex())
 			continue R
 		default:
 		}
 		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)
 	}
 }
 
@@ -193,18 +216,17 @@ R:
 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)
+				log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId)
 				return true
 			}
 		}
 		sp := d.getPeer(spId)
 		if sp == nil {
-			log.Warn("peer not found", "id", spId)
+			log.Warn("Delivery.RequestFromPeers: peer not found", "id", spId)
 			return true
 		}
 		// TODO: skip light nodes that do not accept retrieve requests

swarm/network/stream/delivery_test.go

@@ -378,22 +378,22 @@ 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, errc, quitC)
+			err := sim.CallClient(id, func(client *rpc.Client) error {
+				err := streamTesting.WatchDisconnections(id, client, errc, quitC)
 				if err != nil {
 					return err
 				}
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
-			j++
+				sid := sim.IDs[j+1]
-			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 {
@@ -415,9 +415,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		}()
 		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
@@ -426,22 +424,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],
@@ -449,7 +446,10 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		},
 	}
 	startedAt := time.Now()
-	result, err := sim.Run(conf)
+	timeout := 300 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	result, err := sim.Run(ctx, conf)
 	finishedAt := time.Now()
 	if err != nil {
 		t.Fatalf("Setting up simulation failed: %v", err)
@@ -487,7 +487,13 @@ 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,
@@ -495,12 +501,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 {
@@ -512,18 +518,21 @@ 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)
+	disconnectC := make(chan error)
 	quitC := make(chan struct{})
 
+	initC := make(chan 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
@@ -539,37 +548,32 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 				break
 			}
 		}
-
+		var err error
 		// 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, simErrC, quitC)
+			err = sim.CallClient(id, func(client *rpc.Client) error {
+				err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
 				if err != nil {
 					return err
 				}
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
-			j++
+				sid := sim.IDs[j+1] // the upstream peer's id
-			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
+			if err != nil {
-		return err
+				break
+			}
+		}
+		initC <- err
+		return nil
 	}
 
 	// 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
 	}
 
@@ -586,26 +590,16 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 	// run the simulation in the background
 	errc := make(chan error)
 	go func() {
-		_, err := sim.Run(conf)
+		_, err := sim.Run(ctx, conf)
+		close(quitC)
 		errc <- err
 	}()
 
 	// wait for simulation action to complete stream subscriptions
-	err = <-simErrC
+	err = <-initC
 	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
@@ -618,6 +612,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 	// benchmark loop
 	b.ResetTimer()
 	b.StopTimer()
+Loop:
 	for i := 0; i < b.N; i++ {
 		// uploading chunkCount random chunks to the last node
 		hashes := make([]storage.Key, chunkCount)
@@ -657,13 +652,34 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 			}
 		}
 		b.StopTimer()
+
+		select {
+		case err = <-disconnectC:
+			if err != nil {
+				break Loop
+			}
+		default:
+		}
+
 		if misses > 0 {
-			simErrC <- fmt.Errorf("%v chunk not found out of %v", misses, total)
+			err = fmt.Errorf("%v chunk not found out of %v", misses, total)
+			break Loop
 		}
 	}
-	// benchmark over, trigger the check function to conclude the simulation
+
-	close(quitC)
+	select {
+	case <-quitC:
+	case trigger <- sim.IDs[0]:
+	}
+	if err == nil {
 		err = <-errc
+	} else {
+		if e := <-errc; e != nil {
+			b.Errorf("sim.Run function error: %v", e)
+		}
+	}
+
+	// benchmark over, trigger the check function to conclude the simulation
 	if err != nil {
 		b.Fatalf("expected no error. got %v", err)
 	}

swarm/network/stream/messages.go

@@ -17,7 +17,6 @@
 package stream
 
 import (
-	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -121,6 +120,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	wg := sync.WaitGroup{}
 	for i := 0; i < len(hashes); i += HashSize {
 		hash := hashes[i : i+HashSize]
+
 		if wait := s.NeedData(hash); wait != nil {
 			want.Set(i/HashSize, true)
 			wg.Add(1)
@@ -168,16 +168,14 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	}
 	go func() {
 		select {
-		case <-time.After(1 * time.Second):
+		case <-time.After(30 * time.Second):
-			p.Drop(errors.New("timeout waiting for batch to be delivered"))
+			p.Drop(err)
 			return
 		case err := <-s.next:
 			if err != nil {
 				p.Drop(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)
@@ -227,9 +225,9 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
 	for i := 0; i < l; i++ {
 		if want.Get(i) {
 			hash := hashes[i*HashSize : (i+1)*HashSize]
-			data := s.GetData(hash)
+			data, err := s.GetData(hash)
-			if data == nil {
+			if err != nil {
-				return errors.New("not found")
+				return fmt.Errorf("handleWantedHashesMsg get data %x: %v", hash, err)
 			}
 			chunk := storage.NewChunk(hash, nil)
 			chunk.SData = data

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{},
@@ -97,7 +101,7 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 		Stream:        s.stream,
 		Key:           s.key,
 	}
-	log.Warn("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
+	log.Trace("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)
 }
 
@@ -167,3 +171,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

@@ -199,6 +199,7 @@ func (r *Registry) run(p *protocols.Peer) error {
 	r.setPeer(sp)
 	defer r.deletePeer(sp)
 	defer close(sp.quit)
+	defer sp.close()
 	return sp.Run(sp.HandleMsg)
 }
 
@@ -227,7 +228,7 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 		return p.handleWantedHashesMsg(msg)
 
 	case *ChunkDeliveryMsg:
-		return p.streamer.delivery.handleChunkDeliveryMsg(msg)
+		return p.streamer.delivery.handleChunkDeliveryMsg(p, msg)
 
 	case *RetrieveRequestMsg:
 		return p.streamer.delivery.handleRetrieveRequestMsg(p, msg)
@@ -256,7 +257,8 @@ type server struct {
 // Server interface for outgoing peer Streamer
 type Server interface {
 	SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
-	GetData([]byte) []byte
+	GetData([]byte) ([]byte, error)
+	Close()
 }
 
 type client struct {
@@ -266,7 +268,6 @@ type client struct {
 	live      bool
 	stream    string
 	key       []byte
-	quit      chan struct{}
 	next      chan error
 }
 

swarm/network/stream/streamer_test.go

@@ -81,8 +81,11 @@ func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, ui
 	return make([]byte, HashSize), from + 1, to + 1, nil, nil
 }
 
-func (self *testServer) GetData([]byte) []byte {
+func (self *testServer) GetData([]byte) ([]byte, error) {
-	return nil
+	return nil, nil
+}
+
+func (self *testServer) Close() {
 }
 
 func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {

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,13 +74,20 @@ 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 {
+func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) {
 	chunk, err := s.db.Get(storage.Key(key))
-	if err != nil {
+	if err == storage.ErrFetching {
-		return nil
+		<-chunk.ReqC
+	} else if err != nil {
+		return nil, err
 	}
-	return chunk.SData
+	return chunk.SData, nil
 }
 
 // GetBatch retrieves the next batch of hashes from the dbstore
@@ -93,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++
@@ -109,7 +123,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 	}
 
 	log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po))
-	return batch, from, to + 1, nil, nil
+	return batch, from, to, nil, nil
 }
 
 // SwarmSyncerClient
@@ -184,7 +198,6 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
 	chunk, _ := s.db.GetOrCreateRequest(key)
 	// TODO: we may want to request from this peer anyway even if the request exists
 	if chunk.ReqC == nil {
-		log.Error("oops this is found")
 		return nil
 	}
 	// create request and wait until the chunk data arrives and is stored

swarm/network/stream/syncer_test.go

@@ -37,14 +37,10 @@ import (
 const dataChunkCount = 500
 
 func TestSyncerSimulation(t *testing.T) {
-	// testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1)
+	testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1)
-	// testSyncBetweenNodes(t, 2, 1, dataChunkCount, false, 1)
+	testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1)
-	// testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1)
+	testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 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, chunkCount int, skipCheck bool, po uint8) {
@@ -61,24 +57,42 @@ 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())
 	}
+
+	// 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
@@ -90,22 +104,34 @@ 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
-	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++ {
+
+	// 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-- {
+		if i < nodes-1 {
+			hashCounts[i] = hashCounts[i+1]
+		}
 		dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
-			hashes = append(hashes, key)
+			hashes[i] = append(hashes[i], key)
+			totalHashes++
+			hashCounts[i]++
 			return true
 		})
 	}
 
+	// 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
@@ -122,24 +148,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, 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, errc, quitC)
 				if err != nil {
 					return err
 				}
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
-			j++
+				// start syncing, i.e., subscribe to upstream peers po 1 bin
-			return client.CallContext(ctx, nil, "stream_subscribeStream", sim.IDs[j], "SYNC", []byte{1}, 0, 0, Top, false)
+				sid := sim.IDs[j+1]
-		}, sim.IDs[0:nodes-1]...)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{1}, 0, 0, Top, false)
+			})
+			if err != nil {
+				return err
+			}
+		}
+		return nil
 	}
 
 	// 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
@@ -148,34 +179,36 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		default:
 		}
 
-		var found int
+		i := nodeIndex[id]
-		total := len(hashes)
+		var total, found int
-		for _, key := range hashes {
+		for j := i; j < nodes; j++ {
-			_, err := dbs[0].Get(key)
+			total += len(hashes[j])
-			if err == nil {
+			for _, key := range hashes[j] {
+				chunk, err := dbs[i].Get(key)
+				if err == storage.ErrFetching {
+					<-chunk.ReqC
+				} else if err != nil {
+					continue
+				}
+				// needed for leveldb not to be closed?
+				// chunk.WaitToStore()
 				found++
 			}
 		}
-		log.Error("sync check", "bin", po, "found", found, "total", total)
+		log.Debug("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total)
-		pass := found == total
+		return total == found, nil
-		if !pass {
-			return false, nil
-		}
-		close(quitC)
-		return true, nil
-
 	}
 
 	conf.Step = &simulations.Step{
 		Action:  action,
-		Trigger: streamTesting.PivotTrigger(100*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()
-	result, err := sim.Run(conf)
+	result, err := sim.Run(ctx, conf)
 	finishedAt := time.Now()
 	if err != nil {
 		t.Fatalf("Setting up simulation failed: %v", err)

swarm/network/stream/testing/testing.go

@@ -154,9 +154,9 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	// set nodes number of Stores available
 	stores, storeTeardown, err := SetStores(addrs...)
 	teardown = func() {
-		storeTeardown()
-		adapterTeardown()
 		net.Shutdown()
+		adapterTeardown()
+		storeTeardown()
 	}
 	if err != nil {
 		return nil, teardown, err
@@ -170,7 +170,7 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	return s, teardown, nil
 }
 
-func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) {
+func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.StepResult, error) {
 	// bring up nodes, launch the servive
 	nodes := conf.NodeCount
 	conns := conf.ConnLevel
@@ -204,9 +204,6 @@ func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) {
 
 	// create an only locally retrieving dpa for the pivot node to test
 	// if retriee requests have arrived
-	timeout := 300 * time.Second
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
 	result := simulations.NewSimulation(s.Net).Run(ctx, conf.Step)
 	return result, nil
 }
@@ -218,7 +215,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error
 		return fmt.Errorf("error getting peer events for node %v: %s", id, err)
 	}
 	go func() {
-		defer sub.Unsubscribe()
+		for {
 			select {
 			case <-quitC:
 				return
@@ -229,28 +226,32 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error
 					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 {
+func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID {
 	trigger := make(chan discover.NodeID)
 	go func() {
+		defer close(trigger)
 		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
+				select {
+				case trigger <- id:
+				case <-quitC:
+					return
+				}
 			}
-			<-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)
 	if node == nil {
 		return fmt.Errorf("unknown node: %s", id)
@@ -259,10 +260,5 @@ func (sim *Simulation) CallClient(f func(*rpc.Client) error, ids ...discover.Nod
 	if err != nil {
 		return fmt.Errorf("error getting node client: %s", err)
 	}
-		err = f(client)
+	return f(client)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
 }

swarm/storage/dbstore.go

@@ -81,7 +81,6 @@ type DbStore struct {
 	po       func(Key) uint8
 
 	batchC   chan bool
-	quit     chan struct{}
 	batchesC chan struct{}
 	batch    *leveldb.Batch
 	lock     sync.RWMutex
@@ -105,7 +104,6 @@ func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uin
 	s.hashfunc = hash
 
 	s.batchC = make(chan bool)
-	s.quit = make(chan struct{})
 	s.batchesC = make(chan struct{}, 1)
 	go s.writeBatches()
 	s.batch = new(leveldb.Batch)
@@ -231,6 +229,7 @@ func encodeData(chunk *Chunk) []byte {
 func decodeIndex(data []byte, index *dpaDBIndex) error {
 	dec := rlp.NewStream(bytes.NewReader(data), 0)
 	return dec.Decode(index)
+
 }
 
 func decodeData(data []byte, chunk *Chunk) {
@@ -538,17 +537,21 @@ func (s *DbStore) CurrentStorageIndex() uint64 {
 }
 
 func (s *DbStore) Put(chunk *Chunk) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-
 	ikey := getIndexKey(chunk.Key)
 	var index dpaDBIndex
 
 	po := s.po(chunk.Key)
+	s.lock.Lock()
+	defer s.lock.Unlock()
 
 	idata, err := s.db.Get(ikey)
 	if err != nil {
 		s.doPut(chunk, ikey, &index, po)
+		batchC := s.batchC
+		go func() {
+			<-batchC
+			close(chunk.dbStored)
+		}()
 	} else {
 		log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access"))
 		decodeIndex(idata, &index)
@@ -559,7 +562,6 @@ func (s *DbStore) Put(chunk *Chunk) {
 	idata = encodeIndex(&index)
 	s.batch.Put(ikey, idata)
 	select {
-	case <-s.quit:
 	case s.batchesC <- struct{}{}:
 	default:
 	}
@@ -579,13 +581,6 @@ func (s *DbStore) doPut(chunk *Chunk, ikey []byte, index *dpaDBIndex, po uint8)
 	cntKey[1] = po
 	s.batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
 
-	batchC := s.batchC
-	go func() {
-		<-batchC
-		close(chunk.dbStored)
-	}()
-
-	log.Trace(fmt.Sprintf("DbStore.Put: %v. db storage counter: %v ", chunk.Key.Log(), s.dataIdx))
 }
 
 func (s *DbStore) writeBatches() {
@@ -691,15 +686,14 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) {
 			hash := hasher.Sum(nil)
 
 			if !bytes.Equal(hash, key) {
-				log.Trace(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
+				log.Error(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
 				s.delete(indx.Idx, getIndexKey(key), s.po(key))
-				log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
+				log.Error("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
 			}
 		}
 
 		chunk = NewChunk(key, nil)
 		decodeData(data, chunk)
-
 	} else {
 		err = ErrNotFound
 	}
@@ -757,30 +751,26 @@ 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 {
-	s.lock.Lock()
+	sincekey := getDataKey(since, po)
-	defer s.lock.Unlock()
 	untilkey := getDataKey(until, po)
-
 	it := s.db.NewIterator()
-	seek := getDataKey(since, po)
-	it.Seek(seek)
 	defer it.Release()
-	for it.Valid() {
+
+	for ok := it.Seek(sincekey); ok; ok = 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
 		}
-		it.Next()
 	}
-	return nil
+	return it.Error()
 }
 
 func databaseExists(path string) bool {

swarm/storage/localstore.go

@@ -94,10 +94,15 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error)
 // LocalStore is itself a chunk store
 // unsafe, in that the data is not integrity checked
 func (self *LocalStore) Put(chunk *Chunk) {
-	self.memStore.Put(chunk)
+	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
-	go func() {
+	c := &Chunk{
-		self.DbStore.Put(chunk)
+		Key:      Key(append([]byte{}, chunk.Key...)),
-	}()
+		SData:    append([]byte{}, chunk.SData...),
+		Size:     chunk.Size,
+		dbStored: chunk.dbStored,
+	}
+	self.memStore.Put(c)
+	self.DbStore.Put(c)
 }
 
 // Get(chunk *Chunk) looks up a chunk in the local stores
@@ -106,7 +111,6 @@ func (self *LocalStore) Put(chunk *Chunk) {
 // ChunkStores are remote and can have long latency
 func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) {
 	chunk, err = self.memStore.Get(key)
-
 	if err == nil {
 		if chunk.ReqC != nil {
 			select {

swarm/storage/memstore.go

@@ -19,7 +19,10 @@
 package storage
 
 import (
+	"fmt"
 	"sync"
+
+	"github.com/ethereum/go-ethereum/log"
 )
 
 const (
@@ -29,318 +32,321 @@ const (
 	defaultCacheCapacity   = 5000
 )
 
-// type MemStore struct {
-// 	memtree            *memTree
-// 	entryCnt, capacity uint   // stored entries
-// 	accessCnt          uint64 // access counter; oldest is thrown away when full
-// 	dbAccessCnt        uint64
-// 	dbStore            *DbStore
-// 	lock               sync.Mutex
-// }
-//
-// /*
-// a hash prefix subtree containing subtrees or one storage entry (but never both)
-//
-// - access[0] stores the smallest (oldest) access count value in this subtree
-// - if it contains more subtrees and its subtree count is at least 4, access[1:2]
-//   stores the smallest access count in the first and second halves of subtrees
-//   (so that access[0] = min(access[1], access[2])
-// - likewise, if subtree count is at least 8,
-//   access[1] = min(access[3], access[4])
-//   access[2] = min(access[5], access[6])
-//   (access[] is a binary tree inside the multi-bit leveled hash tree)
-// */
-//
-// func NewMemStore(d *DbStore, capacity uint) (m *MemStore) {
-// 	m = &MemStore{}
-// 	m.memtree = newMemTree(memTreeFLW, nil, 0)
-// 	m.dbStore = d
-// 	m.setCapacity(capacity)
-// 	return
-// }
-//
-// type memTree struct {
-// 	subtree   []*memTree
-// 	parent    *memTree
-// 	parentIdx uint
-//
-// 	bits  uint // log2(subtree count)
-// 	width uint // subtree count
-//
-// 	entry        *Chunk // if subtrees are present, entry should be nil
-// 	lastDBaccess uint64
-// 	access       []uint64
-// }
-//
-// func newMemTree(b uint, parent *memTree, pidx uint) (node *memTree) {
-// 	node = new(memTree)
-// 	node.bits = b
-// 	node.width = 1 << b
-// 	node.subtree = make([]*memTree, node.width)
-// 	node.access = make([]uint64, node.width-1)
-// 	node.parent = parent
-// 	node.parentIdx = pidx
-// 	if parent != nil {
-// 		parent.subtree[pidx] = node
-// 	}
-//
-// 	return node
-// }
-//
-// func (node *memTree) updateAccess(a uint64) {
-// 	aidx := uint(0)
-// 	var aa uint64
-// 	oa := node.access[0]
-// 	for node.access[aidx] == oa {
-// 		node.access[aidx] = a
-// 		if aidx > 0 {
-// 			aa = node.access[((aidx-1)^1)+1]
-// 			aidx = (aidx - 1) >> 1
-// 		} else {
-// 			pidx := node.parentIdx
-// 			node = node.parent
-// 			if node == nil {
-// 				return
-// 			}
-// 			nn := node.subtree[pidx^1]
-// 			if nn != nil {
-// 				aa = nn.access[0]
-// 			} else {
-// 				aa = 0
-// 			}
-// 			aidx = (node.width + pidx - 2) >> 1
-// 		}
-//
-// 		if (aa != 0) && (aa < a) {
-// 			a = aa
-// 		}
-// 	}
-// }
-//
-// func (s *MemStore) setCapacity(c uint) {
-// 	s.lock.Lock()
-// 	defer s.lock.Unlock()
-//
-// 	for c < s.entryCnt {
-// 		s.removeOldest()
-// 	}
-// 	s.capacity = c
-// }
-//
-// // entry (not its copy) is going to be in MemStore
-// func (s *MemStore) Put(entry *Chunk) {
-// 	if s.capacity == 0 {
-// 		return
-// 	}
-//
-// 	s.lock.Lock()
-// 	defer s.lock.Unlock()
-//
-// 	if s.entryCnt >= s.capacity {
-// 		s.removeOldest()
-// 	}
-//
-// 	s.accessCnt++
-//
-// 	node := s.memtree
-// 	bitpos := uint(0)
-// 	for node.entry == nil {
-// 		l := entry.Key.bits(bitpos, node.bits)
-// 		st := node.subtree[l]
-// 		if st == nil {
-// 			st = newMemTree(memTreeLW, node, l)
-// 			bitpos += node.bits
-// 			node = st
-// 			break
-// 		}
-// 		bitpos += node.bits
-// 		node = st
-// 	}
-//
-// 	if node.entry != nil {
-//
-// 		if node.entry.Key.isEqual(entry.Key) {
-// 			node.updateAccess(s.accessCnt)
-// 			if entry.SData == nil {
-// 				entry.Size = node.entry.Size
-// 				entry.SData = node.entry.SData
-// 			}
-// 			if entry.ReqC == nil {
-// 				entry.ReqC = node.entry.ReqC
-// 			}
-// 			entry.C = node.entry.C
-// 			node.entry = entry
-// 			return
-// 		}
-//
-// 		for node.entry != nil {
-//
-// 			l := node.entry.Key.bits(bitpos, node.bits)
-// 			st := node.subtree[l]
-// 			if st == nil {
-// 				st = newMemTree(memTreeLW, node, l)
-// 			}
-// 			st.entry = node.entry
-// 			node.entry = nil
-// 			st.updateAccess(node.access[0])
-//
-// 			l = entry.Key.bits(bitpos, node.bits)
-// 			st = node.subtree[l]
-// 			if st == nil {
-// 				st = newMemTree(memTreeLW, node, l)
-// 			}
-// 			bitpos += node.bits
-// 			node = st
-//
-// 		}
-// 	}
-//
-// 	node.entry = entry
-// 	node.lastDBaccess = s.dbAccessCnt
-// 	node.updateAccess(s.accessCnt)
-// 	s.entryCnt++
-// }
-//
-// func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
-// 	s.lock.Lock()
-// 	defer s.lock.Unlock()
-//
-// 	node := s.memtree
-// 	bitpos := uint(0)
-// 	for node.entry == nil {
-// 		l := hash.bits(bitpos, node.bits)
-// 		st := node.subtree[l]
-// 		if st == nil {
-// 			return nil, ErrNotFound
-// 		}
-// 		bitpos += node.bits
-// 		node = st
-// 	}
-//
-// 	if node.entry.Key.isEqual(hash) {
-// 		s.accessCnt++
-// 		node.updateAccess(s.accessCnt)
-// 		chunk = node.entry
-// 		if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
-// 			s.dbAccessCnt++
-// 			node.lastDBaccess = s.dbAccessCnt
-// 			if s.dbStore != nil {
-// 				s.dbStore.updateAccessCnt(hash)
-// 			}
-// 		}
-// 	} else {
-// 		err = ErrNotFound
-// 	}
-//
-// 	return
-// }
-//
-// func (s *MemStore) removeOldest() {
-// 	node := s.memtree
-// 	log.Warn("purge memstore")
-// 	for node.entry == nil {
-//
-// 		aidx := uint(0)
-// 		av := node.access[aidx]
-//
-// 		for aidx < node.width/2-1 {
-// 			if av == node.access[aidx*2+1] {
-// 				node.access[aidx] = node.access[aidx*2+2]
-// 				aidx = aidx*2 + 1
-// 			} else if av == node.access[aidx*2+2] {
-// 				node.access[aidx] = node.access[aidx*2+1]
-// 				aidx = aidx*2 + 2
-// 			} else {
-// 				panic(nil)
-// 			}
-// 		}
-// 		pidx := aidx*2 + 2 - node.width
-// 		if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) {
-// 			if node.subtree[pidx+1] != nil {
-// 				node.access[aidx] = node.subtree[pidx+1].access[0]
-// 			} else {
-// 				node.access[aidx] = 0
-// 			}
-// 		} else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) {
-// 			if node.subtree[pidx] != nil {
-// 				node.access[aidx] = node.subtree[pidx].access[0]
-// 			} else {
-// 				node.access[aidx] = 0
-// 			}
-// 			pidx++
-// 		} else {
-// 			panic(nil)
-// 		}
-//
-// 		//fmt.Println(pidx)
-// 		node = node.subtree[pidx]
-//
-// 	}
-//
-// 	log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log()))
-// 	<-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.ReqC == nil {
-// 		node.entry = nil
-// 		s.entryCnt--
-// 	} else {
-// 		return
-// 	}
-//
-// 	node.access[0] = 0
-//
-// 	//---
-//
-// 	aidx := uint(0)
-// 	for {
-// 		aa := node.access[aidx]
-// 		if aidx > 0 {
-// 			aidx = (aidx - 1) >> 1
-// 		} else {
-// 			pidx := node.parentIdx
-// 			node = node.parent
-// 			if node == nil {
-// 				return
-// 			}
-// 			aidx = (node.width + pidx - 2) >> 1
-// 		}
-// 		if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) {
-// 			node.access[aidx] = aa
-// 		}
-// 	}
-// }
-
 type MemStore struct {
-	m  map[string]*Chunk
+	memtree            *memTree
-	mu sync.RWMutex
+	entryCnt, capacity uint   // stored entries
+	accessCnt          uint64 // access counter; oldest is thrown away when full
+	dbAccessCnt        uint64
+	dbStore            *DbStore
+	lock               sync.Mutex
 }
 
+/*
+a hash prefix subtree containing subtrees or one storage entry (but never both)
+
+- access[0] stores the smallest (oldest) access count value in this subtree
+- if it contains more subtrees and its subtree count is at least 4, access[1:2]
+  stores the smallest access count in the first and second halves of subtrees
+  (so that access[0] = min(access[1], access[2])
+- likewise, if subtree count is at least 8,
+  access[1] = min(access[3], access[4])
+  access[2] = min(access[5], access[6])
+  (access[] is a binary tree inside the multi-bit leveled hash tree)
+*/
+
 func NewMemStore(d *DbStore, capacity uint) (m *MemStore) {
-	return &MemStore{
+	m = &MemStore{}
-		m: make(map[string]*Chunk),
+	m.memtree = newMemTree(memTreeFLW, nil, 0)
+	m.dbStore = d
+	m.setCapacity(capacity)
+	return
+}
+
+type memTree struct {
+	subtree   []*memTree
+	parent    *memTree
+	parentIdx uint
+
+	bits  uint // log2(subtree count)
+	width uint // subtree count
+
+	entry        *Chunk // if subtrees are present, entry should be nil
+	lastDBaccess uint64
+	access       []uint64
+}
+
+func newMemTree(b uint, parent *memTree, pidx uint) (node *memTree) {
+	node = new(memTree)
+	node.bits = b
+	node.width = 1 << b
+	node.subtree = make([]*memTree, node.width)
+	node.access = make([]uint64, node.width-1)
+	node.parent = parent
+	node.parentIdx = pidx
+	if parent != nil {
+		parent.subtree[pidx] = node
+	}
+
+	return node
+}
+
+func (node *memTree) updateAccess(a uint64) {
+	aidx := uint(0)
+	var aa uint64
+	oa := node.access[0]
+	for node.access[aidx] == oa {
+		node.access[aidx] = a
+		if aidx > 0 {
+			aa = node.access[((aidx-1)^1)+1]
+			aidx = (aidx - 1) >> 1
+		} else {
+			pidx := node.parentIdx
+			node = node.parent
+			if node == nil {
+				return
+			}
+			nn := node.subtree[pidx^1]
+			if nn != nil {
+				aa = nn.access[0]
+			} else {
+				aa = 0
+			}
+			aidx = (node.width + pidx - 2) >> 1
+		}
+
+		if (aa != 0) && (aa < a) {
+			a = aa
+		}
 	}
 }
 
-func (m *MemStore) Get(key Key) (*Chunk, error) {
+func (s *MemStore) setCapacity(c uint) {
-	m.mu.RLock()
+	s.lock.Lock()
-	defer m.mu.RUnlock()
+	defer s.lock.Unlock()
-	c, ok := m.m[string(key[:])]
+
-	if !ok {
+	for c < s.entryCnt {
+		s.removeOldest()
+	}
+	s.capacity = c
+}
+
+// entry (not its copy) is going to be in MemStore
+func (s *MemStore) Put(entry *Chunk) {
+	if s.capacity == 0 {
+		return
+	}
+
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	if s.entryCnt >= s.capacity {
+		s.removeOldest()
+	}
+
+	s.accessCnt++
+
+	node := s.memtree
+	bitpos := uint(0)
+	for node.entry == nil {
+		l := entry.Key.bits(bitpos, node.bits)
+		st := node.subtree[l]
+		if st == nil {
+			st = newMemTree(memTreeLW, node, l)
+			bitpos += node.bits
+			node = st
+			break
+		}
+		bitpos += node.bits
+		node = st
+	}
+
+	if node.entry != nil {
+
+		if node.entry.Key.isEqual(entry.Key) {
+			node.updateAccess(s.accessCnt)
+			if entry.SData == nil {
+				entry.Size = node.entry.Size
+				entry.SData = node.entry.SData
+			}
+			if entry.ReqC == nil {
+				entry.ReqC = node.entry.ReqC
+			}
+			entry.C = node.entry.C
+			node.entry = entry
+			return
+		}
+
+		for node.entry != nil {
+
+			l := node.entry.Key.bits(bitpos, node.bits)
+			st := node.subtree[l]
+			if st == nil {
+				st = newMemTree(memTreeLW, node, l)
+			}
+			st.entry = node.entry
+			node.entry = nil
+			st.updateAccess(node.access[0])
+
+			l = entry.Key.bits(bitpos, node.bits)
+			st = node.subtree[l]
+			if st == nil {
+				st = newMemTree(memTreeLW, node, l)
+			}
+			bitpos += node.bits
+			node = st
+
+		}
+	}
+
+	node.entry = entry
+	node.lastDBaccess = s.dbAccessCnt
+	node.updateAccess(s.accessCnt)
+	s.entryCnt++
+}
+
+func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	node := s.memtree
+	bitpos := uint(0)
+	for node.entry == nil {
+		l := hash.bits(bitpos, node.bits)
+		st := node.subtree[l]
+		if st == nil {
 			return nil, ErrNotFound
 		}
-	return c, nil
+		bitpos += node.bits
+		node = st
+	}
+
+	if node.entry.Key.isEqual(hash) {
+		s.accessCnt++
+		node.updateAccess(s.accessCnt)
+		chunk = node.entry
+		if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
+			s.dbAccessCnt++
+			node.lastDBaccess = s.dbAccessCnt
+			if s.dbStore != nil {
+				s.dbStore.updateAccessCnt(hash)
+			}
+		}
+	} else {
+		err = ErrNotFound
+	}
+
+	return
 }
 
-func (m *MemStore) Put(c *Chunk) {
+func (s *MemStore) removeOldest() {
-	m.mu.Lock()
+	node := s.memtree
-	defer m.mu.Unlock()
+	log.Warn("purge memstore")
-	m.m[string(c.Key[:])] = c
+	for node.entry == nil {
+
+		aidx := uint(0)
+		av := node.access[aidx]
+
+		for aidx < node.width/2-1 {
+			if av == node.access[aidx*2+1] {
+				node.access[aidx] = node.access[aidx*2+2]
+				aidx = aidx*2 + 1
+			} else if av == node.access[aidx*2+2] {
+				node.access[aidx] = node.access[aidx*2+1]
+				aidx = aidx*2 + 2
+			} else {
+				panic(nil)
+			}
+		}
+		pidx := aidx*2 + 2 - node.width
+		if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) {
+			if node.subtree[pidx+1] != nil {
+				node.access[aidx] = node.subtree[pidx+1].access[0]
+			} else {
+				node.access[aidx] = 0
+			}
+		} else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) {
+			if node.subtree[pidx] != nil {
+				node.access[aidx] = node.subtree[pidx].access[0]
+			} else {
+				node.access[aidx] = 0
+			}
+			pidx++
+		} else {
+			panic(nil)
+		}
+
+		//fmt.Println(pidx)
+		node = node.subtree[pidx]
+
+	}
+
+	log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log()))
+	<-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.ReqC == nil {
+		node.entry = nil
+		s.entryCnt--
+	} else {
+		return
+	}
+
+	node.access[0] = 0
+
+	//---
+
+	aidx := uint(0)
+	for {
+		aa := node.access[aidx]
+		if aidx > 0 {
+			aidx = (aidx - 1) >> 1
+		} else {
+			pidx := node.parentIdx
+			node = node.parent
+			if node == nil {
+				return
+			}
+			aidx = (node.width + pidx - 2) >> 1
+		}
+		if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) {
+			node.access[aidx] = aa
+		}
+	}
 }
 
-func (m *MemStore) setCapacity(n int) {
+// type MemStore struct {
+// 	m  map[string]*Chunk
+// 	mu sync.RWMutex
+// }
 
-}
+// func NewMemStore(d *DbStore, capacity uint) (m *MemStore) {
+// 	return &MemStore{
+// 		m: make(map[string]*Chunk),
+// 	}
+// }
+
+// func (m *MemStore) Get(key Key) (*Chunk, error) {
+// 	m.mu.RLock()
+// 	defer m.mu.RUnlock()
+// 	c, ok := m.m[string(key[:])]
+// 	if !ok {
+// 		return nil, ErrNotFound
+// 	}
+// 	if !bytes.Equal(c.Key, key) {
+// 		panic(fmt.Errorf("MemStore.Get: chunk key %s != req key %s", c.Key.Hex(), key.Hex()))
+// 	}
+// 	return c, nil
+// }
+
+// func (m *MemStore) Put(c *Chunk) {
+// 	m.mu.Lock()
+// 	defer m.mu.Unlock()
+// 	m.m[string(c.Key[:])] = c
+// }
+
+// func (m *MemStore) setCapacity(n int) {
+
+// }
 
 // Close memstore
 func (s *MemStore) Close() {}

swarm/storage/netstore.go

@@ -36,6 +36,15 @@ func NewNetStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *Net
 // Get is the entrypoint for local retrieve requests
 // waits for response or times out
 func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
+	if self.retrieve == nil {
+		chunk, err = self.localStore.Get(key)
+		if err == nil {
+			return chunk, nil
+		}
+		if err != ErrFetching {
+			return nil, err
+		}
+	} else {
 		var created bool
 		chunk, created = self.localStore.GetOrCreateRequest(key)
 		if chunk.ReqC == nil {
@@ -47,6 +56,8 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
 				return nil, err
 			}
 		}
+	}
+
 	t := time.NewTicker(searchTimeout)
 	defer t.Stop()
 

swarm/swarm.go

@@ -132,7 +132,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 
 	db := storage.NewDBAPI(self.lstore)
 	delivery := stream.NewDelivery(to, db)
-	self.streamer = stream.NewRegistry(addr, delivery)
+	self.streamer = stream.NewRegistry(addr, delivery, self.lstore, false)
 	stream.RegisterSwarmSyncerServer(self.streamer, db)
 	stream.RegisterSwarmSyncerClient(self.streamer, db)