fluencelabs/go-libp2p-kad-dht
1
0
Fork 0
mirror of https://github.com/fluencelabs/go-libp2p-kad-dht synced 2025-07-02 06:01:38 +00:00
Code Issues Projects Releases Wiki Activity

Fix races with DialQueue variables

Browse Source
This commit is contained in:
Matt Joiner
2019-01-31 13:08:45 +11:00
parent 0998477db1
commit 7246a3b0f4
3 changed files with 24 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
dial_queue_test.go
View File

@ -2,22 +2,16 @@ package dht
import (
"context"
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
peer "github.com/libp2p/go-libp2p-peer"
queue "github.com/libp2p/go-libp2p-peerstore/queue"
"github.com/libp2p/go-libp2p-peer"
"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{})
@ -35,7 +29,7 @@ func TestDialQueueGrowsOnSlowDials(t *testing.T) {
}
// remove the mute period to grow faster.
dq := newDialQueue(context.Background(), "test", in, dialFn)
dq := newDialQueue(context.Background(), "test", in, dialFn, 10*time.Minute, 0)
for i := 0; i < 4; i++ {
_ = dq.Consume()
@ -55,7 +49,6 @@ 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{})
@ -68,12 +61,7 @@ func TestDialQueueShrinksWithNoConsumers(t *testing.T) {
return nil
}
dq := newDialQueue(context.Background(), "test", in, dialFn)
defer func() {
recover()
fmt.Println(dq.nWorkers)
}()
dq := newDialQueue(context.Background(), "test", in, dialFn, 10*time.Minute, 0)
// acquire 3 consumers, everytime we acquire a consumer, we will grow the pool because no dial job is completed
// and immediately returnable.
@ -117,8 +105,6 @@ 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{})
@ -135,7 +121,7 @@ func TestDialQueueShrinksWithWhenIdle(t *testing.T) {
in.EnqChan <- peer.ID(i)
}
dq := newDialQueue(context.Background(), "test", in, dialFn)
dq := newDialQueue(context.Background(), "test", in, dialFn, time.Second, 0)
// keep up to speed with backlog by releasing the dial function every time we acquire a channel.
for i := 0; i < 13; i++ {
@ -161,8 +147,6 @@ 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
@ -178,7 +162,7 @@ func TestDialQueueMutePeriodHonored(t *testing.T) {
in.EnqChan <- peer.ID(i)
}
dq := newDialQueue(context.Background(), "test", in, dialFn)
dq := newDialQueue(context.Background(), "test", in, dialFn, DialQueueMaxIdle, 2*time.Second)
// pick up three consumers.
for i := 0; i < 3; i++ {