diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go index f0dd134..b87eef1 100644 --- a/dht/DHT Lite.go +++ b/dht/DHT Lite.go @@ -37,16 +37,28 @@ type DHT struct { // SendRequestFindValue sends an information request to find data. nodes are the nodes to send the request to. SendRequestFindValue func(request *InformationRequest) + // FilterSearchStatus is called with updates of searches in the DHT + FilterSearchStatus func(client *SearchClient, function, format string, v ...interface{}) + // The maximum time to wait for a response to any message in Store, Get, FindNode TMsgTimeout time.Duration + + // TimeoutSearch is the maximum time a search may take. + TimeoutSearch time.Duration + + // TimeoutIR is the maximum an information request to a node may take. + TimeoutIR time.Duration } // NewDHT initializes a new DHT node with default values. func NewDHT(self *Node, bits, bucketSize, alpha int) *DHT { return &DHT{ - ht: newHashTable(self, bits, bucketSize), - alpha: alpha, - TMsgTimeout: 2 * time.Second, + ht: newHashTable(self, bits, bucketSize), + alpha: alpha, + TMsgTimeout: 2 * time.Second, + FilterSearchStatus: func(client *SearchClient, function, format string, v ...interface{}) {}, + TimeoutSearch: 10 * time.Second, + TimeoutIR: 6 * time.Second, } } @@ -123,10 +135,6 @@ func (dht *DHT) Store(key []byte, dataSize uint64) (err error) { results := info.CollectResults(dht.TMsgTimeout) for _, result := range results { - if result.Error != nil { - sl.RemoveNode(result.SenderID) - continue - } sl.AppendUniqueNodes(result.Closest...) } @@ -169,10 +177,6 @@ func (dht *DHT) Get(key []byte) (value []byte, senderID []byte, found bool, err 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, result.SenderID, true, nil } @@ -191,51 +195,24 @@ func (dht *DHT) Get(key []byte) (value []byte, senderID []byte, found bool, err } } -// FindNode finds the target node in the network -func (dht *DHT) FindNode(key []byte) (value []byte, found bool, err error) { +// FindNode finds the target node in the network. Blocking! +// The caller may use dht.NewSearch directly and take advantage of the asynchronous response and custom timeouts. +func (dht *DHT) FindNode(key []byte) (node *Node, err error) { if len(key)*8 != dht.ht.bBits { - return nil, false, errors.New("invalid key size") + return nil, 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 + search := dht.NewSearch(ActionFindNode, key, dht.TimeoutSearch, dht.TimeoutIR) + search.LogStatus = func(function, format string, v ...interface{}) { + dht.FilterSearchStatus(search, function, format, v...) } + search.SearchAway() - // 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 { - info := dht.NewInformationRequest(ActionFindNode, key, sl.GetUncontacted(dht.alpha, !queryRest)) - 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...) - - // TODO: Check if node was found. - } - - sort.Sort(sl) - - // If closestNode is unchanged then we are done - if bytes.Equal(sl.Nodes[0].ID, closestNode.ID) || queryRest { - if !queryRest { - queryRest = true - continue - } - return nil, false, nil - } - - closestNode = sl.Nodes[0] + select { + case <-search.TerminateSignal: + return nil, nil + case result := <-search.Results: + return result.TargetNode, nil } } diff --git a/dht/Information Request.go b/dht/Information Request.go index d88f74a..fa5e0f8 100644 --- a/dht/Information Request.go +++ b/dht/Information Request.go @@ -19,6 +19,7 @@ type InformationRequest struct { Action int // ActionX Key []byte // Key that is being queried ResultChan chan *NodeMessage // Result channel + ResultChanExt chan *NodeMessage // External result channel to use instead 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 @@ -94,12 +95,17 @@ func (ir *InformationRequest) Done() { } } -// QueueResult accepts incoming results +// QueueResult accepts incoming results and queues them to the result channel. Non-blocking. func (ir *InformationRequest) QueueResult(message *NodeMessage) { ir.Lock() if !ir.IsTerminated { + targetChan := ir.ResultChan + if ir.ResultChanExt != nil { + targetChan = ir.ResultChanExt + } + select { - case ir.ResultChan <- message: + case targetChan <- message: default: } } diff --git a/dht/Node.go b/dht/Node.go index 22611a4..0510071 100644 --- a/dht/Node.go +++ b/dht/Node.go @@ -127,7 +127,6 @@ type NodeMessage struct { 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 diff --git a/dht/Search Client.go b/dht/Search Client.go new file mode 100644 index 0000000..ed3240a --- /dev/null +++ b/dht/Search Client.go @@ -0,0 +1,324 @@ +/* +File Name: Search Client.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner + +A search client runs concurrent information requests for a single query. It solves the query efficiently by using levels. +Any result that is closer to the target gets pushed down into a new lower level, which contacts nodes closer to the result. +Level are running concurrently. +*/ + +package dht + +import ( + "bytes" + "encoding/hex" + "fmt" + "sync" + "sync/atomic" + "time" +) + +// MaxConcurrentRequestsPerLevel is the default max count of active concurrent requests per level. +const MaxConcurrentRequestsPerLevel = 5 + +// MaxConcurrentRequestKnownStore is the default max count of nodes to request the data for ActionFindValue. +const MaxConcurrentRequestKnownStore = 7 + +// MaxAcceptKnownStore is maximum count accepted of known peers that store the value +const MaxAcceptKnownStore = 10 + +// MaxClosest is maximum number of closest peers accepted +const MaxClosest = 3 + +// MaxLevel defines the max level. +const MaxLevel = 10 + +// SearchClient defines a search in the distributed hash table involving multiple information requests. +// The search can be created for a node or for a value, both identified by the hash (known as the key). +type SearchClient struct { + Action int // ActionX + Key []byte // Key that is being queried + IsTerminated bool // If true, it was signaled for termination + TerminateSignal chan struct{} // gets closed on termination signal, can be used in select via "case _ = <- TerminateSignal:" + sync.Mutex // for sychronized closing + timeStart time.Time // When the search started. + timeEnd time.Time // When the search ended. + dht *DHT // DHT used + timeoutTotal time.Duration // Timeout after the entire search will be terminated client-side. + timeoutIR time.Duration // Timeout for information requests (entire roundtrip). + Results chan *SearchResult // Result channel + list *shortList // List of nodes to contact + contactedNodesMap map[string]struct{} // List of nodes already contacted + contactedNodesMutex sync.RWMutex // Sync map access + storing chan []*Node // Internal channel to signal nodes that indicate storing the searched value. + activeLevels uint64 // demo + LogStatus func(function, format string, v ...interface{}) // Filter function for status output +} + +// SearchResult is a single result to the search. Depending on the search type and parameters, multiple results may be sent. +type SearchResult struct { + Key []byte // Original key that was searched for + Action int // Original action + SenderID []byte // Sender node ID of the result + + // data for ActionFindNode + TargetNode *Node // The node that was requested. + + // data for ActionFindValue + Data []byte // Actual data +} + +// NewSearch creates a new search client. +// Action indicates the action to take (from ActionX constants), to either find a node, or a value. +// Timeout is the total time the search may take, covering all information requests. TimeoutIR is the time an information request may take. +func (dht *DHT) NewSearch(Action int, Key []byte, Timeout, TimeoutIR time.Duration) (client *SearchClient) { + client = &SearchClient{ + Action: Action, + Key: Key, + dht: dht, + timeoutTotal: Timeout, + timeoutIR: TimeoutIR, + contactedNodesMap: make(map[string]struct{}), + storing: make(chan []*Node, MaxConcurrentRequestKnownStore*2), + TerminateSignal: make(chan struct{}), + Results: make(chan *SearchResult), + LogStatus: func(function, format string, v ...interface{}) {}, + } + fmt.Printf("New search for key %s action %d\n", hex.EncodeToString(Key), Action) + + return +} + +// Terminate sends the termination signal to all workers. It is safe to call Terminate multiple times. +func (client *SearchClient) Terminate() { + client.Lock() + defer client.Unlock() + + if client.IsTerminated { + return + } + + // set the termination signal + client.IsTerminated = true + close(client.TerminateSignal) // safety guaranteed via lock + + client.timeEnd = time.Now() + + close(client.Results) + close(client.storing) +} + +// isContactedNode checks if a node was contacted +func (client *SearchClient) isContactedNode(ID []byte, Set bool) (contacted bool) { + client.contactedNodesMutex.Lock() + _, contacted = client.contactedNodesMap[string(ID)] + if Set { + client.contactedNodesMap[string(ID)] = struct{}{} + } + client.contactedNodesMutex.Unlock() + return contacted +} + +// filterUncontactedNodes returns only nodes that were not contacted so far. All nodes will be set to contacted. Limit is optional (0 for no limit). +func (client *SearchClient) filterUncontactedNodes(input []*Node, limit int) (output []*Node) { + client.contactedNodesMutex.Lock() + + for _, node := range input { + if _, ok := client.contactedNodesMap[string(node.ID)]; !ok { + output = append(output, node) + client.contactedNodesMap[string(node.ID)] = struct{}{} + + if limit > 0 { + limit-- + if limit == 0 { + break + } + } + } + } + + client.contactedNodesMutex.Unlock() + return output +} + +// SearchAway starts the search. Non-blocking! +func (client *SearchClient) SearchAway() { + client.timeStart = time.Now() + + // create the first search level and start it + client.list = client.dht.ht.getClosestContacts(MaxConcurrentRequestsPerLevel, client.Key, nil) + if len(client.list.Nodes) == 0 { + client.Terminate() + return + } + + go client.queryNodesKnownStore() + + // start the first information request + go client.startSearch(0) + + // start an automated termination function for the timeout + go func(client *SearchClient) { + // sleep + watch for closing + select { + case <-client.TerminateSignal: // exit the function on other signal + return + case <-time.After(client.timeoutTotal): + client.Terminate() + } + }(client) +} + +// sendInfoRequest sends out a new info request to the nodes +func (client *SearchClient) sendInfoRequest(nodes []*Node, resultChan chan *NodeMessage) (info *InformationRequest) { + if client.IsTerminated { + return nil + } + + for _, node := range nodes { + client.LogStatus("sendInfoRequest", "contact node %s\n", hex.EncodeToString(node.ID)) + } + + info = client.dht.NewInformationRequest(client.Action, client.Key, nodes) + info.ResultChanExt = resultChan + client.dht.SendRequestFindNode(info) + + go func() { + select { + case <-client.TerminateSignal: + case <-time.After(client.timeoutIR): + } + info.Terminate() + }() + + return info +} + +// queryNodesKnownStore queries nodes that are known to store the value. Only for ActionFindValue. +// Returned 'closest nodes' are ignored, as the queried nodes are expected to store the value. This might be adjusted in the future. +func (client *SearchClient) queryNodesKnownStore() { + // all results are redirected to a single channel + resultChan := make(chan *NodeMessage, MaxConcurrentRequestKnownStore) + + for { + select { + case <-client.TerminateSignal: + return + + case nodes := <-client.storing: + client.sendInfoRequest(nodes, resultChan) + + case result := <-resultChan: + if len(result.Data) > 0 { + client.Results <- &SearchResult{Key: client.Key, Action: client.Action, SenderID: result.SenderID, Data: result.Data} + client.Terminate() + return + } + } + } +} + +func (client *SearchClient) startSearch(level int) { + atomic.AddUint64(&client.activeLevels, 1) + defer atomic.AddUint64(&client.activeLevels, ^uint64(0)) + nestedStarted := false + + results := make(chan *NodeMessage, MaxConcurrentRequestsPerLevel*2) + + closestNode := client.list.Nodes[0] + + // start an info request + startInfoRequest := func() (info *InformationRequest) { + nodes := client.list.GetUncontacted(MaxConcurrentRequestsPerLevel, true) + if len(nodes) == 0 { + client.LogStatus("startSearch", "search in level %d aborted, no new nodes to contact\n", level) + return nil + } + client.LogStatus("startSearch", "start search in level %d contacting %d nodes\n", level, len(nodes)) + return client.sendInfoRequest(nodes, results) + } + + info := startInfoRequest() + if info == nil { + return + } + + for { + select { + case <-client.TerminateSignal: + client.LogStatus("startSearch", "search in level %d aborted, search client termination signal\n", level) + return + + case result := <-results: + + switch client.Action { + case ActionFindValue: + // search for value and it was found? + if len(result.Data) > 0 { + client.LogStatus("startSearch", "find value: sender %s: data found (%d bytes)\n", hex.EncodeToString(result.SenderID), len(result.Data)) + client.Results <- &SearchResult{Key: client.Key, Action: client.Action, SenderID: result.SenderID, Data: result.Data} + client.Terminate() + return + } + + result.Storing = client.filterUncontactedNodes(result.Storing, MaxAcceptKnownStore) + result.Closest = client.filterUncontactedNodes(result.Closest, MaxClosest) + + client.LogStatus("startSearch", "find value: sender %s: %d uncontacted nodes store and %d nodes are close to value\n", hex.EncodeToString(result.SenderID), len(result.Storing), len(result.Closest)) + + // Find value: Nodes known to store the value are queried in a separate function. + if len(result.Storing) > 0 { + client.storing <- result.Storing + } + + case ActionFindNode: + // search for node and it was found? + for _, closePeer := range result.Closest { + if bytes.Equal(closePeer.ID, client.Key) { + client.LogStatus("startSearch", "find node: sender %s: node found!\n", hex.EncodeToString(result.SenderID)) + client.Results <- &SearchResult{Key: client.Key, Action: client.Action, SenderID: result.SenderID, TargetNode: closePeer} + client.Terminate() + return + } + } + + result.Closest = client.filterUncontactedNodes(result.Closest, MaxClosest) + + client.LogStatus("startSearch", "find node: sender %s: %d nodes are close to value\n", hex.EncodeToString(result.SenderID), len(result.Closest)) + + } + + // Add closest to list + client.list.AppendUniqueNodes(result.Closest...) + + // If no subsequent level, and there's closer nodes, start one! + if !nestedStarted && !bytes.Equal(client.list.Nodes[0].ID, closestNode.ID) && level < MaxLevel { + nestedStarted = true + go client.startSearch(level + 1) + } + + case <-info.TerminateSignal: + // If highest level (= not nested), and there was no conclusive result, try one more round. + // This helps against result poisoning. + + if !nestedStarted { + client.LogStatus("startSearch", "search in level %d aborted, info request termination signal. Final try.\n", level) + if info = startInfoRequest(); info != nil { + continue + } + } + + if client.activeLevels == 1 { // if this was the last level, no more results will appear + client.LogStatus("startSearch", "level %d last active level, not found, terminate search\n", level) + client.Terminate() + } else { + client.LogStatus("startSearch", "level %d end, info request termination signal\n", level) + } + return + + //case <-time.After(time.Second): + // Future todo: Launch another routine with the with uncontacted nodes if any, to speed up the query + } + } +}