Peernet-Archive/core
1
0
Fork 0
mirror of https://github.com/PeernetOfficial/core.git synced 2026-07-17 02:47:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add information request list to DHT structure. It keeps track of requests and matches responses.

Browse Source 
Break up SendRequest into individual functions SendRequestStore and SendRequestFindNode, which improves readability.
Alpha parameter must be now passed. MarkNodeAsSeen is exported.
Breaking up dht.iterate and embedding into 3 functions Get, FindNode, and FindValue.
This commit is contained in:
Kleissner
2021-04-09 16:07:20 +02:00
parent 7927e18572
commit c44e57a0a0
3 changed files with 202 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
dht/DHT Lite.go
View File

@@ -12,16 +12,10 @@ import (
"bytes"
"errors"
"sort"
"sync"
"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
@@ -30,28 +24,35 @@ type DHT struct {
// The alpha amount of nodes will be contacted in parallel for finding the target.
alpha int
// ListIR keeps track of all active information requests.
ListIR map[string]*InformationRequest
listIRmutex sync.RWMutex
// 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)
// SendRequestStore sends a store message to the remote node. I.e. asking it to store the given key-value.
SendRequestStore 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)
// SendRequestFindNode sends an information request to find a particular node. nodes are the nodes to send the request to.
SendRequestFindNode func(request *InformationRequest)
// The maximum time to wait for a response to any message
// SendRequestFindValue sends an information request to find data. nodes are the nodes to send the request to.
SendRequestFindValue func(request *InformationRequest)
// The maximum time to wait for a response to any message in Store, Get, FindNode
TMsgTimeout time.Duration
}
// NewDHT initializes a new DHT node with default values.
func NewDHT(self *Node, bits, bucketSize int) *DHT {
func NewDHT(self *Node, bits, bucketSize, alpha int) *DHT {
return &DHT{
ht: newHashTable(self, bits, bucketSize),
alpha: 3,
alpha: alpha,
TMsgTimeout: 2 * time.Second,
ListIR: make(map[string]*InformationRequest),
}
}
@@ -88,41 +89,110 @@ func (dht *DHT) GetClosestContacts(count int, target []byte, filterFunc NodeFilt
return closest.Nodes
}
// Store stores data on the network. This will trigger an IterateStore message.
// MarkNodeAsSeen marks a node as seen, which pushes it to the top in the bucket list.
func (dht *DHT) MarkNodeAsSeen(ID []byte) {
dht.ht.markNodeAsSeen(dht.ht.getBucketIndexFromDifferingBit(ID), ID)
}
// ---- Synchronous network query functions below ----
// Store stores data on the network.
func (dht *DHT) Store(key, data []byte) (err error) {
_, _, err = dht.iterate(IterateStore, key[:], data)
return err
if len(key) != dht.ht.bBits {
return errors.New("invalid key size")
}
// Keep a reference to closestNode. If after performing a search we do not find a closer node, we stop searching.
sl := dht.ht.getClosestContacts(dht.alpha, key, nil)
if len(sl.Nodes) == 0 {
return nil
}
closestNode := sl.Nodes[0]
for {
info := dht.NewInformationRequest(ActionFindNode, key, sl.GetUncontacted(dht.alpha, true))
dht.SendRequestFindNode(info)
results := info.CollectResults(dht.TMsgTimeout)
for _, result := range results {
if result.Error != nil {
sl.RemoveNode(result.SenderID)
continue
}
sl.AppendUniqueNodes(result.Closest...)
}
sort.Sort(sl)
// If closestNode is unchanged then we are done
if bytes.Equal(sl.Nodes[0].ID, closestNode.ID) {
for i, node := range sl.Nodes {
if i >= dht.ht.bSize {
break
}
dht.SendRequestStore(node, key, data)
}
return nil
}
closestNode = sl.Nodes[0]
}
}
// 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
if len(key) != dht.ht.bBits {
return nil, false, errors.New("invalid key size")
}
// Keep a reference to closestNode. If after performing a search we do not find a closer node, we stop searching.
sl := dht.ht.getClosestContacts(dht.alpha, key, nil)
if len(sl.Nodes) == 0 {
return nil, false, nil
}
closestNode := sl.Nodes[0]
for {
info := dht.NewInformationRequest(ActionFindValue, key, sl.GetUncontacted(dht.alpha, true))
dht.SendRequestFindValue(info)
results := info.CollectResults(dht.TMsgTimeout)
for _, result := range results {
if result.Error != nil {
sl.RemoveNode(result.SenderID)
continue
}
if len(result.Data) > 0 {
return result.Data, false, nil
}
sl.AppendUniqueNodes(result.Storing...) // TODO: Assign higher priority, skip closesNode check.
sl.AppendUniqueNodes(result.Closest...)
}
sort.Sort(sl)
// If closestNode is unchanged then we are done
if bytes.Equal(sl.Nodes[0].ID, closestNode.ID) {
return nil, false, nil
}
closestNode = sl.Nodes[0]
}
}
// 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(action 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 action < IterateStore || action > IterateFindValue {
return nil, nil, errors.New("unknown iterate type")
if len(key) != dht.ht.bBits {
return nil, false, errors.New("invalid key size")
}
sl := dht.ht.getClosestContacts(dht.alpha, target, nil)
// We keep a reference to the closestNode. If after performing a search we do not find a closer node, we stop searching.
// Keep a reference to closestNode. If after performing a search we do not find a closer node, we stop searching.
sl := dht.ht.getClosestContacts(dht.alpha, key, nil)
if len(sl.Nodes) == 0 {
return nil, nil, nil
return nil, false, 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
@@ -132,8 +202,8 @@ func (dht *DHT) iterate(action int, target []byte, data []byte) (value []byte, c
closestNode := sl.Nodes[0]
for {
info := NewInformationRequest(action, target)
dht.SendRequest(info, sl.GetUncontacted(dht.alpha, !queryRest))
info := dht.NewInformationRequest(ActionFindNode, key, sl.GetUncontacted(dht.alpha, !queryRest))
dht.SendRequestFindNode(info)
results := info.CollectResults(dht.TMsgTimeout)
for _, result := range results {
@@ -141,47 +211,20 @@ func (dht *DHT) iterate(action int, target []byte, data []byte) (value []byte, c
sl.RemoveNode(result.SenderID)
continue
}
switch action {
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...)
}
sl.AppendUniqueNodes(result.Closest...)
// TODO: Check if node was found.
}
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 action {
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
if bytes.Equal(sl.Nodes[0].ID, closestNode.ID) || queryRest {
if !queryRest {
queryRest = true
continue
}
return nil, false, nil
}
closestNode = sl.Nodes[0]

91
dht/Information Request.go
View File

@@ -10,26 +10,45 @@ package dht
import (
"sync"
"sync/atomic"
"time"
)
// InformationRequest is an asynchronous request sent. It tracks any asynchronous replies and handles timeouts.
// InformationRequest is an asynchronous request sent to nodes. It tracks any asynchronous replies and handles timeouts.
type InformationRequest struct {
Action int // IterateFindNode or IterateFindValue
Key []byte // Target key
Action int // ActionX
Key []byte // Key that is being queried
ResultChan chan *NodeMessage // Result channel
ActiveNodes uint64 // Number of nodes actively handling the request.
Nodes []*Node // Nodes that are receiving the request.
IsTerminated bool // If true, it was signaled for termination
TerminateSignal chan interface{} // gets closed on termination signal, can be used in select via "case _ = <- network.terminateSignal:"
TerminateSignal chan struct{} // gets closed on termination signal, can be used in select via "case _ = <- network.terminateSignal:"
sync.Mutex // for sychronized closing
}
// NewInformationRequest creates a new information request
func NewInformationRequest(Action int, Key []byte) (ir *InformationRequest) {
return &InformationRequest{
ResultChan: make(chan *NodeMessage),
Action: Action,
Key: Key,
// Actions for performing the information request
const (
ActionFindNode = iota // Find a node
ActionFindValue // Find a value
)
// NewInformationRequest creates a new information request and adds it to the list.
// It marks the count of nodes as active, meaning the caller should later decrease it via ActiveNodesSub.
func (dht *DHT) NewInformationRequest(Action int, Key []byte, Nodes []*Node) (ir *InformationRequest) {
ir = &InformationRequest{
ResultChan: make(chan *NodeMessage),
TerminateSignal: make(chan struct{}),
Action: Action,
Key: Key,
Nodes: Nodes,
ActiveNodes: uint64(len(Nodes)),
}
for _, node := range Nodes {
dht.irAdd(node.ID, ir)
}
return
}
// CollectResults collects all information request responses within the given timeout.
@@ -65,5 +84,53 @@ func (ir *InformationRequest) Terminate() {
close(ir.ResultChan)
}
// TODO: Incoming information request responses should be handled in batches, i.e. every X ms (for example 100ms) without waiting for all results to finish.
// This could seriously speed up discovery.
// ActiveNodesAdd increases the number of active nodes handling this request
func (ir *InformationRequest) ActiveNodesAdd(count uint64) {
atomic.AddUint64(&ir.ActiveNodes, count)
}
// ActiveNodesSub decreases the number of active nodes handling this request
func (ir *InformationRequest) ActiveNodesSub(count uint64) {
if atomic.AddUint64(&ir.ActiveNodes, ^uint64(count-1)) <= 0 {
// If the counter reaches 0, it means no nodes are handling this request anymore -> terminate it.
ir.Terminate()
}
}
// ---- keep track of information requests ----
// irRemove add the information request to the list
// If a request to the same node with the same key exists, it is overwritten. This will be improved. TODO: A nonce could easily fix that?
func (dht *DHT) irAdd(nodeID []byte, request *InformationRequest) {
dht.listIRmutex.Lock()
defer dht.listIRmutex.Unlock()
// The list only supports one request per target node and key. This should be improved.
lookupKey := string(nodeID) + string(request.Key)
dht.ListIR[lookupKey] = request
// auto remove from list upon termination
go func(lookupKey string, terminateChan <-chan struct{}) {
<-terminateChan
dht.irRemove(lookupKey)
}(lookupKey, request.TerminateSignal)
}
// irRemove removes the information request from the list
func (dht *DHT) irRemove(lookupKey string) {
dht.listIRmutex.Lock()
defer dht.listIRmutex.Unlock()
delete(dht.ListIR, lookupKey)
}
// IRLookup looks up the information request based on the peers public key and hash
func (dht *DHT) IRLookup(nodeID []byte, hash []byte) (request *InformationRequest) {
dht.listIRmutex.RLock()
defer dht.listIRmutex.RUnlock()
lookupKey := string(nodeID) + string(request.Key)
request = dht.ListIR[lookupKey]
return request
}

9
dht/Node.go
View File

@@ -127,10 +127,11 @@ func (n *shortList) GetUncontacted(count int, useCount bool) (Nodes []*Node) {
// NodeMessage is a message sent by a node
type NodeMessage struct {
SenderID []byte // Sender of this message
Data []byte
Closest []*Node
Error error
SenderID []byte // Sender of the message
Data []byte // FIND_VALUE: Actual data
Closest []*Node // FIND_VALUE, FIND_NODE: Closest nodes to the requested key
Storing []*Node // FIND_VALUE: Nodes known to store the value
Error error // To be removed
}
// NodeFilterFunc is called to filter nodes based on the callers choice