swarm: request streamer implementation

swarm/network/protocol.go

@@ -115,9 +115,9 @@ type Bzz struct {
 // * bzz config
 // * overlay driver
 // * peer store
-func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
+func NewBzz(config *BzzConfig, kad Overlay, store StateStore, streamer *Streamer) *Bzz {
 	return &Bzz{
-		Streamer:   NewStreamer(),
+		Streamer:   streamer,
 		Hive:       NewHive(config.HiveParams, kad, store),
 		localAddr:  &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
 		handshakes: make(map[discover.NodeID]*HandshakeMsg),

swarm/network/requests.go

@@ -16,122 +16,122 @@
 
 package network
 
-import (
+// import (
-	"fmt"
+// 	"fmt"
 
-	"github.com/ethereum/go-ethereum/log"
+// 	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/swarm/storage"
+// 	"github.com/ethereum/go-ethereum/swarm/storage"
-)
+// )
 
-/*
+// /*
- Retrieve Request and store Request handling
+//  Retrieve Request and store Request handling
-*/
+// */
 
-// Handler for storage/retrieval related protocol requests
+// // Handler for storage/retrieval related protocol requests
-type RequestHandler struct {
+// type RequestHandler struct {
-	netStore *storage.NetStore
+// 	netStore *storage.NetStore
-}
+// }
 
-// NewEwquestHandler creates a new RequestHandler
+// // NewEwquestHandler creates a new RequestHandler
-// netStore to
+// // netStore to
-func NewRequestHandler(netStore *storage.NetStore) *RequestHandler {
+// func NewRequestHandler(netStore *storage.NetStore) *RequestHandler {
-	return &RequestHandler{
+// 	return &RequestHandler{
-		netStore: netStore, // entrypoint internal
+// 		netStore: netStore, // entrypoint internal
-	}
+// 	}
-}
+// }
 
-/*
+// /*
-Retrieve request
+// Retrieve request
 
-MaxSize specifies the maximum size that the peer will accept. This is useful in
+// 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
+// particular if we allow storage and delivery of multichunk payload representing
-the entire or partial subtree unfolding from the requested root key.
+// 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
+// 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.
+// 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.
+// Request ID can be newly generated or kept from the request originator.
 
-*/
+// */
-type retrieveRequestMsg struct {
+// type retrieveRequestMsg struct {
-	Key     storage.Key   // target Key address of chunk to be retrieved
+// 	Key     storage.Key   // target Key address of chunk to be retrieved
-	Id      uint64        // request id, request is a lookup if missing or zero
+// 	Id      uint64        // request id, request is a lookup if missing or zero
-	MaxSize uint64        // maximum size of delivery accepted
+// 	MaxSize uint64        // maximum size of delivery accepted
-	from    *StreamerPeer //
+// 	from    *StreamerPeer //
-}
+// }
 
-func (self retrieveRequestMsg) String() string {
+// func (self retrieveRequestMsg) String() string {
-	var from string
+// 	var from string
-	if self.from == nil {
+// 	if self.from == nil {
-		from = "ourselves"
+// 		from = "ourselves"
-	} else {
+// 	} else {
-		from = fmt.Sprintf("%x", self.from.Over())
+// 		from = fmt.Sprintf("%x", self.from.Over())
-	}
+// 	}
-	var target []byte
+// 	var target []byte
-	if len(self.Key) > 3 {
+// 	if len(self.Key) > 3 {
-		target = self.Key[:4]
+// 		target = self.Key[:4]
-	}
+// 	}
-	return fmt.Sprintf("Requester: %v, Key: %x; ID: %v, MaxSize: %v", from, target, self.Id, self.MaxSize)
+// 	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 *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error {
+// func (self *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error {
-	req := msg.(*retrieveRequestMsg)
+// 	req := msg.(*retrieveRequestMsg)
-	req.from = self
+// 	req.from = self
-	// TODO:
+// 	// TODO:
-	// swap - record credit for 1 request
+// 	// swap - record credit for 1 request
-	// note that only charge actual reqsearches
+// 	// note that only charge actual reqsearches
 
-	// call storage.NetStore#Get which
+// 	// call storage.NetStore#Get which
-	// blocks until local retrieval finished
+// 	// blocks until local retrieval finished
-	// launches cloud retrieval
+// 	// launches cloud retrieval
-	chunk, _ := self.netStore.Get(req.Key)
+// 	chunk, _ := self.netStore.Get(req.Key)
-	rs := chunk.Req
+// 	rs := chunk.Req
-	if rs != nil {
+// 	if rs != nil {
-		rs = storage.NewRequestStatus(req.Key)
+// 		rs = storage.NewRequestStatus(req.Key)
-		addRequester(rs, req)
+// 		addRequester(rs, req)
-		chunk.Req = rs
+// 		chunk.Req = rs
-	}
+// 	}
 
-	// check if we can immediately deliver
+// 	// check if we can immediately deliver
-	if chunk.SData != nil {
+// 	if chunk.SData != nil {
-		if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size {
+// 		if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size {
-			err := self.Deliver(chunk, Top)
+// 			err := self.Deliver(chunk, Top)
-			if err != nil {
+// 			if err != nil {
-				log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err))
+// 				log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err))
-				return nil
+// 				return nil
-			}
+// 			}
-			log.Trace(fmt.Sprintf("%v - content found, delivering...", req.Key.Log()))
+// 			log.Trace(fmt.Sprintf("%v - content found, delivering...", req.Key.Log()))
-		} else {
+// 		} else {
-			log.Trace(fmt.Sprintf("%v - content found, not wanted", req.Key.Log()))
+// 			log.Trace(fmt.Sprintf("%v - content found, not wanted", req.Key.Log()))
-		}
+// 		}
-	} else {
+// 	} else {
-		log.Trace(fmt.Sprintf("content not found locally, retrieve via bzz", req.Key.Log()))
+// 		log.Trace(fmt.Sprintf("content not found locally, retrieve via bzz", req.Key.Log()))
-	}
+// 	}
-	return nil
+// 	return nil
-}
+// }
 
-/*
+// /*
-adds a new peer to an existing open request
+// adds a new peer to an existing open request
-only add if less than requesterCount peers forwarded the same request id so far
+// only add if less than requesterCount peers forwarded the same request id so far
-note this is done irrespective of status (searching or found)
+// note this is done irrespective of status (searching or found)
-*/
+// */
-func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) {
+// 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))
+// 	log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id))
-	list := rs.Requesters[req.Id]
+// 	list := rs.Requesters[req.Id]
-	rs.Requesters[req.Id] = append(list, req)
+// 	rs.Requesters[req.Id] = append(list, req)
-}
+// }
 
-func (self storeRequestMsg) String() string {
+// func (self storeRequestMsg) String() string {
-	var from string
+// 	var from string
-	if self.from == nil {
+// 	if self.from == nil {
-		from = "self"
+// 		from = "self"
-	} else {
+// 	} else {
-		from = fmt.Sprintf("%x", self.from.Over())
+// 		from = fmt.Sprintf("%x", self.from.Over())
-	}
+// 	}
-	end := len(self.SData)
+// 	end := len(self.SData)
-	if len(self.SData) > 10 {
+// 	if len(self.SData) > 10 {
-		end = 10
+// 		end = 10
-	}
+// 	}
-	return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end])
+// 	return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end])
-}
+// }

swarm/network/streamer.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -128,13 +129,20 @@ type Streamer struct {
 	outgoingLock sync.RWMutex
 	outgoing     map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)
 	incoming     map[Stream]func(*StreamerPeer) (IncomingStreamer, error)
+
+	dbAccess *DbAccess
+	overlay  Overlay
+	receiveC chan *ChunkDeliveryMsg
 }
 
 // NewStreamer is Streamer constructor
-func NewStreamer() *Streamer {
+func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer {
 	return &Streamer{
 		outgoing: make(map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)),
 		incoming: make(map[Stream]func(*StreamerPeer) (IncomingStreamer, error)),
+		dbAccess: dbAccess,
+		overlay:  overlay,
+		receiveC: make(chan *ChunkDeliveryMsg, 10),
 	}
 }
 
@@ -183,15 +191,15 @@ func (self *Streamer) PeerInfo(id discover.NodeID) interface{} {
 }
 
 // OutgoingStreamer interface for outgoing peer Streamer
-type OutgoingStreamerBackend interface {
+type OutgoingStreamer interface {
 	CurrentBatch() []byte
-	SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof, error)
+	SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
 	GetData([]byte) []byte
 	Priority() int
 }
 
 // IncomingStreamer interface for incoming peer Streamer
