Initial code of DHT Lite. Requires more testing and iterations (pun intended).

dht/DHT Lite.go

@@ -0,0 +1,189 @@
+/*
+File Name:  DHT Lite.go
+Copyright:  2021 Peernet s.r.o.
+Author:     Peter Kleissner
+
+A "lite" DHT implementation without any direct network and store code. There is really no reason for any of the heavy network implementation to be part of this.
+*/
+
+package dht
+
+import (
+	"bytes"
+	"errors"
+	"sort"
+	"time"
+)
+
+// IterateX are actions on the DHT
+const (
+	IterateStore     = iota // Store information in the network
+	IterateFindNode         // Find a node
+	IterateFindValue        // Find a value
+)
+
+// DHT represents the state of the local node in the distributed hash table
+type DHT struct {
+	ht *hashTable
+
+	// A small number representing the degree of parallelism in network calls.
+	// The alpha amount of nodes will be contacted in parallel for finding the target.
+	alpha int
+
+	// Functions below must be set and provided by the caller.
+
+	// ShouldEvict determines whether the given node shall be evicted
+	ShouldEvict func(node *Node) bool
+
+	// SendStore sends a store message to the remote node. I.e. asking it to store the given key-value.
+	SendStore func(node *Node, key []byte, value []byte)
+
+	// SendRequest sends an information request to the remote node. I.e. requesting information.
+	// The returned results channel will be closed when no more results are to be expected.
+	SendRequest func(request *InformationRequest, nodes []*Node) (results chan *message2)
+
+	// The maximum time to wait for a response to any message
+	TMsgTimeout time.Duration
+}
+
+// NewDHT initializes a new DHT node with default values.
+// Store must be set by the caller.
+func NewDHT(store Store, self Node, bits, bucketSize int) *DHT {
+	return &DHT{
+		ht:          newHashTable(self, bits, bucketSize),
+		alpha:       3,
+		TMsgTimeout: 2 * time.Second,
+	}
+}
+
+// NumNodes returns the total number of nodes stored in the local routing table
+func (dht *DHT) NumNodes() int {
+	return dht.ht.totalNodes()
+}
+
+// Nodes returns the nodes themselves sotred in the routing table.
+func (dht *DHT) Nodes() []*Node {
+	return dht.ht.Nodes()
+}
+
+// GetSelfID returns the identifier of the local node
+func (dht *DHT) GetSelfID() []byte {
+	return dht.ht.Self.ID
+}
+
+// AddNode adds a node into the appropriate k bucket. These buckets are stored in big-endian order so we look at the bits from right to left in order to find the appropriate bucket.
+func (dht *DHT) AddNode(node *Node) {
+	// The previous code made an immediate ping to the oldest node to "ping the oldest node to find out if it responds back in a reasonable amount of time. If not - remove it."
+	// In DHT Lite, however, it will be up to the caller to determine nodes to remove.
+	dht.ht.insertNode(node, dht.ShouldEvict)
+}
+
+// RemoveNode removes a node
+func (dht *DHT) RemoveNode(ID []byte) {
+	dht.ht.removeNode(ID)
+}
+
+// GetClosestContacts returns the closes contacts in the hash table
+func (dht *DHT) GetClosestContacts(count int, target []byte, ignoredNodes ...Node) []*Node {
+	closest := dht.ht.getClosestContacts(count, target, ignoredNodes...)
+	return closest.Nodes
+}
+
+// Store stores data on the network. This will trigger an IterateStore message.
+func (dht *DHT) Store(key, data []byte) (err error) {
+	_, _, err = dht.iterate(IterateStore, key[:], data)
+	return err
+}
+
+// Get retrieves data from the network using key
+func (dht *DHT) Get(key []byte) (value []byte, found bool, err error) {
+	value, _, err = dht.iterate(IterateFindValue, key, nil)
+	return value, value != nil, err
+}
+
+// FindNode finds the target node in the network
+func (dht *DHT) FindNode(key []byte) (value []byte, found bool, err error) {
+	value, _, err = dht.iterate(IterateFindNode, key, nil)
+	return value, value != nil, err
+}
+
+// Iterate does an iterative search through the network. This can be done
+// for multiple reasons. These reasons include:
+//     IterateStore - Used to store new information in the network.
+//     IterateFindNode - Used to bootstrap the network.
+//     IterateFindValue - Used to find a value among the network given a key.
+func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closest []*Node, err error) {
+	if len(target) != dht.ht.bBits {
+		return nil, nil, errors.New("invalid key")
+	} else if t < IterateStore || t > IterateFindValue {
+		return nil, nil, errors.New("unknown iterate type")
+	}
+
+	sl := dht.ht.getClosestContacts(dht.alpha, target)
+
+	// We keep a reference to the closestNode. If after performing a search we do not find a closer node, we stop searching.
+	if len(sl.Nodes) == 0 {
+		return nil, nil, nil
+	}
+
+	// According to the Kademlia white paper, after a round of FIND_NODE RPCs fails to provide a node closer than closestNode, we should send a
+	// FIND_NODE RPC to all remaining nodes in the shortlist that have not yet been contacted.
+	queryRest := false
+
+	closestNode := sl.Nodes[0]
+
+	for {
+		resultsChan := dht.SendRequest(&InformationRequest{Action: t, Key: target}, sl.GetUncontacted(dht.alpha, !queryRest))
+		results := infoCollectResults(resultsChan, dht.TMsgTimeout)
+
+		for _, result := range results {
+			if result.Error != nil {
+				sl.RemoveNode(result.SenderID)
+				continue
+			}
+			switch t {
+			case IterateFindNode:
+				sl.AppendUniqueNodes(result.Closest...)
+				// TODO: Accept contact info?
+			case IterateFindValue:
+				// When an IterateFindValue succeeds, the initiator COULD store the key/value pair at the closest node seen which did not return the value.
+				if len(result.Data) > 0 {
+					return result.Data, nil, nil
+				}
+				sl.AppendUniqueNodes(result.Closest...)
+			case IterateStore:
+				sl.AppendUniqueNodes(result.Closest...)
+			}
+		}
+
+		sort.Sort(sl)
+
+		// If closestNode is unchanged then we are done
+		if bytes.Compare(sl.Nodes[0].ID, closestNode.ID) == 0 || queryRest {
+			// We are done
+			switch t {
+			case IterateFindNode:
+				if !queryRest {
+					queryRest = true
+					continue
+				}
+				return nil, sl.Nodes, nil
+
+			case IterateFindValue:
+				return nil, sl.Nodes, nil
+
+			case IterateStore:
+				for i, node := range sl.Nodes {
+					if i >= dht.ht.bSize {
+						break
+					}
+
+					dht.SendStore(node, target, data)
+				}
+				return nil, nil, nil
+			}
+		}
+
+		closestNode = sl.Nodes[0]
+	}
+}

dht/Hash Table.go

@@ -8,7 +8,6 @@ package dht
 
 import (
 	"bytes"
-	"errors"
 	"math"
 	"math/big"
 	"math/rand"
@@ -34,7 +33,7 @@ type hashTable struct {
 	// [ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ]
 	//  ^                                                           ^
 	//  └ Least recently seen                    Most recently seen ┘
-	RoutingTable [][]Node // bBits x bSize
+	RoutingTable [][]*Node // bBits x bSize
 
 	mutex *sync.Mutex
 }
@@ -47,25 +46,26 @@ func newHashTable(n Node, bits, bucketSize int) *hashTable {
 		Self:  n,
 	}
 
-	ht.RoutingTable = make([][]Node, ht.bBits)
+	ht.RoutingTable = make([][]*Node, ht.bBits)
 	return ht
 }
 
-func (ht *hashTable) markNodeAsSeen(node []byte) {
+func (ht *hashTable) markNodeAsSeen(ID []byte) {
 	ht.mutex.Lock()
 	defer ht.mutex.Unlock()
-	index := getBucketIndexFromDifferingBit(ht.bBits, ht.Self.ID, node)
+	index := ht.getBucketIndexFromDifferingBit(ID)
 	bucket := ht.RoutingTable[index]
 	nodeIndex := -1
 	for i, v := range bucket {
-		if bytes.Compare(v.ID, node) == 0 {
+		if bytes.Compare(v.ID, ID) == 0 {
 			nodeIndex = i
 			break
 		}
 	}
 
 	if nodeIndex == -1 {
-		panic(errors.New("Tried to mark nonexistent node as seen"))
+		//errors.New("Tried to mark nonexistent node as seen")
+		return
 	}
 
 	n := bucket[nodeIndex]
@@ -90,9 +90,8 @@ func (ht *hashTable) doesNodeExistInBucket(bucket int, node []byte) bool {
 func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ...Node) *shortList {
 	ht.mutex.Lock()
 	defer ht.mutex.Unlock()
-	// First we need to build the list of adjacent indices to our target
-	// in order
-	index := getBucketIndexFromDifferingBit(ht.bBits, ht.Self.ID, target)
+	// First we need to build the list of adjacent indices to our target in order
+	index := ht.getBucketIndexFromDifferingBit(target)
 	indexList := []int{index}
 	for i, j := index-1, index+1; len(indexList) < ht.bBits; i, j = i-1, j+1 {
 		if j < ht.bBits {
@@ -103,7 +102,7 @@ func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ...
 		}
 	}
 
-	sl := &shortList{}
+	sl := newShortList()
 
 	leftToAdd := num
 
@@ -119,7 +118,7 @@ func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ...
 				}
 			}
 			if !ignored {
-				sl.AppendUnique(ht.RoutingTable[index][i])
+				sl.AppendUniqueNodes(ht.RoutingTable[index][i])
 				leftToAdd--
 				if leftToAdd == 0 {
 					break
@@ -133,11 +132,10 @@ func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ...
 	return sl
 }
 
-func (ht *hashTable) insertNode(node Node, pinger func(Node) error) {
-	index := getBucketIndexFromDifferingBit(ht.bBits, ht.Self.ID, node.ID)
+func (ht *hashTable) insertNode(node *Node, shouldEvict func(*Node) bool) {
+	index := ht.getBucketIndexFromDifferingBit(node.ID)
 
-	// Make sure node doesn't already exist
-	// If it does, mark it as seen
+	// If the node already exist, mark it as seen
 	if ht.doesNodeExistInBucket(index, node.ID) {
 		ht.markNodeAsSeen(node.ID)
 		return
@@ -151,7 +149,7 @@ func (ht *hashTable) insertNode(node Node, pinger func(Node) error) {
 	bucket := ht.RoutingTable[index]
 
 	if len(bucket) == ht.bSize {
-		if pinger(bucket[0]) != nil {
+		if shouldEvict(bucket[0]) {
 			bucket = append(bucket, node)
 			bucket = bucket[1:]
 		}
@@ -166,7 +164,7 @@ func (ht *hashTable) removeNode(ID []byte) {
 	ht.mutex.Lock()
 	defer ht.mutex.Unlock()
 
-	index := getBucketIndexFromDifferingBit(ht.bBits, ht.Self.ID, ID)
+	index := ht.getBucketIndexFromDifferingBit(ID)
 	bucket := ht.RoutingTable[index]
 
 	for i, v := range bucket {
@@ -250,10 +248,10 @@ func (ht *hashTable) getRandomIDFromBucket(bucket int) []byte {
 	return id
 }
 
-func (ht *hashTable) lastSeenBefore(cutoff time.Time) (nodes []Node) {
+func (ht *hashTable) lastSeenBefore(cutoff time.Time) (nodes []*Node) {
 	ht.mutex.Lock()
 	defer ht.mutex.Unlock()
-	nodes = make([]Node, 0, ht.bSize)
+	nodes = make([]*Node, 0, ht.bSize)
 	for _, v := range ht.RoutingTable {
 		for _, n := range v {
 			if n.LastSeen.Before(cutoff) {
@@ -267,18 +265,18 @@ func (ht *hashTable) lastSeenBefore(cutoff time.Time) (nodes []Node) {
 	return nodes
 }
 
-func getBucketIndexFromDifferingBit(b int, id1 []byte, id2 []byte) int {
+func (ht *hashTable) getBucketIndexFromDifferingBit(id1 []byte) int {
 	// Look at each byte from left to right
 	for j := 0; j < len(id1); j++ {
 		// xor the byte
-		xor := id1[j] ^ id2[j]
+		xor := id1[j] ^ ht.Self.ID[j]
 
 		// check each bit on the xored result from left to right in order
 		for i := 0; i < 8; i++ {
 			if hasBit(xor, uint(i)) {
 				byteIndex := j * 8
 				bitIndex := i
-				return b - (byteIndex + bitIndex) - 1
+				return ht.bBits - (byteIndex + bitIndex) - 1
 			}
 		}
 	}
@@ -298,10 +296,10 @@ func (ht *hashTable) totalNodes() int {
 	return total
 }
 
-func (ht *hashTable) Nodes() (nodes []Node) {
+func (ht *hashTable) Nodes() (nodes []*Node) {
 	ht.mutex.Lock()
 	defer ht.mutex.Unlock()
-	nodes = make([]Node, 0, ht.bSize)
+	nodes = make([]*Node, 0, ht.bSize)
 	for _, v := range ht.RoutingTable {
 		nodes = append(nodes, v...)
 	}

dht/Information Request.go

@@ -0,0 +1,43 @@
+/*
+File Name:  Information Request.go
+Copyright:  2021 Peernet s.r.o.
+Author:     Peter Kleissner
+
+Information requests are asynchronous queries sent to nodes.
+*/
+
+package dht
+
+import "time"
+
+// InformationRequest is an asynchronous request sent. It tracks any asynchronous replies and handles timeouts.
+type InformationRequest struct {
+	Action int    // IterateFindNode or IterateFindValue
+	Key    []byte // Target key
+
+	// TODO: Include results channel? Timeout settings? Cancelation?
+}
+
+type message2 struct {
+	SenderID []byte // Sender of this message
+	Data     []byte
+	Closest  []*Node
+	Error    error
+}
+
+// infoCollectResults collects all information request responses within the given timeout.
+func infoCollectResults(resultChan chan *message2, timeout time.Duration) (results []*message2) {
+	for {
+		select {
+		case result, ok := <-resultChan:
+			if !ok {
+				return
+			}
+			results = append(results, result)
+		case <-time.After(timeout):
+			// send cancelation signal ?
+			//close(resultChan)
+			return
+		}
+	}
+}

dht/Node.go

@@ -21,14 +21,22 @@ type Node struct {
 	LastSeen time.Time
 }
 
-// nodeList is used in order to sort a list of arbitrary nodes against a
-// comparator. These nodes are sorted by xor distance
+// shortList is used in order to sort a list of arbitrary nodes against a comparator. These nodes are sorted by xor distance
 type shortList struct {
 	// Nodes are a list of nodes to be compared
-	Nodes []Node
+	Nodes []*Node
 
 	// Comparator is the ID to compare to
 	Comparator []byte
+
+	// Contacted is a list of nodes that are considered contacted
+	Contacted map[string]bool
+}
+
+func newShortList() *shortList {
+	return &shortList{
+		Contacted: make(map[string]bool),
+	}
 }
 
 func areNodesEqual(n1 *Node, n2 *Node, allowNilID bool) bool {
@@ -46,30 +54,16 @@ func areNodesEqual(n1 *Node, n2 *Node, allowNilID bool) bool {
 	return true
 }
 
-func (n *shortList) RemoveNode(node Node) {
+func (n *shortList) RemoveNode(ID []byte) {
 	for i := 0; i < n.Len(); i++ {
-		if bytes.Compare(n.Nodes[i].ID, node.ID) == 0 {
+		if bytes.Compare(n.Nodes[i].ID, ID) == 0 {
 			n.Nodes = append(n.Nodes[:i], n.Nodes[i+1:]...)
 			return
 		}
 	}
 }
 
-func (n *shortList) AppendUniqueNodes(nodes ...Node) {
-	for _, vv := range nodes {
-		exists := false
-		for _, v := range n.Nodes {
-			if bytes.Compare(v.ID, vv.ID) == 0 {
-				exists = true
-			}
-		}
-		if !exists {
-			n.Nodes = append(n.Nodes, vv)
-		}
-	}
-}
-
-func (n *shortList) AppendUnique(nodes ...Node) {
+func (n *shortList) AppendUniqueNodes(nodes ...*Node) {
 	for _, vv := range nodes {
 		exists := false
 		for _, v := range n.Nodes {
@@ -108,3 +102,24 @@ func getDistance(id1 []byte, id2 []byte) *big.Int {
 	result := new(big.Int).Xor(buf1, buf2)
 	return result
 }
+
+// GetUncontacted returns a list of uncontacted nodes. Each returned node will be marked as contacted.
+func (n *shortList) GetUncontacted(count int, useCount bool) (Nodes []*Node) {
+	for _, node := range n.Nodes {
+		if useCount && count <= 0 {
+			break
+		}
+
+		// Don't contact nodes already contacted
+		if n.Contacted[string(node.ID)] == true {
+			continue
+		}
+
+		n.Contacted[string(node.ID)] = true
+		Nodes = append(Nodes, node)
+
+		count--
+	}
+
+	return Nodes
+}

dht/README.md

@@ -1,3 +1,9 @@
-# DHT
+# DHT Lite
 
-This code is a fork from https://github.com/james-lawrence/kademlia with modifications for proper abstraction. All networking code was removed from the original one. This package shall only provide DHT fuctionality.
+This code is a fork from https://github.com/james-lawrence/kademlia and https://github.com/prettymuchbryce/kademlia with modifications for proper abstraction. All networking code was removed from the original one. This package shall only provide DHT fuctionality.
+
+The following functions are not handled here and must be done by the caller, if desired:
+* Remove nodes that are deemed inactive via `dht.RemoveNode`
+* Provide a function `ShouldEvict` to determine if a node shall be evicted in favor of another one
+* Refreshing buckets: Pick random node in the unrefreshed bucket, ask closest alpha neighbors. This makes sure the bucket remains accessible?
+* The actual store data functions (and associated replication/expiration) are not provided, only the functionality to traverse through the network.

dht/Store.go

@@ -32,7 +32,7 @@ type Store interface {
 }
 
 // NewMemoryStore create a properly initialized memory store.
-func NewMemoryStore(dht *DHT) *MemoryStore {
+func NewMemoryStore() *MemoryStore {
 	return &MemoryStore{
 		data:         make(map[string][]byte),
 		mutex:        &sync.Mutex{},