Files
traefikoidc/vendor/github.com/redis/go-redis/v9/internal/semaphore.go
T
lukaszraczyloandGitHub 2d1b04c637 review fixes apr 2026 (#130)
* Multiple fixes

- refresh coordinator dedup + memory pressure wire
- middleware sse consolidation + timer leak + claim cache
- universal cache sync backfill + isDebug gate
- lazy background task race
- memory monitor stw cached + refresh() api

* fix(auth): suppress OIDC redirects on non-navigation requests

- [x] Add isNonNavigationRequest using Sec-Fetch-Mode and Accept headers
- [x] Add comprehensive TestIsNonNavigationRequest
- [x] Update ServeHTTP to 401 non-navigation and AJAX requests

Fixes #129

* feat(config): add custom CA and insecure skip verify for OIDC TLS

- [x] Add CACertPath, CACertPEM, InsecureSkipVerify to Config
- [x] Implement loadCACertPool for CA bundle loading
- [x] Update HTTPClientConfig with RootCAs and InsecureSkipVerify
- [x] Apply CA pool and skip verify to pooled HTTP clients
- [x] Enhance configKey to distinguish TLS configs
- [x] Add comprehensive ca_cert_test.go

Fixes #125

* feat(oidc): add custom CA certificate support for private OIDC providers

- [x] Add caCertPath, caCertPEM, insecureSkipVerify config options
- [x] Update traefik.yml with new OIDC client config fields
- [x] Add configuration schema descriptions for new options
- [x] Update README table and add Custom CA Certificates section

* Fix the documentation.

* test(redis): add oversized argument rejection test

- [x] Add TestRedisConn_RejectOversizedArgumentBytes
- [x] Import strings package

* Dependencies cleanup
2026-04-19 10:12:00 +01:00

193 lines
4.9 KiB
Go

package internal
import (
"context"
"sync"
"time"
)
var semTimers = sync.Pool{
New: func() interface{} {
t := time.NewTimer(time.Hour)
t.Stop()
return t
},
}
// FastSemaphore is a channel-based semaphore optimized for performance.
// It uses a fast path that avoids timer allocation when tokens are available.
// The channel is pre-filled with tokens: Acquire = receive, Release = send.
// Closing the semaphore unblocks all waiting goroutines.
//
// Performance: ~30 ns/op with zero allocations on fast path.
// Fairness: Eventual fairness (no starvation) but not strict FIFO.
type FastSemaphore struct {
tokens chan struct{}
max int32
}
// NewFastSemaphore creates a new fast semaphore with the given capacity.
func NewFastSemaphore(capacity int32) *FastSemaphore {
ch := make(chan struct{}, capacity)
// Pre-fill with tokens
for i := int32(0); i < capacity; i++ {
ch <- struct{}{}
}
return &FastSemaphore{
tokens: ch,
max: capacity,
}
}
// TryAcquire attempts to acquire a token without blocking.
// Returns true if successful, false if no tokens available.
func (s *FastSemaphore) TryAcquire() bool {
select {
case <-s.tokens:
return true
default:
return false
}
}
// Acquire acquires a token, blocking if necessary until one is available.
// Returns an error if the context is cancelled or the timeout expires.
// Uses a fast path to avoid timer allocation when tokens are immediately available.
func (s *FastSemaphore) Acquire(ctx context.Context, timeout time.Duration, timeoutErr error) error {
// Check context first
select {
case <-ctx.Done():
return ctx.Err()
default:
}
// Try fast path first (no timer needed)
select {
case <-s.tokens:
return nil
default:
}
// Slow path: need to wait with timeout
timer := semTimers.Get().(*time.Timer)
defer semTimers.Put(timer)
timer.Reset(timeout)
select {
case <-s.tokens:
if !timer.Stop() {
<-timer.C
}
return nil
case <-ctx.Done():
if !timer.Stop() {
<-timer.C
}
return ctx.Err()
case <-timer.C:
return timeoutErr
}
}
// AcquireBlocking acquires a token, blocking indefinitely until one is available.
func (s *FastSemaphore) AcquireBlocking() {
<-s.tokens
}
// Release releases a token back to the semaphore.
func (s *FastSemaphore) Release() {
s.tokens <- struct{}{}
}
// Close closes the semaphore, unblocking all waiting goroutines.
// After close, all Acquire calls will receive a closed channel signal.
func (s *FastSemaphore) Close() {
close(s.tokens)
}
// Len returns the current number of acquired tokens.
func (s *FastSemaphore) Len() int32 {
return s.max - int32(len(s.tokens))
}
// FIFOSemaphore is a channel-based semaphore with strict FIFO ordering.
// Unlike FastSemaphore, this guarantees that threads are served in the exact order they call Acquire().
// The channel is pre-filled with tokens: Acquire = receive, Release = send.
// Closing the semaphore unblocks all waiting goroutines.
//
// Performance: ~115 ns/op with zero allocations (slower than FastSemaphore due to timer allocation).
// Fairness: Strict FIFO ordering guaranteed by Go runtime.
type FIFOSemaphore struct {
tokens chan struct{}
max int32
}
// NewFIFOSemaphore creates a new FIFO semaphore with the given capacity.
func NewFIFOSemaphore(capacity int32) *FIFOSemaphore {
ch := make(chan struct{}, capacity)
// Pre-fill with tokens
for i := int32(0); i < capacity; i++ {
ch <- struct{}{}
}
return &FIFOSemaphore{
tokens: ch,
max: capacity,
}
}
// TryAcquire attempts to acquire a token without blocking.
// Returns true if successful, false if no tokens available.
func (s *FIFOSemaphore) TryAcquire() bool {
select {
case <-s.tokens:
return true
default:
return false
}
}
// Acquire acquires a token, blocking if necessary until one is available.
// Returns an error if the context is cancelled or the timeout expires.
// Always uses timer to guarantee FIFO ordering (no fast path).
func (s *FIFOSemaphore) Acquire(ctx context.Context, timeout time.Duration, timeoutErr error) error {
// No fast path - always use timer to guarantee FIFO
timer := semTimers.Get().(*time.Timer)
defer semTimers.Put(timer)
timer.Reset(timeout)
select {
case <-s.tokens:
if !timer.Stop() {
<-timer.C
}
return nil
case <-ctx.Done():
if !timer.Stop() {
<-timer.C
}
return ctx.Err()
case <-timer.C:
return timeoutErr
}
}
// AcquireBlocking acquires a token, blocking indefinitely until one is available.
func (s *FIFOSemaphore) AcquireBlocking() {
<-s.tokens
}
// Release releases a token back to the semaphore.
func (s *FIFOSemaphore) Release() {
s.tokens <- struct{}{}
}
// Close closes the semaphore, unblocking all waiting goroutines.
// After close, all Acquire calls will receive a closed channel signal.
func (s *FIFOSemaphore) Close() {
close(s.tokens)
}
// Len returns the current number of acquired tokens.
func (s *FIFOSemaphore) Len() int32 {
return s.max - int32(len(s.tokens))
}