@@ -230,18 +230,6 @@ type BucketConfig struct {
230230 BucketCount int `yaml:"count"`
231231}
232232
233- const asyncTenantID = "$batch"
234-
235- // RequestQueueName returns the Redis sorted-set name for submitting async requests to the given pool.
236- func RequestQueueName (poolName string ) string {
237- return "llm-d-async:requests:" + poolName
238- }
239-
240- // ResultQueueName returns the Redis list name for collecting async results from the given pool.
241- func ResultQueueName (poolName string ) string {
242- return "llm-d-async:results:" + poolName + ":" + asyncTenantID
243- }
244-
245233// IsAsync returns true when the processor is configured for async dispatch.
246234func (c * ProcessorConfig ) IsAsync () bool {
247235 return c .DispatchMode == DispatchModeAsync
@@ -386,14 +374,21 @@ func (c *ProcessorConfig) Validate() error {
386374 return fmt .Errorf ("progress_ttl_seconds must be > 0" )
387375 }
388376
389- if err := c .validateDispatchMode (); err != nil {
377+ if err := c .validateGateways (); err != nil {
390378 return err
391379 }
392380
393381 return nil
394382}
395383
396- func (c * ProcessorConfig ) validateDispatchMode () error {
384+ func (c * ProcessorConfig ) validateGateways () error {
385+ if c .GlobalInferenceGateway == nil && len (c .ModelGateways ) == 0 {
386+ return fmt .Errorf ("either global_inference_gateway or model_gateways must be configured" )
387+ }
388+ if c .GlobalInferenceGateway != nil && len (c .ModelGateways ) > 0 {
389+ return fmt .Errorf ("global_inference_gateway and model_gateways are mutually exclusive" )
390+ }
391+
397392 switch c .DispatchMode {
398393 case DispatchModeSync , DispatchMode ("" ):
399394 c .DispatchMode = DispatchModeSync
@@ -405,21 +400,7 @@ func (c *ProcessorConfig) validateDispatchMode() error {
405400 }
406401}
407402
408- func (c * ProcessorConfig ) validateGateways () error {
409- if c .GlobalInferenceGateway == nil && len (c .ModelGateways ) == 0 {
410- return fmt .Errorf ("either global_inference_gateway or model_gateways must be configured" )
411- }
412- if c .GlobalInferenceGateway != nil && len (c .ModelGateways ) > 0 {
413- return fmt .Errorf ("global_inference_gateway and model_gateways are mutually exclusive" )
414- }
415- return nil
416- }
417-
418403func (c * ProcessorConfig ) validateSyncDispatchConfig () error {
419- if err := c .validateGateways (); err != nil {
420- return err
421- }
422-
423404 if c .GlobalInferenceGateway != nil {
424405 if err := validateGatewayConfig ("global_inference_gateway" , * c .GlobalInferenceGateway ); err != nil {
425406 return err
@@ -435,15 +416,13 @@ func (c *ProcessorConfig) validateSyncDispatchConfig() error {
435416
436417func (c * ProcessorConfig ) validateAsyncDispatchConfig () error {
437418 if c .AsyncDispatchConfig .ResultPollTimeout <= 0 {
438- return fmt .Errorf ("async.result_poll_timeout must be > 0" )
439- }
440- if err := c .validateGateways (); err != nil {
441- return err
419+ return fmt .Errorf ("async_dispatch.result_poll_timeout must be > 0" )
442420 }
443421 if c .GlobalInferenceGateway != nil {
444- if c .GlobalInferenceGateway .InferencePoolName == "" {
445- return fmt .Errorf ("global_inference_gateway.inference_pool_name must be set when dispatch_mode is %q" , DispatchModeAsync )
446- }
422+ return fmt .Errorf ("global_inference_gateway is not supported with dispatch_mode %q; use model_gateways with inference_pool_name" , DispatchModeAsync )
423+ }
424+ if len (c .ModelGateways ) == 0 {
425+ return fmt .Errorf ("model_gateways must be configured when dispatch_mode is %q" , DispatchModeAsync )
447426 }
448427 for model , gw := range c .ModelGateways {
449428 if gw .InferencePoolName == "" {
@@ -580,6 +559,7 @@ func toGatewayClientConfig(gw ModelGatewayConfig, apiKey string) inference.Gatew
580559type ResolvedGateways struct {
581560 Global * inference.GatewayClientConfig
582561 PerModel map [string ]inference.GatewayClientConfig
562+ Async * inference.AsyncClientConfig
583563}
584564
585565// ResolveModelGateways resolves API keys for all configured gateways and returns
@@ -588,6 +568,17 @@ type ResolvedGateways struct {
588568func ResolveModelGateways (cfg * ProcessorConfig ) (* ResolvedGateways , error ) {
589569 result := & ResolvedGateways {}
590570
571+ if cfg .IsAsync () {
572+ models := make (map [string ]string , len (cfg .ModelGateways ))
573+ for model , gw := range cfg .ModelGateways {
574+ models [model ] = gw .InferencePoolName
575+ }
576+ result .Async = & inference.AsyncClientConfig {
577+ Models : models ,
578+ }
579+ return result , nil
580+ }
581+
591582 if cfg .GlobalInferenceGateway != nil {
592583 apiKey , err := resolveGatewayAPIKey ("global_inference_gateway" , * cfg .GlobalInferenceGateway )
593584 if err != nil {
0 commit comments