-type IncomingStreamerBackend interface {
+type IncomingStreamer interface {
 	NextBatch(uint64) (uint64, uint64)
 	NeedData([]byte) func()
 	Priority() int
@@ -200,9 +208,10 @@ type IncomingStreamerBackend interface {
 // StreamerPeer is the Peer extention for the streaming protocol
 type StreamerPeer struct {
 	Peer
-	streamer     *Streamer
+	streamer *Streamer
-	pq           *pq.PriorityQueue
+	pq       *pq.PriorityQueue
-	netStore     storage.ChunkStore
+	//netStore     storage.ChunkStore
+	dbAccess     *DbAccess
 	outgoingLock sync.RWMutex
 	incomingLock sync.RWMutex
 	outgoing     map[Stream]OutgoingStreamer
@@ -210,15 +219,15 @@ type StreamerPeer struct {
 	quit         chan struct{}
 }
 
-type IncomingStreamer struct {
+// type IncomingStreamer struct {
-	priority uint8
+// 	priority uint8
-	peer     *StreamerPeer
+// 	peer     *StreamerPeer
-}
+// }
 
-type OutgoingStreamer struct {
+// type OutgoingStreamer struct {
-	priority uint8
+// 	priority uint8
-	peer     *StreamerPeer
+// 	peer     *StreamerPeer
-}
+// }
 
 // NewStreamerPeer is the constructor for StreamerPeer
 func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
@@ -238,6 +247,142 @@ func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
 	return self
 }
 
+type RetrieveRequestMsg struct {
+	Key storage.Key
+}
+
+// RetrieveRequestStreamer implements OutgoingStreamer
+type RetrieveRequestStreamer struct {
+	deliveryC    chan *storage.Chunk
+	batchC       chan []byte
+	dbAccess     *DbAccess
+	currentBatch []byte
+	currentLen   uint64
+}
+
+func RegisterRequestStreamer(streamer *Streamer, dbAccess *DbAccess) {
+	streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer) (OutgoingStreamer, error) {
+		return NewRetrieveRequestStreamer(dbAccess), nil
+	})
+	streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer) (IncomingStreamer, error) {
+		return NewIncomingSwarmSyncer(Top, nil, p, dbAccess, nil)
+	})
+}
+
+func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer {
+	s := &RetrieveRequestStreamer{
+		deliveryC: make(chan *storage.Chunk),
+		batchC:    make(chan []byte),
+		dbAccess:  dbAccess,
+	}
+	go s.processDeliveries()
+	return s
+}
+
+func (s *RetrieveRequestStreamer) processDeliveries() {
+	var hashes []byte
+	for {
+		select {
+		case delivery := <-s.deliveryC:
+			hashes = append(hashes, delivery.Key[:]...)
+		case s.batchC <- hashes:
+			hashes = nil
+		}
+	}
+}
+
+func (s *RetrieveRequestStreamer) CurrentBatch() []byte {
+	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
+}
+
+func (s *RetrieveRequestStreamer) GetData(key []byte) []byte {
+	chunk, _ := s.dbAccess.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 {
+	chunk, created := self.dbAccess.getOrCreateRequest(req.Key)
+	s, err := self.getOutgoingStreamer(retrieveRequestStream)
+	if err != nil {
+		return err
+	}
+	streamer := s.(*RetrieveRequestStreamer)
+	if chunk.ReqC != nil {
+		if created {
+			if err := self.streamer.Retrieve(chunk); err != nil {
+				return err
+			}
+		}
+		go func() {
+			t := time.NewTicker(3 * time.Minute)
+			defer t.Stop()
+
+			select {
+			case <-chunk.ReqC:
+			case <-self.quit:
+				return
+			case <-t.C:
+				return
+			}
+
+			streamer.deliveryC <- chunk
+		}()
+		return nil
+	}
+	// TODO: call the retrieve function of the outgoing syncer
+	streamer.deliveryC <- chunk
+	return nil
+}
+
+func (self *Streamer) Retrieve(chunk *storage.Chunk) error {
+	// TODO: using the overlay find the closes peer to send the retrieve
+	//       request to.
+	//       self.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool {})
+	return nil
+}
+
+func (self *StreamerPeer) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
+	chunk, err := self.dbAccess.get(req.Key)
+	if err != nil {
+		return err
+	}
+
+	self.streamer.receiveC <- req
+
+	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, self))
+	return nil
+}
+
+func (self *Streamer) processReceivedChunks() {
+	for {
+		select {
+		case req := <-self.receiveC:
+			chunk, err := self.dbAccess.get(req.Key)
+			if err != nil {
+				continue
+			}
+			chunk.SData = req.SData
+			self.dbAccess.put(chunk)
+			close(chunk.ReqC)
+		}
+	}
+}
+
 func (self *StreamerPeer) getOutgoingStreamer(s Stream) (OutgoingStreamer, error) {
 	self.outgoingLock.RLock()
 	defer self.outgoingLock.RUnlock()
@@ -278,10 +423,6 @@ func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer) erro
 	return nil
 }
 
-func (self *OutgoingStreamer) Subscribe(s Stream) OutgoingStreamerBackend {
-
-}
-
 // Subscribe initiates the streamer
 func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
 	f, err := self.streamer.GetIncomingStreamer(s)
@@ -303,8 +444,7 @@ func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
 	return nil
 }
 
-func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error {
+func (self *StreamerPeer) handleSubscribeMsg(req *SubscribeMsg) error {
-	req := msg.(*SubscribeMsg)
 	f, err := self.streamer.GetOutgoingStreamer(req.Stream)
 	if err != nil {
 		return err
@@ -322,8 +462,7 @@ func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error {
 
 // handleUnsyncedKeysMsg protocol msg handler calls the incoming streamer interface
 // Filter method
-func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error {
+func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error {
-	req := msg.(*UnsyncedKeysMsg)
 	s, err := self.getIncomingStreamer(req.Stream)
 	if err != nil {
 		return err
@@ -357,7 +496,7 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error {
 	if from == to {
 		return nil
 	}
-	msg = &WantedKeysMsg{
+	msg := &WantedKeysMsg{
 		Stream: req.Stream,
 		Want:   want.Bytes(),
 		From:   from,
@@ -370,8 +509,7 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error {
 // 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 {
+func (self *StreamerPeer) handleWantedKeysMsg(req *WantedKeysMsg) error {
-	req := msg.(*WantedKeysMsg)
 	s, err := self.getOutgoingStreamer(req.Stream)
 	if err != nil {
 		return err
@@ -401,8 +539,7 @@ func (self *StreamerPeer) handleWantedKeysMsg(msg interface{}) error {
 	return nil
 }
 
-func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error {
+func (self *StreamerPeer) handleTakeoverProofMsg(req *TakeoverProofMsg) error {
-	req := msg.(*TakeoverProofMsg)
 	_, err := self.getOutgoingStreamer(req.Stream)
 	if err != nil {
 		return err
@@ -411,21 +548,9 @@ func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error {
 	return nil
 }
 
-func (self *StreamerPeer) handleChunkDeliveryMsg(msg interface{}) error {
-	req := msg.(*chunkDeliveryMsg)
-	req.from = self
-	// TODO: chunk validation
-	chunk := storage.NewChunk(req.Key, nil)
-	chunk.SData = req.SData
-	chunk.Source = p
-	self.netStore.Put(chunk)
-	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p))
-	return nil
-}
-
 // Deliver sends a storeRequestMsg protocol message to the peer
 func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority int) error {
-	msg := &storeRequestMsg{
+	msg := &ChunkDeliveryMsg{
 		Key:   chunk.Key,
 		SData: chunk.SData,
 	}
@@ -470,6 +595,10 @@ var StreamerSpec = &protocols.Spec{
 func (s *Streamer) Run(p *bzzPeer) error {
 	sp := NewStreamerPeer(p, s)
 	// load saved intervals
+	sp.handleSubscribeMsg(&SubscribeMsg{
+		Stream:   retrieveRequestStream,
+		Priority: uint8(Top),
+	})
 	defer close(sp.quit)
 	return sp.Run(sp.HandleMsg)
 }

swarm/network/syncer.go

@@ -55,6 +55,16 @@ func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage.
 	return self.db.SyncIterator(from, to, po, f)
 }
 
+// to obtain the chunks from key or request db entry only
+func (self *DbAccess) getOrCreateRequest(key storage.Key) (*storage.Chunk, bool) {
+	return self.loc.GetOrCreateRequest(key)
+}
+
+// to obtain the chunks from key or request db entry only
+func (self *DbAccess) put(chunk *storage.Chunk) {
+	self.loc.Put(chunk)
+}
+
 // OutgoingSwarmSyncer implements an OutgoingStreamer for history syncing on bins
 // offered streams:
 // * live request delivery with or without checkback
@@ -133,7 +143,6 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64,
 
 // IncomingSwarmSyncer
 type IncomingSwarmSyncer struct {
-	po            uint8
 	priority      int
 	sessionAt     uint64
 	nextC         chan struct{}
@@ -142,23 +151,19 @@ type IncomingSwarmSyncer struct {
 	sessionReader storage.LazySectionReader
 	retrieveC     chan *storage.Chunk
 	storeC        chan *storage.Chunk
-	store         storage.ChunkStore
+	dbAccess      *DbAccess
 	chunker       storage.Chunker
 	currentRoot   storage.Key
+	requestFunc   func(chunk *storage.Chunk)
 	end, start    uint64
 }
 
-type {
-	Subscribe()
-}
-
 // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
-func NewIncomingSwarmSyncer(po uint8, priority int, intervals []uint64, p Peer, store storage.ChunkStore, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
+func NewIncomingSwarmSyncer(priority int, intervals []uint64, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
 	self := &IncomingSwarmSyncer{
-		po:        po,
 		priority:  priority,
 		intervals: intervals,
-		store:     store,
+		dbAccess:  dbAccess,
 		chunker:   chunker,
 	}
 	return self, nil
@@ -194,12 +199,12 @@ func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
 	for po := uint8(0); po < maxPO; po++ {
 		stream := Stream(fmt.Sprintf("SYNC-%02d-live", po))
 		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) {
-			return NewIncomingSwarmSyncer(po, High, nil, p, nil, nil)
+			return NewIncomingSwarmSyncer(High, nil, p, nil, nil)
 		})
 		stream = Stream(fmt.Sprintf("SYNC-%02d-history", po))
 		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) {
 			//intervals := loadIntervals(p, po, false)
-			return NewIncomingSwarmSyncer(po, Mid, nil, p, nil, nil)
+			return NewIncomingSwarmSyncer(Mid, nil, p, nil, nil)
 		})
 		// stream = fmt.Sprintf("SYNC-%02d-delete", po)
 		// streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
@@ -210,13 +215,11 @@ func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
 }
 
 // NeedData
-func (self *IncomingSwarmSyncer) NeedData(key []byte) func() {
+func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
-	chunk, err := self.store.Get(key)
+	chunk, created := self.dbAccess.getOrCreateRequest(key)
-	if err == nil {
+	// TODO: we may want to request from this peer anyway even if the request exists
-		if chunk.SData == nil {
+	if chunk.ReqC == nil || !created {
-			// send a request instead
+		return nil
-			return nil
-		}
 	}
 	// create request and wait until the chunk data arrives and is stored
 	return chunk.WaitToStore

swarm/storage/dpa.go

@@ -179,53 +179,44 @@ func (self *DPA) storeWorker() {
 // access by calling network is blocking with a timeout
 
 type dpaChunkStore struct {
-	n          int
+	localStore *LocalStore
-	localStore ChunkStore
+	retrieve   func(chunk *Chunk) error
-	netStore   ChunkStore
 }
 
-func NewDpaChunkStore(localStore, netStore ChunkStore) *dpaChunkStore {
+func NewDpaChunkStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *dpaChunkStore {
-	return &dpaChunkStore{0, localStore, netStore}
+	return &dpaChunkStore{localStore, retrieve}
 }
 
 // Get is the entrypoint for local retrieve requests
 // waits for response or times out
 func (self *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) {
-	chunk, err = self.netStore.Get(key)
+	var created bool
-	if chunk.SData != nil {
+	chunk, created = self.localStore.GetOrCreateRequest(key)
+	if chunk.ReqC == nil {
 		log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData)))
 		return
 	}
-	// TODO: use self.timer time.Timer and reset with defer disableTimer
+
-	timer := time.After(searchTimeout)
+	if created {
-	select {
+		if err := self.retrieve(chunk); err != nil {
-	case <-timer:
+			return nil, err
-		log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log()))
+		}
-		err = notFound
-	case <-chunk.Req.C:
-		log.Trace(fmt.Sprintf("DPA.Get: %v retrieved, %d bytes (%p)", key.Log(), len(chunk.SData), chunk))
 	}
-	return
+	t := time.NewTicker(searchTimeout)
+	defer t.Stop()
+
+	select {
+	case <-t.C:
+		log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log()))
+		return nil, notFound
+	case <-chunk.ReqC:
+	}
+	return chunk, nil
 }
 
 // Put is the entrypoint for local store requests coming from storeLoop
-func (self *dpaChunkStore) Put(entry *Chunk) {
+func (self *dpaChunkStore) Put(chunk *Chunk) {
-	chunk, err := self.localStore.Get(entry.Key)
+	self.localStore.Put(chunk)
-	if err != nil {
-		log.Trace(fmt.Sprintf("DPA.Put: %v new chunk. call netStore.Put", entry.Key.Log()))
-		chunk = entry
-	} else if chunk.SData == nil {
-		log.Trace(fmt.Sprintf("DPA.Put: %v request entry found", entry.Key.Log()))
-		chunk.SData = entry.SData
-		chunk.Size = entry.Size
-	} else {
-		log.Trace(fmt.Sprintf("DPA.Put: %v chunk already known", entry.Key.Log()))
-		return
-	}
-	// from this point on the storage logic is the same with network storage requests
-	log.Trace(fmt.Sprintf("DPA.Put %v: %v", self.n, chunk.Key.Log()))
-	self.n++
-	self.netStore.Put(chunk)
 }
 
 // Close chunk store

swarm/storage/forwarder.go

@@ -1,16 +0,0 @@
-package storage
-
-// implements CloudStore
-// noop placeholder for netstore functionality
-
-type Forwarder struct {
-}
-
-func (self *Forwarder) Store(chunk *Chunk) {
-}
-
-func (self *Forwarder) Retrieve(chunk *Chunk) {
-}
-
-func (self *Forwarder) Deliver(chunk *Chunk) {
-}

swarm/storage/localstore.go

@@ -18,6 +18,9 @@ package storage
 
 import (
 	"encoding/binary"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/log"
 )
 
 // LocalStore is a combination of inmemory db over a disk persisted db
@@ -66,6 +69,26 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) {
 	return
 }
 
+// retrieve logic common for local and network chunk retrieval requests
+func (self *LocalStore) GetOrCreateRequest(key Key) (chunk *Chunk, created bool) {
+	var err error
+	chunk, err = self.Get(key)
+	if err == nil {
+		if chunk.ReqC == nil {
+			log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v found locally", key))
+		} else {
+			log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v hit on an existing request", key))
+			// no need to launch again
+		}
+		return chunk, false
+	}
+	// no data and no request status
+	log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v not found locally. open new request", key))
+	chunk = NewChunk(key, make(chan bool))
+	self.memStore.Put(chunk)
+	return chunk, true
+}
+
 // Close local store
 func (self *LocalStore) Close() {
 	self.DbStore.Close()

swarm/storage/memstore.go

@@ -168,8 +168,8 @@ func (s *MemStore) Put(entry *Chunk) {
 				entry.Size = node.entry.Size
 				entry.SData = node.entry.SData
 			}
-			if entry.Req == nil {
+			if entry.ReqC == nil {
-				entry.Req = node.entry.Req
+				entry.ReqC = node.entry.ReqC
 			}
 			entry.C = node.entry.C
 			node.entry = entry

swarm/storage/netstore.go

@@ -17,38 +17,43 @@
 package storage
 
 import (
-	"fmt"
 	"path/filepath"
 	"time"
-
-	"github.com/ethereum/go-ethereum/log"
 )
 
-/*
+// import (
-NetStore is a cloud storage access abstaction layer for swarm
+// 	"fmt"
-it contains the shared logic of network served chunk store/retrieval requests
+// 	"path/filepath"
-both local (coming from DPA api) and remote (coming from peers via bzz protocol)
+// 	"time"
-it implements the ChunkStore interface and embeds LocalStore
 
-It is called by the bzz protocol instances via Depo (the store/retrieve request handler)
+// 	"github.com/ethereum/go-ethereum/log"
-a protocol instance is running on each peer, so this is heavily parallelised.
+// )
-NetStore falls back to a backend (CloudStorage interface)
-implemented by bzz/network/forwarder. forwarder or IPFS or IPΞS
-*/
-type NetStore struct {
-	hashfunc   SwarmHasher
-	localStore *LocalStore
-	cloud      CloudStore
-}
 
-// backend engine for cloud store
+// /*
-// It can be aggregate dispatching to several parallel implementations:
+// NetStore is a cloud storage access abstaction layer for swarm
-// bzz/network/forwarder. forwarder or IPFS or IPΞS
+// it contains the shared logic of network served chunk store/retrieval requests
-type CloudStore interface {
+// both local (coming from DPA api) and remote (coming from peers via bzz protocol)
-	Store(*Chunk)
+// it implements the ChunkStore interface and embeds LocalStore
-	Deliver(*Chunk)
+
-	Retrieve(*Chunk)
+// It is called by the bzz protocol instances via Depo (the store/retrieve request handler)
-}
+// a protocol instance is running on each peer, so this is heavily parallelised.
+// NetStore falls back to a backend (CloudStorage interface)
+// implemented by bzz/network/forwarder. forwarder or IPFS or IPΞS
+// */
+// type NetStore struct {
+// 	hashfunc   SwarmHasher
+// 	localStore *LocalStore
+// 	cloud      CloudStore
+// }
+
+// // backend engine for cloud store
+// // It can be aggregate dispatching to several parallel implementations:
+// // bzz/network/forwarder. forwarder or IPFS or IPΞS
+// type CloudStore interface {
+// 	Store(*Chunk)
+// 	Deliver(*Chunk)
+// 	Retrieve(*Chunk)
+// }
 
 type StoreParams struct {
 	ChunkDbPath   string
@@ -72,70 +77,56 @@ func (self *StoreParams) Init(path string) {
 	self.ChunkDbPath = filepath.Join(path, "chunks")
 }
 
-// netstore contructor, takes path argument that is used to initialise dbStore,
+// // netstore contructor, takes path argument that is used to initialise dbStore,
-// the persistent (disk) storage component of LocalStore
+// // the persistent (disk) storage component of LocalStore
-// the second argument is the hive, the connection/logistics manager for the node
+// // the second argument is the hive, the connection/logistics manager for the node
-func NewNetStore(hash SwarmHasher, lstore *LocalStore, cloud CloudStore, params *StoreParams) *NetStore {
+// func NewNetStore(hash SwarmHasher, lstore *LocalStore, cloud CloudStore, params *StoreParams) *NetStore {
-	return &NetStore{
+// 	return &NetStore{
-		hashfunc:   hash,
+// 		hashfunc:   hash,
-		localStore: lstore,
+// 		localStore: lstore,
-		cloud:      cloud,
+// 		cloud:      cloud,
-	}
+// 	}
-}
+// }
 
-const (
+// const (
-	// maximum number of peers that a retrieved message is delivered to
+// 	// maximum number of peers that a retrieved message is delivered to
-	requesterCount = 3
+// 	requesterCount = 3
-)
+// )
 
 var (
 	// timeout interval before retrieval is timed out
 	searchTimeout = 3 * time.Second
 )
 
-// store logic common to local and network chunk store requests
+// // store logic common to local and network chunk store requests
-// ~ unsafe put in localdb no check if exists no extra copy no hash validation
+// // ~ unsafe put in localdb no check if exists no extra copy no hash validation
-// the chunk is forced to propagate (Cloud.Store) even if locally found!
+// // the chunk is forced to propagate (Cloud.Store) even if locally found!
-// caller needs to make sure if that is wanted
+// // caller needs to make sure if that is wanted
-func (self *NetStore) Put(entry *Chunk) {
+// func (self *NetStore) Put(entry *Chunk) {
-	self.localStore.Put(entry)
+// 	self.localStore.Put(entry)
 
-	// handle deliveries
+// 	// handle deliveries
-	if entry.Req != nil {
+// 	if entry.ReqC != nil {
-		log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v hit existing request...delivering", entry.Key.Log()))
+// 		log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v hit existing request...delivering", entry.Key.Log()))
-		// closing C signals to other routines (local requests)
+// 		// closing C signals to other routines (local requests)
-		// that the chunk is has been retrieved
+// 		// that the chunk is has been retrieved
-		close(entry.Req.C)
+// 		close(entry.ReqC)
-		// deliver the chunk to requesters upstream
+// 		// deliver the chunk to requesters upstream
-		go self.cloud.Deliver(entry)
+// 		go self.cloud.Deliver(entry)
-	} else {
+// 	} else {
-		log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v stored locally", entry.Key.Log()))
+// 		log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v stored locally", entry.Key.Log()))
-		// handle propagating store requests
+// 		// handle propagating store requests
-		// go self.cloud.Store(entry)
+// 		// go self.cloud.Store(entry)
-		go self.cloud.Store(entry)
+// 		go self.cloud.Store(entry)
-	}
+// 	}
-}
+// }
 
-// retrieve logic common for local and network chunk retrieval requests
+// // retrieve logic common for local and network chunk retrieval requests
-func (self *NetStore) Get(key Key) (*Chunk, error) {
+// func (self *NetStore) Get(key Key) (*Chunk, error) {
-	var err error
+// 	chunk, _ := self.localStore.GetOrCreateRequest(key)
-	chunk, err := self.localStore.Get(key)
+// 	go self.cloud.Retrieve(chunk)
-	if err == nil {
+// 	return chunk, nil
-		if chunk.Req == nil {
+// }
-			log.Trace(fmt.Sprintf("NetStore.Get: %v found locally", key))
-		} else {
-			log.Trace(fmt.Sprintf("NetStore.Get: %v hit on an existing request", key))
-			// no need to launch again
-		}
-		return chunk, err
-	}
-	// no data and no request status
-	log.Trace(fmt.Sprintf("NetStore.Get: %v not found locally. open new request", key))
-	chunk = NewChunk(key, NewRequestStatus(key))
-	self.localStore.memStore.Put(chunk)
-	go self.cloud.Retrieve(chunk)
-	return chunk, nil
-}
 
-// Close netstore
+// // Close netstore
-func (self *NetStore) Close() {}
+// func (self *NetStore) Close() {}

swarm/storage/types.go

@@ -166,44 +166,23 @@ func (c KeyCollection) Swap(i, j int) {
 	c[i], c[j] = c[j], c[i]
 }
 
-// each chunk when first requested opens a record associated with the request
-// next time a request for the same chunk arrives, this record is updated
-// this request status keeps track of the request ID-s as well as the requesting
-// peers and has a channel that is closed when the chunk is retrieved. Multiple
-// local callers can wait on this channel (or combined with a timeout, block with a
-// select).
-type RequestStatus struct {
-	Key        Key
-	Source     Peer
-	C          chan bool
-	Requesters map[uint64][]interface{}
-}
-
-func NewRequestStatus(key Key) *RequestStatus {
-	return &RequestStatus{
-		Key:        key,
-		Requesters: make(map[uint64][]interface{}),
-		C:          make(chan bool),
-	}
-}
-
 // Chunk also serves as a request object passed to ChunkStores
 // in case it is a retrieval request, Data is nil and Size is 0
 // Note that Size is not the size of the data chunk, which is Data.Size()
 // but the size of the subtree encoded in the chunk
 // 0 if request, to be supplied by the dpa
 type Chunk struct {
-	Key      Key            // always
+	Key   Key    // always
-	SData    []byte         // nil if request, to be supplied by dpa
+	SData []byte // nil if request, to be supplied by dpa
-	Size     int64          // size of the data covered by the subtree encoded in this chunk
+	Size  int64  // size of the data covered by the subtree encoded in this chunk
-	Source   Peer           // peer
+	//Source   Peer           // peer
-	C        chan bool      // to signal data delivery by the dpa
+	C        chan bool // to signal data delivery by the dpa
-	Req      *RequestStatus // request Status needed by netStore
+	ReqC     chan bool // to signal the request done
-	dbStored chan bool      // never remove a chunk from memStore before it is written to dbStore
+	dbStored chan bool // never remove a chunk from memStore before it is written to dbStore
 }
 
-func NewChunk(key Key, rs *RequestStatus) *Chunk {
+func NewChunk(key Key, reqC chan bool) *Chunk {
-	return &Chunk{Key: key, Req: rs, dbStored: make(chan bool)}
+	return &Chunk{Key: key, ReqC: reqC, dbStored: make(chan bool)}
 }
 
 func (c *Chunk) WaitToStore() {

swarm/swarm.go

@@ -49,10 +49,11 @@ type Swarm struct {
 	api    *api.Api     // high level api layer (fs/manifest)
 	dns    api.Resolver // DNS registrar
 	//dbAccess    *network.DbAccess      // access to local chunk db iterator and storage counter
-	storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
+	//storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
-	dpa     *storage.DPA       // distributed preimage archive, the local API to the storage with document level storage/retrieval support
+	dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support
 	//depo        network.StorageHandler // remote request handler, interface between bzz protocol and the storage
-	cloud       storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
+	streamer *network.Streamer
+	//cloud       storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
 	bzz         *network.Bzz       // the logistic manager
 	backend     chequebook.Backend // simple blockchain Backend
 	privateKey  *ecdsa.PrivateKey
@@ -114,8 +115,8 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 	config.HiveParams.Discovery = true
 
 	// setup cloud storage internal access layer
-	self.cloud = &storage.Forwarder{}
+	//self.cloud = &storage.Forwarder{}
-	self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams)
+	//self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams)
 	log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store"))
 	nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey)))
 	addr := network.NewAddrFromNodeID(nodeid)
@@ -124,10 +125,16 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 		UnderlayAddr: addr.UAddr,
 		HiveParams:   config.HiveParams,
 	}
-	self.bzz = network.NewBzz(bzzconfig, to, nil)
+
+	dbAccess := network.NewDbAccess(self.lstore)
+	self.streamer = network.NewStreamer(to, dbAccess)
+	network.RegisterOutgoingSyncers(self.streamer, dbAccess)
+	network.RegisterIncomingSyncers(self.streamer, dbAccess)
+
+	self.bzz = network.NewBzz(bzzconfig, to, nil, self.streamer)
 
 	// set up DPA, the cloud storage local access layer
-	dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.storage)
+	dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.streamer.Retrieve)
 	log.Debug(fmt.Sprintf("-> Local Access to Swarm"))
 	// Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage
 	self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams)