common/kademlia first stab, tests pass

common/kademlia/kademlia.go

@@ -0,0 +1,465 @@
+package kademlia
+
+import (
+	"fmt"
+	"sort"
+	// "math"
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/logger"
+)
+
+var kadlogger = logger.NewLogger("KΛÐ")
+
+const (
+	bucketSize = 20
+	maxProx    = 255
+)
+
+type Kademlia struct {
+	// immutable baseparam
+	addr Address
+
+	// adjustable parameters
+	BucketSize     int
+	MaxProx        int
+	MaxProxBinSize int
+	nodeDB         [][]*nodeRecord
+	nodeIndex      map[Address]*nodeRecord
+
+	// state
+	proxLimit int
+	proxSize  int
+
+	//
+	count   int
+	buckets []*bucket
+
+	lock  sync.RWMutex
+	quitC chan bool
+}
+
+type Address common.Hash
+
+type Node interface {
+	Addr() Address
+	// Url()
+	LastActive() time.Time
+}
+
+type nodeRecord struct {
+	Address Address `json:address`
+	Active  int64   `json:active`
+	node    Node
+}
+
+func (self *nodeRecord) setActive() {
+	if self.node != nil {
+		self.Active = self.node.LastActive().UnixNano()
+	}
+}
+
+type kadDB struct {
+	Address Address         `json:address`
+	Nodes   [][]*nodeRecord `json:nodes`
+}
+
+// public constructor with compulsory arguments
+// hash is a byte slice of length equal to self.HashBytes
+func New(a Address) *Kademlia {
+	return &Kademlia{
+		addr: a, // compulsory fields without default
+	}
+}
+
+// accessor for KAD self address
+func (self *Kademlia) Addr() Address {
+	return self.addr
+}
+
+// accessor for KAD self count
+func (self *Kademlia) Count() int {
+	return self.count
+}
+
+// Start brings up a pool of entries potentially from an offline persisted source
+// and sets default values for optional parameters
+func (self *Kademlia) Start() error {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	if self.quitC != nil {
+		return nil
+	}
+	if self.MaxProx == 0 {
+		self.MaxProx = maxProx
+	}
+	if self.BucketSize == 0 {
+		self.BucketSize = bucketSize
+	}
+	// runtime parameters
+	if self.MaxProxBinSize == 0 {
+		self.MaxProxBinSize = self.BucketSize
+	}
+
+	self.buckets = make([]*bucket, self.MaxProx+1)
+	for i, _ := range self.buckets {
+		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.quitC = make(chan bool)
+	return nil
+}
+
+// Stop saves the routing table into a persistant form
+func (self *Kademlia) Stop(path string) (err error) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	if self.quitC == nil {
+		return
+	}
+	close(self.quitC)
+	self.quitC = nil
+
+	if len(path) > 0 {
+		err = self.Save(path)
+		if err != nil {
+			kadlogger.Warnf("unable to save node records: %v", err)
+		}
+	}
+	return
+}
+
+// RemoveNode is the entrypoint where nodes are taken offline
+func (self *Kademlia) RemoveNode(node Node) (err error) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	index := self.proximityBin(node.Addr())
+	bucket := self.buckets[index]
+	for i := 0; i < len(bucket.nodes); i++ {
+		if node.Addr() == bucket.nodes[i].Addr() {
+			bucket.nodes = append(bucket.nodes[:i], bucket.nodes[(i+1):]...)
+		}
+	}
+	self.count--
+	if len(bucket.nodes) < bucket.size {
+		err = fmt.Errorf("insufficient nodes (%v) in bucket %v", len(bucket.nodes), index)
+	}
+	if len(bucket.nodes) == 0 {
+		self.adjustProx(index, -1)
+	}
+	// async callback to notify user that bucket needs filling
+	// action is left to the user
+	// go self.getNode(index)
+	return
+}
+
+// AddNode is the entry point where new nodes are registered
+func (self *Kademlia) AddNode(node Node) (err error) {
+
+	self.lock.Lock()
+	defer self.lock.Unlock()
+
+	index := self.proximityBin(node.Addr())
+	kadlogger.Debugf("bin %d, len: %d\n", index, len(self.buckets))
+
+	bucket := self.buckets[index]
+	err = bucket.insert(node)
+	if err != nil {
+		return
+	}
+	self.count++
+	if index >= self.proxLimit {
+		self.adjustProx(index, 1)
+	}
+
+	go func() {
+		record, found := self.nodeIndex[node.Addr()]
+		if found {
+			record.node = node
+		} else {
+			record = &nodeRecord{
+				Address: node.Addr(),
+				// Url:     node.Url(),
+				Active: node.LastActive().UnixNano(),
+				node:   node,
+			}
+			self.nodeIndex[node.Addr()] = record
+			self.nodeDB[index] = append(self.nodeDB[index], record)
+		}
+	}()
+
+	kadlogger.Infof("add peer %v...", node)
+	return
+
+}
+
+// adjust Prox (proxLimit and proxSize after an insertion of add nodes into bucket r)
+func (self *Kademlia) adjustProx(r int, add int) {
+	switch {
+	case add > 0 && r == self.proxLimit:
+		self.proxLimit += add
+		for ; self.proxLimit < self.MaxProx && len(self.buckets[self.proxLimit].nodes) > 0; self.proxLimit++ {
+			self.proxSize -= len(self.buckets[self.proxLimit].nodes)
+		}
+	case add > 0 && r > self.proxLimit && self.proxSize+add > self.MaxProxBinSize:
+		self.proxLimit++
+		self.proxSize -= len(self.buckets[r].nodes) - add
+	case add > 0 && r > self.proxLimit:
+		self.proxSize += add
+	case add < 0 && r < self.proxLimit && len(self.buckets[r].nodes) == 0:
+		for i := self.proxLimit - 1; i > r; i-- {
+			self.proxSize += len(self.buckets[i].nodes)
+		}
+		self.proxLimit = r
+	}
+}
+
+/*
+GetNodes(target) returns the list of nodes belonging to the same proximity bin
+as the target. The most proximate bin will be the union of the bins between
+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) {
+	self.lock.RLock()
+	defer self.lock.RUnlock()
+	r.target = target
+	index := self.proximityBin(target)
+	start := index
+	var down bool
+	if index >= self.proxLimit {
+		start = self.MaxProx
+		down = true
+	}
+	var n int
+	limit := max
+	if max == 0 {
+		limit = 1000
+	}
+	for {
+		bucket := self.buckets[start].nodes
+		for i := 0; i < len(bucket); i++ {
+			r.push(bucket[i], limit)
+			n++
+		}
+		if max == 0 && start == index ||
+			max > 0 && down && start <= index && (n >= max || n == self.Count() || start == 0) {
+			break
+		}
+		if down {
+			start--
+		} else {
+			if start == self.MaxProx {
+				if index == 0 {
+					break
+				}
+				start = index - 1
+				down = true
+			} else {
+				start++
+			}
+		}
+	}
+	return
+}
+
+// in situ mutable bucket
+type bucket struct {
+	size  int
+	nodes []Node
+	lock  sync.RWMutex
+}
+
+func (a Address) Bin() string {
+	var bs []string
+	for _, b := range a[:] {
+		bs = append(bs, fmt.Sprintf("%08b", b))
+	}
+	return strings.Join(bs, "")
+}
+
+// nodesByDistance is a list of nodes, ordered by distance to target.
+type nodesByDistance struct {
+	nodes  []Node
+	target Address
+}
+
+func sortedByDistanceTo(target Address, slice []Node) bool {
+	var last Address
+	for i, node := range slice {
+		if i > 0 {
+			if proxCmp(target, node.Addr(), last) < 0 {
+				return false
+			}
+		}
+		last = node.Addr()
+	}
+	return true
+}
+
+// push(node, max) adds the given node to the list, keeping the total size
+// below max elements.
+func (h *nodesByDistance) push(node Node, max int) {
+	// returns the firt index ix such that func(i) returns true
+	ix := sort.Search(len(h.nodes), func(i int) bool {
+		return proxCmp(h.target, h.nodes[i].Addr(), node.Addr()) >= 0
+	})
+
+	if len(h.nodes) < max {
+		h.nodes = append(h.nodes, node)
+	}
+	if ix < len(h.nodes) {
+		copy(h.nodes[ix+1:], h.nodes[ix:])
+		h.nodes[ix] = node
+	}
+}
+
+// insert adds a peer to a bucket either by appending to existing items if
+// bucket length does not exceed bucketLength, or by replacing the worst
+// Node in the bucket
+func (self *bucket) insert(node Node) (err error) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	if len(self.nodes) >= self.size { // >= allows us to add peers beyond the bucketsize limitation
+		worst := self.worstNode()
+		self.nodes[worst] = node
+	} else {
+		self.nodes = append(self.nodes, node)
+	}
+	return
+}
+
+// worst expunges the single worst entry in a row, where worst entry is with a peer that has not been active the longests
+func (self *bucket) worstNode() (index int) {
+	var oldest time.Time
+	for i, node := range self.nodes {
+		if (oldest == time.Time{}) || node.LastActive().Before(oldest) {
+			oldest = node.LastActive()
+			index = i
+		}
+	}
+	return
+}
+
+/*
+Taking the proximity value relative to a fix point x classifies the points in
+the space (n byte long byte sequences) into bins the items in which are each at
+most half as distant from x as items in the previous bin. Given a sample of
+uniformly distrbuted items (a hash function over arbitrary sequence) the
+proximity scale maps onto series of subsets with cardinalities on a negative
+exponential scale.
+
+It also has the property that any two item belonging to the same bin are at
+most half as distant from each other as they are from x.
+
+If we think of random sample of items in the bins as connections in a network of interconnected nodes than relative proximity can serve as the basis for local
+decisions for graph traversal where the task is to find a route between two
+points. Since in every step of forwarding, the finite distance halves, there is
+a guaranteed constant maximum limit on the number of hops needed to reach one
+node from the other.
+*/
+
+func (self *Kademlia) proximityBin(other Address) (ret int) {
+	ret = proximity(self.addr, other)
+	if ret > self.MaxProx {
+		ret = self.MaxProx
+	}
+	return
+}
+
+/*
+The distance metric MSB(x, y) of two equal length byte sequences x an y is the
+value of the binary integer cast of the xor-ed byte sequence (most significant
+bit first).
+proximity(x, y) counts the common zeros in the front of this distance measure.
+*/
+func proximity(one, other Address) (ret int) {
+	for i := 0; i < len(one); i++ {
+		oxo := one[i] ^ other[i]
+		for j := 0; j < 8; j++ {
+			if (oxo>>uint8(7-j))&0x1 != 0 {
+				return i*8 + j
+			}
+		}
+	}
+	return len(one)*8 - 1
+}
+
+// proxCmp compares the distances a->target and b->target.
+// Returns -1 if a is closer to target, 1 if b is closer to target
+// and 0 if they are equal.
+func proxCmp(target, a, b Address) int {
+	for i := range target {
+		da := a[i] ^ target[i]
+		db := b[i] ^ target[i]
+		if da > db {
+			return 1
+		} else if da < db {
+			return -1
+		}
+	}
+	return 0
+}
+
+func (self *Kademlia) DB() [][]*nodeRecord {
+	return self.nodeDB
+}
+
+func (n *nodeRecord) bumpActive() {
+	stamp := time.Now().Unix()
+	atomic.StoreInt64(&n.Active, stamp)
+}
+
+func (n *nodeRecord) LastActive() time.Time {
+	stamp := atomic.LoadInt64(&n.Active)
+	return time.Unix(stamp, 0)
+}
+
+// save persists all peers encountered
+func (self *Kademlia) Save(path string) error {
+
+	kad := kadDB{
+		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
+	}
+	return ioutil.WriteFile(path, data, os.ModePerm)
+}
+
+// loading the idle node record from disk
+func (self *Kademlia) Load(path string) (err error) {
+	var data []byte
+	data, err = ioutil.ReadFile(path)
+	if err != nil {
+		return
+	}
+	var kad kadDB
+	err = json.Unmarshal(data, &kad)
+	if err != nil {
+		return
+	}
+	self.nodeDB = kad.Nodes
+	if self.addr != kad.Address {
+		return fmt.Errorf("invalid kad db: address mismatch, expected %v, got %v", self.addr, kad.Address)
+	}
+	return
+}

