Revert "Merge pull request #241 from libp2p/dq-var-test-races"

This reverts commit 1b1fb7e, reversing
changes made to b598d08.

dial_queue.go

@@ -9,7 +9,7 @@ import (
 	queue "github.com/libp2p/go-libp2p-peerstore/queue"
 )
 
-const (
+var (
 	// DialQueueMinParallelism is the minimum number of worker dial goroutines that will be alive at any time.
 	DialQueueMinParallelism = 6
 	// DialQueueMaxParallelism is the maximum number of worker dial goroutines that can be alive at any time.
@@ -25,10 +25,8 @@ type dialQueue struct {
 	ctx    context.Context
 	dialFn func(context.Context, peer.ID) error
 
-	nWorkers          int
-	scalingFactor     float64
-	scalingMutePeriod time.Duration
-	maxIdle           time.Duration
+	nWorkers      int
+	scalingFactor float64
 
 	in  *queue.ChanQueue
 	out *queue.ChanQueue
@@ -63,16 +61,12 @@ type waitingCh struct {
 // Dialler throttling (e.g. FD limit exceeded) is a concern, as we can easily spin up more workers to compensate, and
 // end up adding fuel to the fire. Since we have no deterministic way to detect this for now, we hard-limit concurrency
 // to DialQueueMaxParallelism.
-func newDialQueue(ctx context.Context, target string, in *queue.ChanQueue, dialFn func(context.Context, peer.ID) error,
-	maxIdle, scalingMutePeriod time.Duration,
-) *dialQueue {
+func newDialQueue(ctx context.Context, target string, in *queue.ChanQueue, dialFn func(context.Context, peer.ID) error) *dialQueue {
 	sq := &dialQueue{
-		ctx:               ctx,
-		dialFn:            dialFn,
-		nWorkers:          DialQueueMinParallelism,
-		scalingFactor:     1.5,
-		scalingMutePeriod: scalingMutePeriod,
-		maxIdle:           maxIdle,
+		ctx:           ctx,
+		dialFn:        dialFn,
+		nWorkers:      DialQueueMinParallelism,
+		scalingFactor: 1.5,
 
 		in:  in,
 		out: queue.NewChanQueue(ctx, queue.NewXORDistancePQ(target)),
@@ -151,13 +145,13 @@ func (dq *dialQueue) control() {
 				dialled = nil
 			}
 		case <-dq.growCh:
-			if time.Since(lastScalingEvt) < dq.scalingMutePeriod {
+			if time.Since(lastScalingEvt) < DialQueueScalingMutePeriod {
 				continue
 			}
 			dq.grow()
 			lastScalingEvt = time.Now()
 		case <-dq.shrinkCh:
-			if time.Since(lastScalingEvt) < dq.scalingMutePeriod {
+			if time.Since(lastScalingEvt) < DialQueueScalingMutePeriod {
 				continue
 			}
 			dq.shrink()
@@ -265,7 +259,7 @@ func (dq *dialQueue) worker() {
 		case <-idleTimer.C:
 		default:
 		}
-		idleTimer.Reset(dq.maxIdle)
+		idleTimer.Reset(DialQueueMaxIdle)
 
 		select {
 		case <-dq.dieCh:

dial_queue_test.go

@@ -2,6 +2,7 @@ package dht
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -11,7 +12,12 @@ import (
 	queue "github.com/libp2p/go-libp2p-peerstore/queue"
 )
 
+func init() {
+	DialQueueScalingMutePeriod = 0
+}
+
 func TestDialQueueGrowsOnSlowDials(t *testing.T) {
+	DialQueueMaxIdle = 10 * time.Minute
 
 	in := queue.NewChanQueue(context.Background(), queue.NewXORDistancePQ("test"))
 	hang := make(chan struct{})
@@ -29,7 +35,7 @@ func TestDialQueueGrowsOnSlowDials(t *testing.T) {
 	}
 
 	// remove the mute period to grow faster.
-	dq := newDialQueue(context.Background(), "test", in, dialFn, 10*time.Minute, 0)
+	dq := newDialQueue(context.Background(), "test", in, dialFn)
 
 	for i := 0; i < 4; i++ {
 		_ = dq.Consume()
@@ -49,6 +55,7 @@ func TestDialQueueGrowsOnSlowDials(t *testing.T) {
 
 func TestDialQueueShrinksWithNoConsumers(t *testing.T) {
 	// reduce interference from the other shrink path.
+	DialQueueMaxIdle = 10 * time.Minute
 
 	in := queue.NewChanQueue(context.Background(), queue.NewXORDistancePQ("test"))
 	hang := make(chan struct{})
@@ -61,7 +68,12 @@ func TestDialQueueShrinksWithNoConsumers(t *testing.T) {
 		return nil
 	}
 
-	dq := newDialQueue(context.Background(), "test", in, dialFn, 10*time.Minute, 0)
+	dq := newDialQueue(context.Background(), "test", in, dialFn)
+
+	defer func() {
+		recover()
+		fmt.Println(dq.nWorkers)
+	}()
 
 	// acquire 3 consumers, everytime we acquire a consumer, we will grow the pool because no dial job is completed
 	// and immediately returnable.
@@ -105,6 +117,8 @@ func TestDialQueueShrinksWithNoConsumers(t *testing.T) {
 
 // Inactivity = workers are idle because the DHT query is progressing slow and is producing too few peers to dial.
 func TestDialQueueShrinksWithWhenIdle(t *testing.T) {
+	DialQueueMaxIdle = 1 * time.Second
+
 	in := queue.NewChanQueue(context.Background(), queue.NewXORDistancePQ("test"))
 	hang := make(chan struct{})
 
@@ -121,7 +135,7 @@ func TestDialQueueShrinksWithWhenIdle(t *testing.T) {
 		in.EnqChan <- peer.ID(i)
 	}
 
-	dq := newDialQueue(context.Background(), "test", in, dialFn, time.Second, 0)
+	dq := newDialQueue(context.Background(), "test", in, dialFn)
 
 	// keep up to speed with backlog by releasing the dial function every time we acquire a channel.
 	for i := 0; i < 13; i++ {
@@ -147,6 +161,8 @@ func TestDialQueueShrinksWithWhenIdle(t *testing.T) {
 }
 
 func TestDialQueueMutePeriodHonored(t *testing.T) {
+	DialQueueScalingMutePeriod = 2 * time.Second
+
 	in := queue.NewChanQueue(context.Background(), queue.NewXORDistancePQ("test"))
 	hang := make(chan struct{})
 	var wg sync.WaitGroup
@@ -162,7 +178,7 @@ func TestDialQueueMutePeriodHonored(t *testing.T) {
 		in.EnqChan <- peer.ID(i)
 	}
 
-	dq := newDialQueue(context.Background(), "test", in, dialFn, DialQueueMaxIdle, 2*time.Second)
+	dq := newDialQueue(context.Background(), "test", in, dialFn)
 
 	// pick up three consumers.
 	for i := 0; i < 3; i++ {

query.go

@@ -103,7 +103,7 @@ func newQueryRunner(q *dhtQuery) *dhtQueryRunner {
 		peersToQuery:   peersToQuery,
 		proc:           proc,
 	}
-	r.peersDialed = newDialQueue(ctx, q.key, peersToQuery, r.dialPeer, DialQueueMaxIdle, DialQueueScalingMutePeriod)
+	r.peersDialed = newDialQueue(ctx, q.key, peersToQuery, r.dialPeer)
 	return r
 }