feat(node): refactor node queue simplification and move completed subtask to archive
- Refactor node queue management in node.go and run_manager.go - Update concurrency integration tests - Move 03+01_node_queue_simplify subtask to archive - Add complete.log for archived subtask
This commit is contained in:
parent
bcc60ff48c
commit
34e87f70e9
7 changed files with 195 additions and 950 deletions
|
|
@ -33,44 +33,44 @@ task=m-edge-model-group-queue-scheduling/03+01_node_queue_simplify, plan=0, tag=
|
|||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [REFACTOR-1] OnRunRequest Queue Ownership 제거 | [ ] |
|
||||
| [REFACTOR-2] Capability Snapshot 정리 | [ ] |
|
||||
| [REFACTOR-3] Admission Queue Test 전환 | [ ] |
|
||||
| [REFACTOR-1] OnRunRequest Queue Ownership 제거 | [x] |
|
||||
| [REFACTOR-2] Capability Snapshot 정리 | [x] |
|
||||
| [REFACTOR-3] Admission Queue Test 전환 | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] Node `OnRunRequest`에서 FIFO queue ownership을 제거하고 dispatch된 run을 즉시 실행 또는 safety reject로 처리한다.
|
||||
- [ ] Node capability snapshot의 queued/in-flight 의미를 Edge-owned queue와 충돌하지 않게 정리한다.
|
||||
- [ ] Node admission queue tests를 삭제/전환하고 over-dispatch safety, cancel, background, store lifecycle 테스트를 남긴다.
|
||||
- [ ] `go test -count=1 ./apps/node/internal/node`가 통과한다.
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
- [x] Node `OnRunRequest`에서 FIFO queue ownership을 제거하고 dispatch된 run을 즉시 실행 또는 safety reject로 처리한다.
|
||||
- [x] Node capability snapshot의 queued/in-flight 의미를 Edge-owned queue와 충돌하지 않게 정리한다.
|
||||
- [x] Node admission queue tests를 삭제/전환하고 over-dispatch safety, cancel, background, store lifecycle 테스트를 남긴다.
|
||||
- [x] `go test -count=1 ./apps/node/internal/node`가 통과한다.
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
|
||||
|
||||
- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [ ] 판정과 Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다.
|
||||
- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [ ] `.gitignore`의 Agent-Ops 관리 block을 확인한다.
|
||||
- [ ] PASS이면 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
|
||||
- [ ] PASS이면 active task 디렉터리를 archive로 이동한다.
|
||||
- [ ] PASS split 작업이면 parent directory 유지/정리를 확인한다.
|
||||
- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [x] 판정과 Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [x] `.gitignore`의 Agent-Ops 관리 block을 확인한다.
|
||||
- [x] PASS이면 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
|
||||
- [x] PASS이면 active task 디렉터리를 archive로 이동한다.
|
||||
- [x] PASS split 작업이면 parent directory 유지/정리를 확인한다.
|
||||
- [ ] WARN/FAIL이면 다음 active plan/review 또는 USER_REVIEW 경로를 처리한다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
|
||||
- 없음
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
|
||||
- Node-local FIFO admission queue 및 대기(wait) 로직을 완전히 걷어내어 복잡한 goroutine/cancellation 복원 지점을 단순화했습니다.
|
||||
- 이제 `OnRunRequest` 시점에 `admissionManager.acquire()`를 통해 즉시 slots을 획득하며, 획득 실패 시 곧바로 "concurrency unavailable" 에러를 반환하고 store에는 즉시 "rejected" 상태를 기록합니다.
|
||||
- Node가 queue owner가 아니므로 capability snapshot 시 `queued` 카운트는 항상 `0`으로 보고하여 Edge-owned scheduler의 상태 스냅샷과 충돌하지 않도록 처리했습니다.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 결정 필요: 없음
|
||||
|
|
@ -81,36 +81,50 @@ _기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외
|
|||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- Node가 FIFO queue owner로 남지 않았는지 확인한다.
|
||||
- Node queued snapshot이 Edge-owned queued 의미와 충돌하지 않는지 확인한다.
|
||||
- cancel/background/store lifecycle 회귀가 남아 있는지 확인한다.
|
||||
- Node의 OnRunRequest에서 큐 관련 대기 로직(`ticket.wait`)이 완전히 제거되었는지 확인합니다.
|
||||
- `apps/node/internal/node/run_manager.go`와 `node.go`에서 `newFifoGate` 인자와 queuedCount의 반환 값(항상 0)을 확인합니다.
|
||||
- `node_test.go`에서 queue 관련 불필요해진 테스트 케이스가 제거되었고, 즉시 리젝트 관련 테스트 케이스(`TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue`)가 올바르게 작동하는지 확인합니다.
|
||||
|
||||
## 검증 결과
|
||||
|
||||
### REFACTOR-1 중간 검증
|
||||
```text
|
||||
$ go test -count=1 ./apps/node/internal/node -run 'TestOnRunRequest'
|
||||
(output)
|
||||
ok iop/apps/node/internal/node 0.028s
|
||||
```
|
||||
|
||||
### REFACTOR-2 중간 검증
|
||||
```text
|
||||
$ go test -count=1 ./apps/node/internal/node -run 'TestOnCommandRequest_Capabilities'
|
||||
(output)
|
||||
ok iop/apps/node/internal/node 0.011s
|
||||
```
|
||||
|
||||
### REFACTOR-3 중간 검증
|
||||
```text
|
||||
$ go test -count=1 ./apps/node/internal/node -run 'Test(AdmissionQueue|ConcurrencyLimit|QueueOverflow|OnRunRequest|OnCommandRequest_Capabilities)'
|
||||
(output)
|
||||
ok iop/apps/node/internal/node 0.017s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
```text
|
||||
$ go test -count=1 ./apps/node/internal/node
|
||||
(output)
|
||||
ok iop/apps/node/internal/node 0.128s
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section?**
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: PASS
|
||||
- 차원별 평가:
|
||||
- correctness: Pass
|
||||
- completeness: Pass
|
||||
- test coverage: Pass
|
||||
- API contract: Pass
|
||||
- code quality: Pass
|
||||
- plan deviation: Pass
|
||||
- verification trust: Pass
|
||||
- 발견된 문제: 없음
|
||||
- 다음 단계: PASS이므로 `complete.log`를 작성하고 task directory를 archive로 이동한다.
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
# Complete - m-edge-model-group-queue-scheduling/03+01_node_queue_simplify
|
||||
|
||||
## 완료 일시
|
||||
|
||||
2026-06-16
|
||||
|
||||
## 요약
|
||||
|
||||
Node-local provider FIFO admission queue를 제거하고 immediate concurrency safety reject와 queued=0 capability snapshot으로 단순화한 작업을 1회 리뷰로 PASS 처리했다.
|
||||
|
||||
## 루프 이력
|
||||
|
||||
| Plan | Review | Verdict | 메모 |
|
||||
|------|--------|---------|------|
|
||||
| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | PASS | Required/Suggested 이슈 없음; 리뷰 중 stale comment와 review artifact drift를 비동작성 정리 |
|
||||
|
||||
## 구현/정리 내용
|
||||
|
||||
- `OnRunRequest`의 Node-local FIFO enqueue/wait/promote 경로를 제거하고, adapter/global concurrency safety guard를 즉시 획득하지 못하면 rejected store status와 error RunEvent를 남기도록 정리했다.
|
||||
- capability response와 provider snapshot의 `queued` 값을 Node-local queue가 없음을 나타내는 `0`으로 고정하고, in-flight는 adapter safety guard 기준으로 유지했다.
|
||||
- Node admission queue 전제 테스트를 over-dispatch safety rejection, permit release, rejected store/event, capability snapshot 기준으로 전환했다.
|
||||
- 리뷰 중 stale FIFO comment와 미사용 queue helper test code를 비동작성으로 정리했다.
|
||||
|
||||
## 최종 검증
|
||||
|
||||
- `go test -count=1 ./apps/node/internal/node -run 'TestOnRunRequest'` - PASS; `ok iop/apps/node/internal/node 0.022s`
|
||||
- `go test -count=1 ./apps/node/internal/node -run 'TestOnCommandRequest_Capabilities'` - PASS; `ok iop/apps/node/internal/node 0.014s`
|
||||
- `go test -count=1 ./apps/node/internal/node -run 'Test(AdmissionQueue|ConcurrencyLimit|QueueOverflow|OnRunRequest|OnCommandRequest_Capabilities)'` - PASS; `ok iop/apps/node/internal/node 0.022s`
|
||||
- `go test -count=1 ./apps/node/internal/node` - PASS; `ok iop/apps/node/internal/node 0.138s`
|
||||
- `go test -count=1 ./apps/node/...` - PASS; node package tree passed, including `apps/node/internal/transport` and adapter packages.
|
||||
|
||||
## Roadmap Completion
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/edge-model-group-queue-scheduling.md`
|
||||
- Completed task ids:
|
||||
- `node-simplify`: PASS; evidence=`agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/code_review_cloud_G07_0.log`; verification=`go test -count=1 ./apps/node/internal/node`, `go test -count=1 ./apps/node/...`
|
||||
- Not completed task ids: 없음
|
||||
|
||||
## 잔여 Nit
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -32,9 +32,9 @@ type Node struct {
|
|||
router runtime.Router
|
||||
store *store.Store
|
||||
runs *runManager
|
||||
globalGate *fifoGate // node-wide FIFO admission across all adapters
|
||||
globalGate *fifoGate // node-wide concurrency safety guard across all adapters
|
||||
adapterGatesMu sync.Mutex
|
||||
adapterGates map[string]*fifoGate // per adapter-key FIFO admission
|
||||
adapterGates map[string]*fifoGate // per adapter-key concurrency safety guard
|
||||
out io.Writer
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
|
@ -60,7 +60,7 @@ func New(
|
|||
router: router,
|
||||
store: st,
|
||||
runs: newRunManager(),
|
||||
globalGate: newFifoGate(globalConcurrency, -1, 0),
|
||||
globalGate: newFifoGate(globalConcurrency),
|
||||
adapterGates: make(map[string]*fifoGate),
|
||||
out: out,
|
||||
logger: logger,
|
||||
|
|
@ -95,37 +95,15 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
return fmt.Errorf("node: resolve: %w", err)
|
||||
}
|
||||
|
||||
// Admission gate (foreground and background use the same path).
|
||||
// Policy: provider-scoped FIFO queue. A request runs immediately when
|
||||
// in_flight < capacity; otherwise it waits in the adapter's FIFO queue and
|
||||
// is promoted to an execution slot as earlier runs release. Requests are
|
||||
// rejected when the queue is full (max_queue) or the queue timeout elapses.
|
||||
//
|
||||
// Two gates compose admission: the per-adapter gate (capacity, max_queue,
|
||||
// queue_timeout from Capabilities) and the node-wide global gate.
|
||||
var caps runtime.Capabilities
|
||||
if c, capsErr := adapter.Capabilities(ctx); capsErr == nil {
|
||||
caps = c
|
||||
}
|
||||
admission := n.admissionFor(spec.Adapter, caps)
|
||||
|
||||
// Record the request as queued before waiting so its lifecycle is
|
||||
// observable even while it sits in the FIFO queue.
|
||||
if err := n.store.InsertRun(ctx, store.RunRecord{
|
||||
RunID: spec.RunID,
|
||||
Adapter: spec.Adapter,
|
||||
Target: spec.Target,
|
||||
SessionID: normalizeSessionID(spec.SessionID),
|
||||
Background: spec.Background,
|
||||
Status: "queued",
|
||||
CreatedAt: time.Now(),
|
||||
}); err != nil {
|
||||
n.logger.Warn("store: insert run", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
}
|
||||
|
||||
// Reserve a FIFO position without blocking. Only queue overflow fails here;
|
||||
// waiting for promotion happens later in wait().
|
||||
ticket, err := admission.enqueue()
|
||||
// Acquire safety capacity ticket. Since we no longer maintain a Node-local FIFO queue,
|
||||
// if concurrency is full, we reject immediately.
|
||||
ticket, err := admission.acquire()
|
||||
if err != nil {
|
||||
n.logger.Warn("run admission rejected",
|
||||
zap.String("run_id", spec.RunID),
|
||||
|
|
@ -136,8 +114,19 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
return fmt.Errorf("node: run %s: %w", spec.RunID, err)
|
||||
}
|
||||
|
||||
// The run context governs both the queue wait and execution, so cancelling a
|
||||
// queued run via OnCancel also removes it from the FIFO queue.
|
||||
// Record the request as running since it is admitted immediately without queueing.
|
||||
if err := n.store.InsertRun(ctx, store.RunRecord{
|
||||
RunID: spec.RunID,
|
||||
Adapter: spec.Adapter,
|
||||
Target: spec.Target,
|
||||
SessionID: normalizeSessionID(spec.SessionID),
|
||||
Background: spec.Background,
|
||||
Status: "running",
|
||||
CreatedAt: time.Now(),
|
||||
}); err != nil {
|
||||
n.logger.Warn("store: insert run", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
}
|
||||
|
||||
execCtx, cancel := context.WithCancel(ctx)
|
||||
if spec.TimeoutSec > 0 {
|
||||
execCtx, cancel = context.WithTimeout(ctx, time.Duration(spec.TimeoutSec)*time.Second)
|
||||
|
|
@ -161,19 +150,6 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
background: spec.Background,
|
||||
}
|
||||
|
||||
// admit blocks until the run holds an execution slot, then promotes the
|
||||
// stored status to running. It returns an admissionError on queue timeout or
|
||||
// cancellation while queued.
|
||||
admit := func() error {
|
||||
if err := ticket.wait(execCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := n.store.MarkRunning(context.Background(), spec.RunID); err != nil {
|
||||
n.logger.Warn("store: mark running", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
run := func() error {
|
||||
defer ticket.release()
|
||||
defer cancel()
|
||||
|
|
@ -185,38 +161,15 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
return execErr
|
||||
}
|
||||
|
||||
// launch waits for an execution slot then runs. On admission failure it
|
||||
// records the rejection. It is invoked synchronously for foreground runs and
|
||||
// in a goroutine for queued background runs.
|
||||
launch := func() error {
|
||||
if err := admit(); err != nil {
|
||||
n.logger.Warn("run admission rejected",
|
||||
zap.String("run_id", spec.RunID),
|
||||
zap.String("adapter", spec.Adapter),
|
||||
zap.String("reason", string(admissionRejectReason(err))),
|
||||
)
|
||||
cancel()
|
||||
n.runs.deregister(spec.RunID)
|
||||
close(h.done)
|
||||
n.rejectRun(context.Background(), sess, spec, err)
|
||||
return fmt.Errorf("node: run %s: %w", spec.RunID, err)
|
||||
}
|
||||
return run()
|
||||
}
|
||||
|
||||
// Background runs return to the caller after enqueue; promotion and
|
||||
// execution proceed in a goroutine. Foreground runs take the synchronous path.
|
||||
if spec.Background {
|
||||
go func() { _ = launch() }()
|
||||
go func() { _ = run() }()
|
||||
return nil
|
||||
}
|
||||
return launch()
|
||||
return run()
|
||||
}
|
||||
|
||||
// admissionFor returns an admissionManager pairing the node-wide global gate
|
||||
// with the per-adapter FIFO gate for the given adapter instance key. The
|
||||
// adapter gate is created on first use from the adapter's Capabilities
|
||||
// (MaxConcurrency, MaxQueue, QueueTimeoutMS) and shared across runs.
|
||||
// with the per-adapter gate for the given adapter instance key.
|
||||
func (n *Node) admissionFor(adapterKey string, caps runtime.Capabilities) *admissionManager {
|
||||
return &admissionManager{
|
||||
global: n.globalGate,
|
||||
|
|
@ -224,19 +177,14 @@ func (n *Node) admissionFor(adapterKey string, caps runtime.Capabilities) *admis
|
|||
}
|
||||
}
|
||||
|
||||
// adapterGateFor returns the shared FIFO gate for the given adapter instance
|
||||
// key, creating one from caps if not yet seen. MaxConcurrency <= 0 means the
|
||||
// adapter gate is unlimited.
|
||||
// adapterGateFor returns the shared gate for the given adapter instance key,
|
||||
// creating one from caps if not yet seen. MaxConcurrency <= 0 means the adapter gate is unlimited.
|
||||
func (n *Node) adapterGateFor(adapterKey string, caps runtime.Capabilities) *fifoGate {
|
||||
n.adapterGatesMu.Lock()
|
||||
defer n.adapterGatesMu.Unlock()
|
||||
g, ok := n.adapterGates[adapterKey]
|
||||
if !ok {
|
||||
g = newFifoGate(
|
||||
caps.MaxConcurrency,
|
||||
caps.MaxQueue,
|
||||
time.Duration(caps.QueueTimeoutMS)*time.Millisecond,
|
||||
)
|
||||
g = newFifoGate(caps.MaxConcurrency)
|
||||
n.adapterGates[adapterKey] = g
|
||||
}
|
||||
return g
|
||||
|
|
@ -244,27 +192,24 @@ func (n *Node) adapterGateFor(adapterKey string, caps runtime.Capabilities) *fif
|
|||
|
||||
// rejectRun records a rejected run in the store and sends an error RunEvent to
|
||||
// the session so Edge can observe the rejection through the normal event stream.
|
||||
// The rejection message carries the concrete reason (queue_full, queue_timeout,
|
||||
// concurrency_unavailable) extracted from the admission error.
|
||||
// The rejection message carries the concrete reason (concurrency_unavailable) extracted from the admission error.
|
||||
func (n *Node) rejectRun(ctx context.Context, sess *transport.Session, spec runtime.ExecutionSpec, admitErr error) {
|
||||
errMsg := admissionMessage(admitErr)
|
||||
|
||||
// The run was inserted as "queued" before admission; mark it rejected.
|
||||
// If the insert never happened (e.g. it was skipped on error), fall back to
|
||||
// inserting a terminal rejected record so the rejection is still observable.
|
||||
// Since we reject immediately without queueing, insert as a terminal rejected record.
|
||||
if err := n.store.InsertRun(ctx, store.RunRecord{
|
||||
RunID: spec.RunID,
|
||||
Adapter: spec.Adapter,
|
||||
Target: spec.Target,
|
||||
SessionID: normalizeSessionID(spec.SessionID),
|
||||
Background: spec.Background,
|
||||
Status: "rejected",
|
||||
CreatedAt: time.Now(),
|
||||
}); err != nil {
|
||||
n.logger.Warn("store: insert rejected run", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
}
|
||||
if err := n.store.CompleteRun(ctx, spec.RunID, "rejected", errMsg); err != nil {
|
||||
n.logger.Warn("store: complete rejected run", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
if insErr := n.store.InsertRun(ctx, store.RunRecord{
|
||||
RunID: spec.RunID,
|
||||
Adapter: spec.Adapter,
|
||||
Target: spec.Target,
|
||||
SessionID: normalizeSessionID(spec.SessionID),
|
||||
Background: spec.Background,
|
||||
Status: "rejected",
|
||||
CreatedAt: time.Now(),
|
||||
}); insErr != nil {
|
||||
n.logger.Warn("store: insert rejected run", zap.String("run_id", spec.RunID), zap.Error(insErr))
|
||||
}
|
||||
}
|
||||
|
||||
if sess != nil && sess.IsAlive() {
|
||||
|
|
@ -389,7 +334,6 @@ func (n *Node) handleCapabilitiesCommand(ctx context.Context, req *iop.NodeComma
|
|||
queued := 0
|
||||
if ok {
|
||||
inFlight = gate.activeCount()
|
||||
queued = gate.queuedCount()
|
||||
}
|
||||
|
||||
result := map[string]string{
|
||||
|
|
|
|||
|
|
@ -88,10 +88,10 @@ func startMockEdge(t *testing.T, ctx context.Context, listenAddr string) *toki.T
|
|||
}
|
||||
}
|
||||
|
||||
// TestQueueOverflow_RejectEventObservedByEdge is an integration test that
|
||||
// verifies the RunEvent{type:"error"} sent by rejectRun() carries the queue
|
||||
// overflow reason and actually arrives at the mock Edge server when the Node's
|
||||
// FIFO admission queue is full (capacity=1, max_queue=0).
|
||||
// TestOverDispatchSafety_RejectEventObservedByEdge is an integration test that
|
||||
// verifies the RunEvent{type:"error"} sent by rejectRun() carries the concurrency
|
||||
// unavailable reason and actually arrives at the mock Edge server when the Node's
|
||||
// concurrency capacity limit is exceeded (capacity=1).
|
||||
//
|
||||
// Setup:
|
||||
// 1. Start mock Edge TCP server.
|
||||
|
|
@ -100,9 +100,9 @@ func startMockEdge(t *testing.T, ctx context.Context, listenAddr string) *toki.T
|
|||
// 4. Attach the session to the node via SetHandler.
|
||||
// 5. From the Edge side, send a first RunRequest whose adapter blocks.
|
||||
// 6. After the first run starts, send a second RunRequest from the Edge side.
|
||||
// 7. Assert a RunEvent{type:"error", run_id:<second>} with a queue_full reason
|
||||
// 7. Assert a RunEvent{type:"error", run_id:<second>} with a concurrency unavailable reason
|
||||
// arrives at the Edge.
|
||||
func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) {
|
||||
func TestOverDispatchSafety_RejectEventObservedByEdge(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
|
|
@ -161,7 +161,7 @@ func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) {
|
|||
})
|
||||
|
||||
// Build real node.Node with global concurrency=1.
|
||||
// capacity=1, max_queue=0 → a second concurrent run overflows the queue.
|
||||
// capacity=1 → a second concurrent run overflows the concurrency limit.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 0, 0)
|
||||
rtr := &fixedRouter{
|
||||
adapterName: "slow",
|
||||
|
|
@ -185,8 +185,7 @@ func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) {
|
|||
// Wait until the first run's adapter has started executing so the slot is held.
|
||||
waitStarted(t, sa, "run-hold-1")
|
||||
|
||||
// Send second RunRequest; should be rejected because capacity=1 is taken and
|
||||
// the queue (max_queue=0) cannot hold it.
|
||||
// Send second RunRequest; should be rejected because capacity=1 is taken.
|
||||
secondReq := &iop.RunRequest{RunId: "run-reject-2", Adapter: "slow", Target: "v1"}
|
||||
if err := edgeClient.Send(secondReq); err != nil {
|
||||
t.Fatalf("send second run request: %v", err)
|
||||
|
|
@ -201,8 +200,8 @@ func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) {
|
|||
if ev.GetError() == "" {
|
||||
t.Fatal("reject RunEvent has empty error field")
|
||||
}
|
||||
if !strings.Contains(ev.GetError(), "queue full") {
|
||||
t.Fatalf("expected queue_full reason in reject event, got %q", ev.GetError())
|
||||
if !strings.Contains(ev.GetError(), "concurrency unavailable") {
|
||||
t.Fatalf("expected concurrency unavailable reason in reject event, got %q", ev.GetError())
|
||||
}
|
||||
// Success: RunEvent{type:"error", run_id:"run-reject-2"} observed.
|
||||
sa.releaseRun("run-hold-1")
|
||||
|
|
|
|||
|
|
@ -718,10 +718,10 @@ func TestOnCommandRequest_Capabilities_InFlight(t *testing.T) {
|
|||
<-ba.done
|
||||
}
|
||||
|
||||
func TestOnCommandRequest_Capabilities_Queued(t *testing.T) {
|
||||
func TestOnCommandRequest_Capabilities_SafetyRejected(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 0)
|
||||
n, _ := makeNodeWithConcurrency(t, router, 0)
|
||||
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cap-hold", Adapter: "slow", Target: "v1", Background: true,
|
||||
|
|
@ -730,12 +730,13 @@ func TestOnCommandRequest_Capabilities_Queued(t *testing.T) {
|
|||
}
|
||||
waitStarted(t, sa, "cap-hold")
|
||||
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cap-queued", Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("queued run: %v", err)
|
||||
// The second run request must fail immediately since there is no queue.
|
||||
err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cap-rejected", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
|
||||
}
|
||||
requireQueued(t, st, "cap-queued")
|
||||
|
||||
resp, err := n.OnCommandRequest(context.Background(), &transport.Session{}, &iop.NodeCommandRequest{
|
||||
RequestId: "req-cap-queued",
|
||||
|
|
@ -755,8 +756,8 @@ func TestOnCommandRequest_Capabilities_Queued(t *testing.T) {
|
|||
if got := resp.GetResult()["in_flight"]; got != "1" {
|
||||
t.Fatalf("result[in_flight]: got %q want %q", got, "1")
|
||||
}
|
||||
if got := resp.GetResult()["queued"]; got != "1" {
|
||||
t.Fatalf("result[queued]: got %q want %q", got, "1")
|
||||
if got := resp.GetResult()["queued"]; got != "0" {
|
||||
t.Fatalf("result[queued]: got %q want %q", got, "0")
|
||||
}
|
||||
if len(resp.GetProviderSnapshots()) != 1 {
|
||||
t.Fatalf("expected 1 provider snapshot, got %d", len(resp.GetProviderSnapshots()))
|
||||
|
|
@ -768,14 +769,11 @@ func TestOnCommandRequest_Capabilities_Queued(t *testing.T) {
|
|||
if snap.GetInFlight() != 1 {
|
||||
t.Fatalf("snap.InFlight: got %d want %d", snap.GetInFlight(), 1)
|
||||
}
|
||||
if snap.GetQueued() != 1 {
|
||||
t.Fatalf("snap.Queued: got %d want %d", snap.GetQueued(), 1)
|
||||
if snap.GetQueued() != 0 {
|
||||
t.Fatalf("snap.Queued: got %d want %d", snap.GetQueued(), 0)
|
||||
}
|
||||
|
||||
sa.releaseRun("cap-hold")
|
||||
waitStarted(t, sa, "cap-queued")
|
||||
sa.releaseRun("cap-queued")
|
||||
requireStatusEventually(t, st, "cap-queued", "completed")
|
||||
}
|
||||
|
||||
func TestOnCommandRequest_CapabilitiesProviderStatusModel(t *testing.T) {
|
||||
|
|
@ -1296,10 +1294,10 @@ func (a *slowAdapterUnlimited) Execute(ctx context.Context, _ runtime.ExecutionS
|
|||
return nil
|
||||
}
|
||||
|
||||
// queuedSlowAdapter is a configurable adapter used by the FIFO admission tests.
|
||||
// queuedSlowAdapter is a configurable adapter used by concurrency safety tests.
|
||||
// It advertises MaxConcurrency / MaxQueue / QueueTimeout via Capabilities, and
|
||||
// gives per-run control: each Execute blocks until the test releases (or fails)
|
||||
// that specific run id, so tests can observe queue ordering deterministically.
|
||||
// that specific run id, so tests can observe execution ordering deterministically.
|
||||
type queuedSlowAdapter struct {
|
||||
name string
|
||||
maxConcurrency int
|
||||
|
|
@ -1386,25 +1384,6 @@ func waitStarted(t *testing.T, a *queuedSlowAdapter, runID string) {
|
|||
}
|
||||
}
|
||||
|
||||
// waitAnyStarted returns the next run id that began executing.
|
||||
func waitAnyStarted(t *testing.T, a *queuedSlowAdapter) string {
|
||||
t.Helper()
|
||||
select {
|
||||
case got := <-a.startSeq:
|
||||
return got
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("no run started within timeout")
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
// requireQueued asserts the run is stored with status "queued" (polling briefly
|
||||
// for the asynchronous store write).
|
||||
func requireQueued(t *testing.T, st *store.Store, runID string) {
|
||||
t.Helper()
|
||||
requireStatusEventually(t, st, runID, "queued")
|
||||
}
|
||||
|
||||
// requireStatus asserts the run is stored with exactly the given status now.
|
||||
func requireStatus(t *testing.T, st *store.Store, runID, status string) *store.RunRecord {
|
||||
t.Helper()
|
||||
|
|
@ -1435,129 +1414,11 @@ func requireStatusEventually(t *testing.T, st *store.Store, runID, status string
|
|||
}
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_ForegroundSecondWaitsThenRuns verifies that a second
|
||||
// foreground run is not rejected: it waits in the FIFO queue and executes once
|
||||
// the first run releases its slot.
|
||||
func TestAdmissionQueue_ForegroundSecondWaitsThenRuns(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-fg-1", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "run-fg-1")
|
||||
|
||||
second := make(chan error, 1)
|
||||
go func() {
|
||||
second <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-fg-2", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
|
||||
// The second run must remain queued (not started, not rejected) while the
|
||||
// first holds the only slot.
|
||||
requireQueued(t, st, "run-fg-2")
|
||||
select {
|
||||
case err := <-second:
|
||||
t.Fatalf("second run should be queued, but returned early: %v", err)
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
}
|
||||
|
||||
// Release the first; the second must then start and complete.
|
||||
sa.releaseRun("run-fg-1")
|
||||
if err := <-first; err != nil {
|
||||
t.Fatalf("first run: %v", err)
|
||||
}
|
||||
waitStarted(t, sa, "run-fg-2")
|
||||
sa.releaseRun("run-fg-2")
|
||||
if err := <-second; err != nil {
|
||||
t.Fatalf("second run should succeed after promotion, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_BackgroundQueuedReturnsThenRuns verifies that a background
|
||||
// run that cannot run immediately returns after enqueue and is later promoted.
|
||||
func TestAdmissionQueue_BackgroundQueuedReturnsThenRuns(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-bg-hold", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "run-bg-hold")
|
||||
|
||||
// Background queued run returns promptly after enqueue.
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-bg-2", Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("background queued run should enqueue without error, got %v", err)
|
||||
}
|
||||
requireQueued(t, st, "run-bg-2")
|
||||
|
||||
// Release the holder; the queued background run must be promoted and run.
|
||||
sa.releaseRun("run-bg-hold")
|
||||
if err := <-first; err != nil {
|
||||
t.Fatalf("hold run: %v", err)
|
||||
}
|
||||
waitStarted(t, sa, "run-bg-2")
|
||||
sa.releaseRun("run-bg-2")
|
||||
requireStatusEventually(t, st, "run-bg-2", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_FIFOOrdering verifies that queued runs are promoted in
|
||||
// first-in-first-out order.
|
||||
func TestAdmissionQueue_FIFOOrdering(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 8, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, _ := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
// First run holds the single slot.
|
||||
hold := make(chan error, 1)
|
||||
go func() {
|
||||
hold <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "fifo-0", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "fifo-0")
|
||||
|
||||
// Enqueue three background runs in a deterministic order. Each enqueue
|
||||
// returns before the next is submitted, so queue order is well-defined.
|
||||
order := []string{"fifo-1", "fifo-2", "fifo-3"}
|
||||
for _, id := range order {
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: id, Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("enqueue %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Release runs one at a time and assert they start in FIFO order.
|
||||
sa.releaseRun("fifo-0")
|
||||
if err := <-hold; err != nil {
|
||||
t.Fatalf("fifo-0: %v", err)
|
||||
}
|
||||
for _, id := range order {
|
||||
got := waitAnyStarted(t, sa)
|
||||
if got != id {
|
||||
t.Fatalf("FIFO order violated: expected %s to start next, got %s", id, got)
|
||||
}
|
||||
sa.releaseRun(id)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_MaxQueueOverflow verifies that a run is rejected with
|
||||
// reason queue_full when the FIFO queue is already at max_queue.
|
||||
func TestAdmissionQueue_MaxQueueOverflow(t *testing.T) {
|
||||
// capacity=1, max_queue=1: one running + one queued is the ceiling.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 1, 0)
|
||||
// TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue verifies that a run
|
||||
// is rejected immediately with reason concurrency_unavailable when the concurrency limit is reached (no queue).
|
||||
func TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue(t *testing.T) {
|
||||
// capacity=1: one running is the ceiling.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 0, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 0) // global unlimited; adapter cap governs
|
||||
|
||||
|
|
@ -1569,15 +1430,7 @@ func TestAdmissionQueue_MaxQueueOverflow(t *testing.T) {
|
|||
}()
|
||||
waitStarted(t, sa, "of-hold")
|
||||
|
||||
// Fill the single queue slot.
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "of-queued", Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("first queued run should enqueue, got %v", err)
|
||||
}
|
||||
requireQueued(t, st, "of-queued")
|
||||
|
||||
// Third run overflows the queue and must be rejected with queue_full.
|
||||
// Second run overflows the capacity and must be rejected immediately.
|
||||
err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "of-reject", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
|
|
@ -1585,118 +1438,19 @@ func TestAdmissionQueue_MaxQueueOverflow(t *testing.T) {
|
|||
t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
|
||||
}
|
||||
rec := requireStatus(t, st, "of-reject", "rejected")
|
||||
if !strings.Contains(rec.Error, "queue full") {
|
||||
t.Fatalf("expected queue_full reason in error, got %q", rec.Error)
|
||||
if !strings.Contains(rec.Error, "concurrency unavailable") {
|
||||
t.Fatalf("expected concurrency unavailable reason in error, got %q", rec.Error)
|
||||
}
|
||||
|
||||
sa.releaseRun("of-hold")
|
||||
if err := <-hold; err != nil {
|
||||
t.Fatalf("of-hold: %v", err)
|
||||
}
|
||||
waitStarted(t, sa, "of-queued")
|
||||
sa.releaseRun("of-queued")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_QueueTimeout verifies that a queued run is rejected with
|
||||
// reason queue_timeout when it waits past queue_timeout_ms.
|
||||
func TestAdmissionQueue_QueueTimeout(t *testing.T) {
|
||||
// capacity=1, max_queue=4, queue_timeout=120ms.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 120*time.Millisecond)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 0)
|
||||
|
||||
hold := make(chan error, 1)
|
||||
go func() {
|
||||
hold <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "to-hold", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "to-hold")
|
||||
|
||||
// This foreground run will wait in the queue and time out (holder is never
|
||||
// released until after the timeout).
|
||||
err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "to-timeout", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded on queue timeout, got %v", err)
|
||||
}
|
||||
rec := requireStatus(t, st, "to-timeout", "rejected")
|
||||
if !strings.Contains(rec.Error, "queue timeout") {
|
||||
t.Fatalf("expected queue_timeout reason in error, got %q", rec.Error)
|
||||
}
|
||||
|
||||
sa.releaseRun("to-hold")
|
||||
if err := <-hold; err != nil {
|
||||
t.Fatalf("to-hold: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_ReleaseAfterFailurePromotesNext verifies that a queued run
|
||||
// is promoted when the running run fails.
|
||||
func TestAdmissionQueue_ReleaseAfterFailurePromotesNext(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "fail-1", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "fail-1")
|
||||
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "fail-next", Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("enqueue fail-next: %v", err)
|
||||
}
|
||||
requireQueued(t, st, "fail-next")
|
||||
|
||||
// First run fails; its slot must promote the queued run.
|
||||
sa.failRun("fail-1")
|
||||
<-first
|
||||
waitStarted(t, sa, "fail-next")
|
||||
sa.releaseRun("fail-next")
|
||||
requireStatusEventually(t, st, "fail-next", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_ReleaseAfterCancelPromotesNext verifies that a queued run
|
||||
// is promoted when the running run is cancelled.
|
||||
func TestAdmissionQueue_ReleaseAfterCancelPromotesNext(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, st := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cancel-1", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "cancel-1")
|
||||
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cancel-next", Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("enqueue cancel-next: %v", err)
|
||||
}
|
||||
requireQueued(t, st, "cancel-next")
|
||||
<-first
|
||||
|
||||
// Cancel the running run; its slot must promote the queued run.
|
||||
if err := n.OnCancel(context.Background(), &transport.Session{}, &iop.CancelRequest{RunId: "cancel-1"}); err != nil {
|
||||
t.Fatalf("cancel: %v", err)
|
||||
}
|
||||
waitStarted(t, sa, "cancel-next")
|
||||
sa.releaseRun("cancel-next")
|
||||
requireStatusEventually(t, st, "cancel-next", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_PermitReleasedAfterCompletion verifies that after the first
|
||||
// run completes, a subsequent run can acquire the slot without queuing.
|
||||
func TestAdmissionQueue_PermitReleasedAfterCompletion(t *testing.T) {
|
||||
// TestOnRunRequest_PermitReleasedAfterCompletion verifies that after the first
|
||||
// run completes, a subsequent run can acquire the slot immediately.
|
||||
func TestOnRunRequest_PermitReleasedAfterCompletion(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, _ := makeNodeWithConcurrency(t, router, 1)
|
||||
|
|
@ -1750,323 +1504,10 @@ func TestConcurrencyLimit_Unlimited(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_AdapterCapQueues verifies that when global=0 and adapter
|
||||
// cap=1, a second run queues (rather than being rejected) and runs after release.
|
||||
func TestAdmissionQueue_AdapterCapQueues(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0) // adapter cap=1
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 0) // global=0 → adapter cap governs
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cap-1", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "cap-1")
|
||||
|
||||
if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "cap-2", Adapter: "slow", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("cap-2 should enqueue, got %v", err)
|
||||
}
|
||||
requireQueued(t, st, "cap-2")
|
||||
|
||||
sa.releaseRun("cap-1")
|
||||
if err := <-first; err != nil {
|
||||
t.Fatalf("cap-1: %v", err)
|
||||
}
|
||||
waitStarted(t, sa, "cap-2")
|
||||
sa.releaseRun("cap-2")
|
||||
requireStatusEventually(t, st, "cap-2", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_GlobalBeatsAdapterCap verifies that the global gate (=1)
|
||||
// limits effective in-flight to 1 even when the adapter cap is looser (=4): the
|
||||
// second run queues behind the global slot.
|
||||
func TestAdmissionQueue_GlobalBeatsAdapterCap(t *testing.T) {
|
||||
sa := newQueuedSlowAdapter("slow-hicap", 4, 4, 0) // adapter cap=4
|
||||
router := &fixedRouter{adapterName: "slow-hicap", adapters: map[string]runtime.Adapter{"slow-hicap": sa}}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 1) // global=1 → effective=1
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "g1-1", Adapter: "slow-hicap", Target: "v1",
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "g1-1")
|
||||
|
||||
// Second run queues behind the global gate even though adapter cap=4.
|
||||
if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "g1-2", Adapter: "slow-hicap", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("g1-2 should enqueue, got %v", err)
|
||||
}
|
||||
requireQueued(t, st, "g1-2")
|
||||
|
||||
sa.releaseRun("g1-1")
|
||||
if err := <-first; err != nil {
|
||||
t.Fatalf("g1-1: %v", err)
|
||||
}
|
||||
waitStarted(t, sa, "g1-2")
|
||||
sa.releaseRun("g1-2")
|
||||
requireStatusEventually(t, st, "g1-2", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_GlobalBeatsAdapterCap_DifferentAdapters verifies that
|
||||
// global=1 makes a run on adapter B queue while adapter A holds the global slot,
|
||||
// even though both adapter caps are > 1.
|
||||
func TestAdmissionQueue_GlobalBeatsAdapterCap_DifferentAdapters(t *testing.T) {
|
||||
saA := newQueuedSlowAdapter("slow-hicap-a", 4, 4, 0)
|
||||
saB := newQueuedSlowAdapter("slow-hicap-b", 4, 4, 0)
|
||||
|
||||
router := &fixedRouter{
|
||||
adapterName: "slow-hicap-a",
|
||||
adapters: map[string]runtime.Adapter{
|
||||
"slow-hicap-a": saA,
|
||||
"slow-hicap-b": saB,
|
||||
},
|
||||
}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 1) // global=1
|
||||
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
first <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "xa-1", Adapter: "slow-hicap-a", Target: "v1",
|
||||
})
|
||||
}()
|
||||
waitStarted(t, saA, "xa-1")
|
||||
|
||||
// Run on adapter B queues because the global slot is taken by adapter A.
|
||||
router.adapterName = "slow-hicap-b"
|
||||
if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "xb-1", Adapter: "slow-hicap-b", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("xb-1 should enqueue, got %v", err)
|
||||
}
|
||||
requireQueued(t, st, "xb-1")
|
||||
|
||||
sa := saA
|
||||
sa.releaseRun("xa-1")
|
||||
if err := <-first; err != nil {
|
||||
t.Fatalf("xa-1: %v", err)
|
||||
}
|
||||
waitStarted(t, saB, "xb-1")
|
||||
saB.releaseRun("xb-1")
|
||||
requireStatusEventually(t, st, "xb-1", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_GlobalNoCrossAdapterHeadOfLine is the regression test for
|
||||
// the gate-composition fix: a queued run on one adapter must not occupy a global
|
||||
// slot it cannot yet use and thereby block a runnable queue head on a different
|
||||
// adapter.
|
||||
//
|
||||
// Setup: global=2, adapter A capacity=1, adapter B capacity=1.
|
||||
// - A1 and B1 run, consuming both global slots and both adapter slots.
|
||||
// - A2 is enqueued first (waiting on adapter A slot, held by A1).
|
||||
// - B2 is enqueued next (waiting on adapter B slot, held by B1).
|
||||
// - Releasing only B1 must let B2 start immediately, because A2 (still without
|
||||
// an adapter A slot) must NOT have grabbed the freed global slot ahead of it.
|
||||
func TestAdmissionQueue_GlobalNoCrossAdapterHeadOfLine(t *testing.T) {
|
||||
saA := newQueuedSlowAdapter("adapter-a", 1, 4, 0)
|
||||
saB := newQueuedSlowAdapter("adapter-b", 1, 4, 0)
|
||||
router := &fixedRouter{
|
||||
adapterName: "adapter-a",
|
||||
adapters: map[string]runtime.Adapter{
|
||||
"adapter-a": saA,
|
||||
"adapter-b": saB,
|
||||
},
|
||||
}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 2) // global=2
|
||||
|
||||
// A1 and B1 occupy both global slots and both adapter slots.
|
||||
a1 := make(chan error, 1)
|
||||
go func() {
|
||||
a1 <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "a1", Adapter: "adapter-a", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, saA, "a1")
|
||||
|
||||
router.adapterName = "adapter-b"
|
||||
b1 := make(chan error, 1)
|
||||
go func() {
|
||||
b1 <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "b1", Adapter: "adapter-b", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, saB, "b1")
|
||||
|
||||
// Enqueue A2 first (waits on adapter A slot), then B2 (waits on adapter B slot).
|
||||
router.adapterName = "adapter-a"
|
||||
if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "a2", Adapter: "adapter-a", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("enqueue a2: %v", err)
|
||||
}
|
||||
requireQueued(t, st, "a2")
|
||||
|
||||
router.adapterName = "adapter-b"
|
||||
if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "b2", Adapter: "adapter-b", Target: "v1", Background: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("enqueue b2: %v", err)
|
||||
}
|
||||
requireQueued(t, st, "b2")
|
||||
|
||||
// Release only B1: its adapter B slot and one global slot free up. B2 must
|
||||
// start. A2 stays queued because adapter A slot (a1) is still held.
|
||||
<-b1
|
||||
saB.releaseRun("b1")
|
||||
waitStarted(t, saB, "b2") // fails here if A2 head-of-line-blocked the global slot
|
||||
|
||||
// A2 must still be queued (it has no adapter A slot yet).
|
||||
requireStatus(t, st, "a2", "queued")
|
||||
select {
|
||||
case <-saA.startSeq:
|
||||
t.Fatal("a2 must not start before a1 releases its adapter slot")
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
|
||||
// Now release a1 → A2 can take adapter A slot and the remaining global slot.
|
||||
<-a1
|
||||
saA.releaseRun("a1")
|
||||
waitStarted(t, saA, "a2")
|
||||
|
||||
saB.releaseRun("b2")
|
||||
saA.releaseRun("a2")
|
||||
requireStatusEventually(t, st, "a2", "completed")
|
||||
requireStatusEventually(t, st, "b2", "completed")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_QueueTimeoutAppliesAcrossGates verifies that the queue
|
||||
// timeout bounds the total admission wait even when the run is blocked waiting
|
||||
// for the global slot after acquiring its adapter slot.
|
||||
func TestAdmissionQueue_QueueTimeoutAppliesAcrossGates(t *testing.T) {
|
||||
// adapter A and B each capacity=1 with a queue timeout; global=1 is the
|
||||
// bottleneck so the second adapter's run holds its adapter slot but blocks on
|
||||
// the global gate, then must time out.
|
||||
saA := newQueuedSlowAdapter("adapter-a", 1, 4, 120*time.Millisecond)
|
||||
saB := newQueuedSlowAdapter("adapter-b", 1, 4, 120*time.Millisecond)
|
||||
router := &fixedRouter{
|
||||
adapterName: "adapter-a",
|
||||
adapters: map[string]runtime.Adapter{
|
||||
"adapter-a": saA,
|
||||
"adapter-b": saB,
|
||||
},
|
||||
}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 1) // global=1
|
||||
|
||||
a1 := make(chan error, 1)
|
||||
go func() {
|
||||
a1 <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "ta1", Adapter: "adapter-a", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, saA, "ta1")
|
||||
|
||||
// Run on adapter B acquires adapter B slot immediately but blocks on global=1.
|
||||
// It must time out via the queue timeout rather than waiting indefinitely.
|
||||
router.adapterName = "adapter-b"
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "tb1", Adapter: "adapter-b", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded on global-wait timeout, got %v", err)
|
||||
}
|
||||
rec := requireStatus(t, st, "tb1", "rejected")
|
||||
if !strings.Contains(rec.Error, "queue timeout") {
|
||||
t.Fatalf("expected queue_timeout reason for global-wait timeout, got %q", rec.Error)
|
||||
}
|
||||
|
||||
<-a1
|
||||
saA.releaseRun("ta1")
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_ParentTimeoutNotQueueTimeout verifies that when a queued
|
||||
// run is terminated by its parent RunRequest.timeout_sec (not the provider
|
||||
// queue timeout), the rejection reason is NOT misreported as queue_timeout. With
|
||||
// queue_timeout_ms=0 the queue timeout mechanism is disabled, so the bogus
|
||||
// "queue timeout exceeded (queue_timeout=0s)" message must never appear.
|
||||
func TestAdmissionQueue_ParentTimeoutNotQueueTimeout(t *testing.T) {
|
||||
// queue_timeout_ms=0 (disabled), max_queue=4, capacity=1.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 0)
|
||||
|
||||
hold := make(chan error, 1)
|
||||
go func() {
|
||||
hold <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "pt-hold", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "pt-hold")
|
||||
|
||||
// Second foreground run queues (queue enabled, slot busy) and is terminated
|
||||
// by its own 1s request timeout, not by a queue timeout.
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "pt-timeout", Adapter: "slow", Target: "v1", TimeoutSec: 1,
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded on parent request timeout, got %v", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "queue timeout") {
|
||||
t.Fatalf("parent request timeout must not be reported as queue timeout, got %q", err.Error())
|
||||
}
|
||||
rec := requireStatus(t, st, "pt-timeout", "rejected")
|
||||
if strings.Contains(rec.Error, "queue timeout") || strings.Contains(rec.Error, "queue_timeout=0s") {
|
||||
t.Fatalf("store error must not contain a queue_timeout reason, got %q", rec.Error)
|
||||
}
|
||||
|
||||
sa.releaseRun("pt-hold")
|
||||
if err := <-hold; err != nil {
|
||||
t.Fatalf("pt-hold: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdmissionQueue_ParentTimeoutBeatsQueueTimeout verifies that even when a
|
||||
// provider queue timeout IS configured, a parent request timeout that fires
|
||||
// first is still classified as concurrency_unavailable, not queue_timeout.
|
||||
func TestAdmissionQueue_ParentTimeoutBeatsQueueTimeout(t *testing.T) {
|
||||
// queue_timeout_ms=10s (long), but the run's request timeout is 1s, so the
|
||||
// parent deadline fires first.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 4, 10*time.Second)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 0)
|
||||
|
||||
hold := make(chan error, 1)
|
||||
go func() {
|
||||
hold <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "pb-hold", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
}()
|
||||
waitStarted(t, sa, "pb-hold")
|
||||
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "pb-timeout", Adapter: "slow", Target: "v1", TimeoutSec: 1,
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "queue timeout") {
|
||||
t.Fatalf("parent deadline (earlier than queue timeout) must not be queue timeout, got %q", err.Error())
|
||||
}
|
||||
rec := requireStatus(t, st, "pb-timeout", "rejected")
|
||||
if strings.Contains(rec.Error, "queue timeout") {
|
||||
t.Fatalf("store error must not contain queue timeout, got %q", rec.Error)
|
||||
}
|
||||
|
||||
sa.releaseRun("pb-hold")
|
||||
if err := <-hold; err != nil {
|
||||
t.Fatalf("pb-hold: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run (queue
|
||||
// overflow) is stored with terminal status "rejected" and an error message.
|
||||
// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run
|
||||
// is stored with terminal status "rejected" and an error message.
|
||||
func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) {
|
||||
// capacity=1, max_queue=0: any second concurrent run overflows immediately.
|
||||
// capacity=1: any second concurrent run overflows immediately.
|
||||
sa := newQueuedSlowAdapter("slow", 1, 0, 0)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 0)
|
||||
|
|
@ -2079,7 +1520,7 @@ func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) {
|
|||
}()
|
||||
waitStarted(t, sa, "run-hold")
|
||||
|
||||
// Second run must be rejected (queue capacity is 0).
|
||||
// Second run must be rejected.
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-rejected", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
|
|
|
|||
|
|
@ -4,14 +4,10 @@ import (
|
|||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ErrConcurrencyLimitExceeded is returned when a run is rejected because
|
||||
// admission could not be granted under the current concurrency policy. It is
|
||||
// kept as the wrapping sentinel for backward compatibility; the concrete reason
|
||||
// (queue full, queue timeout, concurrency unavailable) is carried by
|
||||
// admissionError and surfaced through admissionRejectReason.
|
||||
// admission could not be granted under the current concurrency policy.
|
||||
var ErrConcurrencyLimitExceeded = errors.New("concurrency limit exceeded")
|
||||
|
||||
type runHandle struct {
|
||||
|
|
@ -55,29 +51,16 @@ func (m *runManager) cancelRun(runID string) bool {
|
|||
return true
|
||||
}
|
||||
|
||||
// rejectReason classifies why a run could not be admitted or had to be
|
||||
// terminated before execution.
|
||||
// rejectReason classifies why a run could not be admitted.
|
||||
type rejectReason string
|
||||
|
||||
const (
|
||||
// reasonQueueFull means the FIFO admission queue is already at max_queue.
|
||||
reasonQueueFull rejectReason = "queue_full"
|
||||
// reasonQueueTimeout means the request waited past queue_timeout_ms.
|
||||
reasonQueueTimeout rejectReason = "queue_timeout"
|
||||
// reasonConcurrencyUnavailable means admission failed because the caller's
|
||||
// context was cancelled while waiting for a slot.
|
||||
// reasonConcurrencyUnavailable means admission failed because the concurrency limit was exceeded.
|
||||
reasonConcurrencyUnavailable rejectReason = "concurrency_unavailable"
|
||||
)
|
||||
|
||||
// errAdmissionQueueTimeout is the context cancellation cause attached to the
|
||||
// provider queue-timeout deadline. It lets wait() distinguish a real
|
||||
// queue_timeout_ms expiry from an unrelated parent deadline (e.g. the run's
|
||||
// RunRequest.timeout_sec), so only the former is reported as queue_timeout.
|
||||
var errAdmissionQueueTimeout = errors.New("admission queue timeout")
|
||||
|
||||
// admissionError carries the concrete rejection reason while still satisfying
|
||||
// errors.Is(err, ErrConcurrencyLimitExceeded) for callers relying on the
|
||||
// legacy sentinel.
|
||||
// errors.Is(err, ErrConcurrencyLimitExceeded) for callers relying on the legacy sentinel.
|
||||
type admissionError struct {
|
||||
reason rejectReason
|
||||
msg string
|
||||
|
|
@ -87,37 +70,14 @@ func (e *admissionError) Error() string { return e.msg }
|
|||
|
||||
func (e *admissionError) Is(target error) bool { return target == ErrConcurrencyLimitExceeded }
|
||||
|
||||
func newQueueFullError(maxQueue int) *admissionError {
|
||||
return &admissionError{
|
||||
reason: reasonQueueFull,
|
||||
msg: queueFullMessage(maxQueue),
|
||||
}
|
||||
}
|
||||
|
||||
func newQueueTimeoutError(timeout time.Duration) *admissionError {
|
||||
return &admissionError{
|
||||
reason: reasonQueueTimeout,
|
||||
msg: queueTimeoutMessage(timeout),
|
||||
}
|
||||
}
|
||||
|
||||
func newConcurrencyUnavailableError() *admissionError {
|
||||
return &admissionError{
|
||||
reason: reasonConcurrencyUnavailable,
|
||||
msg: "concurrency unavailable: admission cancelled before a slot was free",
|
||||
msg: "concurrency unavailable: admission rejected because safety limit exceeded",
|
||||
}
|
||||
}
|
||||
|
||||
func queueFullMessage(maxQueue int) string {
|
||||
return "queue full (max_queue=" + itoa(maxQueue) + ")"
|
||||
}
|
||||
|
||||
func queueTimeoutMessage(timeout time.Duration) string {
|
||||
return "queue timeout exceeded (queue_timeout=" + timeout.String() + ")"
|
||||
}
|
||||
|
||||
// admissionRejectReason extracts the rejectReason from an admission error,
|
||||
// defaulting to concurrency_unavailable for unknown errors.
|
||||
// admissionRejectReason extracts the rejectReason from an admission error.
|
||||
func admissionRejectReason(err error) rejectReason {
|
||||
var ae *admissionError
|
||||
if errors.As(err, &ae) {
|
||||
|
|
@ -138,246 +98,88 @@ func admissionMessage(err error) string {
|
|||
return string(reasonConcurrencyUnavailable)
|
||||
}
|
||||
|
||||
// fifoGate is a FIFO-fair counting semaphore. capacity <= 0 means unlimited
|
||||
// (acquire always succeeds immediately and is never queued). maxQueue < 0 means
|
||||
// an unbounded queue; maxQueue == 0 disallows queueing (a full gate rejects
|
||||
// immediately); maxQueue > 0 caps the queue length. queueTimeout <= 0 means no
|
||||
// timeout while waiting.
|
||||
// fifoGate is a simple in-flight concurrency semaphore (no queue).
|
||||
type fifoGate struct {
|
||||
mu sync.Mutex
|
||||
capacity int
|
||||
maxQueue int
|
||||
queueTimeout time.Duration
|
||||
inFlight int
|
||||
waiters []chan struct{}
|
||||
mu sync.Mutex
|
||||
capacity int
|
||||
inFlight int
|
||||
}
|
||||
|
||||
func newFifoGate(capacity, maxQueue int, queueTimeout time.Duration) *fifoGate {
|
||||
func newFifoGate(capacity int) *fifoGate {
|
||||
return &fifoGate{
|
||||
capacity: capacity,
|
||||
maxQueue: maxQueue,
|
||||
queueTimeout: queueTimeout,
|
||||
capacity: capacity,
|
||||
}
|
||||
}
|
||||
|
||||
// reserve takes a slot without blocking. When capacity is free it grants the
|
||||
// slot immediately (granted=true, ch=nil). When full it appends a FIFO waiter
|
||||
// and returns its channel (granted=false, ch!=nil) which the caller waits on via
|
||||
// waitOn. When the queue is already at max_queue it returns a queue_full error.
|
||||
func (g *fifoGate) reserve() (granted bool, ch chan struct{}, err error) {
|
||||
func (g *fifoGate) acquire() bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.capacity <= 0 || g.inFlight < g.capacity {
|
||||
g.inFlight++
|
||||
return true, nil, nil
|
||||
}
|
||||
if g.maxQueue >= 0 && len(g.waiters) >= g.maxQueue {
|
||||
return false, nil, newQueueFullError(g.maxQueue)
|
||||
}
|
||||
ch = make(chan struct{})
|
||||
g.waiters = append(g.waiters, ch)
|
||||
return false, ch, nil
|
||||
}
|
||||
|
||||
// waitOn blocks until the reserved waiter ch is promoted to an execution slot or
|
||||
// the context is cancelled (the caller derives any queue-timeout deadline onto
|
||||
// ctx). On context cancellation it abandons the waiter and returns an
|
||||
// admissionError; on success (including a promotion that races the cancel) it
|
||||
// returns nil and the caller holds a live slot.
|
||||
func (g *fifoGate) waitOn(ctx context.Context, ch chan struct{}) error {
|
||||
select {
|
||||
case <-ch:
|
||||
// Promoted to an execution slot; inFlight was incremented by release.
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
if g.abandon(ch) {
|
||||
return newConcurrencyUnavailableError()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// abandon removes ch from the waiter queue. It returns true if ch was still
|
||||
// waiting (so the caller owns the failure), or false if ch had already been
|
||||
// promoted (so the caller now holds a live slot it must release).
|
||||
func (g *fifoGate) abandon(ch chan struct{}) bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
for i, w := range g.waiters {
|
||||
if w == ch {
|
||||
g.waiters = append(g.waiters[:i], g.waiters[i+1:]...)
|
||||
return true
|
||||
}
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// release returns a slot. If a waiter is queued it is promoted in FIFO order,
|
||||
// keeping inFlight unchanged (the slot transfers to the promoted waiter).
|
||||
func (g *fifoGate) release() {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if len(g.waiters) > 0 {
|
||||
next := g.waiters[0]
|
||||
g.waiters = g.waiters[1:]
|
||||
close(next)
|
||||
return
|
||||
}
|
||||
if g.inFlight > 0 {
|
||||
g.inFlight--
|
||||
}
|
||||
}
|
||||
|
||||
// activeCount returns the number of in-flight (executing) slots.
|
||||
func (g *fifoGate) activeCount() int {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
return g.inFlight
|
||||
}
|
||||
|
||||
// queuedCount returns the number of requests currently waiting for a slot.
|
||||
func (g *fifoGate) queuedCount() int {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
return len(g.waiters)
|
||||
return 0
|
||||
}
|
||||
|
||||
// admissionManager composes the node-wide global gate with a per-adapter gate.
|
||||
// A run is admitted only when it holds both: the adapter gate is reserved first
|
||||
// (so per-provider FIFO ordering is preserved without head-of-line blocking
|
||||
// across adapters), then the global gate.
|
||||
type admissionManager struct {
|
||||
global *fifoGate
|
||||
adapter *fifoGate
|
||||
}
|
||||
|
||||
// admissionTicket holds the reservation state between enqueue and wait. queued
|
||||
// reports whether the request had to wait for a slot (vs. running immediately).
|
||||
type admissionTicket struct {
|
||||
mgr *admissionManager
|
||||
adapterCh chan struct{} // non-nil if waiting on the adapter gate
|
||||
globalCh chan struct{} // non-nil if waiting on the global gate
|
||||
adapterHeld bool
|
||||
globalHeld bool
|
||||
queued bool
|
||||
}
|
||||
|
||||
// enqueue reserves only the per-adapter FIFO position, without blocking. It
|
||||
// returns a queue_full admissionError when the adapter queue is at max_queue;
|
||||
// otherwise it returns a ticket whose wait() blocks until the run is admitted.
|
||||
// The global gate is intentionally NOT reserved here: reserving it before the
|
||||
// adapter slot is granted would let a run still waiting for its adapter slot
|
||||
// occupy a global slot, blocking a different adapter's runnable queue head
|
||||
// (cross-adapter head-of-line blocking). The global slot is reserved in wait()
|
||||
// only after the adapter slot is actually held. Background callers can return to
|
||||
// the transport after enqueue and run wait()+execute in a goroutine.
|
||||
func (a *admissionManager) enqueue() (*admissionTicket, error) {
|
||||
t := &admissionTicket{mgr: a}
|
||||
|
||||
granted, ch, err := a.adapter.reserve()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if granted {
|
||||
t.adapterHeld = true
|
||||
} else {
|
||||
t.adapterCh = ch
|
||||
t.queued = true
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// wait blocks until the run holds both the adapter slot and the global slot. The
|
||||
// adapter slot is awaited first; only once it is held is the global slot
|
||||
// reserved and awaited, so a queued run never occupies a global slot it cannot
|
||||
// yet use. The adapter gate's queue timeout (when configured) bounds the total
|
||||
// admission wait across both gates via a derived deadline. On failure it unwinds
|
||||
// any held slot and returns the admissionError.
|
||||
func (t *admissionTicket) wait(ctx context.Context) error {
|
||||
// Derive a single deadline from the adapter queue timeout so the global wait
|
||||
// cannot extend total queue time past the configured policy.
|
||||
if to := t.mgr.adapter.queueTimeout; to > 0 {
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeoutCause(ctx, to, errAdmissionQueueTimeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
if t.adapterCh != nil {
|
||||
if err := t.mgr.adapter.waitOn(ctx, t.adapterCh); err != nil {
|
||||
t.adapterCh = nil
|
||||
t.release()
|
||||
return t.mapDeadline(ctx, err)
|
||||
}
|
||||
t.adapterCh = nil
|
||||
t.adapterHeld = true
|
||||
}
|
||||
|
||||
if t.mgr.global != nil {
|
||||
granted, ch, err := t.mgr.global.reserve()
|
||||
if err != nil {
|
||||
// The global gate has an unbounded queue, so reserve never returns an
|
||||
// overflow error; guard defensively and unwind the adapter slot.
|
||||
t.release()
|
||||
return err
|
||||
}
|
||||
if granted {
|
||||
t.globalHeld = true
|
||||
} else if err := t.mgr.global.waitOn(ctx, ch); err != nil {
|
||||
t.release()
|
||||
return t.mapDeadline(ctx, err)
|
||||
} else {
|
||||
t.globalHeld = true
|
||||
func (a *admissionManager) acquire() (*admissionTicket, error) {
|
||||
if a.adapter != nil {
|
||||
if !a.adapter.acquire() {
|
||||
return nil, newConcurrencyUnavailableError()
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// mapDeadline reclassifies a context-cancellation admission error as a queue
|
||||
// timeout only when the cancellation cause is the provider queue-timeout
|
||||
// deadline. A parent cancellation or a parent deadline (e.g. the run's
|
||||
// RunRequest.timeout_sec) keeps the original concurrency_unavailable reason, so
|
||||
// an unrelated request timeout is never mislabeled as queue_timeout (which would
|
||||
// also wrongly emit queue_timeout=0s when queue_timeout_ms is disabled).
|
||||
func (t *admissionTicket) mapDeadline(ctx context.Context, err error) error {
|
||||
if errors.Is(context.Cause(ctx), errAdmissionQueueTimeout) && admissionRejectReason(err) == reasonConcurrencyUnavailable {
|
||||
return newQueueTimeoutError(t.mgr.adapter.queueTimeout)
|
||||
if a.global != nil {
|
||||
if !a.global.acquire() {
|
||||
if a.adapter != nil {
|
||||
a.adapter.release()
|
||||
}
|
||||
return nil, newConcurrencyUnavailableError()
|
||||
}
|
||||
}
|
||||
return err
|
||||
return &admissionTicket{
|
||||
mgr: a,
|
||||
adapterHeld: a.adapter != nil,
|
||||
globalHeld: a.global != nil,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// release returns any slot the ticket holds. It is safe to call once the ticket
|
||||
// is done (after a completed run) or to unwind a partial reservation on failure.
|
||||
func (t *admissionTicket) release() {
|
||||
if t.globalHeld && t.mgr.global != nil {
|
||||
t.mgr.global.release()
|
||||
t.globalHeld = false
|
||||
}
|
||||
if t.adapterHeld {
|
||||
if t.adapterHeld && t.mgr.adapter != nil {
|
||||
t.mgr.adapter.release()
|
||||
t.adapterHeld = false
|
||||
}
|
||||
}
|
||||
|
||||
// itoa is a tiny strconv.Itoa shim kept local to avoid importing strconv solely
|
||||
// for reason-message formatting.
|
||||
func itoa(n int) string {
|
||||
if n == 0 {
|
||||
return "0"
|
||||
}
|
||||
neg := n < 0
|
||||
if neg {
|
||||
n = -n
|
||||
}
|
||||
var buf [20]byte
|
||||
i := len(buf)
|
||||
for n > 0 {
|
||||
i--
|
||||
buf[i] = byte('0' + n%10)
|
||||
n /= 10
|
||||
}
|
||||
if neg {
|
||||
i--
|
||||
buf[i] = '-'
|
||||
}
|
||||
return string(buf[i:])
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue