Compare commits

...

3 Commits

Author SHA1 Message Date
oldme 367e7c4a93
Merge 4e99ec9e63 into 080e051124 2024-11-20 13:44:30 +02:00
LINKIWI 080e051124
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>
2024-11-20 13:38:06 +02:00
oldme 4e99ec9e63 up 2024-09-02 15:52:18 +08:00
3 changed files with 69 additions and 2 deletions

View File

@ -109,6 +109,10 @@ func (w *Writer) WriteArg(v interface{}) error {
return w.uint(v)
case *uint64:
return w.uint(*v)
case uintptr:
return w.uint(uint64(v))
case *uintptr:
return w.uint(uint64(*v))
case float32:
return w.float(float64(v))
case *float32:

View File

@ -176,8 +176,6 @@ func (hs *hooksMixin) withProcessPipelineHook(
}
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)
}

View File

@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"net"
"sync"
"testing"
"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))
})
})