- Refactoring: bringing names in correspondence to functionality

- Added netstore Put and Get
- Storage logic

bzz/dpa.go

@@ -110,20 +110,6 @@ func (self *DPA) Store(data SectionReader) (key Key, err error) {
 
 }
 
-// DPA is itself a chunk store , to stores a chunk only
-// its integrity is checked ?
-func (self *DPA) Put(chunk *Chunk) {
-	// rely on storeC
-	return
-}
-
-// Get(chunk *Chunk) looks up a chunk in the local stores
-// This method is blocking until the chunk is retrieved so additional timeout is needed to wrap this call
-func (self *DPA) Get(key Key) (chunk *Chunk, err error) {
-	// rely on retrieveC
-	return
-}
-
 func (self *DPA) Start() {
 	self.lock.Lock()
 	defer self.lock.Unlock()

bzz/netstore.go

@@ -3,7 +3,7 @@ package bzz
 /*
 TODO:
 - put Data -> Reader logic to chunker
-- clarify dpa / hive / netstore naming and division of labour and entry points for local/remote requests
+- clarify dpa / localStore / hive / netstore naming and division of labour and entry points for local/remote requests
 - figure out if its a problem that peers on requester list may disconnect while searching
 - Id (nonce/requester map key) should probs be random byte slice or (hash of) originator's address to avoid collisions
 - rework protocol errors using errs after PR merged
@@ -12,6 +12,7 @@ TODO:
 */
 
 import (
+	"math/rand"
 	"sync"
 	"time"
 )
@@ -22,24 +23,24 @@ type peerPool struct {
 }
 
 func (self *peerPool) addPeer(p peer) {
-	self.pool[p.peer.identity.Pubkey()] = p
+	self.pool[string(p.pubkey)] = p
 }
 
 func (self *peerPool) removePeer(p peer) {
-	delete(self.pool, p.peer.identity.Pubkey)
+	delete(self.pool, string(p.pubkey))
 }
 
 func (self *peerPool) getPeers(target Key) (peers []peer) {
-	for key, value := range self.pool {
+	for _, value := range self.pool {
 		peers = append(peers, value)
 	}
 	return
 }
 
-type Hive struct {
+type netStore struct {
-	dpa      *DPA
+	localStore *localStore
-	memstore *memStore
+	lock       sync.Mutex
-	lock     sync.Mutex
+	peerPool   *peerPool
 }
 
 /*
@@ -61,16 +62,17 @@ const requesterCount = 3
 
 type peer struct {
 	*bzzProtocol
+	pubkey []byte
 }
 
 type requestStatus struct {
 	key        Key
 	status     int
-	requesters map[uint64][]*retrieveRequestMsgData
+	requesters map[int64][]*retrieveRequestMsgData
 }
 
 // it's assumed that caller holds the lock
-func (self *Hive) startSearch(chunk *Chunk) {
+func (self *netStore) startSearch(chunk *Chunk) {
 	chunk.req.status = reqSearching
 	// implement search logic here
 }
@@ -80,13 +82,9 @@ 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/timedOut)
 */
-func (self *Hive) addRequester(rs *requestStatus, req *retrieveRequestMsgData) (added bool) {
+func (self *netStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) {
 	list := rs.requesters[req.Id]
-	if len(list) < requesterCount {
+	rs.requesters[req.Id] = append(list, req)
-		rs.requesters[req.Id] = append(list, req)
-		added = true
-	}
-	return
 }
 
 /*
@@ -101,78 +99,81 @@ this is the most simplistic implementation:
  - respond with peers and timeout if still searching
 ! in the last case as well, we should respond with reject if already got requesterCount peers with that exact id
 */
-func (self *Hive) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (msgTyp int, timeout time.Time) {
+func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (msgTyp int, timeout time.Time) {
 
 	switch rs.status {
 	case reqSearching:
-		if self.addRequester(rs, req) {
+		msgTyp = peersMsg
-			msgTyp = peersMsg
+		timeout = self.searchTimeout(rs, req)
-			timeout = self.searchTimeout(rs, req)
-		}
 	case reqTimedOut:
 		msgTyp = peersMsg
 	case reqFound:
-		if self.addRequester(rs, req) {
+		msgTyp = storeRequestMsg
-			msgTyp = storeRequestMsg
-		}
 	}
 	return
 
 }
 
-func (self *Hive) addStoreRequest(req *storeRequestMsgData) (err error) {
+func (self *netStore) put(entry *Chunk) {
+	self.localStore.Put(entry)
+	self.store(entry)
+	// only send responses once
+	if entry.req != nil && entry.req.status == reqSearching {
+		entry.req.status = reqFound
+		self.propagateResponse(entry)
+	}
+}
 
+func (self *netStore) Put(entry *Chunk) {
+	chunk, err := self.localStore.Get(entry.Key)
+	if err != nil {
+		chunk = entry
+	} else if chunk.Data == nil {
+		chunk.Data = entry.Data
+		chunk.Size = entry.Size
+	} else {
+		return
+	}
+	self.put(chunk)
+}
+
+func (self *netStore) addStoreRequest(req *storeRequestMsgData) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
-	chunk, err := self.dpa.Get(req.Key)
+	chunk, err := self.localStore.Get(req.Key)
 	// we assume that a returned chunk is the one stored in the memory cache
 	if err != nil {
-		s := new(storeRequestStatus)
 		chunk = &Chunk{
-			Key:                req.Key,
+			Key:  req.Key,
-			Data:               req.Data,
+			Data: req.Data,
-			Size:               req.Size,
+			Size: req.Size,
-			storeRequestStatus: s,
-		}
-		self.dpa.Put(chunk)
-		self.store(chunk)
-	} else {
-		// pending retrieval request
-		if chunk.Data != nil {
-			// update access counts not needed, Get takes care of it
-			return
 		}
+	} else if chunk.Data == nil {
 		chunk.Data = req.Data
 		chunk.Size = req.Size
-		// FIXME: breach of memstore contract data is put into storage without checking capacity
+	} else {
-		self.dpa.Put(chunk)
+		return
-		// only send responses once
-		if chunk.req.status == reqSearching {
-			chunk.req.status = reqFound
-			self.propagateResponse(chunk)
-		}
 	}
-
+	self.put(chunk)
-	return
 }
 
-func (self *Hive) propagateResponse(chunk *Chunk) {
+func (self *netStore) propagateResponse(chunk *Chunk) {
 	// send chunk to first requesterCount peer of each Id
 }
 
-func (self *Hive) addRetrieveRequest(req *retrieveRequestMsgData) {
+func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) {
 
 	self.lock.Lock()
 	defer self.lock.Unlock()
 
-	chunk, err := self.dpa.Get(req.Key)
+	chunk, err := self.localStore.Get(req.Key)
 	// we assume that a returned chunk is the one stored in the memory cache
 	if err != nil {
 		// no data and no request status
 		chunk = &Chunk{
 			Key: req.Key,
 		}
-		self.memstore.Put(chunk)
+		self.localStore.memStore.Put(chunk)
 	}
 
 	if chunk.req == nil {
@@ -184,7 +185,7 @@ func (self *Hive) addRetrieveRequest(req *retrieveRequestMsgData) {
 
 	send, timeout := self.strategyUpdateRequest(chunk.req, req) // may change req status
 
-	if send {
+	if send == storeRequestMsg {
 		self.deliver(req, chunk)
 	} else {
 		// we might need chunk.req to cache relevant peers response, or would it expire?
@@ -193,7 +194,7 @@ func (self *Hive) addRetrieveRequest(req *retrieveRequestMsgData) {
 
 }
 
-func (self *Hive) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
+func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
 	storeReq := &storeRequestMsgData{
 		Key:            req.Key,
 		Id:             req.Id,
@@ -206,20 +207,20 @@ func (self *Hive) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
 	req.peer.store(storeReq)
 }
 
-func (self *Hive) store(chunk) {
+func (self *netStore) store(chunk *Chunk) {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	req := storeRequestMsgData{
+	req := &storeRequestMsgData{
 		Key:  chunk.Key,
 		Data: chunk.Data,
 		Id:   r.Int63(),
 		Size: chunk.Size,
 	}
-	for _, peer := range self.peerPool.GetPeers(chunk.Key) {
+	for _, peer := range self.peerPool.getPeers(chunk.Key) {
 		go peer.store(req)
 	}
 }
 
-func (self *Hive) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) {
+func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) {
 	peersData := &peersMsgData{
 		Peers:   []*peerAddr{}, // get proximity bin from cademlia routing table
 		Key:     req.Key,
@@ -229,15 +230,15 @@ func (self *Hive) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.
 	req.peer.peers(peersData)
 }
 
-func (self *Hive) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout time.Time) {
+func (self *netStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout time.Time) {
 	return
 }
 
 // these should go to cademlia
-func (self *Hive) addPeers(req *peersMsgData) (err error) {
+func (self *netStore) addPeers(req *peersMsgData) (err error) {
 	return
 }
 
-func (self *Hive) removePeer(p peer) {
+func (self *netStore) removePeer(p peer) {
 	return
 }

bzz/protocol.go

@@ -32,9 +32,9 @@ const (
 // bzzProtocol represents the swarm wire protocol
 // instance is running on each peer
 type bzzProtocol struct {
-	hive *Hive
+	netStore *netStore
-	peer *p2p.Peer
+	peer     *p2p.Peer
-	rw   p2p.MsgReadWriter
+	rw       p2p.MsgReadWriter
 }
 
 /*
@@ -77,7 +77,7 @@ type storeRequestMsgData struct {
 	Size int64  // size of data in bytes
 	Data []byte // is this needed?
 	// optional
-	Id             uint64    //
+	Id             int64     //
 	RequestTimeout time.Time // expiry for forwarding
 	StorageTimeout time.Time // expiry of content
 	Metadata       metaData  //
@@ -95,7 +95,7 @@ It is unclear if a retrieval request with an empty target is the same as a self
 type retrieveRequestMsgData struct {
 	Key Key
 	// optional
-	Id       uint64    //
+	Id       int64     //
 	MaxSize  int64     //  maximum size of delivery accepted
 	Timeout  time.Time //
 	Metadata metaData  //
@@ -121,7 +121,7 @@ type peersMsgData struct {
 	Peers   []*peerAddr //
 	Timeout time.Time   // indicate whether responder is expected to deliver content
 	Key     Key         // if a response to a retrieval request
-	Id      uint64      // if a response to a retrieval request
+	Id      int64       // if a response to a retrieval request
 	//
 	peer peer
 }
@@ -140,31 +140,31 @@ main entrypoint, wrappers starting a server running the bzz protocol
 use this constructor to attach the protocol ("class") to server caps
 the Dev p2p layer then runs the protocol instance on each peer
 */
-func BzzProtocol(hive *Hive) p2p.Protocol {
+func BzzProtocol(netStore *netStore) p2p.Protocol {
 	return p2p.Protocol{
 		Name:    "bzz",
 		Version: Version,
 		Length:  ProtocolLength,
 		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			return runBzzProtocol(hive, p, rw)
+			return runBzzProtocol(netStore, p, rw)
 		},
 	}
 }
 
 // the main loop that handles incoming messages
 // note RemovePeer in the post-disconnect hook
-func runBzzProtocol(hive *Hive, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+func runBzzProtocol(netStore *netStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
 	self := &bzzProtocol{
-		hive: hive,
+		netStore: netStore,
-		rw:   rw,
+		rw:       rw,
-		peer: p,
+		peer:     p,
 	}
 	err = self.handleStatus()
 	if err == nil {
 		for {
 			err = self.handle()
 			if err != nil {
-				self.hive.removePeer(peer{self})
+				self.netStore.removePeer(peer{bzzProtocol: self})
 				break
 			}
 		}
@@ -197,24 +197,24 @@ func (self *bzzProtocol) handle() error {
 		if err := msg.Decode(&req); err != nil {
 			return self.protoError(ErrDecode, "msg %v: %v", msg, err)
 		}
-		req.peer = peer{self}
+		req.peer = peer{bzzProtocol: self}
-		self.hive.addStoreRequest(&req)
+		self.netStore.addStoreRequest(&req)
 
 	case retrieveRequestMsg:
 		var req retrieveRequestMsgData
 		if err := msg.Decode(&req); err != nil {
 			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
 		}
-		req.peer = peer{self}
+		req.peer = peer{bzzProtocol: self}
-		self.hive.addRetrieveRequest(&req)
+		self.netStore.addRetrieveRequest(&req)
 
 	case peersMsg:
 		var req peersMsgData
 		if err := msg.Decode(&req); err != nil {
 			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
 		}
-		req.peer = peer{self}
+		req.peer = peer{bzzProtocol: self}
-		self.hive.addPeers(&req)
+		self.netStore.addPeers(&req)
 
 	default:
 		return self.protoError(ErrInvalidMsgCode, "%v", msg.Code)
@@ -271,10 +271,10 @@ func (self *bzzProtocol) handleStatus() error {
 	req := &peersMsgData{
 		// Peers: []*peerAddr{self.peer.Address()}, // not implemented in p2p, should be the same as node discovery cademlia
 		// Key:   nil,
-		peer: peer{self},
+		peer: peer{bzzProtocol: self, pubkey: status.NodeID},
 	}
 
-	self.hive.addPeers(req)
+	self.netStore.addPeers(req)
 
 	return nil
 }