dht logic for retrieval request handling

bzz/hive.go

@@ -1,9 +1,177 @@
 package bzz
 
-import ()
+/*
+TODO:
+- put Data -> Reader logic to chunker
+- clarify dpa / 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
+- integrate cademlia as peer pool
+- finish the net/dht logic, startSearch and storage
+*/
+
+import (
+	"sync"
+	"time"
+)
+
+type Hive struct {
+	dpa      *DPA
+	memstore *memStore
+	lock     sync.Mutex
+}
 
 /*
-Hive is the logistic manager at swarm
+request status values:
-It is based on kademlia wisdom and flexible forwarding policies for optimal network health.
+- blank
-Hive implements the PeerPool interface (Thx fjl) and as such plays a role in how peers are selected by the p2p server. Ideally the p2p server regularly polls the registered protocol peer pools for good peers (an ordered wishlist of peers to connect to) and chooses the best one not connected. The Bzz Hive is therefore keeping a persistent record of peers for reputation and proximity considerations (or any other indirect incentive maybe).
+- started searching
+- timed out
+- found
 */
+
+const (
+	reqBlank = iota
+	reqSearching
+	reqTimedOut
+	reqFound
+)
+
+const requesterCount = 3
+
+type peer struct {
+	*bzzProtocol
+}
+
+type requestStatus struct {
+	key        Key
+	status     int
+	requesters map[uint64][]*retrieveRequestMsgData
+}
+
+// it's assumed that caller holds the lock
+func (self *Hive) startSearch(chunk *Chunk) {
+	chunk.req.status = reqSearching
+	// implement search logic here
+}
+
+/*
+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) {
+	list := rs.requesters[req.Id]
+	if len(list) < requesterCount {
+		rs.requesters[req.Id] = append(list, req)
+		added = true
+	}
+	return
+}
+
+/*
+decides how to respond to a retrieval request
+updates the request status if needed
+returns
+send bool: true if chunk is to be delivered, false if respond with peers (as for now)
+timeout: if respond with peers, timeout indicates our bet
+this is the most simplistic implementation:
+ - respond with delivery iff less than requesterCount peers forwarded the same request id so far and chunk is found
+ - respond with reject (peers and zero timeout) if given up
+ - 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) (send bool, timeout time.Time) {
+
+	switch rs.status {
+	case reqSearching:
+		self.addRequester(rs, req)
+		timeout = self.searchTimeout(rs, req)
+	case reqTimedOut:
+	case reqFound:
+		if self.addRequester(rs, req) {
+			send = true
+		} else {
+			// timeout = time.Time(0)
+		}
+	}
+	return
+
+}
+
+func (self *Hive) addStoreRequest(req *storeRequestMsgData) (err error) {
+
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	// TODO:
+	return
+}
+
+func (self *Hive) addRetrieveRequest(req *retrieveRequestMsgData) {
+
+	self.lock.Lock()
+	defer self.lock.Unlock()
+
+	chunk, err := self.dpa.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)
+	}
+
+	if chunk.req == nil {
+		chunk.req = new(requestStatus)
+		if chunk.Data == nil {
+			self.startSearch(chunk)
+		}
+	}
+
+	send, timeout := self.strategyUpdateRequest(chunk.req, req) // may change req status
+
+	if send {
+		self.deliver(req, chunk)
+	} else {
+		// we might need chunk.req to cache relevant peers response, or would it expire?
+		self.peers(req, chunk, timeout)
+	}
+
+}
+
+func (self *Hive) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
+	storeReq := &storeRequestMsgData{
+		Key:            req.Key,
+		Id:             req.Id,
+		Data:           chunk.Data,
+		Size:           chunk.Size,
+		RequestTimeout: req.Timeout, //
+		// StorageTimeout time.Time // expiry of content
+		// Metadata       metaData
+	}
+	req.peer.store(storeReq)
+}
+
+func (self *Hive) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) {
+	peersData := &peersMsgData{
+		Peers:   []*peerAddr{}, // get proximity bin from cademlia routing table
+		Key:     req.Key,
+		Id:      req.Id,
+		Timeout: timeout,
+	}
+	req.peer.peers(peersData)
+}
+
+func (self *Hive) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout time.Time) {
+	return
+}
+
+// these should go to cademlia
+func (self *Hive) addPeers(req *peersMsgData) (err error) {
+	return
+}
+
+func (self *Hive) removePeer(p peer) {
+	return
+}

bzz/netstore.go

@@ -6,5 +6,5 @@ It accumulates requests from peers, keeping a request pool and does forwarding f
 */
 
 // it implements the ChunkStore interface
