llm-d · jtechapps · Jul 23, 2026
@@ -36,6 +36,8 @@ The architecture adheres to the following core principles:
   - [Deployment](#deployment)
   - [Command line parameters](#command-line-parameters)
   - [Dispatch Gates](#dispatch-gates)
+    - [Difference between Queue and Pool Gates](#difference-between-queue-and-pool-gates)
+    - [Worker Pool Gate Budget & Proportional Throttling](#worker-pool-gate-budget--proportional-throttling)
     - [Per-Queue Dispatch Gates](#per-queue-dispatch-gates)
   - [Request Messages and Consumption](#request-messages-and-consumption)
     - [Request Merge Policy](#request-merge-policy)
@@ -140,6 +142,17 @@ The Async Processor supports dispatch gates to control batch processing based on
 * **Queue-level gates** run at the admission phase for a specific queue. When a queue-level gate denies admission (returning `ActionRefuse`), the request is immediately returned to the broker to be retried/re-delivered, freeing the worker to process other queues.
 * **Pool-level gates** run directly inside the worker loop to regulate capacity constraints shared by all queues routing to that worker pool. When a pool-level gate returns `ActionWait`, the worker parks in-memory and polls until capacity is available, avoiding broker nack/retry overhead. If the pool-level gate returns `ActionRefuse`, the request is immediately returned to the broker.
 
+### Worker Pool Gate Budget & Proportional Throttling
+
+When a pool-level gate is configured, `poolGate.Budget(ctx)` dynamically calculates the available capacity budget $B \in [0.0, 1.0]$ for the worker pool. Rather than treating gate capacity as a binary on/off switch, the Async Processor uses `Budget()` for **Proportional Worker Pool Throttling**:
+
+1. **Active Worker Calculation**: The pool dynamically scales active worker concurrency $A$ based on total pool size $W$ and current budget $B$:
+   $$A = \max\left(1, \lfloor W \times B \rfloor\right) \quad \text{(when } W \ge 1\text{)}$$
+2. **Pre-Pop Throttling (Starvation Prevention)**: Each worker goroutine $w \in \{1 \dots W\}$ evaluates $w \le A$ **before** popping a request from the pool's dispatch channel:
+   - **Active workers ($w \le A$)**: Pull messages from the channel and evaluate `poolGate.Apply()` to dispatch inference requests.
+   - **Throttled workers ($w > A$)**: Sleep in a polling loop without pulling messages, preserving queue FIFO order and preventing head-of-line blocking / request starvation.
+3. **Prometheus Metrics**: The pool budget is exported to Prometheus via `async_dispatch_budget{pool_name="..."}`.
+
 ### Per-Queue Dispatch Gates
 
 For more fine-grained control, configure gates per queue in your configuration file. Each queue can have its own gate type and parameters.

@@ -68,9 +68,26 @@ $$F = 0.3, \quad B = 0.1$$
 2. `max_concurrency` is expressed in request units here, but the same approach generalizes to other capacity units (e.g., tokens, bytes) given an equivalent per-endpoint capacity constant.
 3. We may also define system capacity in terms of number of bytes; in this case $D$ would be truncated at the boundary of a valid request; for instance, if we expect to be able to forward a maximum of 1024 KiB of data, then we may forward at most $1024 \times (0.7-0.1) = 614.4$ KiB; which means if we have $r_1, r_2, ...$ enqueued; if $r_1$ is 600 KiB and $r_2$ is 100 KiB, then we would only dispatch $r_1$
 
+### Worker Pool Budget Enforcement & Proportional Throttling
+
+When applied as a **Pool-Level Gate**, the dispatch budget $D \in [0, 1]$ directly regulates worker concurrency across the pool of $W$ total worker goroutines:
+
+1. **Active Worker Formula**:
+   The active worker count $A$ is dynamically calculated from total pool size $W$ and current budget $D$:
+   $$A = \max\left(1, \lfloor W \times D \rfloor\right) \quad (\text{when } W \ge 1\text{)}$$
+
+2. **Pre-Pop Capacity Guard (Starvation & Burst Prevention)**:
+   Worker goroutine $w \in \{1 \dots W\}$ checks $w \le A$ **before** popping a request from the pool's dispatch channel (`requestChannel`):
+   - **Active Workers ($w \le A$)**: Pull messages from `requestChannel` and evaluate `poolGate.Apply()` to dispatch inference requests to the gateway.
+   - **Throttled Workers ($w > A$)**: Sleep in a polling loop without pulling messages. This prevents head-of-line blocking and request starvation while maintaining queue backpressure on the broker reader.
+
+3. **Metrics Integration**:
+   The active dispatch budget $D$ is exported to Prometheus via the gauge:
+   `async_dispatch_budget{queue_id="", queue_name="", pool_name="<poolID>"}`
+
 ### Failure Modes
 
 * **Queue Consumer Failures:** If the Async Processor pod crashes, the messages remain in the persistent Message Queue (Redis/PubSub), ensuring no data loss.
-* **Downstream Router/IGW Backpressure:** If `llm-d-router` (or another inference gateway) returns an HTTP error indicating overload (e.g. HTTP 429\) despite the computed budget, then the **saturation is assumed to be close to 1**, and therefore a **dispatch budget close to 0\.** The Async Processor will not retry until a subsequent update from the metrics store will bring it back within the acceptable limits.
+* **Downstream Router/IGW Backpressure:** If `llm-d-router` (or another inference gateway) returns an HTTP error indicating overload (e.g. HTTP 429) despite the computed budget, then the **saturation is assumed to be close to 1**, and therefore a **dispatch budget close to 0.** The Async Processor will not retry until a subsequent update from the metrics store will bring it back within the acceptable limits.
   * This might happen if a sudden spike of traffic enters the gateway and the metrics did not update on-time.
-* **Unreadable Metrics:** In case of unreadable metrics, the Async Processor cannot take informed decisions, and therefore will assume a **dispatch budget of 0\.**
+* **Unreadable Metrics:** In case of unreadable metrics, the Async Processor cannot take informed decisions, and therefore will assume a **dispatch budget of 0.**
@@ -33,16 +33,51 @@ const (
 	cancellationCheckRetryAfterSecond = 1.0
 )
 
+func computeActiveWorkers(totalWorkers int, budget float64) int {
+	if totalWorkers <= 0 {
+		return 0
+	}
+	if budget <= 0.0 {
+		return 1
+	}
+	limit := int(math.Floor(float64(totalWorkers) * budget))
+	if limit < 1 {
+		return 1
+	}
+	if limit > totalWorkers {
+		return totalWorkers
+	}
+	return limit
+}
+
 func Worker(consumeCtx, requestCtx context.Context, characteristics pipeline.Characteristics, client asyncapi.InferenceClient, requestChannel chan pipeline.EmbelishedRequestMessage,
 	retryChannel chan pipeline.RetryMessage, resultChannel chan asyncapi.ResultMessage, requestTimeout time.Duration, transforms *transform.Chain) {
-	WorkerWithGate(consumeCtx, requestCtx, characteristics, client, requestChannel, retryChannel, resultChannel, requestTimeout, transforms, nil)
+	WorkerWithGate(consumeCtx, requestCtx, characteristics, client, requestChannel, retryChannel, resultChannel, requestTimeout, transforms, nil, 1, 1, "")
 }
 
 func WorkerWithGate(consumeCtx, requestCtx context.Context, characteristics pipeline.Characteristics, client asyncapi.InferenceClient, requestChannel chan pipeline.EmbelishedRequestMessage,
-	retryChannel chan pipeline.RetryMessage, resultChannel chan asyncapi.ResultMessage, requestTimeout time.Duration, transforms *transform.Chain, poolGate pipeline.Gate) {
+	retryChannel chan pipeline.RetryMessage, resultChannel chan asyncapi.ResultMessage, requestTimeout time.Duration, transforms *transform.Chain, poolGate pipeline.Gate, workerID, totalWorkers int, poolID string) {
 
 	logger := log.FromContext(requestCtx)
 	for {
+		if poolGate != nil && totalWorkers > 0 {
+			for {
+				budget := poolGate.Budget(requestCtx)
+				metrics.SetDispatchBudget(budget, "", "", poolID)
+				activeLimit := computeActiveWorkers(totalWorkers, budget)
+				if workerID <= activeLimit {
+					break
+				}
+				select {
+				case <-consumeCtx.Done():
+					logger.V(logutil.DEFAULT).Info("Worker finishing while throttled by pool gate.")
+					return
+				case <-time.After(gateWaitPollInterval):
+					// Poll again until budget opens up enough to allow workerID
+				}
+			}
+		}
+
 		select {
 		case <-consumeCtx.Done():
 			logger.V(logutil.DEFAULT).Info("Worker finishing, draining request channel.")

@@ -470,7 +470,7 @@ func TestWorker_PoolGateActionWaitThrottlesCancellationChecks(t *testing.T) {
 	ctx := WithCancellationChecker(requestCtx, checker)
 	gate := &notifyingWaitGate{applied: make(chan struct{})}
 
-	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate)
+	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate, 1, 1, "")
 
 	requestChannel <- newEmbR(asyncapi.InternalRouting{RequestToken: "token-gate-wait-throttled"}, asyncapi.RequestMessage{
 		ID:       "gate-wait-throttled",
@@ -2198,7 +2198,7 @@ func TestWorker_PoolGateShutdownReenqueues(t *testing.T) {
 
 	done := make(chan struct{})
 	go func() {
-		WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate)
+		WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate, 1, 1, "")
 		close(done)
 	}()
 
@@ -2246,7 +2246,7 @@ func TestWorker_PoolGateActionWaitShutdownReenqueues(t *testing.T) {
 
 	done := make(chan struct{})
 	go func() {
-		WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate)
+		WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate, 1, 1, "")
 		close(done)
 	}()
 
@@ -2297,7 +2297,7 @@ func TestWorker_PoolGateActionWaitHonorsCancellation(t *testing.T) {
 	gate := &notifyingWaitGate{applied: make(chan struct{})}
 	requestToken := "token-gate-wait-cancelled"
 
-	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate)
+	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate, 1, 1, "")
 
 	requestChannel <- newEmbR(asyncapi.InternalRouting{RequestToken: requestToken}, asyncapi.RequestMessage{
 		ID:       msgID,
@@ -2355,7 +2355,7 @@ func TestWorker_RechecksCancellationAfterGateContinue(t *testing.T) {
 	}
 	requestToken := "token-gate-continue-cancelled"
 
-	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate)
+	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, gate, 1, 1, "")
 
 	requestChannel <- newEmbR(asyncapi.InternalRouting{RequestToken: requestToken}, asyncapi.RequestMessage{
 		ID:       msgID,
@@ -2488,7 +2488,7 @@ func TestWorker_QueueGateAndPoolGateRace(t *testing.T) {
 		releaseCalled: &poolReleaseCalled,
 	}
 
-	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, poolGate)
+	go WorkerWithGate(ctx, ctx, pipeline.Characteristics{}, inferenceClient, requestChannel, retryChannel, resultChannel, defaultRequestTimeout, nil, poolGate, 1, 1, "")
 
 	ir := asyncapi.NewInternalRequest(
 		asyncapi.InternalRouting{RequestQueueName: "test-queue"},
@@ -2545,3 +2545,33 @@ func TestWorker_QueueGateAndPoolGateRace(t *testing.T) {
 		t.Errorf("expected pool release to be called exactly once, got %d", poolReleaseCalled)
 	}
 }
+
+func TestComputeActiveWorkers(t *testing.T) {
+	tests := []struct {
+		name         string
+		totalWorkers int
+		budget       float64
+		expected     int
+	}{
+		{"full budget 64 workers", 64, 1.0, 64},
+		{"half budget 64 workers", 64, 0.5, 32},
+		{"quarter budget 64 workers", 64, 0.25, 16},
+		{"5% budget 64 workers", 64, 0.05, 3},
+		{"tiny positive budget 64 workers", 64, 0.001, 1},
+		{"zero budget 64 workers", 64, 0.0, 1},
+		{"negative budget 64 workers", 64, -0.1, 1},
+		{"half budget 10 workers", 10, 0.5, 5},
+		{"half budget 1 worker", 1, 0.5, 1},
+		{"zero budget 1 worker", 1, 0.0, 1},
+		{"zero total workers", 0, 1.0, 0},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := computeActiveWorkers(tt.totalWorkers, tt.budget)
+			if got != tt.expected {
+				t.Errorf("computeActiveWorkers(%d, %g) = %d; want %d", tt.totalWorkers, tt.budget, got, tt.expected)
+			}
+		})
+	}
+}
@@ -156,10 +156,10 @@ func (r *Runner) Run(ctx context.Context) (err error) {
 		setupLog.Info("Spawning workers for pool", "poolID", poolID, "workers", workersCount, "hasGate", poolGate != nil)
 		for w := 1; w <= workersCount; w++ {
 			wg.Add(1)
-			go func(mergedChan chan pipeline.EmbelishedRequestMessage, poolGate pipeline.Gate) {
+			go func(workerID int, mergedChan chan pipeline.EmbelishedRequestMessage, poolGate pipeline.Gate, poolID string) {
 				defer wg.Done()
-				asyncworker.WorkerWithGate(ctx, drainCtx, flow.Characteristics(), inferenceClient, mergedChan, flow.RetryChannel(), flow.ResultChannel(), opts.Worker.RequestTimeout, transforms, poolGate)
-			}(mergedChan, poolGate)
+				asyncworker.WorkerWithGate(ctx, drainCtx, flow.Characteristics(), inferenceClient, mergedChan, flow.RetryChannel(), flow.ResultChannel(), opts.Worker.RequestTimeout, transforms, poolGate, workerID, workersCount, poolID)
+			}(w, mergedChan, poolGate, poolID)
 		}
 	}
 

@@ -7,9 +7,10 @@ ap:
   workerPools:
     - id: "benchmark-pool"
       workers: 64
-      gate_type: "wait-on-refuse"
+      gate_type: "prometheus-saturation"
       gate_params:
-        gate: '{"gate_type":"composite","gate_params":{"gates":[{"gate_type":"prometheus-saturation","gate_params":{"pool":"e2e-pool","threshold":"0.8"}},{"gate_type":"local-max-concurrency","gate_params":{"limit":50}}]}}'
+        pool: "e2e-pool"
+        threshold: "0.8"
   metrics:
     enabled: true
     port: 9090

@@ -4,9 +4,11 @@ package integration_test
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -54,11 +56,11 @@ func TestPoolGating_Blocking(t *testing.T) {
 	var wg sync.WaitGroup
 	for w := 1; w <= 4; w++ {
 		wg.Add(1)
-		go func() {
+		go func(workerID int) {
 			defer wg.Done()
 			asyncworker.WorkerWithGate(ctx, ctx, pipeline.Characteristics{HasExternalBackoff: false},
-				client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, poolGate)
-		}()
+				client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, poolGate, workerID, 4, "test-pool")
+		}(w)
 	}
 
 	// Send 3 requests into the channel
@@ -142,11 +144,11 @@ func TestPoolGating_Timeout(t *testing.T) {
 	var wg sync.WaitGroup
 	for w := 1; w <= 2; w++ {
 		wg.Add(1)
-		go func() {
+		go func(workerID int) {
 			defer wg.Done()
 			asyncworker.WorkerWithGate(ctx, ctx, pipeline.Characteristics{HasExternalBackoff: false},
-				client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, poolGate)
-		}()
+				client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, poolGate, workerID, 2, "test-pool")
+		}(w)
 	}
 
 	// Request 1 takes the capacity slot
@@ -248,7 +250,7 @@ func TestPoolGating_ActionWait(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		asyncworker.WorkerWithGate(ctx, ctx, pipeline.Characteristics{HasExternalBackoff: false},
-			client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate)
+			client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate, 1, 1, "test-pool")
 	}()
 
 	ir := asyncapi.NewInternalRequest(
@@ -312,7 +314,7 @@ func TestPoolGating_ActionRefuse(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		asyncworker.WorkerWithGate(ctx, ctx, pipeline.Characteristics{HasExternalBackoff: false},
-			client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate)
+			client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate, 1, 1, "test-pool")
 	}()
 
 	ir := asyncapi.NewInternalRequest(
@@ -347,3 +349,97 @@ func TestPoolGating_ActionRefuse(t *testing.T) {
 	cancel()
 	wg.Wait()
 }
+
+type dynamicBudgetGate struct {
+	mu     sync.Mutex
+	budget float64
+}
+
+func (g *dynamicBudgetGate) SetBudget(b float64) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.budget = b
+}
+
+func (g *dynamicBudgetGate) Budget(ctx context.Context) float64 {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return g.budget
+}
+
+func (g *dynamicBudgetGate) Apply(ctx context.Context, msg *asyncapi.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error) {
+	if g.Budget(ctx) <= 0 {
+		return pipeline.Wait(), nil
+	}
+	return pipeline.Continue(), nil
+}
+
+func TestPoolGating_ProportionalThrottling(t *testing.T) {
+	var activeWorkers int32
+	var maxObservedActive int32
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		current := atomic.AddInt32(&activeWorkers, 1)
+		for {
+			oldMax := atomic.LoadInt32(&maxObservedActive)
+			if current <= oldMax || atomic.CompareAndSwapInt32(&maxObservedActive, oldMax, current) {
+				break
+			}
+		}
+		time.Sleep(100 * time.Millisecond)
+		atomic.AddInt32(&activeWorkers, -1)
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{"result":"success"}`))
+	}))
+	defer server.Close()
+
+	client := asyncworker.NewHTTPInferenceClient(server.Client())
+	requestChannel := make(chan pipeline.EmbelishedRequestMessage, 20)
+	retryChannel := make(chan pipeline.RetryMessage, 20)
+	resultChannel := make(chan asyncapi.ResultMessage, 20)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	gate := &dynamicBudgetGate{budget: 0.3} // 30% of 10 workers = 3 workers allowed
+
+	const totalWorkers = 10
+	var wg sync.WaitGroup
+	for w := 1; w <= totalWorkers; w++ {
+		wg.Add(1)
+		go func(workerID int) {
+			defer wg.Done()
+			asyncworker.WorkerWithGate(ctx, ctx, pipeline.Characteristics{HasExternalBackoff: false},
+				client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate, workerID, totalWorkers, "test-pool")
+		}(w)
+	}
+
+	for i := 1; i <= 10; i++ {
+		ir := asyncapi.NewInternalRequest(
+			asyncapi.InternalRouting{RequestQueueName: "test-queue"},
+			&asyncapi.RequestMessage{
+				ID:       fmt.Sprintf("req-%d", i),
+				Created:  time.Now().Unix(),
+				Deadline: time.Now().Add(5 * time.Minute).Unix(),
+				Payload:  map[string]any{"model": "test"},
+			},
+		)
+		requestChannel <- pipeline.EmbelishedRequestMessage{
+			InternalRequest: ir,
+			RequestURL:      server.URL + "/v1/completions",
+		}
+	}
+
+	time.Sleep(250 * time.Millisecond)
+
+	maxActive := atomic.LoadInt32(&maxObservedActive)
+	assert.LessOrEqual(t, maxActive, int32(3), "expected at most 3 active workers for 0.3 budget on pool of 10")
+	assert.Greater(t, maxActive, int32(0), "expected at least 1 active worker")
+
+	for i := 0; i < 10; i++ {
+		<-resultChannel
+	}
+
+	cancel()
+	wg.Wait()
+}
@@ -55,7 +55,7 @@ func TestTierPriorityGate_Integration(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		asyncworker.WorkerWithGate(ctx, ctx, pipeline.Characteristics{HasExternalBackoff: false},
-			client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate)
+			client, requestChannel, retryChannel, resultChannel, 5*time.Minute, nil, gate, 1, 1, "test-pool")
 	}()
 
 	t.Run("Saturated - Interactive - Overflow => Drop with 429", func(t *testing.T) {