Merge pull request #329 from libp2p/fix/goroutine-leak-empty-routing

query: fix a goroutine leak when the routing table is empty

Author: Stebalien

query.go

@@ -9,6 +9,7 @@ import (
 	todoctr "github.com/ipfs/go-todocounter"
 	process "github.com/jbenet/goprocess"
 	ctxproc "github.com/jbenet/goprocess/context"
+	kb "github.com/libp2p/go-libp2p-kbucket"
 	inet "github.com/libp2p/go-libp2p-net"
 	peer "github.com/libp2p/go-libp2p-peer"
 	pset "github.com/libp2p/go-libp2p-peer/peerset"
@@ -58,6 +59,11 @@ type queryFunc func(context.Context, peer.ID) (*dhtQueryResult, error)
 
 // Run runs the query at hand. pass in a list of peers to use first.
 func (q *dhtQuery) Run(ctx context.Context, peers []peer.ID) (*dhtQueryResult, error) {
+	if len(peers) == 0 {
+		logger.Warning("Running query with no peers!")
+		return nil, kb.ErrLookupFailure
+	}
+
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
@@ -121,11 +127,6 @@ func (r *dhtQueryRunner) Run(ctx context.Context, peers []peer.ID) (*dhtQueryRes
 	r.log = logger
 	r.runCtx = ctx
 
-	if len(peers) == 0 {
-		logger.Warning("Running query with no peers!")
-		return nil, nil
-	}
-
 	// setup concurrency rate limiting
 	for i := 0; i < r.query.concurrency; i++ {
 		r.rateLimit <- struct{}{}

routing.go

@@ -495,6 +495,15 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key cid.Cid,
 		}
 	}
 
+	peers := dht.routingTable.NearestPeers(kb.ConvertKey(key.KeyString()), AlphaValue)
+	if len(peers) == 0 {
+		notif.PublishQueryEvent(ctx, &notif.QueryEvent{
+			Type:  notif.QueryError,
+			Extra: kb.ErrLookupFailure.Error(),
+		})
+		return
+	}
+
 	// setup the Query
 	parent := ctx
 	query := dht.newQuery(key.KeyString(), func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
@@ -545,7 +554,6 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key cid.Cid,
 		return &dhtQueryResult{closerPeers: clpeers}, nil
 	})
 
-	peers := dht.routingTable.NearestPeers(kb.ConvertKey(key.KeyString()), AlphaValue)
 	_, err := query.Run(ctx, peers)
 	if err != nil {
 		logger.Debugf("Query error: %s", err)