-type DHTStore struct {
+type netStore struct {
 }

bzz/protocol.go

@@ -2,12 +2,12 @@ package bzz
 
 /*
 BZZ implements the bzz wire protocol of swarm
-routing decoded storage and retrieval requests to DPA
+routing decoded storage and retrieval requests
-and registering peers with the DHT
+registering peers with the DHT
 */
 
 import (
-	"fmt"
+	"net"
 	"time"
 
 	"github.com/ethereum/go-ethereum/p2p"
@@ -23,30 +23,17 @@ const (
 
 // bzz protocol message codes
 const (
-	StatusMsg          = iota // 0x01
+	statusMsg          = iota // 0x01
-	StoreRequestMsg           // 0x02
+	storeRequestMsg           // 0x02
-	RetrieveRequestMsg        // 0x03
+	retrieveRequestMsg        // 0x03
-	GetBlockHashesMsg         // 0x04
+	peersMsg                  // 0x04
-	PeersMsg                  // 0x05
-	DeliveryMsg               // 0x06
 )
 
-type BzzHive interface {
-	AddStoreRequest(*StoreRequestMsgData, *p2p.Peer) error
-	AddRetrieveRequest(*RetrieveRequestMsgData, *p2p.Peer) error
-	AddPeers([]*p2p.Peer) error
-	AddDelivery(*DeliveryMsgData, *p2p.Peer) error
-	RemovePeer(*p2p.Peer)
-}
-
 // bzzProtocol represents the swarm wire protocol
 // instance is running on each peer
 type bzzProtocol struct {
-	Hive BzzHive
+	hive *Hive
-	DHT  *DHTStore
-	// CAD *p2p.Cademlia
 	peer *p2p.Peer
-	id   string
 	rw   p2p.MsgReadWriter
 }
 
@@ -68,18 +55,15 @@ Peers
 
 [0x04, key: B_256, timeout: B_64, peers: [[peer], [peer], .... ]] the encoding of a peer is identical to that in the devp2p base protocol peers messages: [IP, Port, NodeID] note that a node's DPA address is not the NodeID but the hash of the NodeID. Timeout serves to indicate whether the responder is forwarding the query within the timeout or not.
 
-Delivery
-
-[0x05, key: B_256, metadata: [], data: B_4k]: the delivery response to retrieval queries.
 */
 
-type StatusMsgData struct {
+type statusMsgData struct {
 	Version   uint64
 	ID        string
 	NodeID    []byte
 	NetworkId uint64
 	Caps      []p2p.Cap
-	Strategy  uint64
+	// Strategy  uint64
 }
 
 /*
@@ -88,15 +72,17 @@ type StatusMsgData struct {
  if they are within our storage radius or have any incentive to store it then attach your nodeID to the metadata
  if the storage request is sufficiently close (within our proximity range (the last row of the routing table), then sending it to all peers will not guarantee convergence, so there needs to be an absolute expiry of the request too. Maybe the protocol should specify a forward probability exponentially declining with age.
 */
-type StoreRequestMsgData struct {
+type storeRequestMsgData struct {
-	Key    Key           // hash of datasize | data
+	Key  Key    // hash of datasize | data
-	Size   uint64        // size of data in bytes
+	Size int64  // size of data in bytes
-	Data   []byte        // is this needed?
+	Data []byte // is this needed?
-	Reader SectionReader // is the underlying byte slice already buffered within the app somewhere?
 	// optional
+	Id             uint64    //
 	RequestTimeout time.Time // expiry for forwarding
 	StorageTimeout time.Time // expiry of content
-	Metadata       MetaData
+	Metadata       metaData  //
+	//
+	peer peer
 }
 
 /*
@@ -106,75 +92,79 @@ MaxSize specifies the maximum size that the peer will accept. This is useful in
 In the special case that the key is identical to the peers own address (hash of NodeID) the message is to be handled as a self lookup. The response is a PeersMsg with the peers in the cademlia proximity bin corresponding to the address.
 It is unclear if a retrieval request with an empty target is the same as a self lookup
 */
-type RetrieveRequestMsgData struct {
+type retrieveRequestMsgData struct {
-	Key      Key
+	Key Key
-	MaxSize  uint64    // optional maximum size of delivery accepted
+	// optional
-	Timeout  time.Time //optional, if missing or
+	Id       uint64    //
-	Metadata MetaData
+	MaxSize  int64     //  maximum size of delivery accepted
+	Timeout  time.Time //
+	Metadata metaData  //
+	//
+	peer peer
 }
 
-/* one response to retrieval, always encouraged after a retrieval request to respond with a list of peers in the same cademlia proximity bin.
+type peerAddr struct {
+	IP     net.IP
+	Port   uint64
+	Pubkey []byte
+}
+
+/*
+one response to retrieval, always encouraged after a retrieval request to respond with a list of peers in the same cademlia proximity bin.
 The encoding of a peer is identical to that in the devp2p base protocol peers messages: [IP, Port, NodeID]
 note that a node's DPA address is not the NodeID but the hash of the NodeID.
 Timeout serves to indicate whether the responder is forwarding the query within the timeout or not.
 The Key is the target (if response to a retrieval request) or peers address (hash of NodeID) if retrieval request was a self lookup.
 It is unclear if PeersMsg with an empty Key has a special meaning or just mean the same as with the peers address as Key (cademlia bin)
 */
-type PeersMsgData struct {
+type peersMsgData struct {
-	Peers []*p2p.Peer
+	Peers   []*peerAddr //
-	Key   Key // if a response to a retrieval request
+	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
+	//
+	peer peer
 }
 
 /*
-Delivery and storeRequest messages could be lumped together or maybe distinguished by their request timeout ?
+metadata is as yet a placeholder
+it will likely contain info about hops or the entire forward chain of node IDs
+this may allow some interesting schemes to evolve optimal routing strategies
+metadata for storage and retrieval requests could specify format parameters relevant for the (blockhashing) chunking scheme used (for chunks corresponding to a treenode). For instance all runtime params for the chunker (hashing algorithm used, branching etc.)
+Finally metadata can hold info relevant to some reward or compensation scheme that may be used to incentivise peers.
 */
-// wonder if we should use Chunk here directly or keep loosely coupled
+type metaData struct{}
-type DeliveryMsgData struct {
-	Key      Key
-	Data     SectionReader
-	Metadata MetaData
-}
 
 /*
- metadata is as yet a placeholder
+main entrypoint, wrappers starting a server running the bzz protocol
- it will likely contain info about hops or the entire forward chain of node IDs
+use this constructor to attach the protocol ("class") to server caps
- this may allow some interesting schemes to evolve optimal routing strategies
+the Dev p2p layer then runs the protocol instance on each peer
-  metadata for storage and retrieval requests could specify format parameters relevant for the (blockhashing) chunking scheme used (for chunks corresponding to a treenode). For instance all runtime params for the chunker (hashing algorithm used, branching etc.)
-  Finally metadata can hold info relevant to some reward or compensation scheme that may be used to incentivise peers.
 */
-type MetaData struct{}
+func BzzProtocol(hive *Hive) p2p.Protocol {
-
-// 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 BzzHive) p2p.Protocol {
 	return p2p.Protocol{
 		Name:    "bzz",
 		Version: Version,
 		Length:  ProtocolLength,
-		Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
+		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			return runBzzProtocol(hive, peer, rw)
+			return runBzzProtocol(hive, p, rw)
 		},
 	}
 }
 
 // the main loop that handles incoming messages
 // note RemovePeer in the post-disconnect hook
-func runBzzProtocol(hive BzzHive, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+func runBzzProtocol(hive *Hive, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
 	self := &bzzProtocol{
-		Hive: hive,
+		hive: hive,
-		// DHT: dht,
-		// CAD:  cad,
 		rw:   rw,
-		peer: peer,
+		peer: p,
-		id:   fmt.Sprintf("%x", peer.Identity().Pubkey()[:8]),
 	}
 	err = self.handleStatus()
 	if err == nil {
 		for {
 			err = self.handle()
 			if err != nil {
-				self.Hive.RemovePeer(self.peer)
+				self.hive.removePeer(peer{self})
 				break
 			}
 		}
@@ -193,43 +183,38 @@ func (self *bzzProtocol) handle() error {
 	// make sure that the payload has been fully consumed
 	defer msg.Discard()
 	/*
-	   StatusMsg          = iota // 0x01
+	   statusMsg          = iota // 0x01
-	   StoreRequestMsg           // 0x02
+	   storeRequestMsg           // 0x02
-	   RetrieveRequestMsg        // 0x03
+	   retrieveRequestMsg        // 0x03
-	   PeersMsg                  // 0x04
+	   peersMsg                  // 0x04
-	   DeliveryMsg               // 0x05
 	*/
 	switch msg.Code {
-	case StatusMsg:
+	case statusMsg:
 		return self.protoError(ErrExtraStatusMsg, "")
 
-	case StoreRequestMsg:
+	case storeRequestMsg:
-		var req StoreRequestMsgData
+		var req storeRequestMsgData
 		if err := msg.Decode(&req); err != nil {
 			return self.protoError(ErrDecode, "msg %v: %v", msg, err)
 		}
-		self.Hive.AddStoreRequest(&req, self.peer)
+		req.peer = peer{self}
+		self.hive.addStoreRequest(&req)
 
-	case RetrieveRequestMsg:
+	case retrieveRequestMsg:
-		var req RetrieveRequestMsgData
+		var req retrieveRequestMsgData
 		if err := msg.Decode(&req); err != nil {
 			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
 		}
-		self.Hive.AddRetrieveRequest(&req, self.peer)
+		req.peer = peer{self}
+		self.hive.addRetrieveRequest(&req)
 
-	case PeersMsg:
+	case peersMsg:
-		var req PeersMsgData
+		var req peersMsgData
 		if err := msg.Decode(&req); err != nil {
 			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
 		}
-		self.Hive.AddPeers(req.Peers)
+		req.peer = peer{self}
-
+		self.hive.addPeers(&req)
-	case DeliveryMsg:
-		var req DeliveryMsgData
-		if err := msg.Decode(&req); err != nil {
-			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
-		}
-		self.Hive.AddDelivery(&req, self.peer)
 
 	default:
 		return self.protoError(ErrInvalidMsgCode, "%v", msg.Code)
@@ -237,19 +222,9 @@ func (self *bzzProtocol) handle() error {
 	return nil
 }
 
-/*
-type StatusMsg struct {
-  Version  uint64
-  ID       string
-  NodeID   []byte
-  Caps     []Cap
-  Strategy uint64
-}
-*/
-
 func (self *bzzProtocol) statusMsg() p2p.Msg {
 
-	return p2p.NewMsg(StatusMsg,
+	return p2p.NewMsg(statusMsg,
 		uint32(Version),
 		uint32(NetworkId),
 		"honey",
@@ -270,15 +245,15 @@ func (self *bzzProtocol) handleStatus() error {
 		return err
 	}
 
-	if msg.Code != StatusMsg {
+	if msg.Code != statusMsg {
-		return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
+		return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, statusMsg)
 	}
 
 	if msg.Size > ProtocolMaxMsgSize {
 		return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
 	}
 
-	var status StatusMsgData
+	var status statusMsgData
 	if err := msg.Decode(&status); err != nil {
 		return self.protoError(ErrDecode, "msg %v: %v", msg, err)
 	}
@@ -293,27 +268,32 @@ func (self *bzzProtocol) handleStatus() error {
 
 	self.peer.Infof("Peer is [bzz] capable (%d/%d)\n", status.Version, status.NetworkId)
 
-	self.Hive.AddPeers([]*p2p.Peer{self.peer})
+	req := &peersMsgData{
+		// Peers: []*peerAddr{self.peer.Address()}, // not implemented in p2p, should be the same as node discovery cademlia
+		// Key:   nil,
+		peer: peer{self},
+	}
+
+	self.hive.addPeers(req)
 
 	return nil
 }
 
-func (self *bzzProtocol) Retrieve(req RetrieveRequestMsgData) error {
+// outgoing messages
-	return p2p.EncodeMsg(self.rw, RetrieveRequestMsg, req)
+func (self *bzzProtocol) retrieve(req *retrieveRequestMsgData) {
+	p2p.EncodeMsg(self.rw, retrieveRequestMsg, req)
 }
 
-func (self *bzzProtocol) Store(req StoreRequestMsgData) error {
+func (self *bzzProtocol) store(req *storeRequestMsgData) {
-	return p2p.EncodeMsg(self.rw, StoreRequestMsg, req)
+	p2p.EncodeMsg(self.rw, storeRequestMsg, req)
 }
 
-func (self *bzzProtocol) DeliveryMsg(req DeliveryMsgData) error {
+func (self *bzzProtocol) peers(req *peersMsgData) {
-	return p2p.EncodeMsg(self.rw, DeliveryMsg, req)
+	p2p.EncodeMsg(self.rw, peersMsg, req)
-}
-
-func (self *bzzProtocol) Peers(req PeersMsgData) error {
-	return p2p.EncodeMsg(self.rw, PeersMsg, req)
 }
 
+// errors
+// TODO: should be reworked using errs pkg
 func (self *bzzProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) {
 	err = ProtocolError(code, format, params...)
 	if err.Fatal() {