From f1209719447d75edea0f90013f9308c69b041ce9 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 8 Mar 2021 03:15:52 +0100 Subject: [PATCH] Initial code of DHT Lite. Requires more testing and iterations (pun intended). --- dht/DHT Lite.go | 189 +++++++++++++++++++++++++++++++++++++ dht/Hash Table.go | 48 +++++----- dht/Information Request.go | 43 +++++++++ dht/Node.go | 55 +++++++---- dht/README.md | 10 +- dht/Store.go | 2 +- 6 files changed, 299 insertions(+), 48 deletions(-) create mode 100644 dht/DHT Lite.go create mode 100644 dht/Information Request.go diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go new file mode 100644 index 0000000..49d3b05 --- /dev/null +++ b/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] + } +} diff --git a/dht/Hash Table.go b/dht/Hash Table.go index c170558..9f5bffe 100644 --- a/dht/Hash Table.go +++ b/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...) } diff --git a/dht/Information Request.go b/dht/Information Request.go new file mode 100644 index 0000000..d32cebe --- /dev/null +++ b/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 + } + } +} diff --git a/dht/Node.go b/dht/Node.go index 671838b..18dfd54 100644 --- a/dht/Node.go +++ b/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 +} diff --git a/dht/README.md b/dht/README.md index b8c8418..fbc94e2 100644 --- a/dht/README.md +++ b/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. diff --git a/dht/Store.go b/dht/Store.go index 548d31c..46a7769 100644 --- a/dht/Store.go +++ b/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{},