diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go index 4e33a4a..4e13bbd 100644 --- a/dht/DHT Lite.go +++ b/dht/DHT Lite.go @@ -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] diff --git a/dht/Information Request.go b/dht/Information Request.go index 4dd5aea..678eeeb 100644 --- a/dht/Information Request.go +++ b/dht/Information Request.go @@ -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 +} diff --git a/dht/Node.go b/dht/Node.go index ff079f0..ff849c8 100644 --- a/dht/Node.go +++ b/dht/Node.go @@ -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