Merge remote-tracking branch 'myxce/chatbot' into chatbot
This commit is contained in:
lei.chen
commit
c3acd62a17
14 changed files with 584 additions and 14 deletions
|
|
@ -122,7 +122,7 @@ spec 声称 36/36 任务完成，但 re-audit（research.md）发现缺陷。来
|
|||
| Phase 2：SSE 事件流 | 计划 | apiserver SSE endpoint + EventBroker，schedulerclient `WatchJobs()`，mcp-server 接入。ADR-004 设计完成待实施。估 2-3 天。 |
|
||||
| Phase 3：LSF adapter | 计划 | CLI 解析优先，Python Sidecar 后续。估 +1 月。 |
|
||||
| Phase 4：Slurm + K8s | 计划 | slurmrestd + CLI fallback；client-go。估 +1 月。 |
|
||||
| modify 工具 | 推迟 | apiserver 当前无 modify 端点（control 仅 kill/stop/resume/checkpoint），需新增后端端点。 |
|
||||
| modify 工具 | 推迟 | `scheduler_modify(id, changes)` 全链路空白：controller `job_ctrl` 只有 siginfo+checkpoint 无改作业字段 action → 需 controller 新 opcode → apiserver 新端点 → schedulerclient → mcp 工具（见 §八.4）。kill/stop/resume 工具已有。 |
|
||||
| 未决决策 | 未解决 | LLM 模型选型（Claude API/自建/混合）；多集群 URM 源识别；air-gapped LLM 访问。 |
|
||||
|
||||
---
|
||||
|
|
@ -198,13 +198,15 @@ spec 声称 36/36 任务完成，但 re-audit（research.md）发现缺陷。来
|
|||
- Phase 4：**Slurm**（slurmrestd）+ **K8s**（client-go）（估 +1 月）
|
||||
|
||||
### 4. 推迟 / 未决
|
||||
- `scheduler_modify` 工具推迟——apiserver 当前无 modify 端点（control 仅 kill/stop/resume/checkpoint），需新增后端端点
|
||||
- `scheduler_modify(id, changes)` 工具推迟（MCP 缺口）——整条链路空白：controller `job_ctrl`（src/controller/job.c:957）仅 siginfo（SIGKILL/SIGSTOP/SIGCONT）+ checkpoint，无改作业字段的 action → 需 **controller 新 job-update opcode**（写回 queue/priority/resreq/user_priority + 触发调度器重新评估，大头）→ **apiserver** `POST /jobs/:jobids/modify`（routers.go 现有 create/get/list/output/control/restart，无 modify）+ 字段/状态校验（哪些字段可改、是否仅 PEND）→ **schedulerclient** `ModifyJob`（write.go 仅 Submit/Cancel/Hold/Release）→ **mcp-server** `scheduler_modify` 工具（~30 行，最小）。注：kill/stop/resume 三件套（`scheduler_cancel`/`hold`/`release`）已有；`scheduler_checkpoint` + `scheduler_restart` 已补（2026-08-30，17 工具，复用控制端 checkpoint action 与 `/jobs/:id/restart` 端点，单测通过）。
|
||||
- 3 个未决决策：LLM 模型选型（Claude API/自建/混合）、多集群 URM 来源识别、air-gapped LLM 访问
|
||||
- **PEND reason 分类体系（进行中，2026-08-14 已实施 P0+P1 部分）**：见 `docs/pend-reason-investigation-design.md`。**已完成（commit 4e2ac86）**：
|
||||
- **code 策略已定**：扩展 PREASON_* 枚举到 9 个（BAD_RESREQ/NO_RESOURCE/MATCH_FAIL/LIMIT_GLOBAL/LIMIT_HOST/QUEUE_CLOSED/QUEUE_DRAINING/TIME_WINDOW/HOST_UNAVAIL），capnp UInt32 无限制。
|
||||
- **P0**：main_reason 修复（首个 detail code 设为主因，集群验证 PEND 作业 main_reason 非 0 + reason_name）+ limit 插件 attach（host 限额→LIMIT_HOST+limit 名，全局限额→LIMIT_GLOBAL+limit 名）。**集群验证**：PEND 作业 API 显示 `main_reason:3 main_reason_name:"match_fail"` + 每条 host/res reason 带 reason_name。
|
||||
- **P1 部分**：队列状态（sched_queue_is_ready 非 stub，closed/inactive 队列不参选 + fcfs election attach QUEUE_CLOSED/DRAINING）；worker 过滤（sched_try_alloc_on_host_ 跳过 UNAVAIL/CLOSED worker + HOST_UNAVAIL）。
|
||||
- **剩余**：P1 时间窗（TIME_WINDOW——scheduler 需保留 dispatchwindow 配置 + 自写"now in window"评估，controller 不计算窗口状态）；P2 依赖/license/保留/exclusive/fairshare（机制本身未实现）；MCP `bjobs -p` 等价工具（未做）。
|
||||
- **⚠️ scheduler-framework 合并（2026-08-25）回归掉 P1**：P1 改动点（4e2ac86/ee79179）在旧 `fcfs.c`，框架合并用 `builtin_fcfs` 取代后旧 fcfs.c 成死路径；`resource.c`/`limit.c` 冲突取 theirs 直接丢；merge-fix 只补 dispatchwindow **字段**不补逻辑；验证从未跑。合并后 `sched_queue_is_ready`/`sched_job_is_ready` 是 `return 1` stub（core.c），仅 3 处 preason emit。
|
||||
- **✅ P1 已恢复（2026-08-30）**：core.c 恢复 job 状态门 + queue 状态门；scheduler.c allocator 循环注入 `sched_queue_status_reason`（状态→QUEUE_CLOSED/DRAINING，时间窗→TIME_WINDOW，用 `xce_core_now` 虚拟时钟）；resource.c 恢复 worker 过滤+HOST_UNAVAIL；limit.c 恢复 LIMIT_GLOBAL/HOST attach。**全部 reason attach 走 allocator 线程（bug-1521 安全）**。单测通过（新增 `test_core_ready_gates.c`，scheduler 套件 52/53，唯一失败是既有的 admin-request 校验顺序问题与 P1 无关）。遗留：持续 closed 队列的作业被 picker 跳过（spec 契约）到不了 allocator → 收不到 reason（合并前 #17 后同样如此，非新回归）。集群验证待做。
|
||||
- **剩余（P2，超出 PEND reason 范畴，属 scheduler 特性开发）**：依赖 depend_cond / license / 保留 RsvID / exclusive / fairshare——机制本身在 scheduler 未实现（设计文档：调度器无处理）；机制落地后 reason 自动出现，MCP 侧无需改。已收尾：TIME_WINDOW 已实现（ee79179，2026-08-15）；`scheduler_pend_reasons` 工具（bjobs -p 等价）已上线（67abbfa，2026-08-15）。
|
||||
|
||||
### 5. 设计层加固（文档已写、代码零实现）
|
||||
- **Auto-pilot 模式**：用户显式授权范围内的自主行为
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ const serverInstructions = "XCE scheduler agent — cluster status, not an incid
|
|||
"root causes, do NOT reference \"known bugs\", and do NOT propose diagnostic or repair actions (checking logs, " +
|
||||
"restarting daemons, opening tickets, ssh'ing to nodes, etc.) unless the user explicitly asks for help or " +
|
||||
"remediation. If you believe action may be warranted, say so in at most one line and let the user decide.\n\n" +
|
||||
"Write tools (scheduler_submit/hold/release/cancel) are destructive and require the user's explicit intent."
|
||||
"Write tools (scheduler_submit/cancel/hold/release/checkpoint/restart) are destructive and require the user's explicit intent."
|
||||
|
||||
func main() {
|
||||
baseURL := os.Getenv("XCE_API_URL")
|
||||
|
|
@ -221,6 +221,22 @@ func main() {
|
|||
mcp.WithDestructiveHintAnnotation(true),
|
||||
), app.HandleReleaseJob)
|
||||
|
||||
// ── scheduler_checkpoint ──
|
||||
s.AddTool(mcp.NewTool("scheduler_checkpoint",
|
||||
mcp.WithDescription("Checkpoint a running job by ID (action 'checkpoint', CRIU). Destructive — you will be asked to confirm before execution."),
|
||||
mcp.WithString("id", mcp.Description("Job ID to checkpoint"), mcp.Required()),
|
||||
mcp.WithReadOnlyHintAnnotation(false),
|
||||
mcp.WithDestructiveHintAnnotation(true),
|
||||
), app.HandleCheckpointJob)
|
||||
|
||||
// ── scheduler_restart ──
|
||||
s.AddTool(mcp.NewTool("scheduler_restart",
|
||||
mcp.WithDescription("Restart (re-queue) a finished job as a new job by ID, resuming from its last checkpoint if one exists. Returns the new job ID. Destructive — you will be asked to confirm before execution."),
|
||||
mcp.WithString("id", mcp.Description("Job ID to restart"), mcp.Required()),
|
||||
mcp.WithReadOnlyHintAnnotation(false),
|
||||
mcp.WithDestructiveHintAnnotation(true),
|
||||
), app.HandleRestartJob)
|
||||
|
||||
log.Printf("Starting scheduler-agent MCP server (base-url: %s)", baseURL)
|
||||
if err := mcps.ServeStdio(s); err != nil {
|
||||
log.Fatalf("server error: %v", err)
|
||||
|
|
|
|||
|
|
@ -288,6 +288,29 @@ func (a *App) HandleReleaseJob(ctx context.Context, req mcp.CallToolRequest) (*m
|
|||
return textResultStr(fmt.Sprintf("job %s released (resume accepted)", id)), nil
|
||||
}
|
||||
|
||||
func (a *App) HandleCheckpointJob(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
||||
id := getString(req, "id")
|
||||
if id == "" {
|
||||
return mcp.NewToolResultError("job id is required"), nil
|
||||
}
|
||||
if err := a.client.CheckpointJob(ctx, id); err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("checkpoint job failed: %v", err)), nil
|
||||
}
|
||||
return textResultStr(fmt.Sprintf("job %s checkpoint accepted", id)), nil
|
||||
}
|
||||
|
||||
func (a *App) HandleRestartJob(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
||||
id := getString(req, "id")
|
||||
if id == "" {
|
||||
return mcp.NewToolResultError("job id is required"), nil
|
||||
}
|
||||
res, err := a.client.RestartJob(ctx, id)
|
||||
if err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("restart job failed: %v", err)), nil
|
||||
}
|
||||
return textResult(res)
|
||||
}
|
||||
|
||||
// ── helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
// parseStatuses parses a comma-separated list of URM status strings.
|
||||
|
|
|
|||
|
|
@ -18,12 +18,16 @@ type fakeClient struct {
|
|||
submitSpec *sclient.JobSpec
|
||||
submitErr error
|
||||
|
||||
cancelID string
|
||||
cancelErr error
|
||||
holdID string
|
||||
holdErr error
|
||||
releaseID string
|
||||
releaseErr error
|
||||
cancelID string
|
||||
cancelErr error
|
||||
holdID string
|
||||
holdErr error
|
||||
releaseID string
|
||||
releaseErr error
|
||||
checkpointID string
|
||||
checkpointErr error
|
||||
restartID string
|
||||
restartErr error
|
||||
}
|
||||
|
||||
func (f *fakeClient) SubmitJob(ctx context.Context, spec sclient.JobSpec) (*sclient.SubmitJobResult, error) {
|
||||
|
|
@ -46,6 +50,19 @@ func (f *fakeClient) ReleaseJob(ctx context.Context, id string) error {
|
|||
return f.releaseErr
|
||||
}
|
||||
|
||||
func (f *fakeClient) CheckpointJob(ctx context.Context, id string) error {
|
||||
f.checkpointID = id
|
||||
return f.checkpointErr
|
||||
}
|
||||
|
||||
func (f *fakeClient) RestartJob(ctx context.Context, id string) (*sclient.RestartJobResult, error) {
|
||||
f.restartID = id
|
||||
if f.restartErr != nil {
|
||||
return nil, f.restartErr
|
||||
}
|
||||
return &sclient.RestartJobResult{JobID: 999, QueueName: "normal", Message: "ok"}, nil
|
||||
}
|
||||
|
||||
func newTestApp(client sclient.Client) *App {
|
||||
a := &App{client: client, templates: NewTemplateRegistry()}
|
||||
return a
|
||||
|
|
@ -288,6 +305,92 @@ func TestHandleReleaseJob(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestHandleCheckpointJob(t *testing.T) {
|
||||
fake := &fakeClient{}
|
||||
a := newTestApp(fake)
|
||||
|
||||
res, err := a.HandleCheckpointJob(context.Background(), idRequest("scheduler_checkpoint", "101"))
|
||||
if err != nil {
|
||||
t.Fatalf("HandleCheckpointJob error: %v", err)
|
||||
}
|
||||
if res.IsError {
|
||||
t.Fatalf("expected success, got error: %s", resultText(t, res))
|
||||
}
|
||||
if fake.checkpointID != "101" {
|
||||
t.Errorf("checkpointID = %q, want 101", fake.checkpointID)
|
||||
}
|
||||
if !strings.Contains(resultText(t, res), "checkpoint accepted") {
|
||||
t.Errorf("result should mention checkpoint accepted, got: %s", resultText(t, res))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCheckpointJob_MissingID(t *testing.T) {
|
||||
fake := &fakeClient{}
|
||||
a := newTestApp(fake)
|
||||
|
||||
res, err := a.HandleCheckpointJob(context.Background(), idRequest("scheduler_checkpoint", ""))
|
||||
if err != nil {
|
||||
t.Fatalf("HandleCheckpointJob error: %v", err)
|
||||
}
|
||||
if !res.IsError {
|
||||
t.Fatal("expected error for missing id")
|
||||
}
|
||||
if fake.checkpointID != "" {
|
||||
t.Errorf("client should not be called, got checkpointID = %q", fake.checkpointID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleRestartJob(t *testing.T) {
|
||||
fake := &fakeClient{}
|
||||
a := newTestApp(fake)
|
||||
|
||||
res, err := a.HandleRestartJob(context.Background(), idRequest("scheduler_restart", "123"))
|
||||
if err != nil {
|
||||
t.Fatalf("HandleRestartJob error: %v", err)
|
||||
}
|
||||
if res.IsError {
|
||||
t.Fatalf("expected success, got error: %s", resultText(t, res))
|
||||
}
|
||||
if fake.restartID != "123" {
|
||||
t.Errorf("restartID = %q, want 123", fake.restartID)
|
||||
}
|
||||
if !strings.Contains(resultText(t, res), `"jobid": 999`) {
|
||||
t.Errorf("result should include the new job id 999, got: %s", resultText(t, res))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleRestartJob_ClientError(t *testing.T) {
|
||||
fake := &fakeClient{restartErr: &fakeError{msg: "restart failed"}}
|
||||
a := newTestApp(fake)
|
||||
|
||||
res, err := a.HandleRestartJob(context.Background(), idRequest("scheduler_restart", "123"))
|
||||
if err != nil {
|
||||
t.Fatalf("HandleRestartJob error: %v", err)
|
||||
}
|
||||
if !res.IsError {
|
||||
t.Fatal("expected error result when client fails")
|
||||
}
|
||||
if !strings.Contains(resultText(t, res), "restart failed") {
|
||||
t.Errorf("result should mention the client error, got: %s", resultText(t, res))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleRestartJob_MissingID(t *testing.T) {
|
||||
fake := &fakeClient{}
|
||||
a := newTestApp(fake)
|
||||
|
||||
res, err := a.HandleRestartJob(context.Background(), idRequest("scheduler_restart", ""))
|
||||
if err != nil {
|
||||
t.Fatalf("HandleRestartJob error: %v", err)
|
||||
}
|
||||
if !res.IsError {
|
||||
t.Fatal("expected error for missing id")
|
||||
}
|
||||
if fake.restartID != "" {
|
||||
t.Errorf("client should not be called, got restartID = %q", fake.restartID)
|
||||
}
|
||||
}
|
||||
|
||||
// fakeError is a minimal error for tests.
|
||||
type fakeError struct{ msg string }
|
||||
|
||||
|
|
|
|||
|
|
@ -51,6 +51,13 @@ type Client interface {
|
|||
// ReleaseJob resumes a suspended job (action "resume", default SIGCONT).
|
||||
ReleaseJob(ctx context.Context, id string) error
|
||||
|
||||
// CheckpointJob checkpoints a running job (action "checkpoint", CRIU).
|
||||
CheckpointJob(ctx context.Context, id string) error
|
||||
|
||||
// RestartJob re-queues a finished job as a new job (resuming from its last
|
||||
// checkpoint when one exists) and returns the new job's identity.
|
||||
RestartJob(ctx context.Context, id string) (*RestartJobResult, error)
|
||||
|
||||
// ── Event stream (Phase 2) ──
|
||||
|
||||
// WatchJobs subscribes to the live job-event SSE stream and returns a
|
||||
|
|
|
|||
|
|
@ -51,6 +51,12 @@ type JobControlRequest struct {
|
|||
Method string `json:"method,omitempty"`
|
||||
}
|
||||
|
||||
// JobRestartRequest is the body sent to POST /xce/v1/jobs/:jobids/restart.
|
||||
// All fields are optional; an empty request restarts to any eligible host.
|
||||
type JobRestartRequest struct {
|
||||
AskedHosts string `json:"asked_hosts,omitempty"`
|
||||
}
|
||||
|
||||
// ControlResult is the apiserver's per-item control response.
|
||||
type ControlResult struct {
|
||||
Status string `json:"status"` // accepted | failed | partially_accepted | no_items_processed
|
||||
|
|
@ -58,6 +64,15 @@ type ControlResult struct {
|
|||
Failed []ControlItem `json:"failed"`
|
||||
}
|
||||
|
||||
// RestartJobResult is the apiserver's response to a job restart — a new job
|
||||
// is created (optionally resumed from the old job's last checkpoint).
|
||||
type RestartJobResult struct {
|
||||
JobID int64 `json:"jobid"`
|
||||
JobIDHuman string `json:"job_id_human"`
|
||||
QueueName string `json:"queue_name"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
// ControlItem reports the outcome for one job (or array element).
|
||||
type ControlItem struct {
|
||||
UniqueJobID int64 `json:"unique_job_id"`
|
||||
|
|
@ -223,3 +238,29 @@ func (hc *HTTPClient) HoldJob(ctx context.Context, id string) error {
|
|||
func (hc *HTTPClient) ReleaseJob(ctx context.Context, id string) error {
|
||||
return hc.controlJob(ctx, id, "resume")
|
||||
}
|
||||
|
||||
// CheckpointJob checkpoints a running job (action "checkpoint", CRIU).
|
||||
func (hc *HTTPClient) CheckpointJob(ctx context.Context, id string) error {
|
||||
return hc.controlJob(ctx, id, "checkpoint")
|
||||
}
|
||||
|
||||
// RestartJob re-queues a finished job as a new job, resuming from its last
|
||||
// checkpoint when one exists. Returns the new job's identity. The apiserver
|
||||
// backs this with the same submit path, so a restart is accepted with 201.
|
||||
func (hc *HTTPClient) RestartJob(ctx context.Context, id string) (*RestartJobResult, error) {
|
||||
if strings.TrimSpace(id) == "" {
|
||||
return nil, fmt.Errorf("restart job: job id is required")
|
||||
}
|
||||
|
||||
url := hc.baseURL + "/xce/v1/jobs/" + id + "/restart"
|
||||
env, err := hc.doPost(ctx, url, JobRestartRequest{}, http.StatusCreated)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("restart job %s: %w", id, err)
|
||||
}
|
||||
|
||||
var res RestartJobResult
|
||||
if err := json.Unmarshal(env.Data, &res); err != nil {
|
||||
return nil, fmt.Errorf("restart job %s: parse result: %w", id, err)
|
||||
}
|
||||
return &res, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -282,6 +282,85 @@ func TestReleaseJob_SendsResume(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// ── Checkpoint / Restart ──────────────────────────────────────────────────────
|
||||
|
||||
func TestCheckpointJob_SendsCheckpoint(t *testing.T) {
|
||||
hc, capture := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
io.WriteString(w, `{"code":202,"msg":"accepted","data":{"status":"accepted","accepted":[],"failed":[]}}`)
|
||||
})
|
||||
|
||||
if err := hc.CheckpointJob(context.Background(), "101"); err != nil {
|
||||
t.Fatalf("CheckpointJob returned error: %v", err)
|
||||
}
|
||||
|
||||
_, path, _, body := capture()
|
||||
if path != "/xce/v1/jobs/101/control" {
|
||||
t.Errorf("path = %q, want /xce/v1/jobs/101/control", path)
|
||||
}
|
||||
if strings.TrimSpace(body) != `{"action":"checkpoint"}` {
|
||||
t.Errorf("body = %s, want {\"action\":\"checkpoint\"}", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckpointJob_StatusFailed(t *testing.T) {
|
||||
hc, _ := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
io.WriteString(w, `{"code":202,"msg":"failed","data":{"status":"failed","accepted":[],"failed":[{"unique_job_id":101,"job_id":101,"array_idx":0,"msg":"job 101 is in state DONE, cannot be checkpointed"}]}}`)
|
||||
})
|
||||
|
||||
err := hc.CheckpointJob(context.Background(), "101")
|
||||
if err == nil {
|
||||
t.Fatal("expected error when control status is failed")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "101") {
|
||||
t.Errorf("error = %q, want to mention job id 101", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRestartJob_Success(t *testing.T) {
|
||||
hc, capture := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
io.WriteString(w, `{"code":201,"msg":"success","data":{"jobid":999,"queue_name":"normal","message":"Job <123> will be restarted with new job <999> to default queue <normal>."}}`)
|
||||
})
|
||||
|
||||
res, err := hc.RestartJob(context.Background(), "123")
|
||||
if err != nil {
|
||||
t.Fatalf("RestartJob returned error: %v", err)
|
||||
}
|
||||
|
||||
method, path, _, body := capture()
|
||||
if method != http.MethodPost {
|
||||
t.Errorf("method = %q, want POST", method)
|
||||
}
|
||||
if path != "/xce/v1/jobs/123/restart" {
|
||||
t.Errorf("path = %q, want /xce/v1/jobs/123/restart", path)
|
||||
}
|
||||
if strings.TrimSpace(body) != "{}" {
|
||||
t.Errorf("body = %s, want {}", body)
|
||||
}
|
||||
if res.JobID != 999 {
|
||||
t.Errorf("JobID = %d, want 999", res.JobID)
|
||||
}
|
||||
if res.QueueName != "normal" {
|
||||
t.Errorf("QueueName = %q, want normal", res.QueueName)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRestartJob_EmptyID(t *testing.T) {
|
||||
hc, _ := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Error("server should not be called when id is empty")
|
||||
})
|
||||
|
||||
_, err := hc.RestartJob(context.Background(), " ")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for empty id")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "job id is required") {
|
||||
t.Errorf("error = %q, want 'job id is required'", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ── conversion helpers ───────────────────────────────────────────────────────
|
||||
|
||||
func TestFormatWalltimeSeconds(t *testing.T) {
|
||||
|
|
|
|||
|
|
@ -432,8 +432,41 @@ xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name)
|
|||
return make_sched_project(core, name);
|
||||
}
|
||||
|
||||
/* FIXME this is just stub, we need to add more */
|
||||
int sched_job_is_ready(xce_sched_job_t* j) { return 1; }
|
||||
/* A job is electable only while it is PENDING and not suspended. Once
|
||||
* dispatched (RUN), finished (EXIT/DONE), or suspended (PSUSP/USUSP/SSUSP)
|
||||
* it must never be re-elected — bug-1498 (restored from the pre-framework
|
||||
* scheduler.c after the scheduler-framework merge dropped it). */
|
||||
int sched_job_is_ready(xce_sched_job_t* j)
|
||||
{
|
||||
if (j == NULL) {
|
||||
return 0;
|
||||
}
|
||||
if ((j->state & JOB_STAT_PEND) == 0) {
|
||||
return 0; /* already dispatched/running or terminal */
|
||||
}
|
||||
if (j->state & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
|
||||
return 0; /* suspended — not electable */
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* FIXME this is just stub, we need to add more */
|
||||
int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }
|
||||
/* A queue is electable only while it accepts new jobs. The controller
|
||||
* maintains q->status (closed / inactivated-by-admin / inactive-by-window).
|
||||
* The dispatch window (q->dispatchwindow, from queues.yaml) is NOT evaluated
|
||||
* here: core.c is isolated from wall-clock time (test_core_isolation.c) and
|
||||
* the window can only be judged against the core's virtual clock
|
||||
* (xce_core_now), which this signature has no access to. Windowed-out queues
|
||||
* therefore pass this gate and are re-pended with PREASON_TIME_WINDOW by the
|
||||
* allocator (scheduler.c, where xce_core_now() is reachable) — same net
|
||||
* effect, with a safe (allocator-thread) reason attach. PEND-reason P1. */
|
||||
int sched_queue_is_ready(xce_sched_queue_t* q)
|
||||
{
|
||||
if (q == NULL) {
|
||||
return 0;
|
||||
}
|
||||
if (q->status
|
||||
& (QUEUE_STAT_CLOSED | QUEUE_STAT_INACT_ADMIN | QUEUE_STAT_INACT_WINDOW)) {
|
||||
return 0;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -284,6 +284,15 @@ static xce_sched_queue_t* core_build_queue(xce_queue_t* q)
|
|||
queue->priority = q->priority;
|
||||
queue->accept_interval = q->accept_interval;
|
||||
queue->sched_delay = q->sched_delay;
|
||||
/* PEND-reason P1 fix (2026-08-30): the scheduler-framework merge
|
||||
* dropped the dispatch-window config here — make_sched_queue (schedobj.c)
|
||||
* duplicates q->dispatchwindow_config, but this framework core path did
|
||||
* not, so the TIME_WINDOW PEND reason could never fire on the live
|
||||
* default path. Mirror make_sched_queue: dup the window (freed by
|
||||
* kill_sched_queue via core->queues' datakiller). */
|
||||
if (q->dispatchwindow_config != NULL) {
|
||||
queue->dispatchwindow = dup_timewindow_config(q->dispatchwindow_config);
|
||||
}
|
||||
/* Same scope note as make_sched_queue's own (schedobj.c): project_name/
|
||||
* job_pick_plugin have no wire-config source yet, and (matching
|
||||
* existing behavior) resreq is not populated here either -- wiring
|
||||
|
|
@ -336,6 +345,15 @@ static void core_upsert_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker
|
|||
FREEUP(q->resreq);
|
||||
q->resreq = makeString(updated->resreq);
|
||||
}
|
||||
|
||||
/* PEND-reason P1 fix (2026-08-30): same merge regression as
|
||||
* core_build_queue above — the update branch never carried the
|
||||
* dispatch window. Refresh it so a QUEUE_UPDATE that changes the
|
||||
* window (or a re-push after the initial build) takes effect. */
|
||||
if (updated->dispatchwindow_config != NULL) {
|
||||
free_timewindow_config(q->dispatchwindow);
|
||||
q->dispatchwindow = dup_timewindow_config(updated->dispatchwindow_config);
|
||||
}
|
||||
}
|
||||
|
||||
if (picker != NULL) {
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@
|
|||
|
||||
#include "kvmgr.h"
|
||||
#include "plugin.h"
|
||||
#include "preason.h"
|
||||
#include "sched/types.h"
|
||||
#include "scheduler.h"
|
||||
#include "utils.h"
|
||||
|
|
@ -208,6 +209,17 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin,
|
|||
TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__,
|
||||
limit->name, obj->user, obj->queue, obj->project, rc);
|
||||
if (rc <= 0) {
|
||||
/* global quota hit — report the specific limit (PEND-reason P1,
|
||||
* restored from 4e2ac86). The election return value is ignored by
|
||||
* the main loop, so the reason is carried on ctx->reason and moved
|
||||
* to obj->reason by limit_job_election. */
|
||||
if (ctx->reason == NULL) {
|
||||
ctx->reason = make_pendreason(0);
|
||||
}
|
||||
if (ctx->reason != NULL) {
|
||||
add_job_preason(ctx->reason->reasons, PREASON_LIMIT_GLOBAL, NULL,
|
||||
limit->name);
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
|
|
@ -299,6 +311,14 @@ static int limit_job_election(void* arg, xce_sched_job_t* job)
|
|||
|
||||
iter_destroy(&it);
|
||||
|
||||
/* Carry any global-quota reason (PREASON_LIMIT_GLOBAL) attached by
|
||||
* match_job_global_limit_ onto the job so it survives to the PEND
|
||||
* decision. The election's rc itself is ignored by the main loop.
|
||||
* (PEND-reason P1, restored from 4e2ac86.) */
|
||||
if (ctx.reason != NULL) {
|
||||
MOVE_PTR(obj->reason, ctx.reason);
|
||||
}
|
||||
|
||||
obj->configver = sched->core.configver;
|
||||
|
||||
return rc;
|
||||
|
|
@ -358,6 +378,16 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin,
|
|||
TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__,
|
||||
limit->name, obj->user, obj->queue, obj->project, rc);
|
||||
if (rc <= 0) {
|
||||
/* host-based quota hit — report the specific limit (PEND-reason P1,
|
||||
* restored from 4e2ac86; previously the block was silent and the job
|
||||
* just stayed PEND). */
|
||||
if (ctx->reason == NULL) {
|
||||
ctx->reason = make_pendreason(0);
|
||||
}
|
||||
if (ctx->reason != NULL) {
|
||||
add_job_preason(ctx->reason->reasons, PREASON_LIMIT_HOST,
|
||||
worker->name, limit->name);
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -484,6 +484,23 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
|
|||
int rc;
|
||||
xvec_t* uncomit = NULL;
|
||||
|
||||
/* Worker down/closed — never eligible. Attach HOST_UNAVAIL instead of
|
||||
* trying to allocate on it (PEND-reason P1, E17 — restored from
|
||||
* 4e2ac86 after the scheduler-framework merge dropped it). Runs in the
|
||||
* allocator thread (sched_try_alloc_on_hosts), so ctx->reason is safe
|
||||
* to touch here. */
|
||||
if (w->status
|
||||
& (XCE_WORKER_UNAVAIL | XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN)) {
|
||||
if (ctx->reason == NULL) {
|
||||
ctx->reason = make_pendreason(0);
|
||||
}
|
||||
if (ctx->reason != NULL) {
|
||||
add_job_preason(ctx->reason->reasons, PREASON_HOST_UNAVAIL,
|
||||
w->name, NULL);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
rc = sched_meet_host_selection(sched, w, primary->select);
|
||||
if (rc == 0) {
|
||||
if (ctx->reason == NULL) {
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@
|
|||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
|
||||
#include "client.h"
|
||||
#include "controller.capnp.h"
|
||||
|
|
@ -266,6 +267,39 @@ static int sched_make_alter_allocation(xce_scheduler_t* sched,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* Map a queue to the PEND reason it should report (PEND-reason P1 restore,
|
||||
* resurrected from the pre-framework scheduler.c). Status flags take
|
||||
* precedence, then the dispatch window. The window is evaluated against the
|
||||
* core's virtual clock (xce_core_now, /1e9 → seconds) so live and sim modes
|
||||
* agree; a NULL/empty window is always open. Runs only from the allocator
|
||||
* thread (see the gate at the top of the allocator loop), so its reason
|
||||
* attachment cannot race MOVE_PTR(obj->reason, ...) (bug-1521). */
|
||||
static int sched_queue_status_reason(xce_scheduler_t* sched,
|
||||
xce_sched_queue_t* q)
|
||||
{
|
||||
if (q == NULL) {
|
||||
return 0;
|
||||
}
|
||||
if (q->status & QUEUE_STAT_CLOSED) {
|
||||
return PREASON_QUEUE_CLOSED;
|
||||
}
|
||||
if (q->status & QUEUE_STAT_INACT_ADMIN) {
|
||||
return PREASON_QUEUE_DRAINING;
|
||||
}
|
||||
if (q->status & QUEUE_STAT_INACT_WINDOW) {
|
||||
return PREASON_TIME_WINDOW;
|
||||
}
|
||||
if (q->dispatchwindow != NULL) {
|
||||
uint64_t now_ns = xce_core_now(&(sched->core));
|
||||
|
||||
if (!xce_timewindow_is_open(q->dispatchwindow,
|
||||
(time_t)(now_ns / 1000000000ULL))) {
|
||||
return PREASON_TIME_WINDOW;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_obj_t* obj)
|
||||
{
|
||||
int rc = 0;
|
||||
|
|
@ -326,6 +360,7 @@ static void* sched_allocator(void* arg)
|
|||
xce_decision_t* decision = NULL;
|
||||
xce_sched_job_t* job = NULL;
|
||||
int alloc_rc = 0;
|
||||
int qr = 0;
|
||||
|
||||
pthread_testcancel();
|
||||
|
||||
|
|
@ -346,6 +381,35 @@ static void* sched_allocator(void* arg)
|
|||
|
||||
plugin_job_election(sched, job);
|
||||
|
||||
/* PEND-reason P1 restore (bug-1521-safe): a job whose queue is not
|
||||
* ready (closed / drained / outside its dispatch window) must not be
|
||||
* dispatched. The gate and reason attach live HERE (allocator thread)
|
||||
* rather than in the picker: the picker thread writing
|
||||
* job->schedobj->reason races this thread's MOVE_PTR(obj->reason, ...)
|
||||
* (bug-1521 heap corruption). The picker still skips persistently-
|
||||
* closed queues per its spec contract, so this path fires for queues
|
||||
* that close or window-out between pick and allocation, and for
|
||||
* windowed-out queues that pass the picker's status-only gate. */
|
||||
qr = sched_queue_status_reason(sched, job->queue);
|
||||
if (qr != 0) {
|
||||
xce_pendreason_t* qreason = make_pendreason(0);
|
||||
|
||||
job_picker->picker_iface->on_pend(job_picker, jobid);
|
||||
|
||||
if (qreason != NULL) {
|
||||
qreason->jobid = job->jobid;
|
||||
add_job_preason(qreason->reasons, qr,
|
||||
job->queue != NULL ? job->queue->name : NULL,
|
||||
NULL);
|
||||
decision = make_pend_decision(qreason);
|
||||
}
|
||||
if (decision != NULL) {
|
||||
commit_decision(sched, decision);
|
||||
cbuf_put(&(sched->core.decisions), &decision);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
/* calculate decision for jobid */
|
||||
alloc_rc = sched_make_allocation(sched, job->schedobj);
|
||||
if (alloc_rc < 0) {
|
||||
|
|
|
|||
|
|
@ -264,6 +264,19 @@ sched_core_jbuf_capacity_pin_test = executable(
|
|||
+ '"'],
|
||||
install: false)
|
||||
|
||||
# Regression guard for the PEND-reason P1 readiness gates restored after
|
||||
# the scheduler-framework merge dropped them (sched_job_is_ready /
|
||||
# sched_queue_is_ready were `return 1` stubs in core.c). Same white-box
|
||||
# #include pattern as test_core_jbuf_capacity_pin.c (pulls the real
|
||||
# ../src/core.c into this TU); exercises the REAL gate functions, not a
|
||||
# re-implementation.
|
||||
sched_core_ready_gates_test = executable(
|
||||
'sched_core_ready_gates_test',
|
||||
['test_core_ready_gates.c'],
|
||||
include_directories:[incdir],
|
||||
dependencies: [scheduler_test_deps],
|
||||
install: false)
|
||||
|
||||
# Contract test for the core event interface and batch event push
|
||||
# (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8): xce_core_push_events()
|
||||
# fires the resolved picker's corresponding state-change hook exactly once
|
||||
|
|
@ -1235,6 +1248,9 @@ test('scheduler no bare printf test',
|
|||
test('scheduler core jbuf capacity pin test',
|
||||
sched_core_jbuf_capacity_pin_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
test('scheduler core ready gates test',
|
||||
sched_core_ready_gates_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
test('scheduler core events test',
|
||||
sched_core_events_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
|
|
|
|||
121
src/scheduler/testsuite/test_core_ready_gates.c
Normal file
121
src/scheduler/testsuite/test_core_ready_gates.c
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
/*
|
||||
* test_core_ready_gates.c - contract test for the PEND-reason P1 readiness
|
||||
* gates restored after the scheduler-framework merge dropped them
|
||||
* (sched_job_is_ready / sched_queue_is_ready were `return 1` stubs in
|
||||
* core.c; the pre-framework scheduler.c implementations were lost).
|
||||
*
|
||||
* Observable under test:
|
||||
* - sched_job_is_ready(): a job is electable only while PENDING and not
|
||||
* suspended. NULL / RUN / EXIT / DONE / PSUSP / USUSP / SSUSP -> 0.
|
||||
* PEND -> 1. (bug-1498: a running job lingering in the pendJobs queue
|
||||
* must never be re-elected and re-dispatched.)
|
||||
* - sched_queue_is_ready(): a queue is electable only while it accepts
|
||||
* new jobs. NULL / QUEUE_STAT_CLOSED / QUEUE_STAT_INACT_ADMIN /
|
||||
* QUEUE_STAT_INACT_WINDOW -> 0. Status 0 -> 1. The dispatch-window
|
||||
* half of the gate intentionally lives in the allocator (scheduler.c,
|
||||
* xce_core_now) not here: core.c is isolated from wall-clock time, and
|
||||
* the queue has no back-pointer to the core (see the function's own
|
||||
* comment in core.c).
|
||||
*
|
||||
* White-box #include of ../src/core.c (same convention as
|
||||
* test_core_jbuf_capacity_pin.c): these tests exercise the REAL core.c
|
||||
* functions, not a re-implementation. Stub surface is exactly the seven
|
||||
* kvmgr datakillers/plugin_unregister core.c references as addresses in
|
||||
* xce_sched_core_init() (never actually invoked -- fixtures aren't torn
|
||||
* down, same convention as the sibling core tests).
|
||||
*/
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "sched/types.h"
|
||||
#include "xce/types.h"
|
||||
|
||||
#include "../src/core.c"
|
||||
|
||||
/* Linker stubs for core.c's extern datakillers (never invoked). */
|
||||
void kill_sched_job(void* arg) { (void)arg; }
|
||||
void kill_sched_queue(void* arg) { (void)arg; }
|
||||
void kill_sched_worker(void* arg) { (void)arg; }
|
||||
void kill_sched_user(void* arg) { (void)arg; }
|
||||
void kill_sched_project(void* arg) { (void)arg; }
|
||||
void kill_sched_resval(xce_val_t* val) { (void)val; }
|
||||
void plugin_unregister(void* arg) { (void)arg; }
|
||||
|
||||
void setUp(void) { return; }
|
||||
void tearDown(void) { return; }
|
||||
|
||||
static void test_job_is_ready_rejects_null(void)
|
||||
{
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(NULL));
|
||||
}
|
||||
|
||||
static void test_job_is_ready_accepts_pend(void)
|
||||
{
|
||||
xce_sched_job_t j = { .state = JOB_STAT_PEND };
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(1, sched_job_is_ready(&j));
|
||||
}
|
||||
|
||||
static void test_job_is_ready_rejects_non_pend(void)
|
||||
{
|
||||
const uint32_t states[] = { 0, JOB_STAT_RUN, JOB_STAT_DONE, JOB_STAT_EXIT };
|
||||
size_t i;
|
||||
|
||||
for (i = 0; i < sizeof(states) / sizeof(states[0]); i++) {
|
||||
xce_sched_job_t j = { .state = states[i] };
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, sched_job_is_ready(&j),
|
||||
"non-PEND job must not be electable");
|
||||
}
|
||||
}
|
||||
|
||||
static void test_job_is_ready_rejects_suspended(void)
|
||||
{
|
||||
const uint32_t susp[] = { JOB_STAT_PSUSP, JOB_STAT_USUSP, JOB_STAT_SSUSP };
|
||||
size_t i;
|
||||
|
||||
for (i = 0; i < sizeof(susp) / sizeof(susp[0]); i++) {
|
||||
xce_sched_job_t j = { .state = JOB_STAT_PEND | susp[i] };
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, sched_job_is_ready(&j),
|
||||
"suspended job must not be electable");
|
||||
}
|
||||
}
|
||||
|
||||
static void test_queue_is_ready_rejects_null(void)
|
||||
{
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_queue_is_ready(NULL));
|
||||
}
|
||||
|
||||
static void test_queue_is_ready_accepts_open(void)
|
||||
{
|
||||
xce_sched_queue_t q = { .status = 0 };
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(1, sched_queue_is_ready(&q));
|
||||
}
|
||||
|
||||
static void test_queue_is_ready_rejects_closed_or_inactive(void)
|
||||
{
|
||||
const uint32_t statuses[] = { QUEUE_STAT_CLOSED, QUEUE_STAT_INACT_ADMIN,
|
||||
QUEUE_STAT_INACT_WINDOW };
|
||||
size_t i;
|
||||
|
||||
for (i = 0; i < sizeof(statuses) / sizeof(statuses[0]); i++) {
|
||||
xce_sched_queue_t q = { .status = statuses[i] };
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, sched_queue_is_ready(&q),
|
||||
"closed/inactive queue must not be electable");
|
||||
}
|
||||
}
|
||||
|
||||
int main(void)
|
||||
{
|
||||
UNITY_BEGIN();
|
||||
RUN_TEST(test_job_is_ready_rejects_null);
|
||||
RUN_TEST(test_job_is_ready_accepts_pend);
|
||||
RUN_TEST(test_job_is_ready_rejects_non_pend);
|
||||
RUN_TEST(test_job_is_ready_rejects_suspended);
|
||||
RUN_TEST(test_queue_is_ready_rejects_null);
|
||||
RUN_TEST(test_queue_is_ready_accepts_open);
|
||||
RUN_TEST(test_queue_is_ready_rejects_closed_or_inactive);
|
||||
return UNITY_END();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue