kademlia integration into hive

* peerAddr in peers message
* hive initialises with base (self) address and path to save peers
* hive start() will implement bootstrap/update cycle
* hive--kademlia interplay
* kademlia: added entrypoints for DB: addNodeRecords, getNodeRecords
* hive has corresponding addPeerEntries, getPeerEntries
* hive is regularly polled to suggest peers, which it relays to kad.getNodeRecords
* cycle of add/get implements bootsrapping
* hive.getPeers gets an extra argument for max items
* bzzProtocol now initialised with self address
* statusMsgData contains peerAddr in Addr field (should check at handshake)
* retrieveRequestMsgData now got a MaxPeers field to limit no of peers received
* protocol instance implements kademlia.Node interface
* currentBucketSize added to kademlia to help prioritisation in getNodeRecords
* introduced dblock Mutex
* p2p: export discover.NewNode, use it in p2p.peer.Node()

bzz/database.go

@@ -1,5 +1,8 @@
 package bzz
 
+// this is a clone of an earlier state of the ethereum ethdb/database
+// no need for queueing/caching
+
 import (
 	"fmt"
 

bzz/hive.go

@@ -1,37 +1,102 @@
 package bzz
 
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/kademlia"
+)
+
 type peer struct {
 	*bzzProtocol
-	pubkey []byte
 }
 
-// This is a mock implementation with a fixed peer pool with no distinction between peers
+// peer not necessary here
+// bzz protocol could implement kademlia.Node interface with
+// Addr(), LastActive() and Drop()
+
+// Hive is the logistic manager of the swarm
+// it uses a generic kademlia nodetable to find best peer list
+// for any target
+// this is used by the netstore to search for content in the swarm
+// the bzz protocol peersMsgData exchange is relayed to Kademlia
+// for db storage and filtering
+// connections and disconnections are reported and relayed
+// to keep the nodetable uptodate
+
 type hive struct {
-	pool map[string]peer
+	kad  *kademlia.Kademlia
+	path string
 }
 
