From fc1f0407f3af2a8e128b5adea7eaa844cdaeae04 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Thu, 29 Jul 2021 03:17:14 +0200 Subject: [PATCH] -> Alpha 3. DHT: Store and Get functions use search client. Multiple improvements in search client. Alpha parameter is now provided by init caller. StoreDataDHT add closestCount parameter. --- Config.go | 2 +- Kademlia.go | 11 +++-- dht/DHT Lite.go | 107 ++++++++++++++++++------------------------- dht/Search Client.go | 45 ++++++++---------- 4 files changed, 71 insertions(+), 94 deletions(-) diff --git a/Config.go b/Config.go index c7f15c4..1280b8c 100644 --- a/Config.go +++ b/Config.go @@ -16,7 +16,7 @@ import ( ) // Version is the current core library version -const Version = "Alpha 2/28.06.2021" +const Version = "Alpha 3/28.07.2021" var config struct { LogFile string `yaml:"LogFile"` // Log file diff --git a/Kademlia.go b/Kademlia.go index db7e553..1513848 100644 --- a/Kademlia.go +++ b/Kademlia.go @@ -156,14 +156,14 @@ func StoreDataLocal(data []byte) error { return Warehouse.Store(key, data, time.Time{}, time.Time{}) } -// StoreDataDHT stores data locally and informs peers in the DHT about it. +// StoreDataDHT stores data locally and informs closestCount peers in the DHT about it. // Remote peers may choose to keep a record (in case another peers asks) or mirror the full data. -func StoreDataDHT(data []byte) error { +func StoreDataDHT(data []byte, closestCount int) error { key := hashData(data) if err := Warehouse.Store(key, data, time.Time{}, time.Time{}); err != nil { return err } - return nodesDHT.Store(key, uint64(len(data))) + return nodesDHT.Store(key, uint64(len(data)), closestCount) } // ---- NODE FUNCTIONS ---- @@ -192,6 +192,7 @@ func FindNode(nodeID []byte, Timeout time.Duration) (node *dht.Node, peer *PeerI // AsyncSearch creates an async search for the given key in the DHT. // Timeout is the total time the search may take, covering all information requests. TimeoutIR is the time an information request may take. -func AsyncSearch(Action int, Key []byte, Timeout, TimeoutIR time.Duration) (client *dht.SearchClient) { - return nodesDHT.NewSearch(Action, Key, Timeout, TimeoutIR) +// Alpha is the number of concurrent requests that will be performed. +func AsyncSearch(Action int, Key []byte, Timeout, TimeoutIR time.Duration, Alpha int) (client *dht.SearchClient) { + return nodesDHT.NewSearch(Action, Key, Timeout, TimeoutIR, Alpha) } diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go index b87eef1..1749779 100644 --- a/dht/DHT Lite.go +++ b/dht/DHT Lite.go @@ -9,9 +9,8 @@ A "lite" DHT implementation without any direct network and store code. There is package dht import ( - "bytes" + "encoding/hex" "errors" - "sort" "time" ) @@ -116,44 +115,32 @@ func (dht *DHT) IsNodeContact(ID []byte) (node *Node) { // ---- Synchronous network query functions below ---- // Store informs the network about data stored locally. -func (dht *DHT) Store(key []byte, dataSize uint64) (err error) { +// Data size informs how big the data is without sending the actual data. closestCount is the number of closest nodes to contact. +func (dht *DHT) Store(key []byte, dataSize uint64, closestCount int) (err error) { if len(key)*8 != 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 + // TODO: Introduce ActionFindClosestNodes? + + search := dht.NewSearch(ActionFindNode, key, dht.TimeoutSearch, dht.TimeoutIR, dht.alpha) + search.LogStatus = func(function, format string, v ...interface{}) { + dht.FilterSearchStatus(search, function, format, v...) + } + search.LogStatus("dht.Store", "Search for closest nodes to key %s. Full timeout %s, per node %s. Alpha = %d.\n", hex.EncodeToString(key), dht.TimeoutSearch.String(), dht.TimeoutIR.String(), dht.alpha) + search.SearchAway() + + // search.Results channel is ignored here. Only the closest nodes to the key are of interest. It is not expected to find a match of key and node ID. + <-search.TerminateSignal + + // Contact the closes nodes found. + for n := 0; n < closestCount && n < len(search.list.Nodes); n++ { + node := search.list.Nodes[n] + search.LogStatus("dht.Store", "Send info-store message to node %s\n", hex.EncodeToString(node.ID)) + dht.SendRequestStore(node, key, dataSize) } - 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 { - 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, dataSize) - } - return nil - } - - closestNode = sl.Nodes[0] - } + return nil } // Get retrieves data from the network using key @@ -162,36 +149,18 @@ func (dht *DHT) Get(key []byte) (value []byte, senderID []byte, found bool, err return nil, 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, nil, false, nil + search := dht.NewSearch(ActionFindValue, key, dht.TimeoutSearch, dht.TimeoutIR, dht.alpha) + search.LogStatus = func(function, format string, v ...interface{}) { + dht.FilterSearchStatus(search, function, format, v...) } + search.LogStatus("dht.Get", "Search for node %s. Full timeout %s, per node %s. Alpha = %d.\n", hex.EncodeToString(key), dht.TimeoutSearch.String(), dht.TimeoutIR.String(), dht.alpha) + search.SearchAway() - closestNode := sl.Nodes[0] - - // TODO: Limit max amount of iterations to mitigate malicious responses. - for { - info := dht.NewInformationRequest(ActionFindValue, key, sl.GetUncontacted(dht.alpha, true)) - dht.SendRequestFindValue(info) - results := info.CollectResults(dht.TMsgTimeout) - - for _, result := range results { - if len(result.Data) > 0 { - return result.Data, result.SenderID, true, 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, nil, false, nil - } - - closestNode = sl.Nodes[0] + select { + case <-search.TerminateSignal: + return nil, nil, false, nil + case result := <-search.Results: + return result.Data, result.SenderID, true, nil } } @@ -202,10 +171,11 @@ func (dht *DHT) FindNode(key []byte) (node *Node, err error) { return nil, errors.New("invalid key size") } - search := dht.NewSearch(ActionFindNode, key, dht.TimeoutSearch, dht.TimeoutIR) + search := dht.NewSearch(ActionFindNode, key, dht.TimeoutSearch, dht.TimeoutIR, dht.alpha) search.LogStatus = func(function, format string, v ...interface{}) { dht.FilterSearchStatus(search, function, format, v...) } + search.LogStatus("dht.FindNode", "Search for node %s. Full timeout %s, per node %s. Alpha = %d.\n", hex.EncodeToString(key), dht.TimeoutSearch.String(), dht.TimeoutIR.String(), dht.alpha) search.SearchAway() select { @@ -218,8 +188,15 @@ func (dht *DHT) FindNode(key []byte) (node *Node, err error) { // ---- DHT Health ---- +// DisableBucketRefresh is an option for debug purposes to reduce noise. It can be useful to disable bucket refresh when debugging outgoing DHT searches. +var DisableBucketRefresh = false + // RefreshBuckets refreshes all buckets not meeting the target node number. 0 to refresh all. func (dht *DHT) RefreshBuckets(target int) { + if DisableBucketRefresh { + return + } + for bucket, total := range dht.ht.getTotalNodesPerBucket() { if target == 0 || total < target { nodeR := dht.ht.getRandomIDFromBucket(bucket) @@ -231,5 +208,9 @@ func (dht *DHT) RefreshBuckets(target int) { dht.FindNode(nodeR) } + + if DisableBucketRefresh { // may be disabled while in full refresh which may take some time + return + } } } diff --git a/dht/Search Client.go b/dht/Search Client.go index ed3240a..5c8fa7a 100644 --- a/dht/Search Client.go +++ b/dht/Search Client.go @@ -13,18 +13,11 @@ 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 @@ -47,6 +40,7 @@ type SearchClient struct { 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). + alpha int // Count of concurrent information requests per level. Results chan *SearchResult // Result channel list *shortList // List of nodes to contact contactedNodesMap map[string]struct{} // List of nodes already contacted @@ -72,20 +66,21 @@ type SearchResult struct { // 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) { +// Alpha is the number of concurrent requests that will be performed. +func (dht *DHT) NewSearch(Action int, Key []byte, Timeout, TimeoutIR time.Duration, Alpha int) (client *SearchClient) { client = &SearchClient{ Action: Action, Key: Key, dht: dht, timeoutTotal: Timeout, timeoutIR: TimeoutIR, + alpha: Alpha, contactedNodesMap: make(map[string]struct{}), - storing: make(chan []*Node, MaxConcurrentRequestKnownStore*2), + storing: make(chan []*Node, Alpha*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 } @@ -147,7 +142,7 @@ 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) + client.list = client.dht.ht.getClosestContacts(client.alpha, client.Key, nil) if len(client.list.Nodes) == 0 { client.Terminate() return @@ -177,7 +172,7 @@ func (client *SearchClient) sendInfoRequest(nodes []*Node, resultChan chan *Node } for _, node := range nodes { - client.LogStatus("sendInfoRequest", "contact node %s\n", hex.EncodeToString(node.ID)) + client.LogStatus("search.sendInfoRequest", "contact node %s\n", hex.EncodeToString(node.ID)) } info = client.dht.NewInformationRequest(client.Action, client.Key, nodes) @@ -199,7 +194,7 @@ func (client *SearchClient) sendInfoRequest(nodes []*Node, resultChan chan *Node // 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) + resultChan := make(chan *NodeMessage, client.alpha) for { select { @@ -224,18 +219,18 @@ func (client *SearchClient) startSearch(level int) { defer atomic.AddUint64(&client.activeLevels, ^uint64(0)) nestedStarted := false - results := make(chan *NodeMessage, MaxConcurrentRequestsPerLevel*2) + results := make(chan *NodeMessage, client.alpha*2) closestNode := client.list.Nodes[0] // start an info request startInfoRequest := func() (info *InformationRequest) { - nodes := client.list.GetUncontacted(MaxConcurrentRequestsPerLevel, true) + nodes := client.list.GetUncontacted(client.alpha, true) if len(nodes) == 0 { - client.LogStatus("startSearch", "search in level %d aborted, no new nodes to contact\n", level) + client.LogStatus("search.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)) + client.LogStatus("search.startSearch", "start search in level %d contacting %d nodes\n", level, len(nodes)) return client.sendInfoRequest(nodes, results) } @@ -247,7 +242,7 @@ func (client *SearchClient) startSearch(level int) { for { select { case <-client.TerminateSignal: - client.LogStatus("startSearch", "search in level %d aborted, search client termination signal\n", level) + client.LogStatus("search.startSearch", "search in level %d aborted, search client termination signal\n", level) return case result := <-results: @@ -256,7 +251,7 @@ func (client *SearchClient) startSearch(level int) { 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.LogStatus("search.startSearch", "result: 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 @@ -265,7 +260,7 @@ func (client *SearchClient) startSearch(level int) { 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)) + client.LogStatus("search.startSearch", "result: 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 { @@ -276,7 +271,7 @@ func (client *SearchClient) startSearch(level int) { // 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.LogStatus("search.startSearch", "result: 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 @@ -285,7 +280,7 @@ func (client *SearchClient) startSearch(level int) { 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)) + client.LogStatus("search.startSearch", "find node: sender %s: %d nodes are close to value\n", hex.EncodeToString(result.SenderID), len(result.Closest)) } @@ -303,17 +298,17 @@ func (client *SearchClient) startSearch(level int) { // This helps against result poisoning. if !nestedStarted { - client.LogStatus("startSearch", "search in level %d aborted, info request termination signal. Final try.\n", level) + client.LogStatus("search.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.LogStatus("search.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) + client.LogStatus("search.startSearch", "level %d end, info request termination signal\n", level) } return