From 8a9ec786e096b8586dd285075d4fd374f4d1551a Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:12:36 -0700 Subject: [PATCH] refactor: simplify the init-concurrency limiter to a single shared budget The limiter primitive added in #780 carried an optional per-environment gate and a matching INIT_PER_ENV_MAX_PERCENT config option. Drop per-environment fairness: the limiter now bounds concurrent admissions with one shared budget (a fixed number of held slots plus a bounded FIFO queue of waiters), and Acquire no longer takes an environment key. Also revise the INIT_SEND_TIMEOUT documentation and default to describe an absolute cap on how long one gated delivery may hold a slot. No runtime effect on its own: nothing acquires from the limiter yet. --- config/config.go | 13 ++-- config/config_concurrency_test.go | 8 +-- internal/concurrency/limiter.go | 96 +++++++--------------------- internal/concurrency/limiter_test.go | 45 +++++-------- 4 files changed, 48 insertions(+), 114 deletions(-) diff --git a/config/config.go b/config/config.go index e2d64ef9..f4860665 100644 --- a/config/config.go +++ b/config/config.go @@ -161,14 +161,11 @@ type ConcurrencyConfig struct { // of queueing them. MaxQueued ct.OptInt `conf:"INIT_MAX_QUEUED"` - // PerEnvMaxPercent limits the share of the budget that any single environment may - // use, as a percentage of MaxConcurrent plus MaxQueued. This keeps one busy - // environment from starving the others. A value of 0 applies no per-environment - // limit. - PerEnvMaxPercent ct.OptInt `conf:"INIT_PER_ENV_MAX_PERCENT"` - - // SendTimeout releases a delivery slot if a streaming initialization payload cannot - // make progress to its client within this duration. It defaults to 30s. + // SendTimeout is the absolute cap on how long a single gated delivery may hold a slot. A + // throughput floor (64 KB/s) closes a client that stalls or is slower than the floor well + // before this; the cap only backstops a client stuck right at the floor on a very large + // payload. If a delivery exceeds it, the connection is closed to reclaim the slot (and the + // SDK reconnects). It defaults to 2m. SendTimeout ct.OptDuration `conf:"INIT_SEND_TIMEOUT"` } diff --git a/config/config_concurrency_test.go b/config/config_concurrency_test.go index 38c04c6d..3dd7c864 100644 --- a/config/config_concurrency_test.go +++ b/config/config_concurrency_test.go @@ -13,16 +13,14 @@ import ( // the INIT_* environment variables. func TestConcurrencyConfigFromEnvironment(t *testing.T) { withEnvironment(map[string]string{ - "INIT_MAX_CONCURRENT": "200", - "INIT_MAX_QUEUED": "1000", - "INIT_PER_ENV_MAX_PERCENT": "40", - "INIT_SEND_TIMEOUT": "15s", + "INIT_MAX_CONCURRENT": "200", + "INIT_MAX_QUEUED": "1000", + "INIT_SEND_TIMEOUT": "15s", }, func() { var c Config require.NoError(t, LoadConfigFromEnvironment(&c, slog.Default())) assert.Equal(t, 200, c.Concurrency.MaxConcurrent.GetOrElse(-1)) assert.Equal(t, 1000, c.Concurrency.MaxQueued.GetOrElse(-1)) - assert.Equal(t, 40, c.Concurrency.PerEnvMaxPercent.GetOrElse(-1)) assert.Equal(t, 15*time.Second, c.Concurrency.SendTimeout.GetOrElse(0)) }) } diff --git a/internal/concurrency/limiter.go b/internal/concurrency/limiter.go index e0ae0099..35dde192 100644 --- a/internal/concurrency/limiter.go +++ b/internal/concurrency/limiter.go @@ -1,8 +1,7 @@ // Package concurrency provides an admission limiter that bounds how much concurrent // work a burst of requests or connections can impose on Relay. It uses two limits: a // maximum number of slots held at once, and a bounded FIFO queue of callers waiting for -// a slot. An optional per-environment gate keeps one environment from using the whole -// budget. +// a slot. package concurrency import ( @@ -19,13 +18,9 @@ type Params struct { // MaxQueued is the number of callers that may wait in FIFO order for a slot once all // slots are held. A value of 0 rejects callers immediately instead of waiting. MaxQueued int - // PerEnvMax limits how many of a single environment's callers may participate, - // counting both held and waiting, at once. A value of 0 applies no per-environment - // limit. The per-environment gate never blocks; it only rejects. - PerEnvMax int } -// Stats is a point-in-time snapshot of a Limiter's counters, for logging/metrics. +// Stats is a point-in-time snapshot of a Limiter's counters for logging and metrics. type Stats struct { Enabled bool MaxConcurrent int @@ -36,30 +31,23 @@ type Stats struct { Rejected int64 } -// Limiter bounds concurrency with two limits plus an optional per-environment gate that -// never blocks and only rejects. The zero value is not usable; construct one with New. +// Limiter bounds concurrency with two limits: a maximum number of slots held at once and +// a bounded FIFO queue of waiters. The zero value is not usable; construct one with New. type Limiter struct { name string enabled bool - tokens chan struct{} // holds MaxConcurrent slots; receive to acquire, send to release + tokens chan struct{} // holds the free slots; receive to acquire one, send to release one maxQueued int64 waiting int64 shutdown chan struct{} closeOnce sync.Once - perEnvMax int - perEnv sync.Map // envKey -> *envGate - admitted atomic.Int64 rejected atomic.Int64 } -type envGate struct { - slots chan struct{} // cap = perEnvMax; try-receive to enter, send to leave -} - -// New builds a Limiter. name is used only for logging/metrics identification. +// New builds a Limiter. name identifies the limiter in logs and metrics. func New(name string, p Params) *Limiter { l := &Limiter{name: name} if p.MaxConcurrent <= 0 { @@ -72,9 +60,6 @@ func New(name string, p Params) *Limiter { } l.maxQueued = int64(p.MaxQueued) l.shutdown = make(chan struct{}) - if p.PerEnvMax > 0 { - l.perEnvMax = p.PerEnvMax - } return l } @@ -84,84 +69,51 @@ func (l *Limiter) Name() string { return l.name } // Enabled reports whether the limiter is enforcing a limit. func (l *Limiter) Enabled() bool { return l != nil && l.enabled } -// Acquire attempts to admit one unit of work for the given environment key. -// On success it returns a release func (call exactly once) and ok=true. If the -// per-env gate or the global backlog is full, or ctx is cancelled, or the -// limiter is shut down, it returns a no-op release and ok=false. A disabled or -// nil limiter always admits immediately. -func (l *Limiter) Acquire(ctx context.Context, envKey string) (release func(), ok bool) { +// Acquire attempts to admit one unit of work. On success it returns a release function, +// which the caller must call exactly once, and ok is true. It returns a no-op release and +// ok=false if the queue is full, ctx is cancelled, or the limiter is shut down. A disabled +// or nil limiter always admits immediately. +func (l *Limiter) Acquire(ctx context.Context) (release func(), ok bool) { if !l.Enabled() { return func() {}, true } - // Per-environment gate. It never blocks; it rejects when the environment is over its share. - var releaseEnv func() - if l.perEnvMax > 0 { - g := l.gateFor(envKey) - select { - case g.slots <- struct{}{}: - releaseEnv = func() { <-g.slots } - default: - l.rejected.Add(1) - return func() {}, false - } - } - - reject := func() (func(), bool) { - if releaseEnv != nil { - releaseEnv() - } - l.rejected.Add(1) - return func() {}, false - } - - // Fast path: a token is immediately available. + // Take a free slot if one is available. select { case <-l.tokens: - return l.admit(releaseEnv), true + return l.admit(), true default: } - // No token free: enter the bounded FIFO backlog, or reject. + // No free slot. Join the bounded queue, or reject if it is full. if atomic.AddInt64(&l.waiting, 1) > l.maxQueued { atomic.AddInt64(&l.waiting, -1) - return reject() + l.rejected.Add(1) + return func() {}, false } defer atomic.AddInt64(&l.waiting, -1) select { case <-l.tokens: - return l.admit(releaseEnv), true + return l.admit(), true case <-ctx.Done(): - return reject() + l.rejected.Add(1) + return func() {}, false case <-l.shutdown: - return reject() + l.rejected.Add(1) + return func() {}, false } } -func (l *Limiter) admit(releaseEnv func()) func() { +func (l *Limiter) admit() func() { l.admitted.Add(1) var once sync.Once return func() { - once.Do(func() { - l.tokens <- struct{}{} - if releaseEnv != nil { - releaseEnv() - } - }) - } -} - -func (l *Limiter) gateFor(envKey string) *envGate { - if g, ok := l.perEnv.Load(envKey); ok { - return g.(*envGate) + once.Do(func() { l.tokens <- struct{}{} }) } - g := &envGate{slots: make(chan struct{}, l.perEnvMax)} - actual, _ := l.perEnv.LoadOrStore(envKey, g) - return actual.(*envGate) } -// Close unblocks all waiters (they receive ok=false). Idempotent. +// Close unblocks all waiters, which then receive ok=false. It may be called more than once. func (l *Limiter) Close() { if !l.Enabled() { return diff --git a/internal/concurrency/limiter_test.go b/internal/concurrency/limiter_test.go index 650aafd1..42240c68 100644 --- a/internal/concurrency/limiter_test.go +++ b/internal/concurrency/limiter_test.go @@ -13,7 +13,7 @@ func TestDisabledLimiterAlwaysAdmits(t *testing.T) { t.Fatal("expected disabled") } for i := 0; i < 100; i++ { - release, ok := l.Acquire(context.Background(), "env") + release, ok := l.Acquire(context.Background()) if !ok { t.Fatal("disabled limiter must always admit") } @@ -23,7 +23,7 @@ func TestDisabledLimiterAlwaysAdmits(t *testing.T) { func TestNilLimiterAdmits(t *testing.T) { var l *Limiter - release, ok := l.Acquire(context.Background(), "env") + release, ok := l.Acquire(context.Background()) if !ok { t.Fatal("nil limiter must admit") } @@ -32,9 +32,9 @@ func TestNilLimiterAdmits(t *testing.T) { func TestRejectWhenNoBacklog(t *testing.T) { l := New("t", Params{MaxConcurrent: 2, MaxQueued: 0}) - r1, ok1 := l.Acquire(context.Background(), "e") - r2, ok2 := l.Acquire(context.Background(), "e") - _, ok3 := l.Acquire(context.Background(), "e") + r1, ok1 := l.Acquire(context.Background()) + r2, ok2 := l.Acquire(context.Background()) + _, ok3 := l.Acquire(context.Background()) if !ok1 || !ok2 { t.Fatal("first two should be admitted") } @@ -50,13 +50,13 @@ func TestRejectWhenNoBacklog(t *testing.T) { func TestQueueThenAdmitOnRelease(t *testing.T) { l := New("t", Params{MaxConcurrent: 1, MaxQueued: 1}) - r1, ok1 := l.Acquire(context.Background(), "e") + r1, ok1 := l.Acquire(context.Background()) if !ok1 { t.Fatal("first should be admitted") } admitted := make(chan struct{}) go func() { - r2, ok2 := l.Acquire(context.Background(), "e") // must queue, then admit when r1 releases + r2, ok2 := l.Acquire(context.Background()) // must queue, then admit when r1 releases if ok2 { close(admitted) r2() @@ -79,36 +79,23 @@ func TestQueueThenAdmitOnRelease(t *testing.T) { func TestBacklogFullRejects(t *testing.T) { l := New("t", Params{MaxConcurrent: 1, MaxQueued: 1}) - r1, _ := l.Acquire(context.Background(), "e") // holds the only token + r1, _ := l.Acquire(context.Background()) // holds the only token defer r1() - go l.Acquire(context.Background(), "e") // fills the single backlog slot + go l.Acquire(context.Background()) // fills the single backlog slot time.Sleep(50 * time.Millisecond) - if _, ok := l.Acquire(context.Background(), "e"); ok { + if _, ok := l.Acquire(context.Background()); ok { t.Fatal("expected rejection when backlog is full") } } -func TestPerEnvGateIsolatesEnvironments(t *testing.T) { - // Global room for 10, but each env may hold at most 1 (participant cap). - l := New("t", Params{MaxConcurrent: 10, MaxQueued: 10, PerEnvMax: 1}) - rA, okA := l.Acquire(context.Background(), "A") - _, okA2 := l.Acquire(context.Background(), "A") // second A rejected by per-env gate - rB, okB := l.Acquire(context.Background(), "B") // different env still admitted - if !okA || okA2 || !okB { - t.Fatalf("per-env gate failed: okA=%v okA2=%v okB=%v", okA, okA2, okB) - } - rA() - rB() -} - func TestContextCancelUnblocksWaiter(t *testing.T) { l := New("t", Params{MaxConcurrent: 1, MaxQueued: 5}) - r1, _ := l.Acquire(context.Background(), "e") + r1, _ := l.Acquire(context.Background()) defer r1() ctx, cancel := context.WithCancel(context.Background()) done := make(chan bool) go func() { - _, ok := l.Acquire(ctx, "e") + _, ok := l.Acquire(ctx) done <- ok }() time.Sleep(50 * time.Millisecond) @@ -125,15 +112,15 @@ func TestContextCancelUnblocksWaiter(t *testing.T) { func TestReleaseIsIdempotent(t *testing.T) { l := New("t", Params{MaxConcurrent: 1, MaxQueued: 0}) - r, _ := l.Acquire(context.Background(), "e") + r, _ := l.Acquire(context.Background()) r() r() // must not release a second token // Two acquires should now succeed sequentially, proving only one token exists. - r1, ok1 := l.Acquire(context.Background(), "e") + r1, ok1 := l.Acquire(context.Background()) if !ok1 { t.Fatal("expected admit after release") } - if _, ok2 := l.Acquire(context.Background(), "e"); ok2 { + if _, ok2 := l.Acquire(context.Background()); ok2 { t.Fatal("double release leaked a token") } r1() @@ -149,7 +136,7 @@ func TestConcurrentAcquireBoundsHeld(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - release, ok := l.Acquire(context.Background(), "e") + release, ok := l.Acquire(context.Background()) if !ok { return }