common/kademlia/kademlia_test.go

@@ -0,0 +1,335 @@
+package kademlia
+
+import (
+	"fmt"
+	"log"
+	"math/rand"
+	"os"
+	"reflect"
+	"sync"
+	"testing"
+	"testing/quick"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/logger"
+)
+
+var (
+	quickrand = rand.New(rand.NewSource(time.Now().Unix()))
+	quickcfg  = &quick.Config{MaxCount: 5000, Rand: quickrand}
+)
+
+var once sync.Once
+
+func LogInit(l logger.LogLevel) {
+	once.Do(func() {
+		logger.NewStdLogSystem(os.Stderr, log.LstdFlags, l)
+	})
+}
+
+type testNode struct {
+	addr Address
+}
+
+func (n *testNode) String() string {
+	return fmt.Sprintf("%x", n.addr[:])
+}
+
+func (n *testNode) Addr() Address {
+	return n.addr
+}
+
+func (n *testNode) LastActive() time.Time {
+	return time.Now()
+}
+
+func (n *testNode) Add(a Address) (err error) {
+	return nil
+}
+
+func TestAddNode(t *testing.T) {
+	LogInit(logger.DebugLevel)
+	addr, ok := gen(Address{}, quickrand).(Address)
+	other, ok := gen(Address{}, quickrand).(Address)
+	if !ok {
+		t.Errorf("oops")
+	}
+	kad := New(addr)
+	kad.Start()
+	err := kad.AddNode(&testNode{addr: other})
+	_ = err
+}
+
+func TestGetNodes(t *testing.T) {
+	t.Parallel()
+	LogInit(logger.DebugLevel)
+
+	test := func(test *getNodesTest) bool {
+		// for any node kad.le, Target and N
+		kad := New(test.Self)
+		kad.MaxProx = 10
+		kad.Start()
+		var err error
+		t.Logf("getNodesTest %v: %v\n", len(test.All), test)
+		for _, node := range test.All {
+			err = kad.AddNode(node)
+			if err != nil {
+				t.Errorf("backend not accepting node")
+				return false
+			}
+		}
+
+		if len(test.All) == 0 || test.N == 0 {
+			return true
+		}
+		result := kad.GetNodes(test.Target, test.N)
+
+		// check that the number of results is min(N, kad.len)
+		wantN := test.N
+		if tlen := kad.Count(); tlen < test.N {
+			wantN = tlen
+		}
+
+		if len(result.nodes) != wantN {
+			t.Errorf("wrong number of nodes: got %d, want %d", len(result.nodes), wantN)
+			return false
+		}
+
+		if hasDuplicates(result.nodes) {
+			t.Errorf("result contains duplicates")
+			return false
+		}
+
+		if !sortedByDistanceTo(test.Target, result.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()
+		for i, b := range kad.buckets {
+			for j, n := range b.nodes {
+				if contains(result.nodes, n.Addr()) {
+					continue // don't run the check below for nodes in result
+				}
+				if proxCmp(test.Target, n.Addr(), farthestResult) < 0 {
+					t.Errorf("kad.le contains node that is closer to target but it's not in result")
+					t.Logf("bucket %v, item %v\n", i, j)
+					t.Logf("  Target:          %x", test.Target)
+					t.Logf("  Farthest Result: %x", farthestResult)
+					t.Logf("  ID:              %x (%d)", n.Addr(), kad.proximityBin(n.Addr()))
+					return false
+				}
+			}
+		}
+		return true
+	}
+	if err := quick.Check(test, quickcfg); err != nil {
+		t.Error(err)
+	}
+}
+
+type proxTest struct {
+	add     bool
+	index   int
+	address Address
+}
+
+var (
+	addresses []Address
+)
+
+func TestProxAdjust(t *testing.T) {
+	t.Parallel()
+	LogInit(logger.DebugLevel)
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	self := gen(Address{}, r).(Address)
+
+	kad := New(self)
+	kad.MaxProx = 10
+	kad.Start()
+	var err error
+	for i := 0; i < 100; i++ {
+		a := gen(Address{}, r).(Address)
+		addresses = append(addresses, a)
+		err = kad.AddNode(&testNode{addr: a})
+		fmt.Printf("add node: %x (%v)\n", a, kad.proximityBin(a))
+		if err != nil {
+			t.Errorf("backend not accepting node")
+			return
+		}
+		fmt.Printf("MaxProxBinSize: %d, proxSize: %d, proxLimit: %d, count: %d\n", kad.MaxProxBinSize, kad.proxSize, kad.proxLimit, kad.count)
+		if !kad.proxCheck(t) {
+			return
+		}
+	}
+
+	test := func(test *proxTest) bool {
+		node := &testNode{test.address}
+		a := test.address
+		if test.add {
+			kad.AddNode(node)
+			fmt.Printf("add node: %x (%v)\n", a, kad.proximityBin(a))
+		} else {
+			kad.RemoveNode(node)
+			fmt.Printf("remove node: %x (%v)\n", common.ToHex(a[:]), kad.proximityBin(a))
+		}
+		fmt.Printf("MaxProxBinSize: %d, proxSize: %d, proxLimit: %d, count: %d\n", kad.MaxProxBinSize, kad.proxSize, kad.proxLimit, kad.count)
+		return kad.proxCheck(t)
+	}
+	if err := quick.Check(test, quickcfg); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestSaveLoad(t *testing.T) {
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	addresses := gen([]Address{}, r).([]Address)
+	self := addresses[0]
+	kad := New(self)
+	kad.MaxProx = 10
+	kad.Start()
+	var err error
+	for _, a := range addresses[1:] {
+		err = kad.AddNode(&testNode{addr: a})
+		if err != nil {
+			t.Errorf("backend not accepting node")
+			return
+		}
+	}
+	nodes := kad.GetNodes(self, 100).nodes
+	path := "/tmp/bzz.peers"
+	kad.Stop(path)
+	kad = New(self)
+	kad.Start()
+	kad.Load(path)
+	for _, b := range kad.DB() {
+		for _, node := range b {
+			node.node = &testNode{node.Address}
+			err = kad.AddNode(node.node)
+			if err != nil {
+				t.Errorf("backend not accepting node")
+				return
+			}
+		}
+	}
+	loadednodes := kad.GetNodes(self, 100).nodes
+	for i, node := range loadednodes {
+		if nodes[i].Addr() != node.Addr() {
+			t.Errorf("node mismatch at %d/%d", i, len(nodes))
+		}
+	}
+}
+
+func (self *Kademlia) proxCheck(t *testing.T) bool {
+	var sum, i int
+	var b *bucket
+	for i, b = range self.buckets {
+		l := len(b.nodes)
+		// if we are in the high prox multibucket
+		if i >= self.proxLimit {
+			// unless it starts with an empty bucket, count the size
+			if l > 0 || sum > 0 {
+				sum += l
+			}
+		} else if l == 0 {
+			t.Errorf("bucket %d empty, yet proxLimit is %d", len(b.nodes), self.proxLimit)
+			return false
+		}
+	}
+	// check if merged high prox bucket does not exceed size
+	if sum > 0 {
+		if sum > self.MaxProxBinSize {
+			t.Errorf("bucket %d is empty, yet proxSize is %d", i, self.proxSize)
+			return false
+		}
+		if sum != self.proxSize {
+			t.Errorf("proxSize incorrect, expected %v, got %v", sum, self.proxSize)
+			return false
+		}
+	}
+	return true
+}
+
+type getNodesTest struct {
+	Self   Address
+	Target Address
+	All    []Node
+	N      int
+}
+
+func (c getNodesTest) String() string {
+	return fmt.Sprintf("A: %x\nT: %x\n(%d)\n", c.Self, c.Target, c.N)
+}
+
+func (*getNodesTest) Generate(rand *rand.Rand, size int) reflect.Value {
+	t := &getNodesTest{
+		Self:   gen(Address{}, rand).(Address),
+		Target: gen(Address{}, rand).(Address),
+		N:      rand.Intn(bucketSize),
+	}
+	for _, a := range gen([]Address{}, rand).([]Address) {
+		t.All = append(t.All, &testNode{addr: a})
+	}
+	return reflect.ValueOf(t)
+}
+
+func (*proxTest) Generate(rand *rand.Rand, size int) reflect.Value {
+	var add bool
+	if rand.Intn(1) == 0 {
+		add = true
+	}
+	var t *proxTest
+	if add {
+		t = &proxTest{
+			address: gen(Address{}, rand).(Address),
+			add:     add,
+		}
+	} else {
+		t = &proxTest{
+			index: rand.Intn(len(addresses)),
+			add:   add,
+		}
+	}
+	return reflect.ValueOf(t)
+}
+
+func hasDuplicates(slice []Node) bool {
+	seen := make(map[Address]bool)
+	for _, node := range slice {
+		if seen[node.Addr()] {
+			return true
+		}
+		seen[node.Addr()] = true
+	}
+	return false
+}
+
+func contains(nodes []Node, addr Address) bool {
+	for _, n := range nodes {
+		if n.Addr() == addr {
+			return true
+		}
+	}
+	return false
+}
+
+// gen wraps quick.Value so it's easier to use.
+// it generates a random value of the given value's type.
+func gen(typ interface{}, rand *rand.Rand) interface{} {
+	v, ok := quick.Value(reflect.TypeOf(typ), rand)
+	if !ok {
+		panic(fmt.Sprintf("couldn't generate random value of type %T", typ))
+	}
+	return v.Interface()
+}
+
+func (Address) Generate(rand *rand.Rand, size int) reflect.Value {
+	var id Address
+	// m := rand.Intn(len(id))
+	for i := 0; i < len(id); i++ {
+		id[i] = byte(rand.Uint32())
+	}
+	return reflect.ValueOf(id)
+}