swarm/network: refactor and modify streamer code

swarm/network/README.md

@@ -1,30 +1,67 @@
-# Requirements
+## Streaming
+
+Streaming is a new protocol of the swarm bzz bundle of protocols.
+This protocol provides the basic logic for chunk-based data flow.
+It implements simple retrieve requests and delivery using priority queue.
+A data exchange stream is a directional flow of chunks between peers.
+The source of datachunks is the upstream, the receiver is called the
+downstream peer. Each streaming protocol defines an outgoing streamer
+and an incoming streamer, the former installing on the upstream,
+the latter on the downstream peer.
+
+Subscribe on StreamerPeer launches an incoming streamer that sends
+a subscribe msg upstream. The streamer on the upstream peer
+handles the subscribe msg by installing the relevant outgoing streamer
+. The modules now engage in a process of upstream sending a sequence of hashes of
+chunks downstream (OfferedHashesMsg). The downstream peer evaluates which hashes are needed
+and get it delivered by sending back a msg (WantedHashesMsg).
+
+Historical syncing is supported - currently not the right abstraction --
+state kept across sessions by saving a series of intervals after their last
+batch actually arrived.
+
+Live streaming is also supported, by starting session from the first item
+after the subscription.
+
+Provable data exchange. In case a stream represents a swarm document's data layer
+or higher level chunks, streaming up to a certain index is always provable. It saves on
+sending intermediate chunks.
+
+Using the streamer logic, various stream types are easy to implement:
+
+* light node requests:
+  * url lookup with offset
+  * document download
+  * document upload
+* syncing
+  * live session syncing
+  * historical syncing
+* simple retrieve requests and deliveries
+* mutable resource updates streams
+* receipting for finger pointing
+
+## Syncing
+
+Syncing is the process that makes sure storer nodes end up storing all and only the chunks that are requested from them.
+
+### Requirements
 
 - eventual consistency: so each chunk historical should be syncable
 - since the same chunk can and will arrive from many peers, (network traffic should be
-optimised (only one transfer of data per chunk)
+optimised, only one transfer of data per chunk)
 - explicit request deliveries should be prioritised higher than recent chunks received
 during the ongoing session which in turn should be higher than historical chunks.
 - insured chunks should get receipted for finger pointing litigation, the receipts storage
 should be organised efficiently, upstream peer should also be able to find these
 receipts for a deleted chunk easily to refute their challenge.
 - syncing should be resilient to cut connections, metadata should be persisted that
-keep track of syncing state across sessions.
+keep track of syncing state across sessions, historical syncing state should survive restart
 - extra data structures to support syncing should be kept at minimum
 - syncing is organized separately for chunk types (resource update v content chunk)
+- various types of streams should have common logic abstracted
 
-
+Syncing is now entirely mediated by the localstore, ie., no processes or memory leaks due to network contention.
-When two peers connect, the bidirectional protocol is a result of two identical
+When a new chunk is stored, its chunk hash is index by proximity bin
-syncing protocols mirrored. So take one direction and call the two parties
-upstream and downstream peer.
-
-when a new chunk is stored, its hash is appended to a boundless stream maintained for
-each kademlia bin. This state is always permanently recorded periodically possibly
-using mutable resource update scheme.
-
-At any point in time (with n chunks in total ) the swarm hash of this hash stream state is
-well defined. Upstream peer pushes so that the reader reads sequential hashes and
-periodically calculates the swarm hash of their stream.
 
 peers syncronise by getting the chunks closer to the downstream peer than to the upstream one.
 Consequently peers just sync all stored items for the kad bin the receiving peer falls into.
@@ -32,24 +69,14 @@ The special case of nearest neighbour sets is handled by the downstream peer
 indicating they want to sync all kademlia bins with proximity equal to or higher
 than their depth.
 
-When peers connect upstream peer sends the latest sync state (item index/cursor length)
-for each relevant bin downstream peer is interested in.
 This sync state represents the initial state of a sync connection session.
-Conversely downstream peers maintain the last state (swarm hash with length) which
+Retrieval is dictated by downstream peers simply using a special streamer protocol.
-the ranges of covered offsets.
 
-Retrieval is dictated by downstream peers simply using the the chunker joiner to read certain offsets.
+Syncing chunks created during the session by the upstream peer is called live session syncing
-
-Syncing chunks created during the session by the upstream peer is called session Syncing
 while syncing of earlier chunks is historical syncing.
 
-Historical syncing is simply carried out by iteratively requesting ranges of hash offsets.
-For simplicity we assume that the minimum unit requested is a chunk. If every 128 chunks  
-is considered syncable one can use data chunk index instead of offset in byte length.
-Note that data chunk here is a sequence of hashes above one ground level of stream of content.
-
 Once the relevant chunk is retrieved, downstream peer looks up all hash segments in its localstore
