fix(edge): provider-pool thinking과 queue 정책을 정리한다
vLLM-MLX provider-pool에서 strict output이 thinking을 끄지 않도록 하고, queue_timeout_ms=0을 IOP queue timeout 없음으로 해석해야 한다. long reasoning/long-context 요청은 IOP queue timeout이 아니라 caller cancellation과 backend timeout 정책으로 제어한다.
This commit is contained in:
toki
parent
cb426fe9f5
commit
ca5eb4685c
12 changed files with 267 additions and 54 deletions
|
|
@ -37,6 +37,7 @@ tracked config에는 public 예시와 기본 구조만 두고, 실제 endpoint/c
|
|||
- `nodes[].providers[].adapter`는 같은 Node 안의 enabled adapter instance key를 참조해야 한다. Exact instance key를 우선하고, legacy type-name route는 같은 type의 enabled instance가 정확히 하나일 때만 허용한다. `category: cli` resource는 enabled CLI adapter가 필요하다.
|
||||
- `nodes[].providers[].enabled`: 생략 또는 `true` → provider pool dispatch 후보에 포함. `false` → dispatch pool에서 제외. 비활성화된 provider는 status snapshot에 `status=disabled`, `health=disabled`, `capacity=0`으로 표시된다. adapter process lifecycle 변경 없음. config refresh 시 `enabled` 토글은 live-apply(restart 불필요)로 분류된다. disabled provider의 adapter reference check는 skip되지만 structural validation(type, category, models, numeric bounds)은 수행된다.
|
||||
- `nodes[].providers[].priority`: provider-pool dispatch tie-breaker다. 기본값은 `0`이고 음수는 validation error다. dispatch는 `in_flight < capacity` 후보 중 가장 낮은 `in_flight`를 먼저 선택하며, `in_flight`가 같은 후보에서만 낮은 숫자의 `priority`를 우선한다. `in_flight`와 `priority`가 모두 같으면 기존 순환을 유지한다. priority 변경은 live-apply(restart 불필요)로 분류된다.
|
||||
- `nodes[].providers[].queue_timeout_ms`와 adapter instance `queue_timeout_ms`는 provider-pool 대기열 timeout이다. queue policy가 구성된 상태에서 값이 `0`이면 queue timeout을 두지 않고, caller context cancellation 또는 연결 종료로만 대기 요청을 중단한다. provider가 queue policy를 전혀 제공하지 않으면 runtime 기본 queue timeout을 사용한다.
|
||||
- legacy single-instance adapter 설정은 load 시 named instance slice로 normalize된다.
|
||||
- `NodeConfigPayload`는 Edge가 Node에 내려주는 실행 adapter/runtime payload다.
|
||||
- refresh 결과는 `applied`, `restart_required`, `rejected`를 구분하고, changed node/provider/model/report slice는 안정적으로 non-nil이어야 한다.
|
||||
|
|
|
|||
|
|
@ -207,7 +207,7 @@ Provider pool model catalog의 `models[]` entry가 generation policy를 제공
|
|||
|
||||
- `default_max_tokens`: caller가 출력 token limit을 생략했을 때 `max_tokens` 또는 `max_output_tokens`로 주입한다.
|
||||
- `min_max_tokens`: caller가 너무 작은 출력 token limit을 보냈을 때 해당 값까지 올린다. caller 값이 더 크면 보존한다.
|
||||
- `default_thinking_token_budget`: caller가 `thinking_token_budget`을 생략했고 strict output이 reasoning을 금지하지 않을 때 내부 실행 입력에 주입한다.
|
||||
- `default_thinking_token_budget`: caller가 `thinking_token_budget`을 생략했을 때 내부 실행 입력에 주입한다. provider-pool catalog에 이 값이 있으면 Edge는 `think=true`도 함께 주입해 vLLM/vLLM-MLX 계열 adapter가 `chat_template_kwargs.enable_thinking=true`로 전달하도록 한다.
|
||||
|
||||
Conflict 정책:
|
||||
|
||||
|
|
@ -220,6 +220,7 @@ Conflict 정책:
|
|||
Strict output 모드:
|
||||
|
||||
- strict output가 활성화되면 `think=true`가 명시되지 않은 요청은 내부 실행 입력에서 `think=false`로 낮춘다.
|
||||
- 단, provider-pool `models[]` entry의 `default_thinking_token_budget`가 적용되는 모델은 catalog의 thinking policy가 우선한다. 이 경우 요청이 `think=false` 또는 `reasoning_effort=none`을 명시하지 않았다면 strict output에서도 `think=true`와 `thinking_token_budget`을 내부 실행 입력에 넣는다.
|
||||
|
||||
`tools`가 있는 Chat Completions 요청에서 provider route(`openai_compat`, `vllm`, `ollama`, provider pool)는 forced tool 선택 객체와 `"none"` 같은 명시적 `tool_choice`를 backend에 전달한다. 단, `"auto"`는 OpenAI-compatible 기본값과 같으므로 provider request에서는 생략한다. 일부 vLLM 계열 backend는 explicit/default `"auto"`를 `--enable-auto-tool-choice`/`--tool-call-parser` 없이 400으로 거부한다. 이 400이 발생하고 요청 tool이 정확히 1개이면 Node adapter는 해당 tool에 대한 forced `tool_choice`로 1회 재시도한다. forced tool도 `--tool-call-parser` 요구로 거부되거나 여러 tool이라 forced를 고를 수 없으면, Node adapter는 `tools`/`tool_choice`를 제거하고 text tool-call system instruction을 leading system message에 병합해 1회 재시도하며 완료 metadata에 `openai_text_tool_fallback: "true"`를 싣는다.
|
||||
provider가 native OpenAI-compatible `tool_calls`를 반환하면 Node는 내부 `RunEvent.metadata["openai_tool_calls"]` JSON으로 보존하고, Edge는 이를 OpenAI-compatible `message.tool_calls` 또는 stream `delta.tool_calls`로 반환하며 `finish_reason: "tool_calls"`를 사용한다.
|
||||
|
|
|
|||
|
|
@ -93,6 +93,8 @@ Mac Studio의 `http://192.168.2.3:8005/v1` mlx-vlm DiffusionGemma endpoint는
|
|||
- DGX Spark 01은 FlashInfer FP4 JIT가 `ninja`를 PATH에서 호출하므로 start script에 `PATH=/home/digitalcommerce_dgx_spark_01/vllm_env/bin:$PATH`를 둔다.
|
||||
- Mac Studio provider runtime은 capacity `5`, context window `262144`, requested KV `262144x3` 기준으로 `--continuous-batching`, `--max-num-seqs 5`, `--prefill-batch-size 5`, `--completion-batch-size 5`, `--chunked-prefill-tokens 1024`, `--max-request-tokens 262144`, `--max-kv-size 786432`, `--reasoning-parser gemma4`을 적용했고 `/health`, `/v1/models`, 직접 동시성 `1..5` chat completion benchmark 통과를 확인했다.
|
||||
- 2026-07-02 재확인 기준 DGX Spark 01은 mac-mini 내부 endpoint `http://192.168.2.2:8002`에서 `/health` 200과 `/v1/models`를 통과했다. DGX Spark 02는 mac-mini에서 SSH, provider `/health`, `/v1/models`가 회복됐고 직접 provider benchmark를 통과했다. Edge OpenAI-compatible capacity smoke는 Control Plane status snapshot 기준으로 재검증 완료됐다.
|
||||
- Gemma provider-pool `gemma4:26b`는 `default_thinking_token_budget: 1024` 기준으로 thinking을 기본 활성화한다. Edge strict output이 켜져 있어도 provider-pool catalog의 thinking policy가 우선하며, vLLM/vLLM-MLX adapter에는 `chat_template_kwargs.enable_thinking=true`로 전달되는지 확인한다.
|
||||
- Gemma provider-pool의 IOP queue timeout은 두지 않는다. config 값은 `queue_timeout_ms=0`이며, caller context cancellation 또는 연결 종료로만 queued request를 중단한다. OpenAI run timeout과 backend request timeout은 long reasoning/long-context 요청을 위해 각각 `openai.timeout_sec=1800`, `request_timeout_ms=1800000`으로 둔다.
|
||||
|
||||
## 2026-07-02 dev-corp native runtime 상태
|
||||
|
||||
|
|
@ -136,7 +138,7 @@ Mac Studio의 `http://192.168.2.3:8005/v1` mlx-vlm DiffusionGemma endpoint는
|
|||
- OpenAI-compatible smoke에서 `/healthz`, `/v1/models`, `/v1/responses`가 기대 상태로 응답한다.
|
||||
- dev-corp capacity smoke는 `/v1/responses`와 `/v1/chat/completions` 각각에 provider capacity 총합 + 1개 동시 요청을 보낸다. Edge config에서 기본 후보 capacity `4 + 4 + 5 = 13`을 확정한 경우 Control Plane status의 `provider_snapshots`에서 총 `in_flight=13`, `queued>=1` 관측을 기준으로 한다.
|
||||
- capacity smoke 완료 후 대상 provider의 `in_flight=0`, `queued=0` 회복을 확인한다.
|
||||
- Gemma 계열 provider-pool smoke는 reasoning/tool-parser 관련 텍스트가 포함될 수 있다. exact-output match를 기본 판정으로 쓰지 않는다.
|
||||
- Gemma 계열 provider-pool smoke는 thinking enabled 기준이며 reasoning/tool-parser 관련 텍스트가 포함될 수 있다. exact-output match를 기본 판정으로 쓰지 않는다.
|
||||
|
||||
## 기준 출력 예시
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
test_env: dev-corp
|
||||
profile: dev-corp-provider-pool
|
||||
last_updated_at: "2026-07-02"
|
||||
last_updated_at: "2026-07-06"
|
||||
|
||||
source:
|
||||
remote_runner:
|
||||
|
|
@ -8,7 +8,7 @@ source:
|
|||
repo_root: /Users/fe/agent-work/iop-dev-corp
|
||||
setup_required: true
|
||||
setup_status: deployed
|
||||
current_observation: /Users/fe/agent-work/iop-dev-corp checkout and build/dev-corp-runtime runtime were created on 2026-06-25, then clean-synced/rebuilt at source ref 5b1255a on 2026-07-02; /Users/fe/iop-field remains unrelated legacy field state
|
||||
current_observation: /Users/fe/agent-work/iop-dev-corp checkout and build/dev-corp-runtime runtime were created on 2026-06-25, clean-synced/rebuilt at source ref 2d08d24 on 2026-07-06, then validated with the provider-pool thinking-policy patch and gemma4:26b default_thinking_token_budget config update; /Users/fe/iop-field remains unrelated legacy field state
|
||||
clean_sync:
|
||||
- git fetch origin main
|
||||
- git reset --hard origin/main
|
||||
|
|
@ -50,8 +50,20 @@ model:
|
|||
alias_status: candidate
|
||||
alias_policy: expose one IOP model alias and map provider-specific served_model values per provider
|
||||
provider_capacity_total: 13
|
||||
provider_capacity_status: verified_with_control_plane_provider_snapshots_2026_07_02
|
||||
provider_capacity_status: verified_with_control_plane_provider_snapshots_2026_07_06
|
||||
context_window_max: 262144
|
||||
default_thinking_token_budget: 1024
|
||||
reasoning_policy: bounded_thinking_enabled
|
||||
think_policy:
|
||||
default: enabled
|
||||
strict_output_behavior: provider_pool_catalog_default_overrides_strict_disable
|
||||
adapter_mapping: vllm_and_vllm_mlx_use_chat_template_kwargs_enable_thinking_true
|
||||
timeout_policy:
|
||||
openai_timeout_sec: 1800
|
||||
provider_queue_timeout_ms: 0
|
||||
provider_queue_timeout_policy: disabled_no_iop_queue_timeout
|
||||
backend_request_timeout_ms: 1800000
|
||||
note: provider-pool long reasoning and long-context requests do not use an IOP queue timeout; backend/OpenAI run budgets remain 30 minutes
|
||||
runtime_capacity_targets:
|
||||
dgx_spark:
|
||||
capacity: 4
|
||||
|
|
@ -70,15 +82,18 @@ model:
|
|||
- /v1/chat/completions
|
||||
concurrent_requests: capacity_plus_one
|
||||
expected_total_in_flight: 13
|
||||
expected_total_in_flight_status: verified_2026_07_02
|
||||
expected_total_in_flight_status: verified_2026_07_06
|
||||
expected_min_queued: 1
|
||||
extended_concurrent_requests: 16
|
||||
extended_expected_min_queued: 3
|
||||
prompt_policy: long_reasoning_allowed
|
||||
think_policy: enabled_for_provider_pool
|
||||
exact_output_match: false
|
||||
latest_capacity_verification:
|
||||
date: "2026-07-02"
|
||||
source_ref: 5b1255abc5ad530ebfd7a65deecdf5575425a34c
|
||||
date: "2026-07-06"
|
||||
source_ref: 2d08d240d48a954df6847df2ba08fb8b39fc3f7d
|
||||
edge_runtime_patch: provider_pool_thinking_policy_enabled_under_strict_output
|
||||
default_thinking_token_budget: 1024
|
||||
control_plane_enabled: true
|
||||
control_plane_status_url: http://127.0.0.1:18002/edges/dev-corp-edge/status
|
||||
provider_snapshot_baseline:
|
||||
|
|
@ -99,6 +114,8 @@ model:
|
|||
responses:
|
||||
concurrent_requests: 14
|
||||
ok: 14
|
||||
timeout_errors: 0
|
||||
http_errors: 0
|
||||
max_total_in_flight: 13
|
||||
max_total_queued: 3
|
||||
provider_max:
|
||||
|
|
@ -115,6 +132,8 @@ model:
|
|||
chat_completions:
|
||||
concurrent_requests: 14
|
||||
ok: 14
|
||||
timeout_errors: 0
|
||||
http_errors: 0
|
||||
max_total_in_flight: 13
|
||||
max_total_queued: 3
|
||||
provider_max:
|
||||
|
|
@ -148,7 +167,7 @@ model:
|
|||
- DGX Spark 01 FlashInfer FP4 JIT invokes ninja from PATH; start script now prefixes /home/digitalcommerce_dgx_spark_01/vllm_env/bin
|
||||
- Mac Studio vLLM-MLX pyexpat required DYLD_LIBRARY_PATH=/opt/homebrew/opt/expat/lib
|
||||
verification:
|
||||
last_rechecked_at: "2026-07-02"
|
||||
last_rechecked_at: "2026-07-06"
|
||||
mac_studio_health: passed
|
||||
dgx01_health: passed
|
||||
dgx02_health: passed
|
||||
|
|
@ -156,7 +175,7 @@ model:
|
|||
dgx01: mac-mini http://192.168.2.2:8002 returned /health 200 and /v1/models exposed gemma-4-26B-A4B-it-NVFP4 after vllm_server_8002 restart; startup log reported GPU KV cache size 834,507 tokens and 3.18x concurrency for 262,144-token requests
|
||||
dgx02: node-local http://127.0.0.1:8004 and mac-mini http://192.168.2.4:8004 returned /health 200, /v1/models exposed gemma-4-26B-A4B-it-NVFP4, and direct concurrency 1..4 chat completion benchmark passed
|
||||
mac_studio: node-local http://127.0.0.1:8004 and mac-mini http://192.168.2.3:8004 returned /health 200, /v1/models exposed mlx-community/gemma-4-26b-a4b-it-nvfp4, and direct concurrency 1..5 chat completion benchmark passed
|
||||
edge_openai_smoke: passed_with_control_plane_capacity_snapshot_2026_07_02
|
||||
edge_openai_smoke: passed_with_control_plane_capacity_snapshot_2026_07_06
|
||||
|
||||
nodes:
|
||||
- id: corp-dgx-spark-01-vllm-node
|
||||
|
|
@ -177,6 +196,8 @@ nodes:
|
|||
capacity_status: configured_health_passed_direct_smoke_and_edge_capacity_smoke
|
||||
last_runtime_observation: restarted with gpu_memory_utilization 0.40 on 2026-07-02; mac-mini http://192.168.2.2:8002 returned /health 200 and /v1/models; startup log reported GPU KV cache size 834,507 tokens and 3.18x concurrency for 262,144-token requests
|
||||
capacity_basis: dev-corp target baseline; vLLM --max-num-seqs 4, --max-model-len 262144, --gpu-memory-utilization 0.40; observed KV cache satisfies requested 262144x2 minimum
|
||||
queue_timeout_ms: 0
|
||||
request_timeout_ms: 1800000
|
||||
edge_connectivity:
|
||||
mode: reverse_ssh_tunnel
|
||||
reason: current node.yaml uses tunnel-local Edge addr in the native provider-pool runtime
|
||||
|
|
@ -215,6 +236,8 @@ nodes:
|
|||
context_window_max: 262144
|
||||
kv_size_tokens_effective: 834507
|
||||
full_context_concurrency_observed: 3.18
|
||||
default_chat_template_kwargs:
|
||||
enable_thinking: true
|
||||
|
||||
- id: corp-dgx-spark-02-vllm-node
|
||||
alias: corp-dgx-spark-02-vllm
|
||||
|
|
@ -234,6 +257,8 @@ nodes:
|
|||
capacity_status: configured_health_passed_direct_smoke_and_edge_capacity_smoke
|
||||
last_runtime_observation: Docker vllm-gemma4 running with gpu_memory_utilization 0.40; node-local and mac-mini /health passed, /v1/models exposed gemma-4-26B-A4B-it-NVFP4, and direct concurrency 1..4 chat completion benchmark passed; startup log reported GPU KV cache size 860,222 tokens and 3.28x concurrency for 262,144-token requests
|
||||
capacity_basis: dev-corp target baseline; Docker publishes host 8004 to container 8004; vLLM --max-num-seqs 4, --max-model-len 262144, --gpu-memory-utilization 0.40; observed KV cache satisfies requested 262144x2 minimum
|
||||
queue_timeout_ms: 0
|
||||
request_timeout_ms: 1800000
|
||||
edge_connectivity:
|
||||
mode: reverse_ssh_tunnel
|
||||
reason: node02 cannot reach mac-mini 172.24.63.178:18085/18086 directly from 192.168.2.4
|
||||
|
|
@ -270,6 +295,8 @@ nodes:
|
|||
context_window_max: 262144
|
||||
kv_size_tokens_effective: 860222
|
||||
full_context_concurrency_observed: 3.28
|
||||
default_chat_template_kwargs:
|
||||
enable_thinking: true
|
||||
|
||||
- id: corp-mac-studio-mlx-vllm-node
|
||||
alias: corp-mac-studio-mlx-vllm
|
||||
|
|
@ -290,6 +317,8 @@ nodes:
|
|||
capacity_status: configured_health_passed_direct_smoke_and_edge_capacity_smoke
|
||||
last_runtime_observation: screen vllm_mlx_8004 running with max_num_seqs 5, max_request_tokens 262144, max_kv_size 786432; node-local and mac-mini /health passed, /v1/models exposed mlx-community/gemma-4-26b-a4b-it-nvfp4, and direct concurrency 1..5 chat completion benchmark passed
|
||||
capacity_basis: vllm-mlx --max-num-seqs 5 with requested KV 262144x3 mapped to --max-kv-size 786432
|
||||
queue_timeout_ms: 0
|
||||
request_timeout_ms: 1800000
|
||||
runtime:
|
||||
host: dc-devui-MacStudio.local
|
||||
os: macOS 26.2
|
||||
|
|
@ -320,6 +349,8 @@ nodes:
|
|||
context_window_max: 262144
|
||||
kv_size_tokens_effective: 786432
|
||||
reasoning_parser: gemma4
|
||||
default_chat_template_kwargs:
|
||||
enable_thinking: true
|
||||
secondary_provider_candidate:
|
||||
id: corp-mac-studio-mlx-vlm-diffusiongemma
|
||||
type: mlx-vlm
|
||||
|
|
|
|||
|
|
@ -92,6 +92,8 @@ Mac Studio의 secondary `http://192.168.2.3:8005/v1` endpoint는 기본 Node/pro
|
|||
- DGX Spark에서 explicit `--max-num-batched-tokens 524288`를 사용하는 경우 first profile 중 FP4 MoE 커널 한계에 걸릴 수 있어 `VLLM_MAX_TOKENS_PER_EXPERT_FP4_MOE=4194304` 보정이 필요하다.
|
||||
- DGX Spark 01은 FlashInfer FP4 JIT가 first compile 중 `ninja`를 PATH에서 찾으므로 `/home/digitalcommerce_dgx_spark_01/vllm_env/bin` PATH prefix가 필요하다.
|
||||
- Mac Studio에는 capacity `5`, context window `262144`, requested KV `262144x3` 기준을 `--max-num-seqs 5`, `--max-request-tokens 262144`, `--max-kv-size 786432`로 반영했고 `/health`, `/v1/models`, 직접 동시성 `1..5` chat completion benchmark 통과를 확인했다.
|
||||
- Gemma provider-pool은 `default_thinking_token_budget: 1024` 기준으로 thinking을 기본 활성화한다. vLLM과 vLLM-MLX provider 모두 Edge/Node adapter 요청에서 `chat_template_kwargs.enable_thinking=true`가 적용되는 방향으로 검증한다.
|
||||
- Gemma provider-pool의 provider queue timeout은 두지 않는다. config 값은 `queue_timeout_ms=0`이며, backend request timeout은 30분 기준인 `request_timeout_ms=1800000`으로 맞춘다.
|
||||
- 2026-07-02 재확인 기준 DGX Spark 02는 mac-mini에서 SSH, provider `/health`, `/v1/models`가 회복됐고 직접 provider benchmark를 통과했다. Edge OpenAI-compatible capacity smoke는 Control Plane status snapshot 기준으로 재검증 완료됐다.
|
||||
|
||||
## 2026-07-02 연결/용량 검증 상태
|
||||
|
|
|
|||
|
|
@ -181,8 +181,13 @@ func applyModelCatalogGenerationPolicyToChat(req *chatCompletionRequest, entry c
|
|||
return
|
||||
}
|
||||
applyOutputTokenPolicy(&req.MaxTokens, req.MaxCompletionTokens, entry.DefaultMaxTokens, entry.MinMaxTokens)
|
||||
if entry.DefaultThinkingTokenBudget > 0 && req.ThinkingTokenBudget == nil && chatRequestAllowsDefaultThinkingBudget(*req, strictOutput) {
|
||||
req.ThinkingTokenBudget = intPtr(entry.DefaultThinkingTokenBudget)
|
||||
if entry.DefaultThinkingTokenBudget > 0 && chatRequestAllowsDefaultThinkingBudget(*req) {
|
||||
if req.ThinkingTokenBudget == nil {
|
||||
req.ThinkingTokenBudget = intPtr(entry.DefaultThinkingTokenBudget)
|
||||
}
|
||||
if strictOutput && req.Think == nil {
|
||||
req.Think = boolPtr(true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -216,10 +221,7 @@ func applyOutputTokenPolicy(primary **int, fallback *int, defaultTokens, minToke
|
|||
}
|
||||
}
|
||||
|
||||
func chatRequestAllowsDefaultThinkingBudget(req chatCompletionRequest, strictOutput bool) bool {
|
||||
if strictOutput && (req.Think == nil || !*req.Think) {
|
||||
return false
|
||||
}
|
||||
func chatRequestAllowsDefaultThinkingBudget(req chatCompletionRequest) bool {
|
||||
if req.Think != nil && !*req.Think {
|
||||
return false
|
||||
}
|
||||
|
|
@ -233,6 +235,10 @@ func intPtr(v int) *int {
|
|||
return &v
|
||||
}
|
||||
|
||||
func boolPtr(v bool) *bool {
|
||||
return &v
|
||||
}
|
||||
|
||||
func chatRunMetadata(runMeta map[string]string, req chatCompletionRequest, outputPolicy strictOutputPolicy) map[string]string {
|
||||
if runMeta == nil {
|
||||
runMeta = make(map[string]string)
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
|
|||
var defaultThinkingTokenBudget int
|
||||
if catalogEntry := s.findProviderPoolEntry(req.Model); catalogEntry != nil {
|
||||
applyModelCatalogGenerationPolicyToResponses(&req, *catalogEntry)
|
||||
if !outputPolicy.Strict && catalogEntry.DefaultThinkingTokenBudget > 0 {
|
||||
if catalogEntry.DefaultThinkingTokenBudget > 0 {
|
||||
defaultThinkingTokenBudget = catalogEntry.DefaultThinkingTokenBudget
|
||||
}
|
||||
}
|
||||
|
|
@ -74,10 +74,11 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
|
|||
runMeta["openai_stream"] = fmt.Sprintf("%t", req.Stream)
|
||||
runMeta["strict_output"] = fmt.Sprintf("%t", outputPolicy.Strict)
|
||||
input := map[string]any{"prompt": prompt}
|
||||
if outputPolicy.Strict {
|
||||
input["think"] = false
|
||||
} else if defaultThinkingTokenBudget > 0 {
|
||||
if defaultThinkingTokenBudget > 0 {
|
||||
input["think"] = true
|
||||
input["thinking_token_budget"] = defaultThinkingTokenBudget
|
||||
} else if outputPolicy.Strict {
|
||||
input["think"] = false
|
||||
}
|
||||
if options := req.providerOptions(); len(options) > 0 {
|
||||
input["options"] = options
|
||||
|
|
|
|||
|
|
@ -3730,6 +3730,37 @@ func TestChatCompletionsProviderPoolAppliesGenerationPolicy(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestChatCompletionsProviderPoolThinkingPolicyOverridesStrictOutputDisable(t *testing.T) {
|
||||
fake := &fakeRunService{events: bufferedRunEvents(
|
||||
&iop.RunEvent{Type: "delta", Delta: "ok"},
|
||||
&iop.RunEvent{Type: "complete"},
|
||||
)}
|
||||
catalog := []config.ModelCatalogEntry{{
|
||||
ID: "ornith:35b",
|
||||
DefaultThinkingTokenBudget: 8192,
|
||||
Providers: map[string]string{"prov-vllm": "Ornith-1.0-35B"},
|
||||
}}
|
||||
srv := NewServer(config.EdgeOpenAIConf{StrictOutput: true}, fake, nil)
|
||||
srv.SetModelCatalog(catalog)
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{
|
||||
"model":"ornith:35b",
|
||||
"messages":[{"role":"user","content":"hello"}]
|
||||
}`))
|
||||
w := httptest.NewRecorder()
|
||||
srv.handleChatCompletions(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
|
||||
}
|
||||
if fake.req.Input["think"] != true {
|
||||
t.Fatalf("provider-pool thinking policy should keep thinking enabled under strict output: %+v", fake.req.Input)
|
||||
}
|
||||
if fake.req.Input["thinking_token_budget"].(int) != 8192 {
|
||||
t.Fatalf("thinking_token_budget policy not applied: %+v", fake.req.Input)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatCompletionsProviderPoolPreservesLargerGenerationPolicyValues(t *testing.T) {
|
||||
fake := &fakeRunService{events: bufferedRunEvents(
|
||||
&iop.RunEvent{Type: "delta", Delta: "ok"},
|
||||
|
|
@ -3876,6 +3907,37 @@ func TestResponsesProviderPoolAppliesGenerationPolicy(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestResponsesProviderPoolThinkingPolicyOverridesStrictOutputDisable(t *testing.T) {
|
||||
fake := &fakeRunService{events: bufferedRunEvents(
|
||||
&iop.RunEvent{Type: "delta", Delta: "ok"},
|
||||
&iop.RunEvent{Type: "complete", Usage: &iop.Usage{InputTokens: 1, OutputTokens: 1}},
|
||||
)}
|
||||
catalog := []config.ModelCatalogEntry{{
|
||||
ID: "ornith:35b",
|
||||
DefaultThinkingTokenBudget: 8192,
|
||||
Providers: map[string]string{"prov-vllm": "Ornith-1.0-35B"},
|
||||
}}
|
||||
srv := NewServer(config.EdgeOpenAIConf{TimeoutSec: 5, StrictOutput: true}, fake, nil)
|
||||
srv.SetModelCatalog(catalog)
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{
|
||||
"model":"ornith:35b",
|
||||
"input":"hello"
|
||||
}`))
|
||||
w := httptest.NewRecorder()
|
||||
srv.handleResponses(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
|
||||
}
|
||||
if fake.req.Input["think"] != true {
|
||||
t.Fatalf("provider-pool thinking policy should keep thinking enabled under strict output: %+v", fake.req.Input)
|
||||
}
|
||||
if fake.req.Input["thinking_token_budget"].(int) != 8192 {
|
||||
t.Fatalf("thinking_token_budget policy not applied: %+v", fake.req.Input)
|
||||
}
|
||||
}
|
||||
|
||||
// TestChatCompletionsProviderPoolFallsBackToLegacyRoute verifies that when the
|
||||
// request model does not match the catalog, the legacy model_routes path is used.
|
||||
func TestChatCompletionsProviderPoolFallsBackToLegacyRoute(t *testing.T) {
|
||||
|
|
|
|||
|
|
@ -56,8 +56,9 @@ func (c *candidateNode) slotKey() string {
|
|||
}
|
||||
|
||||
type groupPolicy struct {
|
||||
maxQueue int
|
||||
queueTimeout time.Duration
|
||||
maxQueue int
|
||||
queueTimeout time.Duration
|
||||
queueTimeoutSet bool
|
||||
}
|
||||
|
||||
type inflightRec struct {
|
||||
|
|
@ -150,19 +151,30 @@ func isTerminalRunEvent(e *iop.RunEvent) bool {
|
|||
return t == "complete" || t == "error" || t == "cancelled"
|
||||
}
|
||||
|
||||
// normalizeGroupPolicy applies queue defaults while preserving an explicit
|
||||
// zero queue timeout as "no timeout".
|
||||
func normalizeGroupPolicy(policy groupPolicy) groupPolicy {
|
||||
if policy.maxQueue <= 0 {
|
||||
policy.maxQueue = defaultGroupMaxQueue
|
||||
}
|
||||
if policy.queueTimeout > 0 {
|
||||
policy.queueTimeoutSet = true
|
||||
}
|
||||
if !policy.queueTimeoutSet {
|
||||
policy.queueTimeout = defaultQueueTimeout
|
||||
policy.queueTimeoutSet = true
|
||||
}
|
||||
return policy
|
||||
}
|
||||
|
||||
// getOrCreateGroupLocked returns an existing group or creates one with the
|
||||
// given policy (applying defaults for zero fields). Policy is only applied
|
||||
// given policy (applying defaults for unset fields). Policy is only applied
|
||||
// at creation; existing groups retain their current policy until an explicit
|
||||
// updateGroupPolicyLocked call (used for config-refresh policy propagation).
|
||||
func (m *modelQueueManager) getOrCreateGroupLocked(key string, policy groupPolicy) *modelQueueGroup {
|
||||
g, ok := m.groups[key]
|
||||
if !ok {
|
||||
if policy.maxQueue <= 0 {
|
||||
policy.maxQueue = defaultGroupMaxQueue
|
||||
}
|
||||
if policy.queueTimeout <= 0 {
|
||||
policy.queueTimeout = defaultQueueTimeout
|
||||
}
|
||||
policy = normalizeGroupPolicy(policy)
|
||||
g = &modelQueueGroup{
|
||||
key: key,
|
||||
policy: policy,
|
||||
|
|
@ -176,7 +188,7 @@ func (m *modelQueueManager) getOrCreateGroupLocked(key string, policy groupPolic
|
|||
|
||||
// updateGroupPolicyLocked refreshes an existing group's queue policy so that
|
||||
// provider max_queue / queue_timeout_ms changes applied via config refresh take
|
||||
// effect on subsequent admissions. Only non-zero policy fields override the
|
||||
// effect on subsequent admissions. Only configured policy fields override the
|
||||
// current values; in-flight counts and queued items are preserved untouched.
|
||||
func (m *modelQueueManager) updateGroupPolicyLocked(g *modelQueueGroup, policy groupPolicy) {
|
||||
if g == nil {
|
||||
|
|
@ -187,6 +199,10 @@ func (m *modelQueueManager) updateGroupPolicyLocked(g *modelQueueGroup, policy g
|
|||
}
|
||||
if policy.queueTimeout > 0 {
|
||||
g.policy.queueTimeout = policy.queueTimeout
|
||||
g.policy.queueTimeoutSet = true
|
||||
} else if policy.queueTimeoutSet {
|
||||
g.policy.queueTimeout = policy.queueTimeout
|
||||
g.policy.queueTimeoutSet = true
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -328,10 +344,14 @@ func (m *modelQueueManager) admitWithReason(ctx context.Context, groupKey, adapt
|
|||
}
|
||||
}
|
||||
|
||||
var deadline time.Time
|
||||
if group.policy.queueTimeout > 0 {
|
||||
deadline = time.Now().Add(group.policy.queueTimeout)
|
||||
}
|
||||
item := &queueItem{
|
||||
candidates: candidates,
|
||||
waitCh: make(chan admitResult, 1),
|
||||
deadline: time.Now().Add(group.policy.queueTimeout),
|
||||
deadline: deadline,
|
||||
long: long,
|
||||
reason: reason,
|
||||
}
|
||||
|
|
@ -339,6 +359,35 @@ func (m *modelQueueManager) admitWithReason(ctx context.Context, groupKey, adapt
|
|||
timeout := group.policy.queueTimeout
|
||||
m.mu.Unlock()
|
||||
|
||||
if timeout <= 0 {
|
||||
select {
|
||||
case res := <-item.waitCh:
|
||||
return res.candidate, item.reason, res.err
|
||||
case <-ctx.Done():
|
||||
m.mu.Lock()
|
||||
m.removeItemLocked(groupKey, item)
|
||||
m.mu.Unlock()
|
||||
// Handle race: dispatch may have sent to waitCh just before cancellation.
|
||||
select {
|
||||
case res := <-item.waitCh:
|
||||
if res.candidate != nil {
|
||||
slot := res.candidate.slotKey()
|
||||
var nodeID, providerID string
|
||||
if colonIdx := findLastColon(slot); colonIdx > 0 {
|
||||
nodeID = slot[:colonIdx]
|
||||
providerID = slot[colonIdx+1:]
|
||||
} else {
|
||||
nodeID = slot
|
||||
}
|
||||
longSlot := item.long && res.candidate.longContextCapacity > 0
|
||||
m.releaseSlotWithLong(groupKey, nodeID, providerID, longSlot)
|
||||
}
|
||||
default:
|
||||
}
|
||||
return nil, "", ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
timer := time.NewTimer(timeout)
|
||||
defer timer.Stop()
|
||||
|
||||
|
|
@ -547,7 +596,7 @@ func (m *modelQueueManager) tryDispatchLocked(group *modelQueueGroup) {
|
|||
for i := 0; i < len(group.queue); i++ {
|
||||
item := group.queue[i]
|
||||
|
||||
if time.Now().After(item.deadline) {
|
||||
if !item.deadline.IsZero() && time.Now().After(item.deadline) {
|
||||
group.queue = append(group.queue[:i], group.queue[i+1:]...)
|
||||
select {
|
||||
case item.waitCh <- admitResult{err: fmt.Errorf("model group %q: %w", group.key, errQueueTimeout)}:
|
||||
|
|
|
|||
|
|
@ -495,6 +495,67 @@ func TestModelQueueUsesProviderQueuePolicy(t *testing.T) {
|
|||
t.Fatalf("timed out too fast: %v", elapsed)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("zeroQueueTimeoutWaitsForContextCancellation", func(t *testing.T) {
|
||||
entry := &edgenode.NodeEntry{NodeID: "node-pq3"}
|
||||
cands := []candidateNode{{entry: entry, capacity: 1}}
|
||||
policy := groupPolicy{maxQueue: 16, queueTimeoutSet: true}
|
||||
|
||||
m := newModelQueueManager(nil)
|
||||
|
||||
m.mu.Lock()
|
||||
g := m.getOrCreateGroupLocked("g-pq-no-timeout", policy)
|
||||
g.inflight["node-pq3"] = 1
|
||||
m.mu.Unlock()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
start := time.Now()
|
||||
_, err := m.admit(ctx, "g-pq-no-timeout", "", "", cands, policy)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected context deadline, got: %v", err)
|
||||
}
|
||||
if errors.Is(err, errQueueTimeout) {
|
||||
t.Fatalf("queue timeout should be disabled, got: %v", err)
|
||||
}
|
||||
if elapsed < 25*time.Millisecond {
|
||||
t.Fatalf("returned before context deadline: %v", elapsed)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestGroupPolicyFromStoreZeroQueueTimeoutDisablesTimeout(t *testing.T) {
|
||||
store := edgenode.NewNodeStore()
|
||||
store.Add(&edgenode.NodeRecord{
|
||||
ID: "node-policy-zero-timeout",
|
||||
Adapters: config.AdaptersConf{
|
||||
OpenAICompatInstances: []config.OpenAICompatInstanceConf{
|
||||
{
|
||||
Name: "provider-zero-timeout",
|
||||
Enabled: true,
|
||||
Capacity: 1,
|
||||
MaxQueue: 16,
|
||||
QueueTimeoutMS: 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
entries := []*edgenode.NodeEntry{{NodeID: "node-policy-zero-timeout"}}
|
||||
|
||||
policy := groupPolicyFromStore(store, entries, "provider-zero-timeout", "")
|
||||
|
||||
if policy.maxQueue != 16 {
|
||||
t.Fatalf("maxQueue: got %d, want 16", policy.maxQueue)
|
||||
}
|
||||
if !policy.queueTimeoutSet {
|
||||
t.Fatalf("queueTimeoutSet: got false, want true")
|
||||
}
|
||||
if policy.queueTimeout != 0 {
|
||||
t.Fatalf("queueTimeout: got %v, want no timeout", policy.queueTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
// TestModelQueueProviderInflightSelectionBeatsPriority verifies that in_flight
|
||||
|
|
|
|||
|
|
@ -443,13 +443,10 @@ func groupPolicyFromRequestOrStore(req SubmitRunRequest, store *edgenode.NodeSto
|
|||
if maxQueue <= 0 {
|
||||
maxQueue = defaultGroupMaxQueue
|
||||
}
|
||||
queueTimeout := defaultQueueTimeout
|
||||
if req.QueueTimeoutMS > 0 {
|
||||
queueTimeout = time.Duration(req.QueueTimeoutMS) * time.Millisecond
|
||||
}
|
||||
return groupPolicy{
|
||||
maxQueue: maxQueue,
|
||||
queueTimeout: queueTimeout,
|
||||
maxQueue: maxQueue,
|
||||
queueTimeout: time.Duration(req.QueueTimeoutMS) * time.Millisecond,
|
||||
queueTimeoutSet: true,
|
||||
}
|
||||
}
|
||||
return groupPolicyFromStore(store, entries, req.Adapter, req.Target)
|
||||
|
|
@ -469,17 +466,15 @@ func groupPolicyFromStore(store *edgenode.NodeStore, entries []*edgenode.NodeEnt
|
|||
}
|
||||
if res.maxQueue > 0 || res.queueTimeoutMS > 0 {
|
||||
p := groupPolicy{
|
||||
maxQueue: positiveOr(res.maxQueue, defaultGroupMaxQueue),
|
||||
queueTimeout: time.Duration(positiveOr(res.queueTimeoutMS, 0)) * time.Millisecond,
|
||||
}
|
||||
if p.queueTimeout <= 0 {
|
||||
p.queueTimeout = defaultQueueTimeout
|
||||
maxQueue: positiveOr(res.maxQueue, defaultGroupMaxQueue),
|
||||
queueTimeout: time.Duration(res.queueTimeoutMS) * time.Millisecond,
|
||||
queueTimeoutSet: true,
|
||||
}
|
||||
return p
|
||||
}
|
||||
}
|
||||
}
|
||||
return groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout}
|
||||
return groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout, queueTimeoutSet: true}
|
||||
}
|
||||
|
||||
// providerCanServe checks whether the provider advertises the served model in
|
||||
|
|
@ -640,8 +635,9 @@ func (s *Service) resolveProviderPoolCandidates(req SubmitRunRequest, store *edg
|
|||
})
|
||||
if !policySet && (prov.MaxQueue > 0 || prov.QueueTimeoutMS > 0) {
|
||||
policy = groupPolicy{
|
||||
maxQueue: positiveOr(prov.MaxQueue, defaultGroupMaxQueue),
|
||||
queueTimeout: time.Duration(positiveOr(prov.QueueTimeoutMS, int(defaultQueueTimeout/time.Millisecond))) * time.Millisecond,
|
||||
maxQueue: positiveOr(prov.MaxQueue, defaultGroupMaxQueue),
|
||||
queueTimeout: time.Duration(prov.QueueTimeoutMS) * time.Millisecond,
|
||||
queueTimeoutSet: true,
|
||||
}
|
||||
policySet = true
|
||||
}
|
||||
|
|
@ -652,7 +648,7 @@ func (s *Service) resolveProviderPoolCandidates(req SubmitRunRequest, store *edg
|
|||
return nil, groupPolicy{}, fmt.Errorf("no connected nodes support provider pool model %q", req.ModelGroupKey)
|
||||
}
|
||||
if !policySet {
|
||||
policy = groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout}
|
||||
policy = groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout, queueTimeoutSet: true}
|
||||
}
|
||||
return candidates, policy, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -163,7 +163,8 @@ type NodeProviderConf struct {
|
|||
LongContextCapacity int `mapstructure:"long_context_capacity" yaml:"long_context_capacity,omitempty"`
|
||||
// MaxQueue is the maximum queue depth.
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue,omitempty"`
|
||||
// QueueTimeoutMS is the queue timeout in milliseconds.
|
||||
// QueueTimeoutMS is the queue timeout in milliseconds. Zero means no queue
|
||||
// timeout when a queue policy is configured.
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms,omitempty"`
|
||||
// Priority is the provider dispatch priority. Lower values are preferred
|
||||
// when in_flight counts are equal. Must be non-negative; zero is the
|
||||
|
|
@ -481,7 +482,7 @@ type OllamaInstanceConf struct {
|
|||
ContextSize int `mapstructure:"context_size" yaml:"context_size"`
|
||||
Capacity int `mapstructure:"capacity" yaml:"capacity"`
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout
|
||||
RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"`
|
||||
}
|
||||
|
||||
|
|
@ -492,7 +493,7 @@ type VllmInstanceConf struct {
|
|||
Endpoint string `mapstructure:"endpoint" yaml:"endpoint"`
|
||||
Capacity int `mapstructure:"capacity" yaml:"capacity"`
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout
|
||||
RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"`
|
||||
}
|
||||
|
||||
|
|
@ -503,7 +504,7 @@ type OpenAICompatConf struct {
|
|||
Headers map[string]string `mapstructure:"headers" yaml:"headers"`
|
||||
Capacity int `mapstructure:"capacity" yaml:"capacity"`
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout
|
||||
RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"`
|
||||
}
|
||||
|
||||
|
|
@ -515,7 +516,7 @@ type OpenAICompatInstanceConf struct {
|
|||
Headers map[string]string `mapstructure:"headers" yaml:"headers"`
|
||||
Capacity int `mapstructure:"capacity" yaml:"capacity"`
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout
|
||||
RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"`
|
||||
}
|
||||
|
||||
|
|
@ -525,7 +526,7 @@ type OllamaConf struct {
|
|||
ContextSize int `mapstructure:"context_size" yaml:"context_size"`
|
||||
Capacity int `mapstructure:"capacity" yaml:"capacity"`
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout
|
||||
RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"`
|
||||
}
|
||||
|
||||
|
|
@ -534,7 +535,7 @@ type VllmConf struct {
|
|||
Endpoint string `mapstructure:"endpoint" yaml:"endpoint"`
|
||||
Capacity int `mapstructure:"capacity" yaml:"capacity"`
|
||||
MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"`
|
||||
QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout
|
||||
RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"`
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue