swarm/network: network rewrite basic skeleton

swarm/network/README.md

@@ -0,0 +1,150 @@
+# 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)
+- 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.
+- extra data structures to support syncing should be kept at minimum
+- syncing is organized separately for chunk types (resource update v content chunk)
+
+
+When two peers connect, the bidirectional protocol is a result of two identical
+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.
+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
+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 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
+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.
+
+On sending chunks there is a priority queue system. If during looking up hashes in its localstore,
+downstream peer hits on an open request then a retrieve request is sent immediately to the upstream peer indicating
+that no extra round of checks is needed. If another peers syncer hits the same open request, it is slightly unsafe to not ask
+that peer too: if the first one disconnects before delivering or fails to deliver and therefore gets
+disconnected, we should still be able to continue with the other. The minimum redundant traffic coming from such simultaneous
+eventualities should be sufficiently rare not to warrant more complex treatment.
+
+Session syncing involves downstream peer to request a new state on a bin from upstream.
+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.
+
+If there is no historical backlog, downstream peer is said to be fully synced with the upstream.
+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.
+
+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
+sync stream. In order to be able to generate such an inclusion proof, upstream peer needs to store the hash index (counting consecutive hash-size segments) alongside the chunk data and preserve it even when the chunk data is deleted until the chunk is no longer insured.
+if there is no valid insurance on the files the entry may be deleted.
+As long as the chunk is preserved, no takeover proof will be needed since the node can respond to any challenge.
+However, once the node needs to delete an insured chunk for capacity reasons, a receipt should be available to
+refute the challenge by finger pointing to a downstream peer.
+As part of the deletion protocol then, hashes of insured chunks to be removed are pushed to an infinite stream for every bin.
+
+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.
+
+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
+
+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.
+
+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/bitvector/bitvector.go

@@ -0,0 +1,34 @@
+package bitvector
+
+type BitVector struct {
+	len int
+	b   []byte
+}
+
+func New(l int) *BitVector {
+	return NewFromBytes(make([]byte, l/8+1), l)
+}
+
+func NewFromBytes(b []byte, l int) *BitVector {
+	return &BitVector{
+		len: l,
+		b:   b,
+	}
+}
+
+func (bv *BitVector) Get(i int) bool {
+	bi := i / 8
+	return uint8(bv.b[bi])&0x1>>uint(i%8) != 0
+}
+
+func (bv *BitVector) Set(i int, v bool) {
+	bi := i / 8
+	cv := bv.Get(i)
+	if cv != v {
+		bv.b[bi] ^= 0x1 >> uint8(i%8)
+	}
+}
+
+func (bv *BitVector) Bytes() []byte {
+	return bv.b
+}

swarm/network/priorityqueues/priorityqueues.go

@@ -0,0 +1,95 @@
+// package priority_queues implement a channel based priority queue
+// over arbitrary types. It provides an
+// an autopop loop applying a function to the items always respecting
+// their priority. The structure is only quasi consistent ie., if a lower
+// priority item is autopopped, it is guaranteed that there was a point
+// when no higher priority item was present, ie. it is not guaranteed
+// that there was any point where the lower priority item was present
+// but the higher was not
+
+package priorityqueues
+
+import (
+	"context"
+	"errors"
+)
+
+var (
+	errContention  = errors.New("queue contention")
+	errBadPriority = errors.New("bad priority")
+
+	wakey = struct{}{}
+)
+
+// PriorityQueues is the basic structure
+type PriorityQueues struct {
+	queues []chan interface{}
+	wakeup chan struct{}
+}
+
+// New is the constructor for PriorityQueues
+func New(n int, l int) *PriorityQueues {
+	var queues = make([]chan interface{}, n)
+	for i := range queues {
+		queues[i] = make(chan interface{}, l)
+	}
+	return &PriorityQueues{
+		queues: queues,
+		wakeup: make(chan struct{}, 1),
+	}
+}
+
+// Run is a forever loop popping items from the queues
+func (pq *PriorityQueues) Run(ctx context.Context, f func(interface{})) {
+	top := len(pq.queues) - 1
+	p := top
+	q := pq.queues[p]
+READ:
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case x := <-q:
+			f(x)
+			p = top
+		default:
+			if p > 0 {
+				p--
+				continue READ
+			}
+			p = top
+			select {
+			case <-ctx.Done():
+				return
+			case <-pq.wakeup:
+			}
+		}
+	}
+}
+
+// Push pushes an item to the appropriate queue specified in the priority argument
+// if context is given it waits until either the item is pushed or the Context aborts
+// otherwise returns errContention if the queue is full
+func (pq *PriorityQueues) Push(ctx context.Context, x interface{}, p int) error {
+	if p < 0 || p >= len(pq.queues) {
+		return errBadPriority
+	}
+	if ctx == nil {
+		select {
+		case pq.queues[p] <- x:
+		default:
+			return errContention
+		}
+	} else {
+		select {
+		case pq.queues[p] <- x:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	select {
+	case pq.wakeup <- wakey:
+	default:
+	}
+	return nil
+}

swarm/network/protocol.go

@@ -103,6 +103,7 @@ type BzzConfig struct {
 
 // Bzz is the swarm protocol bundle
 type Bzz struct {
+	Streamer *Streamer
 	*Hive
 	localAddr  *BzzAddr
 	mtx        sync.Mutex
@@ -116,6 +117,7 @@ type Bzz struct {
 // * peer store
 func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
 	return &Bzz{
+		Streamer:   NewStreamer(),
 		Hive:       NewHive(config.HiveParams, kad, store),
 		localAddr:  &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
 		handshakes: make(map[discover.NodeID]*HandshakeMsg),
@@ -157,6 +159,14 @@ func (b *Bzz) Protocols() []p2p.Protocol {
 			NodeInfo: b.Hive.NodeInfo,
 			PeerInfo: b.Hive.PeerInfo,
 		},
+		{
+			Name:     StreamerSpec.Name,
+			Version:  StreamerSpec.Version,
+			Length:   StreamerSpec.Length(),
+			Run:      b.RunProtocol(StreamerSpec, b.Streamer.Run),
+			NodeInfo: b.Streamer.NodeInfo,
+			PeerInfo: b.Streamer.PeerInfo,
+		},
 	}
 }
 

swarm/network/requests.go

@@ -0,0 +1,166 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// 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 (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+/*
+ Retrieve Request and store Request handling
+*/
+
+// Handler for storage/retrieval related protocol requests
+type RequestHandler struct {
+	netStore *storage.NetStore
+}
+
+// NewEwquestHandler creates a new RequestHandler
+// netStore to
+func NewRequestHandler(netStore *storage.NetStore) *RequestHandler {
+	return &RequestHandler{
+		netStore: netStore, // entrypoint internal
+	}
+}
+
+/*
+Retrieve request
+
+MaxSize specifies the maximum size that the peer will accept. This is useful in
+particular if we allow storage and delivery of multichunk payload representing
+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.
+
+*/
+type retrieveRequestMsgData struct {
+	Key     storage.Key // target Key address of chunk to be retrieved
+	Id      uint64      // request id, request is a lookup if missing or zero
+	MaxSize uint64      // maximum size of delivery accepted
+	from    Peer        //
+}
+
+func (self retrieveRequestMsgData) String() string {
+	var from string
+	if self.from == nil {
+		from = "ourselves"
+	} else {
+		from = fmt.Sprintf("%x", self.from.OverlayAddr())
+	}
+	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
+// checks swap balance - return if peer has no credit
+func (self *RequestHandler) handleRetrieveRequestMsg(msg interface{}, p Peer) error {
+	req := msg.(*retrieveRequestMsgData)
+	req.from = p
+	// TODO:
+	// swap - record credit for 1 request
+	// note that only charge actual reqsearches
+
+	// call storage.NetStore#Get which
+	// blocks until local retrieval finished
+	// launches cloud retrieval
+	chunk, _ := self.netStore.Get(req.Key)
+	rs := chunk.Req
+	if rs != nil {
+		rs = storage.NewRequest()
+		self.addRequester(rs, req)
+		chunk.Req = rs
+	}
+
+	// check if we can immediately deliver
+	if chunk.SData != nil {
+		if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size {
+			err := self.netStore.Deliver(chunk)
+			if err != nil {
+				log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err))
+				return nil
+			}
+			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
+}
+
+/*
+adds a new peer to an existing open request
+only add if less than requesterCount peers forwarded the same request id so far
+note this is done irrespective of status (searching or found)
+*/
+func (self *RequestHandler) addRequester(rs *storage.RequestStatus, req *retrieveRequestMsgData) {
+	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)
+}
+
+/*
+ store requests are put in netstore so they are stored and then
+ forwarded to the peers in their kademlia proximity bin by the syncer
+*/
+type storeRequestMsgData struct {
+	SData []byte // the stored chunk Data (incl size)
+	// optional
+	Id   uint64 // request ID. if delivery, the ID is retrieve request ID
+	from Peer   // [not serialised] protocol registers the requester
+}
+
+func (self storeRequestMsgData) String() string {
+	var from string
+	if self.from == nil {
+		from = "self"
+	} else {
+		from = fmt.Sprintf("%x", self.from.OverlayAddr())
+	}
+	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])
+}
+
+// the entrypoint for store requests coming from the bzz wire protocol
+// if key found locally, return. otherwise
+// remote is untrusted, so hash is verified and chunk passed on to NetStore
+func (self *RequestHandler) handleStoreRequestMsg(msg interface{}, p Peer) error {
+	req := msg.(*storeRequestMsgData)
+	req.from = p
+	chunk, err := storage.NewChunkFromData(req.SData)
+	if err != nil {
+		return err
+	}
+	chunk.Source = p
+	self.netStore.Put(chunk)
+	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p))
+	return nil
+}

swarm/network/streamer.go

@@ -0,0 +1,427 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// 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 (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
+	pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueues"
+)
+
+const (
+	HashSize = 32
+
+	Low int = iota
+	Mid
+	High
+	Top
+	PriorityQueues       // number of queues
+	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
+	Start, End uint64      // index of hashes
+	Root       common.Hash // Root hash for indexed segment inclusion proofs
+}
+
+// HandoverProof represents a signed statement that the upstream peer handed over the stream section
+type HandoverProof struct {
+	Sig []byte // Sign(Hash(Serialisation(Handover)))
+	*Handover
+}
+
+// Takeover represents a statement that downstream peer took over (stored all data)
+// handed over
+type Takeover Handover
+
+//  TakeoverProof represents a signed statement that the downstream peer took over
+// the stream section
+type TakeoverProof struct {
+	Sig []byte // Sign(Hash(Serialisation(Takeover)))
+	*Takeover
+}
+
+// TakeoverProofMsg is the protocol msg sent by downstream peer
+type TakeoverProofMsg TakeoverProof
+
+// String pretty prints TakeoverProofMsg
+func (self TakeoverProofMsg) String() string {
+	return fmt.Sprintf("Stream: '%v' [%v-%v], Root: %x, Sig: %x", self.Stream, self.From, self.To, self.Root, self.Sig)
+}
+
+// SubcribeMsg is the protocol msg for requesting a stream(section)
+type SubscribeMsg struct {
+	Stream   Stream
+	From, To uint64
+	Priority uint8 // delivered on priority channel
+}
+
+// UnsyncedKeysMsg is the protocol msg for offering to hand over a
+// stream section
+type UnsyncedKeysMsg struct {
+	Stream         Stream // name of Stream
+	From, To       uint64 // peer and db-specific entry count
+	Hashes         []byte // stream of hashes (128)
+	*HandoverProof        // HandoverProof
+}
+
+// String pretty prints UnsyncedKeysMsg
+func (self UnsyncedKeysMsg) String() string {
+	return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", self.Stream, self.From, self.To, len(self.Hashes)/HashSize)
+}
+
+// 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
+	Want     []byte // bitvector indicating which keys of the batch needed
+	From, To uint64 // next interval offset - empty if not to be continued
+}
+
+// String pretty prints WantedKeysMsg
+func (self WantedKeysMsg) String() string {
+	return fmt.Sprintf("Stream '%v', Want: %x, Next: [%v-%v]", self.Stream, self.Want, self.From, self.To)
+}
+
+// Streamer registry for outgoing and incoming streamer constructors
+type Streamer struct {
+	incomingLock sync.RWMutex
+	outgoingLock sync.RWMutex
+	outgoing     map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)
+	incoming     map[Stream]func(*StreamerPeer) (IncomingStreamer, error)
+}
+
+// NewStreamer is Streamer constructor
+func NewStreamer() *Streamer {
+	return &Streamer{
+		outgoing: make(map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)),
+		incoming: make(map[Stream]func(*StreamerPeer) (IncomingStreamer, error)),
+	}
+}
+
+// RegisterIncomingStreamer registers an incoming streamer constructor
+func (self *Streamer) RegisterIncomingStreamer(stream Stream, f func(*StreamerPeer) (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)) {
+	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) {
+	self.incomingLock.RLock()
+	defer self.incomingLock.RUnlock()
+	f := self.incoming[stream]
+	if f == nil {
+		return nil, fmt.Errorf("stream %v not registered", s)
+	}
+	return f, nil
+}
+
+// GetOutgoingStreamer accessor for incoming streamer constructors
+func (self *Streamer) GetOutgoingStreamer(stream Stream) func(*StreamerPeer) (OutgoingStreamer, error) {
+	self.outgoingLock.RLock()
+	defer self.outgoingLock.RUnlock()
+	f := self.outgoing[stream]
+	if f == nil {
+		return nil, fmt.Errorf("stream %v not registered", s)
+	}
+}
+
+// OutgoingStreamer interface for outgoing peer Streamer
+type OutgoingStreamer interface {
+	CurrentBatch() []byte
+	SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof)
+	GetData([]byte) []byte
+	Priority() int
+}
+
+// IncomingStreamer interface for incoming peer Streamer
+type IncomingStreamer interface {
+	NextBatch(uint64, uint64) (uint64, uint64)
+	NeedData([]byte) func()
+	Priority() int
+}
+
+// StreamerPeer is the Peer extention for the streaming protocol
+type StreamerPeer struct {
+	Peer
+	streamer     *Streamer
+	pq           *pq.PriorityQueues
+	outgoingLock sync.RWMutex
+	incomingLock sync.RWMutex
+	outgoing     map[Stream]OutgoingStreamer
+	incoming     map[Stream]IncomingStreamer
+	quit         chan struct{}
+}
+
+// NewStreamerPeer is the constructor for StreamerPeer
+func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
+	self := &StreamerPeer{
+		pq:       pq.New(PriorityQueues, PriorityQueueCap),
+		streamer: streamer,
+		outgoing: make(map[Stream]OutgoingStreamer),
+		incoming: make(map[Stream]IncomingStreamer),
+		quit:     make(chan struct{}),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	go self.pq.Run(ctx, func(i interface{}) { p.Send(i) })
+	go func() {
+		<-self.quit
+		cancel()
+	}()
+	return self
+}
+
+func (self *StreamerPeer) getOutgoingStreamer(s Stream) (OutgoingStreamer, error) {
+	self.outgoingLock.RLock()
+	defer self.outgoingLock.RUnlock()
+	streamer := self.outgoing[s]
+	if streamer == nil {
+		return nil, fmt.Errorf("stream '%v' not provided", s)
+	}
+	return streamer, nil
+}
+
+func (self *StreamerPeer) getIncomingStreamer(s Stream) (IncomingStreamer, error) {
+	self.incomingLock.RLock()
+	defer self.incomingLock.RUnlock()
+	streamer := self.incoming[s]
+	if streamer == nil {
+		return nil, fmt.Errorf("stream '%v' not provided", s)
+	}
+	return streamer, nil
+}
+
+func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer) error {
+	self.outgoingLock.Lock()
+	defer self.outgoingLock.Unlock()
+	if self.outgoing[s] != nil {
+		return fmt.Errorf("stream %v already registered", s)
+	}
+	self.outgoing[s] = o
+	return nil
+}
+
+func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer) 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
+	return nil
+}
+
+// Subscribe initiates the streamer
+func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
+	f, err := self.streamer.GetIncomingStreamer(s)
+	if err != nil {
+		return err
+	}
+	is := f(self)
+	self.setIncomingStreamer(s, is)
+	msg := &SubscribeMsg{
+		Stream:   s,
+		From:     from,
+		To:       to,
+		Priority: uint8(is.Priority()),
+	}
+	self.Send(msg, is.Priority())
+}
+
+func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error {
+	req := msg.(*SubscribeMsg)
+	f, err := self.streamer.getOutgoingStreamer(req.Stream)
+	if err != nil {
+		return err
+	}
+	s := f(self)
+	if err := self.setOutgoingStreamer(req.Stream, s); err != nil {
+		return nil
+	}
+	self.UnsyncedKeys(s, req.From, req.To)
+	return nil
+}
+
+// handleUnsyncedKeysMsg protocol msg handler calls the incoming streamer interface
+// Filter method
+func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error {
+	req := msg.(*UnsyncedKeysMsg)
+	req.C = make(chan struct{})
+	s, err := self.getIncomingStreamer(req.Stream)
+	if err != nil {
+		return err
+	}
+	hashes := req.Hashes
+	want := bv.New(len(hashes) / HashSize)
+	wg := sync.WaitGroup{}
+	for i := 0; i < len(hashes)/HashSize; i += HashSize {
+		hash := hashes[i : i+HashSize]
+		if wait := s.NeedData(hash); wait != nil {
+			want.Set(i, true)
+			wg.Add(1)
+			// create request and wait until the chunk data arrives and is stored
+			go func(w func()) {
+				w()
+				wg.Done()
+			}(wait)
+		}
+	}
+	go func() {
+		wg.Wait()
+		msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root)
+		self.Send(msg, s.Priority())
+	}()
+	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
+	// except
+	from, to = s.NextBatch(from, to)
+	if from == to {
+		return nil
+	}
+	msg = &WantedKeysMsg{
+		Stream: req.Stream,
+		Want:   want,
+		From:   from,
+		To:     to,
+	}
+	self.Send(msg, s.Priority())
+	return nil
+}
+
+// handleWantedKeysMsg protocol msg handler
+// * sends the next batch of unsynced keys
+// * sends the actual data chunks as per WantedKeysMsg
+func (self *StreamerPeer) handleWantedKeysMsg(msg interface{}) error {
+	req := msg.(*WantedKeysMsg)
+	s, err := self.getOutgoingStreamer(req.Stream)
+	if err != nil {
+		return err
+	}
+	hashes := s.CurrentBatch()
+	// launch in go routine since GetBatch blocks until new hashes arrive
+	go self.UnsyncedKeys(s, req.From, req.To)
+	l := len(hashes) / HashSize
+	want := bv.NewFromBytes(req.Want, l)
+	for i := 0; i < l; i++ {
+		if want.Get(i) {
+			hash := hashes[i*HashSize : (i+1)*HashSize]
+			data := s.GetData(hash)
+			if data == nil {
+				return errNotFound
+			}
+			if err := self.Deliver(data, s.Priority()); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error {
+	req := msg.(*TakeoverProofMsg)
+	s, err := self.getOutgoingStreamer(req.Stream)
+	if err != nil {
+		return err
+	}
+	// store the strongest takeoverproof for the stream in streamer
+	return nil
+}
+
+// Deliver sends a storeRequestMsg protocol message to the peer
+func (self *StreamerPeer) Deliver(data []byte, priority int) error {
+	msg := &storeRequestMsg{
+		SData: data,
+	}
+	return self.pq.Push(nil, msg, priority)
+}
+
+// Deliver sends a storeRequestMsg protocol message to the peer
+func (self *StreamerPeer) Send(msg interface{}, priority int) error {
+	return self.pq.Push(nil, msg, priority)
+}
+
+// UnsyncedKeys sends UnsyncedKeysMsg protocol msg
+func (self *StreamerPeer) SendUnsyncedKeys(s OutgoingStreamer, f, t uint64, priority int) {
+	hashes, from, to, proof := s.SetNextBatch(f, t)
+	msg := &UnsyncedKeysMsg{
+		HandoverProof: proof,
+		Hashes:        hashes,
+		From:          from,
+		To:            to,
+	}
+	self.Send(msg, s.Priority())
+}
+
+// BzzSpec is the spec of the generic swarm handshake
+var StrSpec = &protocols.Spec{
+	Name:       "stream",
+	Version:    1,
+	MaxMsgSize: 10 * 1024 * 1024,
+	Messages: []interface{}{
+		HandshakeMsg{},
+		UnsyncedKeysMsg{},
+		WantedKeysMsg{},
+		TakeoverProofMsg{},
+		SubscribeMsg{},
+	},
+}
+
+// Run protocol run function
+func (s *Streamer) Run(p *bzzPeer) error {
+	sp := NewStreamerPeer(p, s)
+	// load saved intervals
+	defer close(sp.quit)
+	return sp.Run(sp.HandleMsg)
+}
+
+// HandleMsg is the message handler that delegates incoming messages
+func (self *StreamerPeer) HandleMsg(msg interface{}) error {
+	switch msg := msg.(type) {
+
+	case *SubscribeMsg:
+		return self.handleSubscribeMsg(msg)
+
+	case *UnsyncedKeysMsg:
+		return self.handleUnsyncedKeysMsg(msg)
+
+	case *TakeoverProofMsg:
+		return self.handleTakeoverProofMsg(msg)
+
+	case *WantedKeysMsg:
+		return self.handleWantedKeysMsg(msg)
+
+	default:
+		return fmt.Errorf("unknown message type: %T", msg)
+	}
+}

swarm/network/syncer.go

@@ -0,0 +1,266 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// 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 (
+	"bytes"
+	"fmt"
+	"io"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+const (
+	batchSize = 128
+)
+
+// wrapper of db-s to provide mockable custom local chunk store access to syncer
+type DbAccess struct {
+	db  *storage.DbStore
+	loc *storage.LocalStore
+}
+
+func NewDbAccess(loc *storage.LocalStore) *DbAccess {
+	return &DbAccess{loc.DbStore.(*storage.DbStore), loc}
+}
+
+// to obtain the chunks from key or request db entry only
+func (self *DbAccess) get(key storage.Key) (*storage.Chunk, error) {
+	return self.loc.Get(key)
+}
+
+// current storage counter of chunk db
+func (self *DbAccess) currentStorageIndex(po int) uint64 {
+	return self.db.CurrentStorageIndex(po)
+}
+
+// iteration storage counter and proximity order
+func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage.Key, uint64) bool) error {
+	return self.db.SyncIterator(from, to, po, f)
+}
+
+// OutgoingSwarmSyncer implements an OutgoingStreamer for history syncing on bins
+// offered streams:
+// * live request delivery with or without checkback
+// * (live/non-live historical) chunk syncing per proximity bin
+type OutgoingSwarmSyncer struct {
+	po           int
+	db           *DbAccess
+	sessionAt    uint64
+	currentBatch []byte
+	priority     int
+}
+
+// NewOutgoingSwarmSyncer is contructor for OutgoingSwarmSyncer
+func NewOutgoingSwarmSyncer(po int, db *DbAccess) *OutgoingSwarmSyncer {
+	self := &OutgoingSwarmSyncer{
+		po:        po,
+		db:        db,
+		sessionAt: db.currentStorageIndex(po),
+	}
+	return self
+}
+
+func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) {
+	for po := 0; po < maxPO; po++ {
+		stream := fmt.Sprintf("SYNC-%02d-live", po)
+		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+			return NewOutgoingSwarmSyncer(po, db)
+		})
+		stream = fmt.Sprintf("SYNC-%02d-history", po)
+		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+			return NewOutgoingSwarmSyncer(po, db)
+		})
+		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 {
+	chunk, err := self.db.get(Key(key))
+	if err != nil {
+		return nil
+	}
+	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 {
+	var batch []byte
+	i := 0
+	err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool {
+		batch = append(batch, key[:])
+		i++
+		to = idx
+		return i < batchSize
+	})
+	self.currentBatch = batch
+	log.Debug("Swarm batch", "po", self.po, "len", i, "from", from, "to", to)
+	return batch, from, to, proof
+}
+
+// IncomingSwarmSyncer
+type IncomingSwarmSyncer struct {
+	po            int
+	priority      int
+	sessionAt     uint64
+	nextC         chan struct{}
+	intervals     []uint64
+	sessionRoot   storage.Key
+	sessionReader storage.LazySectionReader
+	retrieveC     chan storage.Chunk
+	storeC        chan storage.Chunk
+}
+
+// NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
+func NewIncomingSwarmSyncer(po int, priority int, sessionAt uint64, intervals []uint64, p Peer) *IncomingSwarmSyncer {
+	self := &IncomingSwarmSyncer{
+		po:        po,
+		priority:  priority,
+		sessionAt: sessionAt,
+		nextC:     make(chan struct{}, 1),
+		intervals: intervals,
+	}
+	return self
+}
+
+// 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)
+	RunChunkRequestor(p, retrieveC)
+	storeC := make(storage.Chunk, chunksCap)
+	RunChunkStorer(store, storeC)
+	self := &IncomingSwarmSyncer{
+		po:            po,
+		priority:      priority,
+		sessionAt:     sessionAt,
+		start:         index,
+		end:           index,
+		nextC:         make(chan struct{}, 1),
+		intervals:     intervals,
+		sessionRoot:   sessionRoot,
+		sessionReader: chunker.Join(sessionRoot, retrieveC),
+		retrieveC:     retrieveC,
+		storeC:        storeC,
+	}
+	return self
+}
+
+func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
+	for po := 0; po < maxPO; po++ {
+		stream := fmt.Sprintf("SYNC-%02d-live", po)
+		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+			return NewIncomingSwarmSyncer(po, Mid, sessionAt, nil, p)
+		})
+		stream = fmt.Sprintf("SYNC-%02d-history", po)
+		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+			intervals := loadIntervals(p, po, false)
+			return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p)
+		})
+		stream = fmt.Sprintf("SYNC-%02d-delete", po)
+		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+			intervals := loadIntervals(p, po, true)
+			return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p)
+		})
+	}
+}
+
+// NeedData
+func (self *IncomingSwarmSyncer) NeedData(key []byte) func() {
+	chunk, err := self.store.Get(hash)
+	if err == nil {
+		if chunk.SData == nil {
+			// send a request instead
+			return nil
+		}
+	}
+	// create request and wait until the chunk data arrives and is stored
+	return func() {
+		storedC := <-chunk.storedC
+		<-storedC
+	}
+}
+
+// NextBatch adjusts the indexes by inspecting the intervals
+func (self *IncomingSwarmSyncer) NextBatch(from, to uint64) (uint64, uint64) {
+	if from >= self.sessionAt { // live syncing
+		self.intervals[1] = to
+	} else if to >= self.sessionAt { // history sync complete
+		self.intervals = nil
+		from = 0
+	} else if len(intervals) > 2 && to >= self.intervals[2] { // filled a gap in the intervals
+		self.intervals[1:] = self.intervals[3:]
+		from = self.intervals[1]
+		if len(intervals) > 2 {
+			to = self.intervals[2]
+		}
+	} else {
+		self.intervals[1] = to
+	}
+	return from, to
+}
+
+//
+func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []byte, root storage.Key) error {
+	// for provable syncer currentRoot is non-zero length
+	if self.chunker != nil {
+		if from > self.sessionAt { // for live syncing currentRoot is always updated
+			expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC, self.storeC)
+			if err != nil {
+				return err
+			}
+			if !bytes.Equal(root, expRoot) {
+				return fmt.Errorf("HandoverProof mismatch")
+			}
+			self.currentRoot = currentRoot
+		} else {
+			expHashes := make([]byte, len(hashes))
+			n, err := self.sessionReader.ReadAt(expHashes, int64(self.end*HashSize))
+			if err != nil && err != io.EOF {
+				return err
+			}
+			if !bytes.Equal(expHashes, hashes) {
+				return errInvalidProof
+			}
+		}
+		return nil
+	}
+	self.end += len(hashes) / HashSize
+	takeover := &Takeover{
+		Stream: s,
+		Start:  self.start,
+		End:    self.end,
+		Root:   root,
+	}
+	// serialise and sign
+	return &TakeoverProof{
+		Takeover: takeover,
+		Sig:      nil,
+	}
+}