yggdrasil-go/src/yggdrasil/dht.go

363 lines
10 KiB
Go
Raw Normal View History

2017-12-29 07:16:20 +03:00
package yggdrasil
2018-10-21 08:05:04 +03:00
// TODO signal to predecessor when we replace them?
// Sending a ping with an extra 0 at the end of our coords should be enough to reset our throttle in their table
// That should encorage them to ping us again sooner, and then we can reply with new info
// Maybe remember old predecessor and check this during maintenance?
2018-06-13 01:50:08 +03:00
import (
2018-10-21 08:05:04 +03:00
"fmt"
2018-06-13 01:50:08 +03:00
"sort"
"time"
)
2017-12-29 07:16:20 +03:00
const dht_lookup_size = 16
// dhtInfo represents everything we know about a node in the DHT.
// This includes its key, a cache of it's NodeID, coords, and timing/ping related info for deciding who/when to ping nodes for maintenance.
2017-12-29 07:16:20 +03:00
type dhtInfo struct {
2018-01-05 01:37:51 +03:00
nodeID_hidden *NodeID
key boxPubKey
coords []byte
recv time.Time // When we last received a message
pings int // Time out if at least 3 consecutive maintenance pings drop
2018-10-21 06:06:36 +03:00
throttle time.Duration
2017-12-29 07:16:20 +03:00
}
// Returns the *NodeID associated with dhtInfo.key, calculating it on the fly the first time or from a cache all subsequent times.
2017-12-29 07:16:20 +03:00
func (info *dhtInfo) getNodeID() *NodeID {
2018-01-05 01:37:51 +03:00
if info.nodeID_hidden == nil {
info.nodeID_hidden = getNodeID(&info.key)
}
return info.nodeID_hidden
2017-12-29 07:16:20 +03:00
}
// Request for a node to do a lookup.
// Includes our key and coords so they can send a response back, and the destination NodeID we want to ask about.
2017-12-29 07:16:20 +03:00
type dhtReq struct {
Key boxPubKey // Key of whoever asked
Coords []byte // Coords of whoever asked
Dest NodeID // NodeID they're asking about
2017-12-29 07:16:20 +03:00
}
// Response to a DHT lookup.
// Includes the key and coords of the node that's responding, and the destination they were asked about.
// The main part is Infos []*dhtInfo, the lookup response.
2017-12-29 07:16:20 +03:00
type dhtRes struct {
Key boxPubKey // key of the sender
Coords []byte // coords of the sender
Dest NodeID
Infos []*dhtInfo // response
2017-12-29 07:16:20 +03:00
}
// The main DHT struct.
2017-12-29 07:16:20 +03:00
type dht struct {
2018-10-22 01:40:43 +03:00
core *Core
nodeID NodeID
table map[NodeID]*dhtInfo
peers chan *dhtInfo // other goroutines put incoming dht updates here
reqs map[boxPubKey]map[NodeID]time.Time
2017-12-29 07:16:20 +03:00
}
// Initializes the DHT
2017-12-29 07:16:20 +03:00
func (t *dht) init(c *Core) {
2018-01-05 01:37:51 +03:00
t.core = c
t.nodeID = *t.core.GetNodeID()
2018-06-07 18:58:24 +03:00
t.peers = make(chan *dhtInfo, 1024)
t.reset()
}
// Resets the DHT in response to coord changes
// This empties all info from the DHT and drops outstanding requests
func (t *dht) reset() {
t.reqs = make(map[boxPubKey]map[NodeID]time.Time)
t.table = make(map[NodeID]*dhtInfo)
}
// Does a DHT lookup and returns up to dht_lookup_size results
2018-10-21 23:10:18 +03:00
func (t *dht) lookup(nodeID *NodeID, everything bool) []*dhtInfo {
2018-10-22 01:40:43 +03:00
results := make([]*dhtInfo, 0, len(t.table))
for _, info := range t.table {
results = append(results, info)
}
sort.SliceStable(results, func(i, j int) bool {
return dht_ordered(nodeID, results[i].getNodeID(), results[j].getNodeID())
2018-10-22 01:40:43 +03:00
})
if len(results) > dht_lookup_size {
results = results[:dht_lookup_size]
}
return results
}
// Insert into table, preserving the time we last sent a packet if the node was already in the table, otherwise setting that time to now
func (t *dht) insert(info *dhtInfo) {
if *info.getNodeID() == t.nodeID {
2018-10-21 01:32:54 +03:00
// This shouldn't happen, but don't add it if it does
return
panic("FIXME")
}
info.recv = time.Now()
2018-10-21 06:06:36 +03:00
if oldInfo, isIn := t.table[*info.getNodeID()]; isIn {
sameCoords := true
if len(info.coords) != len(oldInfo.coords) {
sameCoords = false
} else {
for idx := 0; idx < len(info.coords); idx++ {
if info.coords[idx] != oldInfo.coords[idx] {
sameCoords = false
break
}
}
}
if sameCoords {
info.throttle = oldInfo.throttle
}
}
t.table[*info.getNodeID()] = info
}
// Return true if first/second/third are (partially) ordered correctly
// FIXME? maybe total ordering makes more sense
func dht_ordered(first, second, third *NodeID) bool {
2018-10-21 08:05:04 +03:00
lessOrEqual := func(first, second *NodeID) bool {
for idx := 0; idx < NodeIDLen; idx++ {
if first[idx] > second[idx] {
return false
}
if first[idx] < second[idx] {
return true
}
}
2018-10-21 08:05:04 +03:00
return true
}
firstLessThanSecond := lessOrEqual(first, second)
secondLessThanThird := lessOrEqual(second, third)
thirdLessThanFirst := lessOrEqual(third, first)
switch {
case firstLessThanSecond && secondLessThanThird:
// Nothing wrapped around 0, the easy case
return true
case thirdLessThanFirst && firstLessThanSecond:
// Third wrapped around 0
return true
case secondLessThanThird && thirdLessThanFirst:
// Second (and third) wrapped around 0
return true
}
2018-10-21 08:05:04 +03:00
return false
2017-12-29 07:16:20 +03:00
}
// Reads a request, performs a lookup, and responds.
// Update info about the node that sent the request.
2017-12-29 07:16:20 +03:00
func (t *dht) handleReq(req *dhtReq) {
2018-01-05 01:37:51 +03:00
// Send them what they asked for
loc := t.core.switchTable.getLocator()
coords := loc.getCoords()
res := dhtRes{
Key: t.core.boxPub,
Coords: coords,
Dest: req.Dest,
Infos: t.lookup(&req.Dest, false),
2018-01-05 01:37:51 +03:00
}
t.sendRes(&res, req)
// Also add them to our DHT
2018-01-05 01:37:51 +03:00
info := dhtInfo{
key: req.Key,
coords: req.Coords,
2018-01-05 01:37:51 +03:00
}
// For bootstrapping to work, we need to add these nodes to the table
t.insert(&info)
}
// Sends a lookup response to the specified node.
func (t *dht) sendRes(res *dhtRes, req *dhtReq) {
// Send a reply for a dhtReq
bs := res.encode()
shared := t.core.sessions.getSharedKey(&t.core.boxPriv, &req.Key)
payload, nonce := boxSeal(shared, bs, nil)
p := wire_protoTrafficPacket{
Coords: req.Coords,
ToKey: req.Key,
FromKey: t.core.boxPub,
Nonce: *nonce,
Payload: payload,
}
packet := p.encode()
t.core.router.out(packet)
}
// Returns nodeID + 1
func (nodeID NodeID) next() NodeID {
for idx := len(nodeID) - 1; idx >= 0; idx-- {
nodeID[idx] += 1
if nodeID[idx] != 0 {
break
}
}
return nodeID
}
// Returns nodeID - 1
func (nodeID NodeID) prev() NodeID {
for idx := len(nodeID) - 1; idx >= 0; idx-- {
nodeID[idx] -= 1
if nodeID[idx] != 0xff {
break
}
}
return nodeID
2017-12-29 07:16:20 +03:00
}
// Reads a lookup response, checks that we had sent a matching request, and processes the response info.
// This mainly consists of updating the node we asked in our DHT (they responded, so we know they're still alive), and deciding if we want to do anything with their responses
2017-12-29 07:16:20 +03:00
func (t *dht) handleRes(res *dhtRes) {
2018-06-02 07:34:21 +03:00
t.core.searches.handleDHTRes(res)
reqs, isIn := t.reqs[res.Key]
2018-01-05 01:37:51 +03:00
if !isIn {
return
}
_, isIn = reqs[res.Dest]
2018-01-05 01:37:51 +03:00
if !isIn {
return
}
delete(reqs, res.Dest)
2018-01-05 01:37:51 +03:00
rinfo := dhtInfo{
key: res.Key,
coords: res.Coords,
}
t.insert(&rinfo) // Or at the end, after checking successor/predecessor?
2018-10-21 08:05:04 +03:00
if len(res.Infos) > dht_lookup_size {
2018-10-21 23:10:18 +03:00
//res.Infos = res.Infos[:dht_lookup_size] //FIXME debug
2018-10-21 08:05:04 +03:00
}
2018-10-22 01:40:43 +03:00
imp := t.getImportant()
for _, info := range res.Infos {
if *info.getNodeID() == t.nodeID {
continue
} // Skip self
if _, isIn := t.table[*info.getNodeID()]; isIn {
// TODO? don't skip if coords are different?
continue
}
2018-10-22 01:40:43 +03:00
if t.isImportant(info, imp) {
t.ping(info, nil)
2018-01-05 01:37:51 +03:00
}
}
// TODO add everyting else to a rumor mill for later use? (when/how?)
2017-12-29 07:16:20 +03:00
}
// Sends a lookup request to the specified node.
2017-12-29 07:16:20 +03:00
func (t *dht) sendReq(req *dhtReq, dest *dhtInfo) {
2018-01-05 01:37:51 +03:00
// Send a dhtReq to the node in dhtInfo
bs := req.encode()
shared := t.core.sessions.getSharedKey(&t.core.boxPriv, &dest.key)
payload, nonce := boxSeal(shared, bs, nil)
p := wire_protoTrafficPacket{
2018-06-02 23:21:05 +03:00
Coords: dest.coords,
ToKey: dest.key,
FromKey: t.core.boxPub,
Nonce: *nonce,
Payload: payload,
2018-01-05 01:37:51 +03:00
}
packet := p.encode()
t.core.router.out(packet)
reqsToDest, isIn := t.reqs[dest.key]
if !isIn {
t.reqs[dest.key] = make(map[NodeID]time.Time)
reqsToDest, isIn = t.reqs[dest.key]
if !isIn {
panic("This should never happen")
}
}
reqsToDest[req.Dest] = time.Now()
2017-12-29 07:16:20 +03:00
}
func (t *dht) ping(info *dhtInfo, target *NodeID) {
// Creates a req for the node at dhtInfo, asking them about the target (if one is given) or themself (if no target is given)
2018-01-05 01:37:51 +03:00
if target == nil {
target = &t.nodeID
2018-01-05 01:37:51 +03:00
}
loc := t.core.switchTable.getLocator()
coords := loc.getCoords()
req := dhtReq{
Key: t.core.boxPub,
Coords: coords,
Dest: *target,
2018-01-05 01:37:51 +03:00
}
t.sendReq(&req, info)
2017-12-29 07:16:20 +03:00
}
func (t *dht) doMaintenance() {
toPing := make(map[NodeID]*dhtInfo)
now := time.Now()
2018-10-22 01:40:43 +03:00
imp := t.getImportant()
good := make(map[NodeID]*dhtInfo)
for _, info := range imp {
good[*info.getNodeID()] = info
}
for infoID, info := range t.table {
if now.Sub(info.recv) > time.Minute || info.pings > 3 {
delete(t.table, infoID)
2018-10-22 01:40:43 +03:00
} else if t.isImportant(info, imp) {
toPing[infoID] = info
2018-01-05 01:37:51 +03:00
}
}
for _, info := range toPing {
if now.Sub(info.recv) > info.throttle {
t.ping(info, info.getNodeID())
info.pings++
info.throttle += time.Second
if info.throttle > 30*time.Second {
info.throttle = 30 * time.Second
}
2018-10-22 01:40:43 +03:00
continue
fmt.Println("DEBUG self:", t.nodeID[:8], "throttle:", info.throttle, "nodeID:", info.getNodeID()[:8], "coords:", info.coords)
}
}
}
2018-10-22 01:40:43 +03:00
func (t *dht) getImportant() []*dhtInfo {
// Get a list of all known nodes
infos := make([]*dhtInfo, 0, len(t.table))
for _, info := range t.table {
infos = append(infos, info)
}
// Sort them by increasing order in distance along the ring
sort.SliceStable(infos, func(i, j int) bool {
// Sort in order of predecessors (!), reverse from chord normal, becuase it plays nicer with zero bits for unknown parts of target addresses
return dht_ordered(infos[j].getNodeID(), infos[i].getNodeID(), &t.nodeID)
//return dht_ordered(&t.nodeID, infos[i].getNodeID(), infos[j].getNodeID())
2018-10-22 01:40:43 +03:00
})
// Keep the ones that are no further than the closest seen so far
minDist := ^uint64(0)
loc := t.core.switchTable.getLocator()
important := infos[:0]
for _, info := range infos {
dist := uint64(loc.dist(info.coords))
if dist < minDist {
minDist = dist
important = append(important, info)
}
}
return important
}
func (t *dht) isImportant(ninfo *dhtInfo, important []*dhtInfo) bool {
// Check if ninfo is of equal or greater importance to what we already know
loc := t.core.switchTable.getLocator()
ndist := uint64(loc.dist(ninfo.coords))
minDist := ^uint64(0)
for _, info := range important {
dist := uint64(loc.dist(info.coords))
if dist < minDist {
minDist = dist
}
//if dht_ordered(&t.nodeID, ninfo.getNodeID(), info.getNodeID()) && ndist <= minDist {
if dht_ordered(info.getNodeID(), ninfo.getNodeID(), &t.nodeID) && ndist <= minDist {
2018-10-22 01:40:43 +03:00
// This node is at least as close in both key space and tree space
return true
2018-10-21 06:06:36 +03:00
}
}
2018-10-22 01:40:43 +03:00
// We didn't find any important node that ninfo is better than
return false
}