swarm/storage, swarm/network: light mode request streamers

swarm/network/lightnode.go

@@ -0,0 +1,191 @@
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.d
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package network
+
+import (
+	"errors"
+
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// RemoteReader implements IncomingStreamer
+type RemoteSectionReader struct {
+	db            *DbAccess
+	start         uint64
+	end           uint64
+	hashes        chan []byte
+	currentHashes []byte
+	currentData   []byte
+	quit          chan struct{}
+	root          []byte
+}
+
+// NewRemoteReader is the constructor for RemoteReader
+func NewRemoteSectionReader(root []byte, db *DbAccess) *RemoteSectionReader {
+	return &RemoteSectionReader{
+		db:     db,
+		root:   root,
+		hashes: make(chan []byte),
+		quit:   make(chan struct{}),
+	}
+}
+
+func (r *RemoteSectionReader) NeedData(key []byte) func() {
+	chunk, created := r.db.getOrCreateRequest(storage.Key(key))
+	// TODO: we may want to request from this peer anyway even if the request exists
+	if chunk.ReqC == nil || !created {
+		return nil
+	}
+	return func() {}
+}
+
+func (r *RemoteSectionReader) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
+	return from, r.end
+}
+
+func (r *RemoteSectionReader) BatchDone(s Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
+	r.hashes <- hashes
+	return nil
+}
+
+func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
+	l := int64(len(b))
+	m := int64(len(r.currentData))
+	if m > l {
+		m = l
+	}
+	copy(b, r.currentData[:m])
+	if m == l {
+		r.currentData = r.currentData[m:]
+		return l, nil
+	}
+	var end bool
+	for i := 0; !end && i < len(r.currentHashes); i += HashSize {
+		hash := r.currentHashes[i : i+HashSize]
+		chunk, err := r.db.get(hash)
+		if err != nil {
+			return n, err
+		}
+		m := chunk.Size
+		if n+m > l {
+			m = l - n
+			end = true
+		}
+		copy(b[n:], chunk.SData[:m])
+		n += int64(m)
+	}
+
+	for {
+		select {
+		case <-r.quit:
+			return n, errors.New("aborted")
+		case hashes := <-r.hashes:
+			var i int
+			for ; !end && i < len(hashes); i += HashSize {
+				hash := hashes[i : i+HashSize]
+				chunk, err := r.db.get(hash)
+				if err != nil {
+					return n, err
+				}
+				m := chunk.Size
+				if n+m > l {
+					m = l - n
+					end = true
+
+				}
+				copy(b[n:], chunk.SData[:m])
+				n += m
+			}
+			hashes = hashes[i:]
+		}
+	}
+	return n, nil
+}
+
+// RemoteSectionServer implements OutgoingStreamer
+type RemoteSectionServer struct {
+	// quit chan struct{}
+	currentBatch []byte
+	root         []byte
+	db           *DbAccess
+	r            *storage.LazyChunkReader
+}
+
+// NewRemoteReader is the constructor for RemoteReader
+func NewRemoteSectionServer(db *DbAccess, r *storage.LazyChunkReader) *RemoteSectionServer {
+	return &RemoteSectionServer{
+		db: db,
+		r:  r,
+	}
+}
+
+// GetData retrieves the actual chunk from localstore
+func (s *RemoteSectionServer) GetData(key []byte) []byte {
+	chunk, err := s.db.get(storage.Key(key))
+	if err != nil {
+		return nil
+	}
+	return chunk.SData
+}
+
+// GetBatch retrieves the next batch of hashes from the dbstore
+func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
+	if to > from+batchSize {
+		to = from + batchSize
+	}
+	batch := make([]byte, (to-from)*HashSize)
+	s.r.ReadAt(batch, int64(from))
+	s.currentBatch = batch
+	return batch, from, to, nil, nil
+}
+
+// RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node
+func RegisterRemoteSectionReader(s *Streamer, db *DbAccess) {
+	name := Stream("REMOTE_SECTION")
+	s.RegisterIncomingStreamer(name, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+		return NewRemoteSectionReader(t, db), nil
+	})
+}
+
+// RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on
+// upstream light server node
+func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) {
+	name := Stream("REMOTE_SECTION")
+	s.RegisterOutgoingStreamer(name, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+		r := rf(t)
+		return NewRemoteSectionServer(db, r), nil
+	})
+}
+
+// RegisterRemoteDownloader registers RemoteDownloader incoming streamer
+// on downstream light  node
+func RegisterRemoteDownloader(s *Streamer, db *DbAccess) {
+	name := Stream("REMOTE_DOWNLOADER")
+	s.RegisterIncomingStreamer(name, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+		return NewRemoteDownloader(t, db), nil
+	})
+}
+
+// RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on
+// upstream light server node
+func RegisterRemoteDownloadServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) {
+	name := Stream("REMOTE_DOWNLOADER")
+	s.RegisterOutgoingStreamer(name, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+		r := rf(t)
+		return NewRemoteDownloadServer(db, r), nil
+	})
+}

swarm/network/streamer.go