-func newHive() *hive {
+func newHive(address common.Hash, hivepath string) *hive {
 	return &hive{
-		pool: make(map[string]peer),
+		path: hivepath,
+		kad:  kademlia.New(kademlia.Address(address)),
 	}
 }
 
+func (self *hive) start() (err error) {
+	self.kad.Start()
+	err = self.kad.Load(self.path)
+	if err != nil {
+		return
+	}
+	// go func() {
+	// 	for {
+	// 		select {
+	// 		case <-timer:
+	// 		case <-subscr:
+	// 		}
+	// 		maxpeers := 4
+	// 		self.getPeerEntries(maxpeers)
+	// 	}
+	// }()
+	return
+}
+
 func (self *hive) addPeer(p peer) {
-	self.pool[string(p.pubkey)] = p
+	self.kad.AddNode(p)
 }
 
 func (self *hive) removePeer(p peer) {
-	delete(self.pool, string(p.pubkey))
+	self.kad.RemoveNode(p)
 }
 
 // Retrieve a list of live peers that are closer to target than us
-func (self *hive) getPeers(target Key) (peers []peer) {
+func (self *hive) getPeers(target Key, max int) (peers []peer) {
-	for _, value := range self.pool {
+	var addr kademlia.Address
-		peers = append(peers, value)
+	copy(addr[:], target[:])
+	for _, node := range self.kad.GetNodes(addr, max) {
+		peers = append(peers, node.(peer))
 	}
 	return
 }
 
-func (self *hive) addPeers(req *peersMsgData) (err error) {
+func newNodeRecord(addr *peerAddr) *kademlia.NodeRecord {
+	return &kademlia.NodeRecord{
+		Address: addr.addr(),
+		Active:  0,
+		Url:     addr.url(),
+	}
+}
+
+// called by the protocol upon receiving peerset (for target address)
+// peersMsgData is converted to a slice of NodeRecords for Kademlia
+// this is to store all thats needed
+func (self *hive) addPeerEntries(req *peersMsgData) {
+	var nrs []*kademlia.NodeRecord
+	for _, p := range req.Peers {
+		nrs = append(nrs, newNodeRecord(p))
+	}
+	self.kad.AddNodeRecords(nrs)
+}
+
+// called to ask periodically for preferences
+// Kademlia ideally maintains a queue of prioritized nodes
+func (self *hive) getPeerEntries(max int) (resp *peersMsgData, err error) {
+	nrs, err := self.kad.GetNodeRecords(max)
+	for _, n := range nrs {
+		_ = n
+		// resp // build response from kademlia noderecords
+	}
 	return
 }

bzz/netstore.go

@@ -5,6 +5,8 @@ import (
 	"math/rand"
 	"sync"
 	"time"
+
+	"github.com/ethereum/go-ethereum/common"
 )
 
 type NetStore struct {
@@ -43,13 +45,13 @@ type requestStatus struct {
 	C          chan bool
 }
 
-func NewNetStore(path string) *NetStore {
+func NewNetStore(addr common.Hash, path, hivepath string) *NetStore {
 	dbStore, _ := newDbStore(path)
 	return &NetStore{
 		localStore: &localStore{
 			memStore: newMemStore(dbStore),
 			dbStore:  dbStore,
-		}, hive: newHive(),
+		}, hive: newHive(addr, hivepath),
 	}
 }
 
@@ -180,8 +182,8 @@ func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) {
 // it's assumed that caller holds the lock
 func (self *NetStore) startSearch(chunk *Chunk, id int64, timeout *time.Time) {
 	chunk.req.status = reqSearching
-	dpaLogger.Debugf("NetStore.startSearch: %064x - getting peers from cademlia...", chunk.Key)
+	dpaLogger.Debugf("NetStore.startSearch: %064x - getting peers from KΛÐΞMLIΛ...", chunk.Key)
-	peers := self.hive.getPeers(chunk.Key)
+	peers := self.hive.getPeers(chunk.Key, 0)
 	req := &retrieveRequestMsgData{
 		Key:     chunk.Key,
 		Id:      uint64(id),
@@ -279,14 +281,18 @@ func (self *NetStore) store(chunk *Chunk) {
 		SData: chunk.SData,
 		Id:    uint64(id),
 	}
-	for _, peer := range self.hive.getPeers(chunk.Key) {
+	for _, peer := range self.hive.getPeers(chunk.Key, 0) {
 		go peer.store(req)
 	}
 }
 
 func (self *NetStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout *time.Time) {
+	var addrs []*peerAddr
+	for _, peer := range self.hive.getPeers(req.Key, int(req.MaxPeers)) {
+		addrs = append(addrs, peer.peerAddr())
+	}
 	peersData := &peersMsgData{
-		Peers:   []*peerAddr{}, // get proximity bin from cademlia routing table
+		Peers:   addrs,
 		Key:     req.Key,
 		Id:      req.Id,
 		timeout: timeout,

bzz/protocol.go

@@ -10,10 +10,12 @@ import (
 	"net"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common/kademlia"
 	"github.com/ethereum/go-ethereum/errs"
 	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
 const (
@@ -55,6 +57,7 @@ var errorToString = map[int]string{
 // bzzProtocol represents the swarm wire protocol
 // instance is running on each peer
 type bzzProtocol struct {
+	self     *discover.Node
 	netStore *NetStore
 	peer     *p2p.Peer
 	rw       p2p.MsgReadWriter
@@ -85,6 +88,7 @@ type statusMsgData struct {
 	Version   uint64
 	ID        string
 	NodeID    []byte
+	Addr      *peerAddr
 	NetworkId uint64
 	Caps      []p2p.Cap
 	// Strategy  uint64
@@ -118,18 +122,37 @@ 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       uint64     // request id
-	MaxSize uint64     //  maximum size of delivery accepted
+	MaxSize  uint64     // maximum size of delivery accepted
-	timeout *time.Time //
+	MaxPeers uint64     // maximum number of peers returned
+	timeout  *time.Time //
 	//Metadata metaData  //
 	//
-	peer peer
+	peer peer // protocol registers the requester
 }
 
 type peerAddr struct {
-	IP     net.IP
+	IP   net.IP
-	Port   uint64
+	Port uint16
-	Pubkey []byte
+	ID   []byte
+	n    *discover.Node
+}
+
+func (self *peerAddr) node() *discover.Node {
+	if self.n == nil {
+		var nodeid discover.NodeID
+		copy(nodeid[:], self.ID)
+		self.n = discover.NewNode(nodeid, self.IP, self.Port, self.Port)
+	}
+	return self.n
+}
+
+func (self *peerAddr) addr() kademlia.Address {
+	return kademlia.Address(self.node().Sha())
+}
+
+func (self *peerAddr) url() string {
+	return self.node().String()
 }
 
 /*
@@ -163,21 +186,23 @@ 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(netStore *NetStore) p2p.Protocol {
+func BzzProtocol(netStore *NetStore, self *discover.Node) p2p.Protocol {
 	return p2p.Protocol{
 		Name:    "bzz",
 		Version: Version,
 		Length:  ProtocolLength,
 		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			return runBzzProtocol(netStore, p, rw)
+			return runBzzProtocol(netStore, self, p, rw)
 		},
 	}
 }
 
 // the main loop that handles incoming messages
 // note RemovePeer in the post-disconnect hook
-func runBzzProtocol(netStore *NetStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+func runBzzProtocol(netStore *NetStore, selfNode *discover.Node, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
 	self := &bzzProtocol{
+		self: selfNode,
+
 		netStore: netStore,
 		rw:       rw,
 		peer:     p,
@@ -248,7 +273,7 @@ func (self *bzzProtocol) handle() error {
 			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
 		}
 		req.peer = peer{bzzProtocol: self}
-		self.netStore.hive.addPeers(&req)
+		self.netStore.hive.addPeerEntries(&req)
 
 	default:
 		return self.protoError(ErrInvalidMsgCode, "%v", msg.Code)
@@ -258,11 +283,12 @@ func (self *bzzProtocol) handle() error {
 
 func (self *bzzProtocol) handleStatus() (err error) {
 	// send precanned status message
-	sliceNodeID := self.peer.ID()
+	sliceNodeID := self.self.ID
 	handshake := &statusMsgData{
 		Version:   uint64(Version),
 		ID:        "honey",
 		NodeID:    sliceNodeID[:],
+		Addr:      newPeerAddrFromNode(self.self),
 		NetworkId: uint64(NetworkId),
 		Caps:      []p2p.Cap{},
 	}
@@ -301,11 +327,39 @@ func (self *bzzProtocol) handleStatus() (err error) {
 
 	glog.V(logger.Info).Infof("Peer is [bzz] capable (%d/%d)\n", status.Version, status.NetworkId)
 
-	self.netStore.hive.addPeer(peer{bzzProtocol: self, pubkey: status.NodeID})
+	self.netStore.hive.addPeer(peer{bzzProtocol: self})
 
 	return nil
 }
 
+// protocol instance implements kademlia.Node interface (embedded hive.peer)
+func (self *bzzProtocol) Addr() (a kademlia.Address) {
+	return kademlia.Address(self.self.Sha())
+}
+
+func (self *bzzProtocol) Url() string {
+	return self.self.String()
+}
+
+func (self *bzzProtocol) LastActive() time.Time {
+	return time.Now()
+}
+
+func (self *bzzProtocol) Drop() {
+}
+
+func newPeerAddrFromNode(node *discover.Node) *peerAddr {
+	return &peerAddr{
+		ID:   node.ID[:],
+		IP:   node.IP,
+		Port: node.TCP,
+	}
+}
+
+func (self *bzzProtocol) peerAddr() *peerAddr {
+	return newPeerAddrFromNode(self.peer.Node())
+}
+
 // outgoing messages
 func (self *bzzProtocol) retrieve(req *retrieveRequestMsgData) {
 	dpaLogger.Debugf("Request message: %#v", req)

common/kademlia/kademlia.go

@@ -28,11 +28,12 @@ type Kademlia struct {
 	addr Address
 
 	// adjustable parameters
-	BucketSize     int
+	MaxProx              int
-	MaxProx        int
+	MaxProxBinSize       int
-	MaxProxBinSize int
+	BucketSize           int
-	nodeDB         [][]*nodeRecord
+	currentMaxBucketSize int
-	nodeIndex      map[Address]*nodeRecord
+	nodeDB               [][]*NodeRecord
+	nodeIndex            map[Address]*NodeRecord
 
 	GetNode func(int)
 
@@ -44,26 +45,28 @@ type Kademlia struct {
 	count   int
 	buckets []*bucket
 
-	lock  sync.RWMutex
+	dblock sync.RWMutex
-	quitC chan bool
+	lock   sync.RWMutex
+	quitC  chan bool
 }
 
 type Address common.Hash
 
 type Node interface {
 	Addr() Address
-	// Url()
+	Url() string
 	LastActive() time.Time
 	Drop()
 }
 
-type nodeRecord struct {
+type NodeRecord struct {
 	Address Address `json:address`
+	Url     string  `json:url`
 	Active  int64   `json:active`
 	node    Node
 }
 
-func (self *nodeRecord) setActive() {
+func (self *NodeRecord) setActive() {
 	if self.node != nil {
 		self.Active = self.node.LastActive().UnixNano()
 	}
@@ -71,7 +74,7 @@ func (self *nodeRecord) setActive() {
 
 type kadDB struct {
 	Address Address         `json:address`
-	Nodes   [][]*nodeRecord `json:nodes`
+	Nodes   [][]*NodeRecord `json:nodes`
 }
 
 // public constructor with compulsory arguments
@@ -116,8 +119,8 @@ func (self *Kademlia) Start() error {
 		self.buckets[i] = &bucket{size: self.BucketSize} // will initialise bucket{int(0),[]Node(nil),sync.Mutex}
 	}
 
-	self.nodeDB = make([][]*nodeRecord, 8*len(self.addr))
+	self.nodeDB = make([][]*NodeRecord, 8*len(self.addr))
-	self.nodeIndex = make(map[Address]*nodeRecord)
+	self.nodeIndex = make(map[Address]*NodeRecord)
 
 	self.quitC = make(chan bool)
 	return nil
@@ -188,15 +191,17 @@ func (self *Kademlia) AddNode(node Node) (err error) {
 	}
 
 	go func() {
+		self.dblock.Lock()
+		defer self.dblock.Unlock()
 		record, found := self.nodeIndex[node.Addr()]
 		if found {
 			record.node = node
 		} else {
-			record = &nodeRecord{
+			record = &NodeRecord{
 				Address: node.Addr(),
-				// Url:     node.Url(),
+				Url:     node.Url(),
-				Active: node.LastActive().UnixNano(),
+				Active:  node.LastActive().UnixNano(),
-				node:   node,
+				node:    node,
 			}
 			self.nodeIndex[node.Addr()] = record
 			self.nodeDB[index] = append(self.nodeDB[index], record)
@@ -236,7 +241,11 @@ proxLimit and MaxProx. proxLimit is dynamically adjusted so that 1) there is no
 empty buckets in bin < proxLimit and 2) the sum of all items are the maximum
 possible but lower than MaxProxBinSize
 */
-func (self *Kademlia) GetNodes(target Address, max int) (r nodesByDistance) {
+func (self *Kademlia) GetNodes(target Address, max int) []Node {
+	return self.getNodes(target, max).nodes
+}
+
+func (self *Kademlia) getNodes(target Address, max int) (r nodesByDistance) {
 	self.lock.RLock()
 	defer self.lock.RUnlock()
 	r.target = target
@@ -279,6 +288,61 @@ func (self *Kademlia) GetNodes(target Address, max int) (r nodesByDistance) {
 	return
 }
 
+// this is used to add node records to the persisted db
+// TODO: maybe db needs to be purged occasionally (reputation will take care of
+// that)
+func (self *Kademlia) AddNodeRecords(nrs []*NodeRecord) {
+	self.dblock.Lock()
+	defer self.dblock.Unlock()
+	for _, node := range nrs {
+		_, found := self.nodeIndex[node.Address]
+		if !found {
+			self.nodeIndex[node.Address] = node
+			index := self.proximityBin(node.Address)
+			self.nodeDB[index] = append(self.nodeDB[index], node)
+		}
+	}
+}
+
+/*
+GetNodeRecords gives back an at most max length slice of node records
+in order of decreasing priority for desired connection
+Used to pick candidates for live nodes to satisfy Kademlia network for Swarm
+
+Does a round robin on buckets starting from 0 to proxLimit then back
+on each round i we inspect if live-nodes fill the bucket
+if len(nodes) + i < currentMaxBucketSize, then take ith element in corresponding
+db row ordered by reputation (active time?)
+
+This has double role. Starting as naive node with empty db, this implements
+Kademlia bootstrapping
+As a mature node, it manages quickly fill in blanks or short lines
+All on demand
+*/
+func (self *Kademlia) GetNodeRecords(max int) (nrs []*NodeRecord, err error) {
+	var round int
+	for max > 0 {
+		for i, b := range self.buckets {
+			if len(b.nodes)+round < self.currentMaxBucketSize {
+				if nr := self.getNodeRecord(i, round); nr != nil {
+					nrs = append(nrs)
+				}
+			}
+		}
+		round++
+		max--
+	}
+	return
+}
+
+func (self *Kademlia) getNodeRecord(row, col int) (nr *NodeRecord) {
+	if row >= 0 && row < len(self.nodeDB) &&
+		col >= 0 && col < len(self.nodeDB[row]) {
+		nr = self.nodeDB[row][col]
+	}
+	return
+}
+
 // in situ mutable bucket
 type bucket struct {
 	size  int
@@ -417,16 +481,16 @@ func proxCmp(target, a, b Address) int {
 	return 0
 }
 
-func (self *Kademlia) DB() [][]*nodeRecord {
+func (self *Kademlia) DB() [][]*NodeRecord {
 	return self.nodeDB
 }
 
-func (n *nodeRecord) bumpActive() {
+func (n *NodeRecord) bumpActive() {
 	stamp := time.Now().Unix()
 	atomic.StoreInt64(&n.Active, stamp)
 }
 
-func (n *nodeRecord) LastActive() time.Time {
+func (n *NodeRecord) LastActive() time.Time {
 	stamp := atomic.LoadInt64(&n.Active)
 	return time.Unix(stamp, 0)
 }
@@ -438,11 +502,13 @@ func (self *Kademlia) Save(path string) error {
 		Address: self.addr,
 		Nodes:   self.nodeDB,
 	}
+
 	for _, b := range kad.Nodes {
 		for _, node := range b {
 			node.setActive()
 		}
 	}
+
 	data, err := json.MarshalIndent(&kad, "", " ")
 	if err != nil {
 		return err

common/kademlia/kademlia_test.go

@@ -42,6 +42,10 @@ func (n *testNode) Addr() Address {
 func (n *testNode) Drop() {
 }
 
+func (n *testNode) Url() string {
+	return ""
+}
+
 func (n *testNode) LastActive() time.Time {
 	return time.Now()
 }
@@ -85,7 +89,7 @@ func TestGetNodes(t *testing.T) {
 		if len(test.All) == 0 || test.N == 0 {
 			return true
 		}
-		result := kad.GetNodes(test.Target, test.N)
+		nodes := kad.GetNodes(test.Target, test.N)
 
 		// check that the number of results is min(N, kad.len)
 		wantN := test.N
@@ -93,26 +97,26 @@ func TestGetNodes(t *testing.T) {
 			wantN = tlen
 		}
 
-		if len(result.nodes) != wantN {
+		if len(nodes) != wantN {
-			t.Errorf("wrong number of nodes: got %d, want %d", len(result.nodes), wantN)
+			t.Errorf("wrong number of nodes: got %d, want %d", len(nodes), wantN)
 			return false
 		}
 
-		if hasDuplicates(result.nodes) {
+		if hasDuplicates(nodes) {
 			t.Errorf("result contains duplicates")
 			return false
 		}
 
-		if !sortedByDistanceTo(test.Target, result.nodes) {
+		if !sortedByDistanceTo(test.Target, nodes) {
 			t.Errorf("result is not sorted by distance to target")
 			return false
 		}
 
 		// check that the result nodes have minimum distance to target.
-		farthestResult := result.nodes[len(result.nodes)-1].Addr()
+		farthestResult := nodes[len(nodes)-1].Addr()
 		for i, b := range kad.buckets {
 			for j, n := range b.nodes {
-				if contains(result.nodes, n.Addr()) {
+				if contains(nodes, n.Addr()) {
 					continue // don't run the check below for nodes in result
 				}
 				if proxCmp(test.Target, n.Addr(), farthestResult) < 0 {
@@ -239,7 +243,7 @@ func TestSaveLoad(t *testing.T) {
 			return
 		}
 	}
-	nodes := kad.GetNodes(self, 100).nodes
+	nodes := kad.GetNodes(self, 100)
 	path := "/tmp/bzz.peers"
 	kad.Stop(path)
 	kad = New(self)
@@ -255,7 +259,7 @@ func TestSaveLoad(t *testing.T) {
 			}
 		}
 	}
-	loadednodes := kad.GetNodes(self, 100).nodes
+	loadednodes := kad.GetNodes(self, 100)
 	for i, node := range loadednodes {
 		if nodes[i].Addr() != node.Addr() {
 			t.Errorf("node mismatch at %d/%d", i, len(nodes))

p2p/discover/node.go

@@ -47,6 +47,14 @@ func newNode(id NodeID, ip net.IP, udpPort, tcpPort uint16) *Node {
 	}
 }
 
+func NewNode(id NodeID, ip net.IP, udpPort, tcpPort uint16) *Node {
+	return newNode(id, ip, udpPort, tcpPort)
+}
+
+func (n *Node) Sha() common.Hash {
+	return n.sha
+}
+
 func (n *Node) addr() *net.UDPAddr {
 	return &net.UDPAddr{IP: n.IP, Port: int(n.UDP)}
 }

p2p/peer.go

@@ -81,6 +81,16 @@ func (p *Peer) LocalAddr() net.Addr {
 	return p.conn.LocalAddr()
 }
 
+// LocalAddr returns the local address of the network connection.
+func (p *Peer) Node() *discover.Node {
+	return discover.NewNode(
+		p.rw.ID,
+		net.ParseIP(p.conn.RemoteAddr().String()),
+		uint16(p.rw.ListenPort), //
+		uint16(p.rw.ListenPort), //
+	)
+}
+
 // Disconnect terminates the peer connection with the given reason.
 // It returns immediately and does not wait until the connection is closed.
 func (p *Peer) Disconnect(reason DiscReason) {