-and sends to the upstream peer a message with a 128-long bitvector (uint16) to indicate
+and sends to the upstream peer a message with a a bitvector to indicate
 missing chunks (e.g., for chunk `k`, hash with chunk internal index which case )
 new items. In turn upstream peer sends the relevant chunk data alongside their index.
 
@@ -64,20 +91,35 @@ Session syncing involves downstream peer to request a new state on a bin from up
 using the new state, the range (of chunks) between the previous state and the new one are retrieved
 and chunks are requested identical to the historical case. After receiving all the missing chunks
 from the new hashes, downstream peer will request a new range. If this happens before upstream peer updates a new state,
-we say that session syncing is live or the two peers are in sync.
+we say that session syncing is live or the two peers are in sync. In general the time interval passed since downstream peer request up to the current session cursor is a good indication of a permanent (probably increasing) lag.
 
-If there is no historical backlog, downstream peer is said to be fully synced with the upstream.
+If there is no historical backlog, and downstream peer has an acceptable 'last synced' tag, then it is said to be fully synced with the upstream peer.
 If a peer is fully synced with all its storer peers, it can advertise itself as globally fully synced.
 
-For healthy operation, however, it is expected that the session is regularly in sync. If this is
-not the case, that indicates that traffic during the session is continuously more than the peer can cope with and
-downstream peer is effectively accumulating a historical backlog.
-
 The downstream peer persists the record of the last synced offset. When the two peers disconnect and
 reconnect syncing can start from there.
 This situation however can also happen while historical syncing is not yet complete.
 Effectively this means that the peer needs to persist a record of an arbitrary array of offset ranges covered.
 
+### Delivery requests
+
+once the appropriate ranges of the hashstream are retrieved and buffered, downstream peer just scans the hashes, looks them up in localstore, if not found, create a request entry.
+The range is referenced by the chunk index. Alongside the name (indicating the stream, e.g., content chunks for bin 6) and the range
+downstream peer sends a 128 long bitvector indicating which chunks are needed.
+Newly created requests are satisfied bound together in a waitgroup which when done, will promptt sending the next one.
+to be able to do check and storage concurrently, we keep a buffer of one, we start with two batches of hashes.
+If there is nothing to give, upstream peers SetNextBatch is blocking. Subscription ends with an unsubscribe. which removes the syncer from the map.
+
+Canceling requests (for instance the late chunks of an erasure batch) should be a chan closed
+on the request
+
+Simple request is also a subscribe
+different streaming protocols are different p2p protocols with same message types.
+the constructor is the Run function itself. which takes a streamerpeer as argument
+
+
+### provable streams
+
 The swarm  hash over the hash stream has many advantages. It implements a provable data transfer
 and provide efficient storage for receipts in the form of inclusion proofs useable for finger pointing litigation.
 When challenged on a missing chunk, upstream peer will provide an inclusion proof of a chunk hash against the state of the
@@ -91,19 +133,16 @@ As part of the deletion protocol then, hashes of insured chunks to be removed ar
 Downstream peer on the other hand needs to make sure that they can only be finger pointed about a chunk they did receive and store.
 For this the check of a state should be exhaustive. If historical syncing finishes on one state, all hashes before are covered, no
 surprises. In other words historical syncing this process is self verifying. With session syncing however, it is not enough to check going back covering the range from old offset to new. Continuity (i.e., that the new state is extension of the old) needs to be verified: after downstream peer reads the range into a buffer, it appends the buffer the last known state at the last known offset and verifies the resulting hash matches
-the latest state. The same goes with intervals of historical syncing where the state used to verify the preceding interval is different from the one used to cover the current range. In these cases too, verification by append is needed for complete security for downstream peer.
+the latest state. Past intervals of historical syncing are checked via the the session root.
-
 Upstream peer signs the states, downstream peers can use as handover proofs.
 Downstream  peers sign off on a state together with an initial offset.
-latter needed for reasonable sized not ever growing syncer
-possible is eachn chunk needs to be reentered if they remain insured
 
 Once historical syncing is complete and the session does not lag, downstream peer only preserves the latest upstream state and store the signed version.
 
 Upstream peer needs to keep the latest takeover states: each deleted chunk's hash should be covered by takeover proof of at least one peer. If historical syncing is complete, upstream peer typically will store only the latest takeover proof from downstream peer.
 Crucially, the structure is totally independent of the number of peers in the bin, so it scales extremely well.
 
-implementation
+## implementation
 
 The simplest protocol just involves upstream peer to prefix the key with the kademlia proximity order (say 0-15 or 0-31)
 and simply iterate on index per bin when syncing with a peer.
@@ -111,40 +150,3 @@ and simply iterate on index per bin when syncing with a peer.
 priority queues are used for sending chunks so that user triggered requests should be responded to first, session syncing second, and historical with lower priority.
 The request on chunks remains implemented as a dataless entry in the memory store.
 The lifecycle of this object should be more carefully thought through, ie., when it fails to retrieve it should be removed.
-
-
-
-Model 1
-The main appeal in this model is that downstream driven syncing falls back to the exact same retrieval mechanism as the one used when downloading a file. If the chunks of the hash stream (datachunks as well as intermediate chunks of the swarm tree above it) are themselves distributed by upstream peer in the normal way, then requesting them from swarm is viable.
-This requires no extra implementation. However, it is unlikely this is feasible for live syncing since the chunks' delay to arrive at their destination has a lag exactly due to session syncronisation.
-Note that if upstream peer handles chunks of the hash stream as normal chunks, there are issues. One is that some of these chunks will fall in the same bin as the one building leading to a situation where the hash stream grows even though there are no external chunks received. If upstream peer wishes to use finger pointing proofs, it has to either store these chunks themselves or insure them.
-
-
-Model 2
-In this model, when retrieving the hash stream from the state, requests are targeted to the upstream peer.
-The simplest way to generate the right requests from a sync state is to have a  
-peer specific dpa chunkstore for chunker join. This can be used by all chunk types and all bins.
-All it does is when picking retrieve tasks off the chunk channel, it marks the chunk with a reference to the
-upstream peer so that when netstore does not find it, the request is sent to the upstream peer (only).
-(Normally the request would be routed based on its address).
-We need to introduce a special field on the chunk to indicate that the data should be requested from a particular peer.
-Upstream peer could still distribute these chunks used in the hash stream in swarm as usual, e.g., long non-synced early
-history.
-This model does not suffer from the availability lag of the first one, correctly puts the burden on upstream peer to preserve
-chunks of the hash stream either in swarm or not.
-
-Model 3
-in another alternative upstream peer sends only the data level (the hash stream interval) not all the intermediate chunks. This saves on traffic and downstream peer can calculate the state by append to verify against the state (root hash).
-This mode of operation is anyhow feasible in cases where the same peer having the top request will be expected to have all the children chunks
-of an intermediate one. This is the case for syncing and hash stream or when light swarm clients channel all their requests to a proxy node (public database lookup or unencrypted content).
-
-This model would require no peer specific dpa and would involve the chunker split only to get the right hand side for the append for verification.
-
-Delivery requests
-
-once the appropriate ranges of the hashstream are retrieved and buffered, downstream peer just scans the hashes, looks them up in localstore, if not found, create a request entry with specific reference to the upstream peer as source.
-The range is referenced by the chunk index. Alongside the name (indicating the stream, e.g., content chunks for bin 6) and the range
-downstream peer sends a 128 long bitvector indicating which chunks are needed.
-Newly created requests are satisfied bound together in a waitgroup which when done, will prompt sending the next one.
-to be able to do check and storage concurrently, we keep a buffer of one, we start with two chunks.
-For session syncing too, if it has not arrived  sby the time the next chunk.

swarm/network/lightnode.go

@@ -50,14 +50,15 @@ func (r *RemoteSectionReader) NeedData(key []byte) func() {
 	if chunk.ReqC == nil || !created {
 		return nil
 	}
-	return func() {}
+	return func() {
+		select {
+		case <-chunk.ReqC:
+		case <-r.quit:
+		}
+	}
 }
 
-func (r *RemoteSectionReader) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
+func (r *RemoteSectionReader) BatchDone(s string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
-	return from, r.end
-}
-
-func (r *RemoteSectionReader) BatchDone(s Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
 	r.hashes <- hashes
 	return nil
 }
@@ -113,13 +114,11 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
 			hashes = hashes[i:]
 		}
 	}
-	return n, nil
 }
 
 // RemoteSectionServer implements OutgoingStreamer
 type RemoteSectionServer struct {
 	// quit chan struct{}
-	currentBatch []byte
 	root []byte
 	db   *DbAccess
 	r    *storage.LazyChunkReader
@@ -149,14 +148,12 @@ func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uin
 	}
 	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("REMOTE_SECTION", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-	s.RegisterIncomingStreamer(name, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
 		return NewRemoteSectionReader(t, db), nil
 	})
 }
@@ -164,8 +161,7 @@ func RegisterRemoteSectionReader(s *Streamer, db *DbAccess) {
 // 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("REMOTE_SECTION", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
-	s.RegisterOutgoingStreamer(name, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
 		r := rf(t)
 		return NewRemoteSectionServer(db, r), nil
 	})
@@ -174,8 +170,7 @@ func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *sto
 // RegisterRemoteDownloader registers RemoteDownloader incoming streamer
 // on downstream light  node
 func RegisterRemoteDownloader(s *Streamer, db *DbAccess) {
-	name := Stream("REMOTE_DOWNLOADER")
+	s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-	s.RegisterIncomingStreamer(name, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
 		return NewRemoteDownloader(t, db), nil
 	})
 }
@@ -183,9 +178,10 @@ func RegisterRemoteDownloader(s *Streamer, db *DbAccess) {
 // 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("REMOTE_DOWNLOADER", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
-	s.RegisterOutgoingStreamer(name, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
 		r := rf(t)
 		return NewRemoteDownloadServer(db, r), nil
 	})
 }
+
+func NewRemoteDownloader()

swarm/network/requests.go

@@ -16,122 +16,93 @@
 
 package network
 
-// import (
+import "github.com/ethereum/go-ethereum/swarm/storage"
-// 	"fmt"
 
-// 	"github.com/ethereum/go-ethereum/log"
+const retrieveRequestStream = "RETRIEVE_REQUEST"
-// 	"github.com/ethereum/go-ethereum/swarm/storage"
-// )
 
-// /*
+// Intervals is a stream specific history of downloaded intervals
-//  Retrieve Request and store Request handling
+// for historical streams
-// */
+type Intervals struct {
+	streamer *Streamer
+	key      string
+}
 
-// // Handler for storage/retrieval related protocol requests
+func (s *Intervals) load() error {
-// type RequestHandler struct {
+	return s.streamer.load(s.key)
-// 	netStore *storage.NetStore
+}
-// }
 
-// // NewEwquestHandler creates a new RequestHandler
+func (s *Intervals) save() error {
-// // netStore to
+	return s.streamer.save(s.key)
-// func NewRequestHandler(netStore *storage.NetStore) *RequestHandler {
+}
-// 	return &RequestHandler{
-// 		netStore: netStore, // entrypoint internal
-// 	}
-// }
 
-// /*
+func (s *Intervals) get() []uint64 {
-// Retrieve request
+	return s.streamer.get(s.key)
+}
 
-// MaxSize specifies the maximum size that the peer will accept. This is useful in
+func (s *Intervals) set(v []uint64) {
-// particular if we allow storage and delivery of multichunk payload representing
+	s.streamer.set(s.key, v)
-// the entire or partial subtree unfolding from the requested root key.
+}
-// So when only interested in limited part of a stream (infinite trees) or only
-// testing chunk availability etc etc, we can indicate it by limiting the size here.
 
-// Request ID can be newly generated or kept from the request originator.
+func NewIntervals(key string, s *Streamer) *Intervals {
+	return &Intervals{
+		streamer: s,
+		key:      key,
+	}
+}
 
-// */
+// RetrieveRequestStreamer implements OutgoingStreamer
-// type retrieveRequestMsg struct {
+type RetrieveRequestStreamer struct {
-// 	Key     storage.Key   // target Key address of chunk to be retrieved
+	deliveryC  chan *storage.Chunk
-// 	Id      uint64        // request id, request is a lookup if missing or zero
+	batchC     chan []byte
-// 	MaxSize uint64        // maximum size of delivery accepted
+	db         *DbAccess
-// 	from    *StreamerPeer //
+	currentLen uint64
-// }
+}
 
-// func (self retrieveRequestMsg) String() string {
+// RegisterRequestStreamer registers outgoing and incoming streamers for request handling
-// 	var from string
+func RegisterRequestStreamer(streamer *Streamer, db *DbAccess) {
-// 	if self.from == nil {
+	streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer, t []byte) (OutgoingStreamer, error) {
-// 		from = "ourselves"
+		return NewRetrieveRequestStreamer(db), nil
-// 	} else {
+	})
-// 		from = fmt.Sprintf("%x", self.from.Over())
+	streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-// 	}
+		return NewIncomingSwarmSyncer(p, db, nil)
-// 	var target []byte
+	})
-// 	if len(self.Key) > 3 {
+}
-// 		target = self.Key[:4]
-// 	}
-// 	return fmt.Sprintf("Requester: %v, Key: %x; ID: %v, MaxSize: %v", from, target, self.Id, self.MaxSize)
-// }
 
-// entrypoint for retrieve requests coming from the bzz wire protocol
+// NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor
-// checks swap balance - return if peer has no credit
+func NewRetrieveRequestStreamer(db *DbAccess) *RetrieveRequestStreamer {
-// func (self *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error {
+	s := &RetrieveRequestStreamer{
-// 	req := msg.(*retrieveRequestMsg)
+		deliveryC: make(chan *storage.Chunk),
-// 	req.from = self
+		batchC:    make(chan []byte),
-// 	// TODO:
+		db:        db,
-// 	// swap - record credit for 1 request
+	}
-// 	// note that only charge actual reqsearches
+	go s.processDeliveries()
+	return s
+}
 
-// 	// call storage.NetStore#Get which
+// processDeliveries handles delivered chunk hashes
-// 	// blocks until local retrieval finished
+func (s *RetrieveRequestStreamer) processDeliveries() {
-// 	// launches cloud retrieval
+	var hashes []byte
-// 	chunk, _ := self.netStore.Get(req.Key)
+	for {
-// 	rs := chunk.Req
+		select {
-// 	if rs != nil {
+		case delivery := <-s.deliveryC:
-// 		rs = storage.NewRequestStatus(req.Key)
+			hashes = append(hashes, delivery.Key[:]...)
-// 		addRequester(rs, req)
+		case s.batchC <- hashes:
-// 		chunk.Req = rs
+			hashes = nil
-// 	}
+		}
+	}
+}
 
