diff --git a/src/yggdrasil/conn.go b/src/yggdrasil/conn.go index 25605cd..f622903 100644 --- a/src/yggdrasil/conn.go +++ b/src/yggdrasil/conn.go @@ -131,7 +131,7 @@ func (c *Conn) search() error { } } sinfo := c.core.router.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted) - sinfo.continueSearch() + sinfo.startSearch() } else { err = errors.New("search already exists") close(done) @@ -155,7 +155,7 @@ func (c *Conn) doSearch() { sinfo = c.core.router.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted) c.core.log.Debugf("%s DHT search started: %p", c.String(), sinfo) // Start the search - sinfo.continueSearch() + sinfo.startSearch() } } c.core.router.Act(c.session, routerWork) diff --git a/src/yggdrasil/dialer.go b/src/yggdrasil/dialer.go index e9da97a..490502b 100644 --- a/src/yggdrasil/dialer.go +++ b/src/yggdrasil/dialer.go @@ -65,12 +65,15 @@ func (d *Dialer) DialContext(ctx context.Context, network, address string) (net. // DialByNodeIDandMask opens a session to the given node based on raw // NodeID parameters. If ctx is nil or has no timeout, then a default timeout of 6 seconds will apply, beginning *after* the search finishes. func (d *Dialer) DialByNodeIDandMask(ctx context.Context, nodeID, nodeMask *crypto.NodeID) (net.Conn, error) { + startDial := time.Now() conn := newConn(d.core, nodeID, nodeMask, nil) if err := conn.search(); err != nil { // TODO: make searches take a context, so they can be cancelled early conn.Close() return nil, err } + endSearch := time.Now() + d.core.log.Debugln("Dial searched for:", nodeID, "in time:", endSearch.Sub(startDial)) conn.session.setConn(nil, conn) var cancel context.CancelFunc if ctx == nil { @@ -80,6 +83,9 @@ func (d *Dialer) DialByNodeIDandMask(ctx context.Context, nodeID, nodeMask *cryp defer cancel() select { case <-conn.session.init: + endInit := time.Now() + d.core.log.Debugln("Dial initialized session for:", nodeID, "in time:", endInit.Sub(endSearch)) + d.core.log.Debugln("Finished dial for:", nodeID, "in time:", endInit.Sub(startDial)) return conn, nil case <-ctx.Done(): conn.Close() diff --git a/src/yggdrasil/search.go b/src/yggdrasil/search.go index ede4547..584a056 100644 --- a/src/yggdrasil/search.go +++ b/src/yggdrasil/search.go @@ -22,9 +22,6 @@ import ( "github.com/yggdrasil-network/yggdrasil-go/src/crypto" ) -// This defines the maximum number of dhtInfo that we keep track of for nodes to query in an ongoing search. -const search_MAX_SEARCH_SIZE = 16 - // This defines the time after which we time out a search (so it can restart). const search_RETRY_TIME = 3 * time.Second const search_STEP_TIME = 100 * time.Millisecond @@ -36,7 +33,6 @@ type searchInfo struct { dest crypto.NodeID mask crypto.NodeID time time.Time - toVisit []*dhtInfo visited crypto.NodeID // Closest address visited so far callback func(*sessionInfo, error) // TODO context.Context for timeout and cancellation @@ -79,57 +75,26 @@ func (s *searches) createSearch(dest *crypto.NodeID, mask *crypto.NodeID, callba // If there is, it adds the response info to the search and triggers a new search step. // If there's no ongoing search, or we if the dhtRes finished the search (it was from the target node), then don't do anything more. func (sinfo *searchInfo) handleDHTRes(res *dhtRes) { - old := sinfo.visited if res != nil { sinfo.recv++ if sinfo.checkDHTRes(res) { return // Search finished successfully } - // Add results to the search - sinfo.addToSearch(res) - } - if res == nil || sinfo.visited != old { - // Continue the search - sinfo.doSearchStep() - } -} - -// Adds the information from a dhtRes to an ongoing search. -// Info about a node that has already been visited is not re-added to the search. -func (sinfo *searchInfo) addToSearch(res *dhtRes) { - // Add to search - for _, info := range res.Infos { - sinfo.toVisit = append(sinfo.toVisit, info) - } - // Sort - sort.SliceStable(sinfo.toVisit, func(i, j int) bool { - // Should return true if i is closer to the destination than j - return dht_ordered(&sinfo.dest, sinfo.toVisit[i].getNodeID(), sinfo.toVisit[j].getNodeID()) - }) - // Remove anything too far away - for idx, info := range sinfo.toVisit { - if *info.getNodeID() == sinfo.visited || !dht_ordered(&sinfo.dest, info.getNodeID(), &sinfo.visited) { - sinfo.toVisit = sinfo.toVisit[:idx] - break + // Use results to start an additional search thread + infos := append([]*dhtInfo(nil), res.Infos...) + infos = sinfo.getAllowedInfos(infos) + if len(infos) > 0 { + sinfo.continueSearch(infos) } } } -// If there are no nodes left toVisit, then this cleans up the search. +// If there has been no response in too long, then this cleans up the search. // Otherwise, it pops the closest node to the destination (in keyspace) off of the toVisit list and sends a dht ping. -func (sinfo *searchInfo) doSearchStep() { - if len(sinfo.toVisit) == 0 { - if time.Since(sinfo.time) > search_RETRY_TIME { - // Dead end and no response in too long, do cleanup - delete(sinfo.searches.searches, sinfo.dest) - sinfo.callback(nil, errors.New("search reached dead end")) - } - return - } - // Send to the next search target - if len(sinfo.toVisit) > 0 { - next := sinfo.toVisit[0] - sinfo.toVisit = sinfo.toVisit[1:] +func (sinfo *searchInfo) doSearchStep(infos []*dhtInfo) { + if len(infos) > 0 { + // Send to the next search target + next := infos[0] rq := dhtReqKey{next.key, sinfo.dest} sinfo.searches.router.dht.addCallback(&rq, sinfo.handleDHTRes) sinfo.searches.router.dht.ping(next, &sinfo.dest) @@ -137,13 +102,29 @@ func (sinfo *searchInfo) doSearchStep() { } } -// If we've recently sent a ping for this search, do nothing. -// Otherwise, doSearchStep and schedule another continueSearch to happen after search_RETRY_TIME. -func (sinfo *searchInfo) continueSearch() { - sinfo.doSearchStep() - // In case the search dies, try to spawn another thread later - // Note that this will spawn multiple parallel searches as time passes - // Any that die aren't restarted, but a new one will start later +// Get a list of search targets that are close enough to the destination to try +// Requires an initial list as input +func (sinfo *searchInfo) getAllowedInfos(infos []*dhtInfo) []*dhtInfo { + sort.SliceStable(infos, func(i, j int) bool { + // Should return true if i is closer to the destination than j + return dht_ordered(&sinfo.dest, infos[i].getNodeID(), infos[j].getNodeID()) + }) + // Remove anything too far away to be useful + for idx, info := range infos { + if !dht_ordered(&sinfo.dest, info.getNodeID(), &sinfo.visited) { + infos = infos[:idx] + break + } + } + return infos +} + +// Run doSearchStep and schedule another continueSearch to happen after search_RETRY_TIME. +// Must not be called with an empty list of infos +func (sinfo *searchInfo) continueSearch(infos []*dhtInfo) { + sinfo.doSearchStep(infos) + infos = infos[1:] // Remove the node we just tried + // In case there's no response, try the next node in infos later time.AfterFunc(search_STEP_TIME, func() { sinfo.searches.router.Act(nil, func() { // FIXME this keeps the search alive forever if not for the searches map, fix that @@ -151,20 +132,51 @@ func (sinfo *searchInfo) continueSearch() { if newSearchInfo != sinfo { return } - sinfo.continueSearch() + // Get good infos here instead of at the top, to make sure we can always start things off with a continueSearch call to ourself + infos = sinfo.getAllowedInfos(infos) + if len(infos) > 0 { + sinfo.continueSearch(infos) + } }) }) } +// Initially start a search +func (sinfo *searchInfo) startSearch() { + loc := sinfo.searches.router.core.switchTable.getLocator() + var infos []*dhtInfo + infos = append(infos, &dhtInfo{ + key: sinfo.searches.router.core.boxPub, + coords: loc.getCoords(), + }) + // Start the search by asking ourself, useful if we're the destination + sinfo.continueSearch(infos) + // Start a timer to clean up the search if everything times out + var cleanupFunc func() + cleanupFunc = func() { + sinfo.searches.router.Act(nil, func() { + // FIXME this keeps the search alive forever if not for the searches map, fix that + newSearchInfo := sinfo.searches.searches[sinfo.dest] + if newSearchInfo != sinfo { + return + } + elapsed := time.Since(sinfo.time) + if elapsed > search_RETRY_TIME { + // cleanup + delete(sinfo.searches.searches, sinfo.dest) + sinfo.callback(nil, errors.New("search reached dead end")) + return + } + time.AfterFunc(search_RETRY_TIME-elapsed, cleanupFunc) + }) + } + time.AfterFunc(search_RETRY_TIME, cleanupFunc) +} + // Calls create search, and initializes the iterative search parts of the struct before returning it. func (s *searches) newIterSearch(dest *crypto.NodeID, mask *crypto.NodeID, callback func(*sessionInfo, error)) *searchInfo { sinfo := s.createSearch(dest, mask, callback) sinfo.visited = s.router.dht.nodeID - loc := s.router.core.switchTable.getLocator() - sinfo.toVisit = append(sinfo.toVisit, &dhtInfo{ - key: s.router.core.boxPub, - coords: loc.getCoords(), - }) // Start the search by asking ourself, useful if we're the destination return sinfo } @@ -175,7 +187,7 @@ func (sinfo *searchInfo) checkDHTRes(res *dhtRes) bool { from := dhtInfo{key: res.Key, coords: res.Coords} if *from.getNodeID() != sinfo.visited && dht_ordered(&sinfo.dest, from.getNodeID(), &sinfo.visited) { // Closer to the destination, so update visited - sinfo.searches.router.core.log.Debugln("Updating search:", sinfo.dest, *from.getNodeID(), sinfo.send, sinfo.recv) + sinfo.searches.router.core.log.Debugln("Updating search:", &sinfo.dest, from.getNodeID(), sinfo.send, sinfo.recv) sinfo.visited = *from.getNodeID() sinfo.time = time.Now() } @@ -202,7 +214,7 @@ func (sinfo *searchInfo) checkDHTRes(res *dhtRes) bool { } // Cleanup if _, isIn := sinfo.searches.searches[sinfo.dest]; isIn { - sinfo.searches.router.core.log.Debugln("Finished search:", sinfo.dest, sinfo.send, sinfo.recv) + sinfo.searches.router.core.log.Debugln("Finished search:", &sinfo.dest, sinfo.send, sinfo.recv) delete(sinfo.searches.searches, res.Dest) } } diff --git a/src/yggdrasil/session.go b/src/yggdrasil/session.go index 91c530d..eaa67fd 100644 --- a/src/yggdrasil/session.go +++ b/src/yggdrasil/session.go @@ -43,7 +43,6 @@ type sessionInfo struct { time time.Time // Time we last received a packet mtuTime time.Time // time myMTU was last changed pingTime time.Time // time the first ping was sent since the last received packet - pingSend time.Time // time the last ping was sent coords []byte // coords of destination reset bool // reset if coords change tstamp int64 // ATOMIC - tstamp from their last session ping, replay attack mitigation @@ -197,7 +196,6 @@ func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo { sinfo.time = now sinfo.mtuTime = now sinfo.pingTime = now - sinfo.pingSend = now sinfo.init = make(chan struct{}) sinfo.cancel = util.NewCancellation() higher := false