@@ -34,7 +34,7 @@ import (
 const (
 	HashSize = 32
 
-	Low int = iota
+	Low uint8 = iota
 	Mid
 	High
 	Top
@@ -80,6 +80,7 @@ func (self TakeoverProofMsg) String() string {
 // SubcribeMsg is the protocol msg for requesting a stream(section)
 type SubscribeMsg struct {
 	Stream   Stream
+	Key      []byte
 	From, To uint64
 	Priority uint8 // delivered on priority channel
 }
@@ -88,6 +89,7 @@ type SubscribeMsg struct {
 // stream section
 type UnsyncedKeysMsg struct {
 	Stream         Stream // name of Stream
+	Key            []byte // subtype or key
 	From, To       uint64 // peer and db-specific entry count
 	Hashes         []byte // stream of hashes (128)
 	*HandoverProof        // HandoverProof
@@ -114,6 +116,7 @@ func (self UnsyncedKeysMsg) String() string {
 // offered in UnsyncedKeysMsg downstream peer actually wants sent over
 type WantedKeysMsg struct {
 	Stream   Stream // name of stream
+	Key      []byte // subtype or key
 	Want     []byte // bitvector indicating which keys of the batch needed
 	From, To uint64 // next interval offset - empty if not to be continued
 }
@@ -127,8 +130,8 @@ func (self WantedKeysMsg) String() string {
 type Streamer struct {
 	incomingLock sync.RWMutex
 	outgoingLock sync.RWMutex
-	outgoing     map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)
+	outgoing     map[Stream]func(*StreamerPeer, []byte) (OutgoingStreamer, error)
-	incoming     map[Stream]func(*StreamerPeer) (IncomingStreamer, error)
+	incoming     map[Stream]func(*StreamerPeer, []byte) (IncomingStreamer, error)
 
 	dbAccess *DbAccess
 	overlay  Overlay
@@ -138,8 +141,8 @@ type Streamer struct {
 // NewStreamer is Streamer constructor
 func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer {
 	return &Streamer{
-		outgoing: make(map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)),
+		outgoing: make(map[Stream]func(*StreamerPeer, []byte) (OutgoingStreamer, error)),
-		incoming: make(map[Stream]func(*StreamerPeer) (IncomingStreamer, error)),
+		incoming: make(map[Stream]func(*StreamerPeer, []byte) (IncomingStreamer, error)),
 		dbAccess: dbAccess,
 		overlay:  overlay,
 		receiveC: make(chan *ChunkDeliveryMsg, 10),
@@ -147,21 +150,21 @@ func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer {
 }
 
 // RegisterIncomingStreamer registers an incoming streamer constructor
-func (self *Streamer) RegisterIncomingStreamer(stream Stream, f func(*StreamerPeer) (IncomingStreamer, error)) {
+func (self *Streamer) RegisterIncomingStreamer(stream Stream, f func(*StreamerPeer, []byte) (IncomingStreamer, error)) {
 	self.incomingLock.Lock()
 	defer self.incomingLock.Unlock()
 	self.incoming[stream] = f
 }
 
 // RegisterOutgoingStreamer registers an outgoing streamer constructor
-func (self *Streamer) RegisterOutgoingStreamer(stream Stream, f func(*StreamerPeer) (OutgoingStreamer, error)) {
+func (self *Streamer) RegisterOutgoingStreamer(stream Stream, f func(*StreamerPeer, []byte) (OutgoingStreamer, error)) {
 	self.outgoingLock.Lock()
 	defer self.outgoingLock.Unlock()
 	self.outgoing[stream] = f
 }
 
 // GetIncomingStreamer accessor for incoming streamer constructors
-func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer) (IncomingStreamer, error), error) {
+func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer, []byte) (IncomingStreamer, error), error) {
 	self.incomingLock.RLock()
 	defer self.incomingLock.RUnlock()
 	f := self.incoming[stream]
@@ -172,7 +175,7 @@ func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer) (I
 }
 
 // GetOutgoingStreamer accessor for incoming streamer constructors
-func (self *Streamer) GetOutgoingStreamer(stream Stream) (func(*StreamerPeer) (OutgoingStreamer, error), error) {
+func (self *Streamer) GetOutgoingStreamer(stream Stream) (func(*StreamerPeer, []byte) (OutgoingStreamer, error), error) {
 	self.outgoingLock.RLock()
 	defer self.outgoingLock.RUnlock()
 	f := self.outgoing[stream]
@@ -190,19 +193,30 @@ func (self *Streamer) PeerInfo(id discover.NodeID) interface{} {
 	return nil
 }
 
+type outgoingStreamer struct {
+	OutgoingStreamer
+	priority     uint8
+	currentBatch []byte
+}
+
 // OutgoingStreamer interface for outgoing peer Streamer
 type OutgoingStreamer interface {
-	CurrentBatch() []byte
 	SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
 	GetData([]byte) []byte
-	Priority() int
+}
+
+type incomingStreamer struct {
+	IncomingStreamer
+	priority uint8
+	quit     chan struct{}
+	next     chan struct{}
 }
 
 // IncomingStreamer interface for incoming peer Streamer
 type IncomingStreamer interface {
 	NextBatch(uint64) (uint64, uint64)
 	NeedData([]byte) func()
-	Priority() int
+	BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error)
 }
 
 // StreamerPeer is the Peer extention for the streaming protocol
@@ -214,8 +228,8 @@ type StreamerPeer struct {
 	dbAccess     *DbAccess
 	outgoingLock sync.RWMutex
 	incomingLock sync.RWMutex
-	outgoing     map[Stream]OutgoingStreamer
+	outgoing     map[Stream]*outgoingStreamer
-	incoming     map[Stream]IncomingStreamer
+	incoming     map[Stream]*incomingStreamer
 	quit         chan struct{}
 }
 
@@ -232,10 +246,10 @@ type StreamerPeer struct {
 // NewStreamerPeer is the constructor for StreamerPeer
 func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
 	self := &StreamerPeer{
-		pq:       pq.New(PriorityQueue, PriorityQueueCap),
+		pq:       pq.New(int(PriorityQueue), PriorityQueueCap),
 		streamer: streamer,
-		outgoing: make(map[Stream]OutgoingStreamer),
+		outgoing: make(map[Stream]*outgoingStreamer),
-		incoming: make(map[Stream]IncomingStreamer),
+		incoming: make(map[Stream]*incomingStreamer),
 		quit:     make(chan struct{}),
 	}
 	ctx, cancel := context.WithCancel(context.Background())
@@ -247,6 +261,7 @@ func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
 	return self
 }
 
+// RetrieveRequestMsg is the protocol msg for chunk retrieve requests
 type RetrieveRequestMsg struct {
 	Key storage.Key
 }
@@ -255,30 +270,32 @@ type RetrieveRequestMsg struct {
 type RetrieveRequestStreamer struct {
 	deliveryC  chan *storage.Chunk
 	batchC     chan []byte
-	dbAccess     *DbAccess
+	db         *DbAccess
-	currentBatch []byte
 	currentLen uint64
 }
 
-func RegisterRequestStreamer(streamer *Streamer, dbAccess *DbAccess) {
+// RegisterRequestStreamer registers outgoing and incoming streamers for request handling
-	streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer) (OutgoingStreamer, error) {
+func RegisterRequestStreamer(streamer *Streamer, db *DbAccess) {
-		return NewRetrieveRequestStreamer(dbAccess), nil
+	streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+		return NewRetrieveRequestStreamer(db), nil
 	})
-	streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer) (IncomingStreamer, error) {
+	streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-		return NewIncomingSwarmSyncer(Top, nil, p, dbAccess, nil)
+		return NewIncomingSwarmSyncer(nil, p, db, nil)
 	})
 }
 
-func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer {
+// NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor
+func NewRetrieveRequestStreamer(db *DbAccess) *RetrieveRequestStreamer {
 	s := &RetrieveRequestStreamer{
 		deliveryC: make(chan *storage.Chunk),
 		batchC:    make(chan []byte),
-		dbAccess:  dbAccess,
+		db:        db,
 	}
 	go s.processDeliveries()
 	return s
 }
 
+// processDeliveries handles delivered chunk hashes
 func (s *RetrieveRequestStreamer) processDeliveries() {
 	var hashes []byte
 	for {
@@ -291,28 +308,21 @@ func (s *RetrieveRequestStreamer) processDeliveries() {
 	}
 }
 
-func (s *RetrieveRequestStreamer) CurrentBatch() []byte {
+// SetNextBatch
-	return s.currentBatch
-}
-
 func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) {
 	hashes = <-s.batchC
-	s.currentBatch = hashes
 	from = s.currentLen
 	s.currentLen += uint64(len(hashes))
 	to = s.currentLen
 	return
 }
 
+// GetData retrives chunk data from db store
 func (s *RetrieveRequestStreamer) GetData(key []byte) []byte {
-	chunk, _ := s.dbAccess.get(storage.Key(key))
+	chunk, _ := s.db.get(storage.Key(key))
 	return chunk.SData
 }
 
-func (s *RetrieveRequestStreamer) Priority() int {
-	return Top
-}
-
 const retrieveRequestStream = Stream("RETRIEVE_REQUEST")
 
 func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) error {
@@ -321,7 +331,7 @@ func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) erro
 	if err != nil {
 		return err
 	}
-	streamer := s.(*RetrieveRequestStreamer)
+	streamer := s.OutgoingStreamer.(*RetrieveRequestStreamer)
 	if chunk.ReqC != nil {
 		if created {
 			if err := self.streamer.Retrieve(chunk); err != nil {
@@ -349,10 +359,16 @@ func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) erro
 	return nil
 }
 
+// Retrieve sends a chunk retrieve request to
 func (self *Streamer) Retrieve(chunk *storage.Chunk) error {
-	// TODO: using the overlay find the closes peer to send the retrieve
+	self.overlay.EachConn(chunk.Key[:], 255, func(p OverlayConn, po int, nn bool) bool {
-	//       request to.
+		sp := p.(*StreamerPeer)
-	//       self.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool {})
+		// TODO: skip light nodes that do not accept retrieve requests
+		sp.SendPriority(&RetrieveRequestMsg{
+			Key: chunk.Key[:],
+		}, Top)
+		return false
+	})
 	return nil
 }
 
@@ -383,7 +399,7 @@ func (self *Streamer) processReceivedChunks() {
 	}
 }
 
-func (self *StreamerPeer) getOutgoingStreamer(s Stream) (OutgoingStreamer, error) {
+func (self *StreamerPeer) getOutgoingStreamer(s Stream) (*outgoingStreamer, error) {
 	self.outgoingLock.RLock()
 	defer self.outgoingLock.RUnlock()
 	streamer := self.outgoing[s]
@@ -393,7 +409,7 @@ func (self *StreamerPeer) getOutgoingStreamer(s Stream) (OutgoingStreamer, error
 	return streamer, nil
 }
 
-func (self *StreamerPeer) getIncomingStreamer(s Stream) (IncomingStreamer, error) {
+func (self *StreamerPeer) getIncomingStreamer(s Stream) (*incomingStreamer, error) {
 	self.incomingLock.RLock()
 	defer self.incomingLock.RUnlock()
 	streamer := self.incoming[s]
@@ -403,44 +419,59 @@ func (self *StreamerPeer) getIncomingStreamer(s Stream) (IncomingStreamer, error
 	return streamer, nil
 }
 
-func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer) error {
+func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer, priority uint8) (*outgoingStreamer, error) {
 	self.outgoingLock.Lock()
 	defer self.outgoingLock.Unlock()
 	if self.outgoing[s] != nil {
-		return fmt.Errorf("stream %v already registered", s)
+		return nil, fmt.Errorf("stream %v already registered", s)
 	}
-	self.outgoing[s] = o
+	os := &outgoingStreamer{
-	return nil
+		OutgoingStreamer: o,
+		priority:         priority,
+	}
+	self.outgoing[s] = os
+	return os, nil
 }
 
-func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer) error {
+func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer, priority uint8) error {
 	self.incomingLock.Lock()
 	defer self.incomingLock.Unlock()
 	if self.incoming[s] != nil {
 		return fmt.Errorf("stream %v already registered", s)
 	}
-	self.incoming[s] = i
+	next := make(chan struct{}, 1)
+	self.incoming[s] = &incomingStreamer{
+		IncomingStreamer: i,
+		priority:         priority,
+		next:             next,
+	}
+	next <- struct{}{} // this is to allow wantedKeysMsg before first batch arrives
 	return nil
 }
 
 // Subscribe initiates the streamer
-func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
+func (self *StreamerPeer) Subscribe(s Stream, t []byte, from, to uint64, priority uint8) error {
 	f, err := self.streamer.GetIncomingStreamer(s)
 	if err != nil {
 		return err
 	}
-	is, err := f(self)
+	is, err := f(self, t)
 	if err != nil {
 		return err
 	}
-	self.setIncomingStreamer(s, is)
+	err = self.setIncomingStreamer(s, is, priority)
+	if err != nil {
+		return err
+	}
+
 	msg := &SubscribeMsg{
 		Stream:   s,
+		Key:      t,
 		From:     from,
 		To:       to,
-		Priority: uint8(is.Priority()),
+		Priority: priority,
 	}
-	self.SendPriority(msg, is.Priority())
+	self.SendPriority(msg, priority)
 	return nil
 }
 
@@ -449,14 +480,16 @@ func (self *StreamerPeer) handleSubscribeMsg(req *SubscribeMsg) error {
 	if err != nil {
 		return err
 	}
-	s, err := f(self)
+	s, err := f(self, req.Key)
 	if err != nil {
 		return err
 	}
-	if err := self.setOutgoingStreamer(req.Stream, s); err != nil {
+	key := string(req.Stream) + string(req.Key)
+	os, err := self.setOutgoingStreamer(Stream(key), s, req.Priority)
+	if err != nil {
 		return nil
 	}
-	self.SendUnsyncedKeys(s, req.From, req.To, int(req.Priority))
+	go self.SendUnsyncedKeys(os, req.From, req.To)
 	return nil
 }
 
@@ -485,11 +518,17 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error {
 			}(wait)
 		}
 	}
-	// go func() {
+	go func() {
-	// 	wg.Wait()
+		wg.Wait()
-	// 	msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root)
+		if tf := s.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil {
-	// 	self.Send(msg, s.Priority())
+			tp, err := tf()
-	// }()
+			if err != nil {
+				return
+			}
+			self.SendPriority(tp, s.priority)
+		}
+		s.next <- struct{}{}
+	}()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
 	from, to := s.NextBatch(req.To)
@@ -502,7 +541,14 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error {
 		From:   from,
 		To:     to,
 	}
-	self.SendPriority(msg, s.Priority())
+	go func() {
+		select {
+		case <-s.next:
+		case <-s.quit:
+			return
+		}
+		self.SendPriority(msg, s.priority)
+	}()
 	return nil
 }
 
@@ -514,9 +560,9 @@ func (self *StreamerPeer) handleWantedKeysMsg(req *WantedKeysMsg) error {
 	if err != nil {
 		return err
 	}
-	hashes := s.CurrentBatch()
+	hashes := s.currentBatch
 	// launch in go routine since GetBatch blocks until new hashes arrive
-	go self.SendUnsyncedKeys(s, req.From, req.To, s.Priority())
+	go self.SendUnsyncedKeys(s, req.From, req.To)
 	l := len(hashes) / HashSize
 	want, err := bv.NewFromBytes(req.Want, l)
 	if err != nil {
@@ -531,7 +577,7 @@ func (self *StreamerPeer) handleWantedKeysMsg(req *WantedKeysMsg) error {
 			}
 			chunk := storage.NewChunk(hash, nil)
 			chunk.SData = data
-			if err := self.Deliver(chunk, s.Priority()); err != nil {
+			if err := self.Deliver(chunk, s.priority); err != nil {
 				return err
 			}
 		}
@@ -549,32 +595,33 @@ func (self *StreamerPeer) handleTakeoverProofMsg(req *TakeoverProofMsg) error {
 }
 
 // Deliver sends a storeRequestMsg protocol message to the peer
-func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority int) error {
+func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority uint8) error {
 	msg := &ChunkDeliveryMsg{
 		Key:   chunk.Key,
 		SData: chunk.SData,
 	}
-	return self.pq.Push(nil, msg, priority)
+	return self.pq.Push(nil, msg, int(priority))
 }
 
 // Deliver sends a storeRequestMsg protocol message to the peer
-func (self *StreamerPeer) SendPriority(msg interface{}, priority int) error {
+func (self *StreamerPeer) SendPriority(msg interface{}, priority uint8) error {
-	return self.pq.Push(nil, msg, priority)
+	return self.pq.Push(nil, msg, int(priority))
 }
 
 // UnsyncedKeys sends UnsyncedKeysMsg protocol msg
-func (self *StreamerPeer) SendUnsyncedKeys(s OutgoingStreamer, f, t uint64, priority int) error {
+func (self *StreamerPeer) SendUnsyncedKeys(s *outgoingStreamer, f, t uint64) error {
 	hashes, from, to, proof, err := s.SetNextBatch(f, t)
 	if err != nil {
 		return err
 	}
+	s.currentBatch = hashes
 	msg := &UnsyncedKeysMsg{
 		HandoverProof: proof,
 		Hashes:        hashes,
 		From:          from,
 		To:            to,
 	}
-	return self.SendPriority(msg, s.Priority())
+	return self.SendPriority(msg, s.priority)
 }
 
 // StreamerSpec is the spec of the streamer protocol.
@@ -595,10 +642,13 @@ var StreamerSpec = &protocols.Spec{
 func (s *Streamer) Run(p *bzzPeer) error {
 	sp := NewStreamerPeer(p, s)
 	// load saved intervals
-	sp.handleSubscribeMsg(&SubscribeMsg{
+	// autosubscribe to request handler to serve request only for non-light nodes
-		Stream:   retrieveRequestStream,
+	// sp.handleSubscribeMsg(&SubscribeMsg{
-		Priority: uint8(Top),
+	// 	Stream:   retrieveRequestStream,
-	})
+	// 	Priority: uint8(Top),
+	// })
+	// subscribe to request handling ; only with non-light nodes
+	sp.Subscribe(retrieveRequestStream, nil, 0, 0, Top)
 	defer close(sp.quit)
 	return sp.Run(sp.HandleMsg)
 }

swarm/network/syncer.go

@@ -73,16 +73,21 @@ type OutgoingSwarmSyncer struct {
 	po        uint8
 	db        *DbAccess
 	sessionAt uint64
-	currentBatch []byte
+	start     uint64
-	priority     int
 }
 
 // NewOutgoingSwarmSyncer is contructor for OutgoingSwarmSyncer
-func NewOutgoingSwarmSyncer(po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) {
+func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) {
+	sessionAt := db.currentBucketStorageIndex(po)
+	var start uint64
+	if live {
+		start = sessionAt
+	}
 	self := &OutgoingSwarmSyncer{
 		po:        po,
 		db:        db,
-		sessionAt: db.currentBucketStorageIndex(po),
+		sessionAt: sessionAt,
+		start:     start,
 	}
 	return self, nil
 }
@@ -90,21 +95,23 @@ func NewOutgoingSwarmSyncer(po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error
 const maxPO = 32
 
 func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) {
-	for po := uint8(0); po < maxPO; po++ {
+	stream := Stream("SYNC")
-		stream := Stream(fmt.Sprintf("SYNC-%02d-live", po))
+	streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
-		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+		syncType, po := parseSyncLabel(t)
-			return NewOutgoingSwarmSyncer(po, db)
+		switch syncType {
-		})
+		case "LIVE":
-		stream = Stream(fmt.Sprintf("SYNC-%02d-history", po))
+			return NewOutgoingSwarmSyncer(true, po, db)
-		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+		case "HISTORY":
-			return NewOutgoingSwarmSyncer(po, db)
+			return NewOutgoingSwarmSyncer(false, po, db)
+		default:
+			return nil, errors.New("invalid sync type")
+		}
 	})
 	// stream = Stream(fmt.Sprintf("SYNC-%02d-delete", po))
 	// streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
 	// 	return NewOutgoingProvableSwarmSyncer(po, db)
 	// })
 }
-}
 
 // GetSection retrieves the actual chunk from localstore
 func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte {
@@ -115,18 +122,13 @@ func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte {
 	return chunk.SData
 }
 
-func (self *OutgoingSwarmSyncer) CurrentBatch() []byte {
-	return self.currentBatch
-}
-
-func (self *OutgoingSwarmSyncer) Priority() int {
-	return self.priority
-}
-
 // GetBatch retrieves the next batch of hashes from the dbstore
 func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
 	var batch []byte
 	i := 0
+	if from == 0 {
+		from = self.start
+	}
 	err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool {
 		batch = append(batch, key[:]...)
 		i++
@@ -136,17 +138,15 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64,
 	if err != nil {
 		return nil, 0, 0, nil, err
 	}
-	self.currentBatch = batch
 	log.Debug("Swarm batch", "po", self.po, "len", i, "from", from, "to", to)
 	return batch, from, to, nil, nil
 }
 
 // IncomingSwarmSyncer
 type IncomingSwarmSyncer struct {
-	priority      int
 	sessionAt     uint64
 	nextC         chan struct{}
-	intervals     []uint64
+	intervals     *Intervals
 	sessionRoot   storage.Key
 	sessionReader storage.LazySectionReader
 	retrieveC     chan *storage.Chunk
@@ -159,9 +159,8 @@ type IncomingSwarmSyncer struct {
 }
 
 // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
-func NewIncomingSwarmSyncer(priority int, intervals []uint64, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
+func NewIncomingSwarmSyncer(intervals *Intervals, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
 	self := &IncomingSwarmSyncer{
-		priority:  priority,
 		intervals: intervals,
 		dbAccess:  dbAccess,
 		chunker:   chunker,
@@ -169,10 +168,6 @@ func NewIncomingSwarmSyncer(priority int, intervals []uint64, p Peer, dbAccess *
 	return self, nil
 }
 
-func (s *IncomingSwarmSyncer) Priority() int {
-	return s.priority
-}
-
 // // NewIncomingProvableSwarmSyncer is a contructor for provable data exchange syncer
 // func NewIncomingProvableSwarmSyncer(po int, priority int, index uint64, sessionAt uint64, intervals []uint64, sessionRoot storage.Key, chunker *storage.PyramidChunker, store storage.ChunkStore, p Peer) *IncomingSwarmSyncer {
 // 	retrieveC := make(storage.Chunk, chunksCap)
@@ -195,16 +190,78 @@ func (s *IncomingSwarmSyncer) Priority() int {
 // 	return self
 // }
 
-func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
+type Syncer struct {
-	for po := uint8(0); po < maxPO; po++ {
+	intervals map[string][]uint64
-		stream := Stream(fmt.Sprintf("SYNC-%02d-live", po))
+}
-		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) {
+
-			return NewIncomingSwarmSyncer(High, nil, p, nil, nil)
+func NewSyncer() *Syncer {
-		})
+	return &Syncer{}
-		stream = Stream(fmt.Sprintf("SYNC-%02d-history", po))
+}
-		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) {
+
-			//intervals := loadIntervals(p, po, false)
+type Intervals struct {
-			return NewIncomingSwarmSyncer(Mid, nil, p, nil, nil)
+	syncer *Syncer
+	key    []byte
+}
+
+func (s *Intervals) load() error {
+	return s.syncer.load(s.key)
+}
+
+func (s *Intervals) save() error {
+	return s.syncer.save(s.key)
+}
+
+func (s *Intervals) get() []uint64 {
+	return s.syncer.get(s.key)
+}
+
+func (s *Intervals) set(v []uint64) {
+	s.syncer.set(s.key, v)
+}
+
+func (s *Syncer) NewIntervals(key []byte) *Intervals {
+	return &Intervals{
+		syncer: s,
+		key:    key,
+	}
+}
+
+func newSyncLabel(typ string, po uint8) []byte {
+	t := []byte(typ)
+	t = append(t, byte(po))
+	return t
+}
+
+func parseSyncLabel(t []byte) (string, uint8) {
+	l := len(t) - 1
+	return sstring(t[:l]), uint8(t[l])
+}
+
+// StartSyncing is called on the StreamerPeer to start the syncing process
+// the idea is that it is called only after kademlia is close to healthy
+func StartSyncing(s *StreamerPeer, po uint8, nn bool) {
+	lastPO := po
+	if nn {
+		lastPO = maxPO
+	}
+	for i := po; i <= lastPO; i++ {
+		s.Subscribe(Stream("SYNC"), newSyncLabel("LIVE", po), 0, 0, High)
+		s.Subscribe(Stream("SYNC"), newSyncLabel("HISTORY", po), 0, 0, Mid)
+	}
+}
+
+func RegisterIncomingSyncers(streamer *Streamer, syncer *Syncer, db *DbAccess) {
+	stream := Stream("SYNC")
+	streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+		syncType, po := parseSyncLabel(t)
+		switch syncType {
+		case "LIVE":
+			return NewIncomingSwarmSyncer(nil, p, nil, nil)
+		case "HISTORY":
+			intervals := syncer.NewIntervals(t)
+			return NewIncomingSwarmSyncer(intervals, p, nil, nil)
+		}
+		return nil, fmt.Errorf("unknown sync type %q", syncType)
 	})
 	// stream = fmt.Sprintf("SYNC-%02d-delete", po)
 	// streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
@@ -212,13 +269,12 @@ func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
 	// 	return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p)
 	// })
 }
-}
 
 // NeedData
 func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
-	chunk, created := self.dbAccess.getOrCreateRequest(key)
+	chunk, _ := self.dbAccess.getOrCreateRequest(key)
 	// TODO: we may want to request from this peer anyway even if the request exists
-	if chunk.ReqC == nil || !created {
+	if chunk.ReqC == nil {
 		return nil
 	}
 	// create request and wait until the chunk data arrives and is stored
@@ -227,28 +283,37 @@ func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
 
 // NextBatch adjusts the indexes by inspecting the intervals
 func (self *IncomingSwarmSyncer) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
-	if self.intervals[0] >= self.sessionAt { // live syncing
+	intervals := self.intervals.get()
+	if intervals[0] >= self.sessionAt { // live syncing
 		nextFrom = from
-		self.intervals[1] = from
+		intervals[1] = from
 	} else if from >= self.sessionAt { // history sync complete
-		self.intervals = nil
+		intervals = nil
-	} else if len(self.intervals) > 2 && from >= self.intervals[2] { // filled a gap in the intervals
+	} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
-		self.intervals = append(self.intervals[:1], self.intervals[3:]...)
+		intervals = append(intervals[:1], intervals[3:]...)
-		nextFrom = self.intervals[1]
+		nextFrom = intervals[1]
-		if len(self.intervals) > 2 {
+		if len(intervals) > 2 {
-			nextTo = self.intervals[2]
+			nextTo = intervals[2]
 		} else {
 			nextTo = self.sessionAt
 		}
 	} else {
 		nextFrom = from
-		self.intervals[1] = from
+		intervals[1] = from
 		nextTo = self.sessionAt
 	}
+	self.intervals.set(intervals)
 	return nextFrom, nextTo
 }
 
-//
+// BatchDone
+func (self *IncomingSwarmSyncer) BatchDone(s Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
+	if self.chunker != nil {
+		return func() (*TakeoverProof, error) { return self.TakeoverProof(s, from, hashes, root) }
+	}
+	return nil
+}
+
 func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
 	// for provable syncer currentRoot is non-zero length
 	if self.chunker != nil {

swarm/storage/chunker.go

@@ -302,16 +302,18 @@ type LazyChunkReader struct {
 	chunkSize int64       // inherit from chunker
 	branches  int64       // inherit from chunker
 	hashSize  int64       // inherit from chunker
+	depth     int
 }
 
 // implements the Joiner interface
-func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader {
+func (self *TreeChunker) Join(key Key, chunkC chan *Chunk, depth int) LazySectionReader {
 	return &LazyChunkReader{
 		key:       key,
 		chunkC:    chunkC,
 		chunkSize: self.chunkSize,
 		branches:  self.branches,
 		hashSize:  self.hashSize,
+		depth:     depth,
 	}
 }
 
@@ -358,8 +360,13 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 		depth++
 	}
 	wg := sync.WaitGroup{}
+	length := int64(len(b))
+	for d := 0; d < self.depth; d++ {
+		off *= self.chunkSize
+		length *= self.chunkSize
+	}
 	wg.Add(1)
-	go self.join(b, off, off+int64(len(b)), depth, treeSize/self.branches, self.chunk, &wg, errC, quitC)
+	go self.join(b, off, off+length, depth, treeSize/self.branches, self.chunk, &wg, errC, quitC)
 	go func() {
 		wg.Wait()
 		close(errC)
@@ -379,18 +386,14 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 
 func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) {
 	defer parentWg.Done()
-	// return NewDPA(&LocalStore{})
-
-	// chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
-
 	// find appropriate block level
-	for chunk.Size < treeSize && depth > 0 {
+	for chunk.Size < treeSize && depth > self.depth {
 		treeSize /= self.branches
 		depth--
 	}
 
 	// leaf chunk found
-	if depth == 0 {
+	if depth == self.depth {
 		extra := 8 + eoff - int64(len(chunk.SData))
 		if extra > 0 {
 			eoff -= extra

swarm/storage/chunker_test.go

@@ -163,7 +163,7 @@ func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Ch
 		}
 	}()
 
-	reader := chunker.Join(key, chunkC)
+	reader := chunker.Join(key, chunkC, 0)
 	return reader
 }
 
@@ -489,7 +489,8 @@ func BenchmarkJoin_4(t *testing.B) { benchmarkJoin(10000, t) }
 func BenchmarkJoin_5(t *testing.B) { benchmarkJoin(100000, t) }
 func BenchmarkJoin_6(t *testing.B) { benchmarkJoin(1000000, t) }
 func BenchmarkJoin_7(t *testing.B) { benchmarkJoin(10000000, t) }
-func BenchmarkJoin_8(t *testing.B) { benchmarkJoin(100000000, t) }
+
+// func BenchmarkJoin_8(t *testing.B) { benchmarkJoin(100000000, t) }
 
 func BenchmarkSplitTreeSHA3_2(t *testing.B)  { benchmarkSplitTreeSHA3(100, t) }
 func BenchmarkSplitTreeSHA3_2h(t *testing.B) { benchmarkSplitTreeSHA3(500, t) }
@@ -500,7 +501,8 @@ func BenchmarkSplitTreeSHA3_4h(t *testing.B) { benchmarkSplitTreeSHA3(50000, t)
 func BenchmarkSplitTreeSHA3_5(t *testing.B)  { benchmarkSplitTreeSHA3(100000, t) }
 func BenchmarkSplitTreeSHA3_6(t *testing.B)  { benchmarkSplitTreeSHA3(1000000, t) }
 func BenchmarkSplitTreeSHA3_7(t *testing.B)  { benchmarkSplitTreeSHA3(10000000, t) }
-func BenchmarkSplitTreeSHA3_8(t *testing.B)  { benchmarkSplitTreeSHA3(100000000, t) }
+
+// func BenchmarkSplitTreeSHA3_8(t *testing.B)  { benchmarkSplitTreeSHA3(100000000, t) }
 
 func BenchmarkSplitTreeBMT_2(t *testing.B)  { benchmarkSplitTreeBMT(100, t) }
 func BenchmarkSplitTreeBMT_2h(t *testing.B) { benchmarkSplitTreeBMT(500, t) }
@@ -511,7 +513,8 @@ func BenchmarkSplitTreeBMT_4h(t *testing.B) { benchmarkSplitTreeBMT(50000, t) }
 func BenchmarkSplitTreeBMT_5(t *testing.B)  { benchmarkSplitTreeBMT(100000, t) }
 func BenchmarkSplitTreeBMT_6(t *testing.B)  { benchmarkSplitTreeBMT(1000000, t) }
 func BenchmarkSplitTreeBMT_7(t *testing.B)  { benchmarkSplitTreeBMT(10000000, t) }
-func BenchmarkSplitTreeBMT_8(t *testing.B)  { benchmarkSplitTreeBMT(100000000, t) }
+
+// func BenchmarkSplitTreeBMT_8(t *testing.B)  { benchmarkSplitTreeBMT(100000000, t) }
 
 func BenchmarkSplitPyramidSHA3_2(t *testing.B)  { benchmarkSplitPyramidSHA3(100, t) }
 func BenchmarkSplitPyramidSHA3_2h(t *testing.B) { benchmarkSplitPyramidSHA3(500, t) }
@@ -522,7 +525,8 @@ func BenchmarkSplitPyramidSHA3_4h(t *testing.B) { benchmarkSplitPyramidSHA3(5000
 func BenchmarkSplitPyramidSHA3_5(t *testing.B)  { benchmarkSplitPyramidSHA3(100000, t) }
 func BenchmarkSplitPyramidSHA3_6(t *testing.B)  { benchmarkSplitPyramidSHA3(1000000, t) }
 func BenchmarkSplitPyramidSHA3_7(t *testing.B)  { benchmarkSplitPyramidSHA3(10000000, t) }
-func BenchmarkSplitPyramidSHA3_8(t *testing.B)  { benchmarkSplitPyramidSHA3(100000000, t) }
+
+// func BenchmarkSplitPyramidSHA3_8(t *testing.B)  { benchmarkSplitPyramidSHA3(100000000, t) }
 
 func BenchmarkSplitPyramidBMT_2(t *testing.B)  { benchmarkSplitPyramidBMT(100, t) }
 func BenchmarkSplitPyramidBMT_2h(t *testing.B) { benchmarkSplitPyramidBMT(500, t) }
@@ -533,7 +537,8 @@ func BenchmarkSplitPyramidBMT_4h(t *testing.B) { benchmarkSplitPyramidBMT(50000,
 func BenchmarkSplitPyramidBMT_5(t *testing.B)  { benchmarkSplitPyramidBMT(100000, t) }
 func BenchmarkSplitPyramidBMT_6(t *testing.B)  { benchmarkSplitPyramidBMT(1000000, t) }
 func BenchmarkSplitPyramidBMT_7(t *testing.B)  { benchmarkSplitPyramidBMT(10000000, t) }
-func BenchmarkSplitPyramidBMT_8(t *testing.B)  { benchmarkSplitPyramidBMT(100000000, t) }
+
+// func BenchmarkSplitPyramidBMT_8(t *testing.B)  { benchmarkSplitPyramidBMT(100000000, t) }
 
 func BenchmarkAppendPyramid_2(t *testing.B)  { benchmarkAppendPyramid(100, 1000, t) }
 func BenchmarkAppendPyramid_2h(t *testing.B) { benchmarkAppendPyramid(500, 1000, t) }
@@ -543,7 +548,8 @@ func BenchmarkAppendPyramid_4h(t *testing.B) { benchmarkAppendPyramid(50000, 100
 func BenchmarkAppendPyramid_5(t *testing.B)  { benchmarkAppendPyramid(1000000, 1000, t) }
 func BenchmarkAppendPyramid_6(t *testing.B)  { benchmarkAppendPyramid(1000000, 1000, t) }
 func BenchmarkAppendPyramid_7(t *testing.B)  { benchmarkAppendPyramid(10000000, 1000, t) }
-func BenchmarkAppendPyramid_8(t *testing.B)  { benchmarkAppendPyramid(100000000, 1000, t) }
+
+// func BenchmarkAppendPyramid_8(t *testing.B)  { benchmarkAppendPyramid(100000000, 1000, t) }
 
 // go test -timeout 20m -cpu 4 -bench=./swarm/storage -run no
 // If you dont add the timeout argument above .. the benchmark will timeout and dump

swarm/storage/dpa.go

@@ -92,7 +92,7 @@ func NewDPA(store ChunkStore, params *ChunkerParams) *DPA {
 // Chunk retrieval blocks on netStore requests with a timeout so reader will
 // report error if retrieval of chunks within requested range time out.
 func (self *DPA) Retrieve(key Key) LazySectionReader {
-	return self.Chunker.Join(key, self.retrieveC)
+	return self.Chunker.Join(key, self.retrieveC, 0)
 }
 
 // Public API. Main entry point for document storage directly. Used by the

swarm/storage/types.go

@@ -273,7 +273,7 @@ type Joiner interface {
 	   The chunks are not meant to be validated by the chunker when joining. This
 	   is because it is left to the DPA to decide which sources are trusted.
 	*/
-	Join(key Key, chunkC chan *Chunk) LazySectionReader
+	Join(key Key, chunkC chan *Chunk, depth int) LazySectionReader
 }
 
 type Chunker interface {