From d5330b0dcd78fd4f17e67077cacdd036ca43f284 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 12 May 2015 13:53:01 +0200 Subject: [PATCH] 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 | 3 + bzz/hive.go | 87 +++++++++++++++++++++---- bzz/netstore.go | 18 ++++-- bzz/protocol.go | 80 +++++++++++++++++++---- common/kademlia/kademlia.go | 108 +++++++++++++++++++++++++------ common/kademlia/kademlia_test.go | 22 ++++--- p2p/discover/node.go | 8 +++ p2p/peer.go | 10 +++ 8 files changed, 276 insertions(+), 60 deletions(-) diff --git a/bzz/database.go b/bzz/database.go index bd7e7e6d6a..7510380326 100644 --- a/bzz/database.go +++ b/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" diff --git a/bzz/hive.go b/bzz/hive.go index 4ee17cb190..04236b946b 100644 --- a/bzz/hive.go +++ b/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) { - for _, value := range self.pool { - peers = append(peers, value) +func (self *hive) getPeers(target Key, max int) (peers []peer) { + var addr kademlia.Address + 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 } diff --git a/bzz/netstore.go b/bzz/netstore.go index 7053ddd596..73f241a7e6 100644 --- a/bzz/netstore.go +++ b/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) - peers := self.hive.getPeers(chunk.Key) + dpaLogger.Debugf("NetStore.startSearch: %064x - getting peers from KΛÐΞMLIΛ...", 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, diff --git a/bzz/protocol.go b/bzz/protocol.go index 8d7a02489a..a0dac74e40 100644 --- a/bzz/protocol.go +++ b/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 // - MaxSize uint64 // maximum size of delivery accepted - timeout *time.Time // + Id uint64 // request id + MaxSize uint64 // maximum size of delivery accepted + 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 - Port uint64 - Pubkey []byte + IP net.IP + Port uint16 + 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) diff --git a/common/kademlia/kademlia.go b/common/kademlia/kademlia.go index fe0d2b4b27..42abf60e49 100644 --- a/common/kademlia/kademlia.go +++ b/common/kademlia/kademlia.go @@ -28,11 +28,12 @@ type Kademlia struct { addr Address // adjustable parameters - BucketSize int - MaxProx int - MaxProxBinSize int - nodeDB [][]*nodeRecord - nodeIndex map[Address]*nodeRecord + MaxProx int + MaxProxBinSize int + BucketSize int + currentMaxBucketSize int + nodeDB [][]*NodeRecord + nodeIndex map[Address]*NodeRecord GetNode func(int) @@ -44,26 +45,28 @@ type Kademlia struct { count int buckets []*bucket - lock sync.RWMutex - quitC chan bool + dblock sync.RWMutex + 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.nodeIndex = make(map[Address]*nodeRecord) + self.nodeDB = make([][]*NodeRecord, 8*len(self.addr)) + 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(), - Active: node.LastActive().UnixNano(), - node: node, + Url: node.Url(), + Active: node.LastActive().UnixNano(), + 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 diff --git a/common/kademlia/kademlia_test.go b/common/kademlia/kademlia_test.go index 3e05c0ec51..5a7a66fff2 100644 --- a/common/kademlia/kademlia_test.go +++ b/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 { - t.Errorf("wrong number of nodes: got %d, want %d", len(result.nodes), wantN) + if len(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)) diff --git a/p2p/discover/node.go b/p2p/discover/node.go index a365ade159..4ad5158d2a 100644 --- a/p2p/discover/node.go +++ b/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)} } diff --git a/p2p/peer.go b/p2p/peer.go index ac691f2ce8..bc211748ce 100644 --- a/p2p/peer.go +++ b/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) {