Merge pull request #651 from Arceliar/search

Search

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)

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()

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
+		// Use results to start an additional search thread
-		sinfo.addToSearch(res)
+		infos := append([]*dhtInfo(nil), res.Infos...)
-	}
+		infos = sinfo.getAllowedInfos(infos)
-	if res == nil || sinfo.visited != old {
+		if len(infos) > 0 {
-		// Continue the search
+			sinfo.continueSearch(infos)
-		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
 		}
 	}
 }
 
-// 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() {
+func (sinfo *searchInfo) doSearchStep(infos []*dhtInfo) {
-	if len(sinfo.toVisit) == 0 {
+	if len(infos) > 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 := infos[0]
-		next := sinfo.toVisit[0]
-		sinfo.toVisit = sinfo.toVisit[1:]
 		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.
+// Get a list of search targets that are close enough to the destination to try
-// Otherwise, doSearchStep and schedule another continueSearch to happen after search_RETRY_TIME.
+// Requires an initial list as input
-func (sinfo *searchInfo) continueSearch() {
+func (sinfo *searchInfo) getAllowedInfos(infos []*dhtInfo) []*dhtInfo {
-	sinfo.doSearchStep()
+	sort.SliceStable(infos, func(i, j int) bool {
-	// In case the search dies, try to spawn another thread later
+		// Should return true if i is closer to the destination than j
-	// Note that this will spawn multiple parallel searches as time passes
+		return dht_ordered(&sinfo.dest, infos[i].getNodeID(), infos[j].getNodeID())
-	// Any that die aren't restarted, but a new one will start later
+	})
+	// 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)
 		}
 	}

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