mirror of https://github.com/go-redis/redis.git
Eliminate redundant dial mutex causing unbounded connection queue contention (#3088)
* Eliminate redundant dial mutex causing unbounded connection queue contention * Dialer connection timeouts unit test --------- Co-authored-by: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com>
This commit is contained in:
parent
930d904205
commit
080e051124
2
redis.go
2
redis.go
|
@ -176,8 +176,6 @@ func (hs *hooksMixin) withProcessPipelineHook(
|
||||||
}
|
}
|
||||||
|
|
||||||
func (hs *hooksMixin) dialHook(ctx context.Context, network, addr string) (net.Conn, error) {
|
func (hs *hooksMixin) dialHook(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||||
hs.hooksMu.Lock()
|
|
||||||
defer hs.hooksMu.Unlock()
|
|
||||||
return hs.current.dial(ctx, network, addr)
|
return hs.current.dial(ctx, network, addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -6,6 +6,7 @@ import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
@ -633,3 +634,67 @@ var _ = Describe("Hook with MinIdleConns", func() {
|
||||||
}))
|
}))
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
var _ = Describe("Dialer connection timeouts", func() {
|
||||||
|
var client *redis.Client
|
||||||
|
|
||||||
|
const dialSimulatedDelay = 1 * time.Second
|
||||||
|
|
||||||
|
BeforeEach(func() {
|
||||||
|
options := redisOptions()
|
||||||
|
options.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||||
|
// Simulated slow dialer.
|
||||||
|
// Note that the following sleep is deliberately not context-aware.
|
||||||
|
time.Sleep(dialSimulatedDelay)
|
||||||
|
return net.Dial("tcp", options.Addr)
|
||||||
|
}
|
||||||
|
options.MinIdleConns = 1
|
||||||
|
client = redis.NewClient(options)
|
||||||
|
})
|
||||||
|
|
||||||
|
AfterEach(func() {
|
||||||
|
err := client.Close()
|
||||||
|
Expect(err).NotTo(HaveOccurred())
|
||||||
|
})
|
||||||
|
|
||||||
|
It("does not contend on connection dial for concurrent commands", func() {
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
|
||||||
|
const concurrency = 10
|
||||||
|
|
||||||
|
durations := make(chan time.Duration, concurrency)
|
||||||
|
errs := make(chan error, concurrency)
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
wg.Add(concurrency)
|
||||||
|
|
||||||
|
for i := 0; i < concurrency; i++ {
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
err := client.Ping(ctx).Err()
|
||||||
|
durations <- time.Since(start)
|
||||||
|
errs <- err
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
wg.Wait()
|
||||||
|
close(durations)
|
||||||
|
close(errs)
|
||||||
|
|
||||||
|
// All commands should eventually succeed, after acquiring a connection.
|
||||||
|
for err := range errs {
|
||||||
|
Expect(err).NotTo(HaveOccurred())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Each individual command should complete within the simulated dial duration bound.
|
||||||
|
for duration := range durations {
|
||||||
|
Expect(duration).To(BeNumerically("<", 2*dialSimulatedDelay))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Due to concurrent execution, the entire test suite should also complete within
|
||||||
|
// the same dial duration bound applied for individual commands.
|
||||||
|
Expect(time.Since(start)).To(BeNumerically("<", 2*dialSimulatedDelay))
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
Loading…
Reference in New Issue