-// 	// check if we can immediately deliver
+// SetNextBatch
-// 	if chunk.SData != nil {
+func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) {
-// 		if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size {
+	hashes = <-s.batchC
-// 			err := self.Deliver(chunk, Top)
+	from = s.currentLen
-// 			if err != nil {
+	s.currentLen += uint64(len(hashes))
-// 				log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err))
+	to = s.currentLen
-// 				return nil
+	return
-// 			}
+}
-// 			log.Trace(fmt.Sprintf("%v - content found, delivering...", req.Key.Log()))
-// 		} else {
-// 			log.Trace(fmt.Sprintf("%v - content found, not wanted", req.Key.Log()))
-// 		}
-// 	} else {
-// 		log.Trace(fmt.Sprintf("content not found locally, retrieve via bzz", req.Key.Log()))
-// 	}
-// 	return nil
-// }
 
-// /*
+// GetData retrives chunk data from db store
-// adds a new peer to an existing open request
+func (s *RetrieveRequestStreamer) GetData(key []byte) []byte {
-// only add if less than requesterCount peers forwarded the same request id so far
+	chunk, _ := s.db.get(storage.Key(key))
-// note this is done irrespective of status (searching or found)
+	return chunk.SData
-// */
+}
-// func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) {
-// 	log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id))
-// 	list := rs.Requesters[req.Id]
-// 	rs.Requesters[req.Id] = append(list, req)
-// }
-
-// func (self storeRequestMsg) String() string {
-// 	var from string
-// 	if self.from == nil {
-// 		from = "self"
-// 	} else {
-// 		from = fmt.Sprintf("%x", self.from.Over())
-// 	}
-// 	end := len(self.SData)
-// 	if len(self.SData) > 10 {
-// 		end = 10
-// 	}
-// 	return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end])
-// }

swarm/network/streamer.go

@@ -42,12 +42,9 @@ const (
 	PriorityQueueCap = 3 // queue capacity
 )
 
-// Stream is string descriptor of the stream
-type Stream string
-
 // Handover represents a statement that the upstream peer hands over the stream section
 type Handover struct {
-	Stream     Stream // name of stream
+	Stream     string // name of stream
 	Start, End uint64 // index of hashes
 	Root       []byte // Root hash for indexed segment inclusion proofs
 }
