refactor: restructure async config with explicit queue names (#526)

benchmarks/helm-values/scenario-5-async.yaml

@@ -3,6 +3,6 @@ processor:
     dispatchMode: async
     asyncDispatch:
       resultPollTimeout: "30s"
-    modelGateways:
-      Qwen/Qwen3-8B:
-        inferencePoolName: "optimized-baseline"
+      models:
+        Qwen/Qwen3-8B:
+          inferencePoolName: "optimized-baseline"

charts/batch-gateway/templates/processor-configmap.yaml

@@ -48,6 +48,22 @@ data:
     {{- if .Values.processor.config.asyncDispatch }}
     async_dispatch:
       result_poll_timeout: {{ .Values.processor.config.asyncDispatch.resultPollTimeout | quote }}
+      {{- if .Values.processor.config.asyncDispatch.models }}
+      models:
+        {{- range $model, $cfg := .Values.processor.config.asyncDispatch.models }}
+        {{ $model | quote }}:
+          inference_pool_name: {{ $cfg.inferencePoolName | quote }}
+          {{- if $cfg.inferenceObjective }}
+          inference_objective: {{ $cfg.inferenceObjective | quote }}
+          {{- end }}
+          {{- if $cfg.requestQueueName }}
+          request_queue_name: {{ $cfg.requestQueueName | quote }}
+          {{- end }}
+          {{- if $cfg.resultQueueName }}
+          result_queue_name: {{ $cfg.resultQueueName | quote }}
+          {{- end }}
+        {{- end }}
+      {{- end }}
     {{- end }}
 
     {{- with .Values.processor.config.globalInferenceGateway }}
@@ -56,9 +72,6 @@ data:
       {{- if .inferenceObjective }}
       inference_objective: {{ .inferenceObjective | quote }}
       {{- end }}
-      {{- if .inferencePoolName }}
-      inference_pool_name: {{ .inferencePoolName | quote }}
-      {{- end }}
       request_timeout: {{ .requestTimeout | quote }}
       max_retries: {{ .maxRetries }}
       initial_backoff: {{ .initialBackoff | quote }}
@@ -85,9 +98,6 @@ data:
     model_gateways:
       {{- range $model, $cfg := .Values.processor.config.modelGateways }}
       {{ $model | quote }}:
-        {{- if $cfg.inferencePoolName }}
-        inference_pool_name: {{ $cfg.inferencePoolName | quote }}
-        {{- end }}
         {{- if $cfg.url }}
         url: {{ $cfg.url | quote }}
         request_timeout: {{ $cfg.requestTimeout | quote }}

charts/batch-gateway/tests/processor-configmap_test.yaml

@@ -353,6 +353,27 @@ tests:
           path: data["config.yaml"]
           pattern: 'global_inference_gateway:[\s\S]*inference_objective:'
 
+  - it: should render async_dispatch.models when configured
+    set:
+      processor.config.dispatchMode: async
+      processor.config.asyncDispatch:
+        resultPollTimeout: "30s"
+        models:
+          "sim-model":
+            inferencePoolName: "sim-pool"
+          "sim-model-inject":
+            inferencePoolName: "sim-pool-inject"
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'dispatch_mode: "async"'
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'async_dispatch:[\s\S]*models:[\s\S]*"sim-model":[\s\S]*inference_pool_name: "sim-pool"'
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: '"sim-model-inject":[\s\S]*inference_pool_name: "sim-pool-inject"'
+
   - it: should not render send_fairness_header when false (default)
     asserts:
       - notMatchRegex:

charts/batch-gateway/values.yaml

@@ -393,6 +393,27 @@ processor:
     #     inferenceObjective: "batch-sheddable-b"   # references gie-b pool
     #     requestTimeout: "2m"
     #     maxRetries: 1
+
+    # Async dispatch mode (alternative to sync).
+    # Set dispatchMode: "async" and configure asyncDispatch.models instead
+    # of modelGateways. Each model maps to a Redis-backed inference pool.
+    # Queue names can be set explicitly; when omitted they are derived from
+    # inferencePoolName (deprecated — set explicit names for new deployments).
+    # dispatchMode: "async"
+    # asyncDispatch:
+    #   resultPollTimeout: "30s"
+    #   models:
+    #     "llama-3":
+    #       inferencePoolName: "pool-a"
+    #       inferenceObjective: "batch-sheddable-a"  # optional: GIE InferenceObjective CRD name
+    #       requestQueueName: "llm-d-async:requests:pool-a"  # optional: explicit request queue
+    #       resultQueueName: "llm-d-async:results:pool-a"    # optional: explicit result queue
+    #     "mistral":
+    #       inferencePoolName: "pool-b"
+    #       inferenceObjective: "batch-sheddable-b"
+    #       requestQueueName: "llm-d-async:requests:pool-b"
+    #       resultQueueName: "llm-d-async:results:pool-b"
+
     defaultOutputExpirationSeconds: 7776000  # 90 days
     progressTTLSeconds: 86400  # 24 hours
     # Whether to send x-gateway-inference-fairness-id on inference requests.

internal/processor/config/config.go

@@ -81,6 +81,31 @@ const (
 	DispatchModeAsync DispatchMode = "async"
 )
 
+// AsyncModelConfig describes the async dispatch target for a single model.
+type AsyncModelConfig struct {
+	// InferencePoolName identifies the async dispatch pool for this model.
+	InferencePoolName string `yaml:"inference_pool_name"`
+
+	// InferenceObjective is the name of a GIE InferenceObjective CRD sent in
+	// the x-gateway-inference-objective header on inference requests.
+	// When empty, the header is not sent.
+	InferenceObjective string `yaml:"inference_objective"`
+
+	// RequestQueueName overrides the Redis sorted-set queue name used to
+	// submit inference requests. When empty, the name is derived from
+	// InferencePoolName as "llm-d-async:requests:<pool>".
+	// Deprecated fallback: the derived naming convention will be removed in
+	// a future release. Set explicit queue names for new deployments.
+	RequestQueueName string `yaml:"request_queue_name"`
+
+	// ResultQueueName overrides the Redis sorted-set queue name used to
+	// read inference results. When empty, the name is derived from
+	// InferencePoolName as "llm-d-async:results:<pool>".
+	// Deprecated fallback: the derived naming convention will be removed in
+	// a future release. Set explicit queue names for new deployments.
+	ResultQueueName string `yaml:"result_queue_name"`
+}
+
 // AsyncDispatchConfig holds configuration for the llm-d-async dispatch backend.
 // Only used when DispatchMode == "async".
 type AsyncDispatchConfig struct {
@@ -92,6 +117,10 @@ type AsyncDispatchConfig struct {
 	// ResultPollTimeout is the timeout per GetResult poll cycle.
 	// Controls how long each blocking poll waits before retrying.
 	ResultPollTimeout time.Duration `yaml:"result_poll_timeout"`
+
+	// Models maps model names to their async dispatch targets.
+	// Required when DispatchMode == "async".
+	Models map[string]AsyncModelConfig `yaml:"models"`
 }
 
 type ProcessorConfig struct {
@@ -219,8 +248,8 @@ type ModelGatewayConfig struct {
 	TLSClientCertFile     string `yaml:"tls_client_cert_file,omitempty"`
 	TLSClientKeyFile      string `yaml:"tls_client_key_file,omitempty"`
 
-	// InferencePoolName identifies the async dispatch pool for this model/gateway.
-	// Required when dispatch_mode is "async". Ignored in sync mode.
+	// Deprecated: use AsyncDispatchConfig.Models instead.
+	// InferencePoolName was used to identify the async dispatch pool for this model.
 	InferencePoolName string `yaml:"inference_pool_name"`
 }
 
@@ -239,6 +268,12 @@ func (c *ProcessorConfig) IsAsync() bool {
 // gateway that will handle requests for modelID.
 // Returns "" when no objective is configured, which means the header is not sent.
 func (c *ProcessorConfig) InferenceObjectiveFor(modelID string) string {
+	if c.IsAsync() {
+		if m, ok := c.AsyncDispatchConfig.Models[modelID]; ok {
+			return m.InferenceObjective
+		}
+		return ""
+	}
 	if c.GlobalInferenceGateway != nil {
 		return c.GlobalInferenceGateway.InferenceObjective
 	}
@@ -382,16 +417,15 @@ func (c *ProcessorConfig) Validate() error {
 }
 
 func (c *ProcessorConfig) validateGateways() error {
-	if c.GlobalInferenceGateway == nil && len(c.ModelGateways) == 0 {
-		return fmt.Errorf("either global_inference_gateway or model_gateways must be configured")
-	}
-	if c.GlobalInferenceGateway != nil && len(c.ModelGateways) > 0 {
-		return fmt.Errorf("global_inference_gateway and model_gateways are mutually exclusive")
-	}
-
 	switch c.DispatchMode {
 	case DispatchModeSync, DispatchMode(""):
 		c.DispatchMode = DispatchModeSync
+		if c.GlobalInferenceGateway == nil && len(c.ModelGateways) == 0 {
+			return fmt.Errorf("either global_inference_gateway or model_gateways must be configured")
+		}
+		if c.GlobalInferenceGateway != nil && len(c.ModelGateways) > 0 {
+			return fmt.Errorf("global_inference_gateway and model_gateways are mutually exclusive")
+		}
 		return c.validateSyncDispatchConfig()
 	case DispatchModeAsync:
 		return c.validateAsyncDispatchConfig()
@@ -419,14 +453,17 @@ func (c *ProcessorConfig) validateAsyncDispatchConfig() error {
 		return fmt.Errorf("async_dispatch.result_poll_timeout must be > 0")
 	}
 	if c.GlobalInferenceGateway != nil {
-		return fmt.Errorf("global_inference_gateway is not supported with dispatch_mode %q; use model_gateways with inference_pool_name", DispatchModeAsync)
+		return fmt.Errorf("global_inference_gateway is not supported with dispatch_mode %q; use async_dispatch.models", DispatchModeAsync)
 	}
-	if len(c.ModelGateways) == 0 {
-		return fmt.Errorf("model_gateways must be configured when dispatch_mode is %q", DispatchModeAsync)
+	if len(c.AsyncDispatchConfig.Models) == 0 {
+		return fmt.Errorf("async_dispatch.models must be configured when dispatch_mode is %q", DispatchModeAsync)
 	}
-	for model, gw := range c.ModelGateways {
-		if gw.InferencePoolName == "" {
-			return fmt.Errorf("model_gateways[%s].inference_pool_name must be set when dispatch_mode is %q", model, DispatchModeAsync)
+	for model, m := range c.AsyncDispatchConfig.Models {
+		if m.InferencePoolName == "" {
+			return fmt.Errorf("async_dispatch.models[%s].inference_pool_name must be set", model)
+		}
+		if (m.RequestQueueName == "") != (m.ResultQueueName == "") {
+			return fmt.Errorf("async_dispatch.models[%s]: request_queue_name and result_queue_name must both be set or both be empty", model)
 		}
 	}
 	return nil
@@ -569,9 +606,13 @@ func ResolveModelGateways(cfg *ProcessorConfig) (*ResolvedGateways, error) {
 	result := &ResolvedGateways{}
 
 	if cfg.IsAsync() {
-		models := make(map[string]string, len(cfg.ModelGateways))
-		for model, gw := range cfg.ModelGateways {
-			models[model] = gw.InferencePoolName
+		models := make(map[string]inference.AsyncModelPoolConfig, len(cfg.AsyncDispatchConfig.Models))
+		for model, m := range cfg.AsyncDispatchConfig.Models {
+			models[model] = inference.AsyncModelPoolConfig{
+				PoolName:         m.InferencePoolName,
+				RequestQueueName: m.RequestQueueName,
+				ResultQueueName:  m.ResultQueueName,
+			}
 		}
 		result.Async = &inference.AsyncClientConfig{
 			Models:            models,