@@ -79,7 +76,7 @@ func (self TakeoverProofMsg) String() string {
 
 // SubcribeMsg is the protocol msg for requesting a stream(section)
 type SubscribeMsg struct {
-	Stream   Stream
+	Stream   string
 	Key      []byte
 	From, To uint64
 	Priority uint8 // delivered on priority channel
@@ -88,7 +85,7 @@ type SubscribeMsg struct {
 // UnsyncedKeysMsg is the protocol msg for offering to hand over a
 // stream section
 type UnsyncedKeysMsg struct {
-	Stream         Stream // name of Stream
+	Stream         string // name of Stream
 	Key            []byte // subtype or key
 	From, To       uint64 // peer and db-specific entry count
 	Hashes         []byte // stream of hashes (128)
@@ -115,7 +112,7 @@ func (self UnsyncedKeysMsg) String() string {
 // WantedKeysMsg is the protocol msg data for signaling which hashes
 // offered in UnsyncedKeysMsg downstream peer actually wants sent over
 type WantedKeysMsg struct {
-	Stream   Stream // name of stream
+	Stream   string // 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
@@ -130,8 +127,8 @@ func (self WantedKeysMsg) String() string {
 type Streamer struct {
 	incomingLock sync.RWMutex
 	outgoingLock sync.RWMutex
-	outgoing     map[Stream]func(*StreamerPeer, []byte) (OutgoingStreamer, error)
+	outgoing     map[string]func(*StreamerPeer, []byte) (OutgoingStreamer, error)
-	incoming     map[Stream]func(*StreamerPeer, []byte) (IncomingStreamer, error)
+	incoming     map[string]func(*StreamerPeer, []byte) (IncomingStreamer, error)
 
 	dbAccess *DbAccess
 	overlay  Overlay
@@ -141,8 +138,8 @@ type Streamer struct {
 // NewStreamer is Streamer constructor
 func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer {
 	return &Streamer{
-		outgoing: make(map[Stream]func(*StreamerPeer, []byte) (OutgoingStreamer, error)),
+		outgoing: make(map[string]func(*StreamerPeer, []byte) (OutgoingStreamer, error)),
-		incoming: make(map[Stream]func(*StreamerPeer, []byte) (IncomingStreamer, error)),
+		incoming: make(map[string]func(*StreamerPeer, []byte) (IncomingStreamer, error)),
 		dbAccess: dbAccess,
 		overlay:  overlay,
 		receiveC: make(chan *ChunkDeliveryMsg, 10),
@@ -150,21 +147,21 @@ func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer {
 }
 
 // RegisterIncomingStreamer registers an incoming streamer constructor
-func (self *Streamer) RegisterIncomingStreamer(stream Stream, f func(*StreamerPeer, []byte) (IncomingStreamer, error)) {
+func (self *Streamer) RegisterIncomingStreamer(stream string, 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, []byte) (OutgoingStreamer, error)) {
+func (self *Streamer) RegisterOutgoingStreamer(stream string, 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, []byte) (IncomingStreamer, error), error) {
+func (self *Streamer) GetIncomingStreamer(stream string) (func(*StreamerPeer, []byte) (IncomingStreamer, error), error) {
 	self.incomingLock.RLock()
 	defer self.incomingLock.RUnlock()
 	f := self.incoming[stream]
@@ -175,7 +172,7 @@ func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer, []
 }
 
 // GetOutgoingStreamer accessor for incoming streamer constructors
-func (self *Streamer) GetOutgoingStreamer(stream Stream) (func(*StreamerPeer, []byte) (OutgoingStreamer, error), error) {
+func (self *Streamer) GetOutgoingStreamer(stream string) (func(*StreamerPeer, []byte) (OutgoingStreamer, error), error) {
 	self.outgoingLock.RLock()
 	defer self.outgoingLock.RUnlock()
 	f := self.outgoing[stream]
@@ -208,15 +205,17 @@ type OutgoingStreamer interface {
 type incomingStreamer struct {
 	IncomingStreamer
 	priority  uint8
+	intervals *Intervals
+	sessionAt uint64
+	live      bool
 	quit      chan struct{}
 	next      chan struct{}
 }
 
 // IncomingStreamer interface for incoming peer Streamer
 type IncomingStreamer interface {
-	NextBatch(uint64) (uint64, uint64)
 	NeedData([]byte) func()
-	BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error)
+	BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error)
 }
 
 // StreamerPeer is the Peer extention for the streaming protocol
@@ -228,28 +227,18 @@ type StreamerPeer struct {
 	dbAccess     *DbAccess
 	outgoingLock sync.RWMutex
 	incomingLock sync.RWMutex
-	outgoing     map[Stream]*outgoingStreamer
+	outgoing     map[string]*outgoingStreamer
-	incoming     map[Stream]*incomingStreamer
+	incoming     map[string]*incomingStreamer
 	quit         chan struct{}
 }
 
-// type IncomingStreamer struct {
-// 	priority uint8
-// 	peer     *StreamerPeer
-// }
-
-// type OutgoingStreamer struct {
-// 	priority uint8
-// 	peer     *StreamerPeer
-// }
-
 // NewStreamerPeer is the constructor for StreamerPeer
 func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
 	self := &StreamerPeer{
 		pq:       pq.New(int(PriorityQueue), PriorityQueueCap),
 		streamer: streamer,
-		outgoing: make(map[Stream]*outgoingStreamer),
+		outgoing: make(map[string]*outgoingStreamer),
-		incoming: make(map[Stream]*incomingStreamer),
+		incoming: make(map[string]*incomingStreamer),
 		quit:     make(chan struct{}),
 	}
 	ctx, cancel := context.WithCancel(context.Background())
@@ -266,65 +255,6 @@ type RetrieveRequestMsg struct {
 	Key storage.Key
 }
 
-// RetrieveRequestStreamer implements OutgoingStreamer
-type RetrieveRequestStreamer struct {
-	deliveryC  chan *storage.Chunk
-	batchC     chan []byte
-	db         *DbAccess
-	currentLen uint64
-}
-
-// RegisterRequestStreamer registers outgoing and incoming streamers for request handling
-func RegisterRequestStreamer(streamer *Streamer, db *DbAccess) {
-	streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer, t []byte) (OutgoingStreamer, error) {
-		return NewRetrieveRequestStreamer(db), nil
-	})
-	streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-		return NewIncomingSwarmSyncer(nil, p, db, nil)
-	})
-}
-
-// NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor
-func NewRetrieveRequestStreamer(db *DbAccess) *RetrieveRequestStreamer {
-	s := &RetrieveRequestStreamer{
-		deliveryC: make(chan *storage.Chunk),
-		batchC:    make(chan []byte),
-		db:        db,
-	}
-	go s.processDeliveries()
-	return s
-}
-
-// processDeliveries handles delivered chunk hashes
-func (s *RetrieveRequestStreamer) processDeliveries() {
-	var hashes []byte
-	for {
-		select {
-		case delivery := <-s.deliveryC:
-			hashes = append(hashes, delivery.Key[:]...)
-		case s.batchC <- hashes:
-			hashes = nil
-		}
-	}
-}
-
-// SetNextBatch
-func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) {
-	hashes = <-s.batchC
-	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.db.get(storage.Key(key))
-	return chunk.SData
-}
-
-const retrieveRequestStream = Stream("RETRIEVE_REQUEST")
-
 func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) error {
 	chunk, created := self.dbAccess.getOrCreateRequest(req.Key)
 	s, err := self.getOutgoingStreamer(retrieveRequestStream)
@@ -399,7 +329,7 @@ func (self *Streamer) processReceivedChunks() {
 	}
 }
 
-func (self *StreamerPeer) getOutgoingStreamer(s Stream) (*outgoingStreamer, error) {
+func (self *StreamerPeer) getOutgoingStreamer(s string) (*outgoingStreamer, error) {
 	self.outgoingLock.RLock()
 	defer self.outgoingLock.RUnlock()
 	streamer := self.outgoing[s]
@@ -409,7 +339,7 @@ func (self *StreamerPeer) getOutgoingStreamer(s Stream) (*outgoingStreamer, erro
 	return streamer, nil
 }
 
-func (self *StreamerPeer) getIncomingStreamer(s Stream) (*incomingStreamer, error) {
+func (self *StreamerPeer) getIncomingStreamer(s string) (*incomingStreamer, error) {
 	self.incomingLock.RLock()
 	defer self.incomingLock.RUnlock()
 	streamer := self.incoming[s]
@@ -419,7 +349,7 @@ func (self *StreamerPeer) getIncomingStreamer(s Stream) (*incomingStreamer, erro
 	return streamer, nil
 }
 
-func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer, priority uint8) (*outgoingStreamer, error) {
+func (self *StreamerPeer) setOutgoingStreamer(s string, o OutgoingStreamer, priority uint8) (*outgoingStreamer, error) {
 	self.outgoingLock.Lock()
 	defer self.outgoingLock.Unlock()
 	if self.outgoing[s] != nil {
@@ -433,15 +363,22 @@ func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer, prio
 	return os, nil
 }
 
-func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer, priority uint8) error {
+func (self *StreamerPeer) setIncomingStreamer(s string, i IncomingStreamer, priority uint8, live bool) error {
 	self.incomingLock.Lock()
 	defer self.incomingLock.Unlock()
 	if self.incoming[s] != nil {
 		return fmt.Errorf("stream %v already registered", s)
 	}
 	next := make(chan struct{}, 1)
+	var intervals *Intervals
+	if !live {
+		key := s + self.ID().String()
+		intervals = NewIntervals(key, self.streamer)
+	}
 	self.incoming[s] = &incomingStreamer{
 		IncomingStreamer: i,
+		intervals:        intervals,
+		live:             live,
 		priority:         priority,
 		next:             next,
 	}
@@ -449,8 +386,37 @@ func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer, prio
 	return nil
 }
 
+// NextBatch adjusts the indexes by inspecting the intervals
+func (self *incomingStreamer) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
+	intervals := self.intervals.get()
+	if self.live {
+		if len(intervals) == 0 {
+			intervals = []uint64{self.sessionAt, from}
+		} else {
+			intervals[1] = from
+		}
+		nextFrom = from
+	} else if from >= self.sessionAt { // history sync complete
+		intervals = nil
+	} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
+		intervals = append(intervals[:1], intervals[3:]...)
+		nextFrom = intervals[1]
+		if len(intervals) > 2 {
+			nextTo = intervals[2]
+		} else {
+			nextTo = self.sessionAt
+		}
+	} else {
+		nextFrom = from
+		intervals[1] = from
+		nextTo = self.sessionAt
+	}
+	self.intervals.set(intervals)
+	return nextFrom, nextTo
+}
+
 // Subscribe initiates the streamer
-func (self *StreamerPeer) Subscribe(s Stream, t []byte, from, to uint64, priority uint8) error {
+func (self *StreamerPeer) Subscribe(s string, t []byte, from, to uint64, priority uint8, live bool) error {
 	f, err := self.streamer.GetIncomingStreamer(s)
 	if err != nil {
 		return err
@@ -459,7 +425,7 @@ func (self *StreamerPeer) Subscribe(s Stream, t []byte, from, to uint64, priorit
 	if err != nil {
 		return err
 	}
-	err = self.setIncomingStreamer(s, is, priority)
+	err = self.setIncomingStreamer(s, is, priority, live)
 	if err != nil {
 		return err
 	}
@@ -484,8 +450,8 @@ func (self *StreamerPeer) handleSubscribeMsg(req *SubscribeMsg) error {
 	if err != nil {
 		return err
 	}
-	key := string(req.Stream) + string(req.Key)
+	key := req.Stream + string(req.Key)
-	os, err := self.setOutgoingStreamer(Stream(key), s, req.Priority)
+	os, err := self.setOutgoingStreamer(key, s, req.Priority)
 	if err != nil {
 		return nil
 	}
@@ -531,7 +497,10 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error {
 	}()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
-	from, to := s.NextBatch(req.To)
+	if s.live {
+		s.sessionAt = req.From
+	}
+	from, to := s.nextBatch(req.To)
 	if from == to {
 		return nil
 	}
@@ -644,11 +613,11 @@ func (s *Streamer) Run(p *bzzPeer) error {
 	// load saved intervals
 	// autosubscribe to request handler to serve request only for non-light nodes
 	// sp.handleSubscribeMsg(&SubscribeMsg{
-	// 	Stream:   retrieveRequestStream,
+	// 	Stream:   retrieveRequeststring,
 	// 	Priority: uint8(Top),
 	// })
 	// subscribe to request handling ; only with non-light nodes
-	sp.Subscribe(retrieveRequestStream, nil, 0, 0, Top)
+	sp.Subscribe(retrieveRequestStream, nil, 0, 0, Top, true)
 	defer close(sp.quit)
 	return sp.Run(sp.HandleMsg)
 }

swarm/network/syncer.go

@@ -95,8 +95,7 @@ func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSy
 const maxPO = 32
 
 func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) {
-	stream := Stream("SYNC")
+	streamer.RegisterOutgoingStreamer("SYNC", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
-	streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
 		syncType, po := parseSyncLabel(t)
 		switch syncType {
 		case "LIVE":
@@ -107,7 +106,6 @@ func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) {
 			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)
 	// })
@@ -146,7 +144,6 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64,
 type IncomingSwarmSyncer struct {
 	sessionAt     uint64
 	nextC         chan struct{}
-	intervals     *Intervals
 	sessionRoot   storage.Key
 	sessionReader storage.LazySectionReader
 	retrieveC     chan *storage.Chunk
@@ -159,9 +156,8 @@ type IncomingSwarmSyncer struct {
 }
 
 // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
-func NewIncomingSwarmSyncer(intervals *Intervals, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
+func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
 	self := &IncomingSwarmSyncer{
-		intervals: intervals,
 		dbAccess: dbAccess,
 		chunker:  chunker,
 	}
@@ -190,42 +186,6 @@ func NewIncomingSwarmSyncer(intervals *Intervals, p Peer, dbAccess *DbAccess, ch
 // 	return self
 // }
 
-type Syncer struct {
-	intervals map[string][]uint64
-}
-
-func NewSyncer() *Syncer {
-	return &Syncer{}
-}
-
-type Intervals struct {
-	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))
@@ -234,7 +194,7 @@ func newSyncLabel(typ string, po uint8) []byte {
 
 func parseSyncLabel(t []byte) (string, uint8) {
 	l := len(t) - 1
-	return sstring(t[:l]), uint8(t[l])
+	return string(t[:l]), uint8(t[l])
 }
 
 // StartSyncing is called on the StreamerPeer to start the syncing process
@@ -245,21 +205,19 @@ func StartSyncing(s *StreamerPeer, po uint8, nn bool) {
 		lastPO = maxPO
 	}
 	for i := po; i <= lastPO; i++ {
-		s.Subscribe(Stream("SYNC"), newSyncLabel("LIVE", po), 0, 0, High)
+		s.Subscribe("SYNC", newSyncLabel("LIVE", po), 0, 0, High, true)
-		s.Subscribe(Stream("SYNC"), newSyncLabel("HISTORY", po), 0, 0, Mid)
+		s.Subscribe("SYNC", newSyncLabel("HISTORY", po), 0, 0, Mid, false)
 	}
 }
 
-func RegisterIncomingSyncers(streamer *Streamer, syncer *Syncer, db *DbAccess) {
+func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
-	stream := Stream("SYNC")
+	streamer.RegisterIncomingStreamer("SYNC", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
-	streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
 		syncType, po := parseSyncLabel(t)
 		switch syncType {
 		case "LIVE":
-			return NewIncomingSwarmSyncer(nil, p, nil, nil)
+			return NewIncomingSwarmSyncer(p, nil, nil)
 		case "HISTORY":
-			intervals := syncer.NewIntervals(t)
+			return NewIncomingSwarmSyncer(p, nil, nil)
-			return NewIncomingSwarmSyncer(intervals, p, nil, nil)
 		}
 		return nil, fmt.Errorf("unknown sync type %q", syncType)
 	})
@@ -281,40 +239,15 @@ func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
 	return chunk.WaitToStore
 }
 
-// NextBatch adjusts the indexes by inspecting the intervals
-func (self *IncomingSwarmSyncer) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
-	intervals := self.intervals.get()
-	if intervals[0] >= self.sessionAt { // live syncing
-		nextFrom = from
-		intervals[1] = from
-	} else if from >= self.sessionAt { // history sync complete
-		intervals = nil
-	} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
-		intervals = append(intervals[:1], intervals[3:]...)
-		nextFrom = intervals[1]
-		if len(intervals) > 2 {
-			nextTo = intervals[2]
-		} else {
-			nextTo = self.sessionAt
-		}
-	} else {
-		nextFrom = 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) {
+func (self *IncomingSwarmSyncer) BatchDone(s string, 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) {
+func (self *IncomingSwarmSyncer) TakeoverProof(s string, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
 	// for provable syncer currentRoot is non-zero length
 	if self.chunker != nil {
 		if from > self.sessionAt { // for live syncing currentRoot is always updated