diff --git a/src/controller/commit.c b/src/controller/commit.c index e49dde14..9b40368a 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -749,13 +749,23 @@ int commit_submit_async(commit_engine_t* ce, uint64_t lsn, ce->continuations[slot].fn = fn; ce->continuations[slot].ctx = ctx; /* raft-apply-decoupling task 1.3: arm this registration's timeout with - * the SAME deadline computation and SAME default constant - * commit_submit's blocking WAIT path uses, so an async registration - * observes identical failure timing to today's blocking path (Req - * 11.1). The dedicated timeout-sweep thread (commit_sweep_worker) - * resolves this continuation FAILED/-ETIMEDOUT if it is still pending - * once this deadline passes. */ - wait_deadline(&ce->continuations[slot].deadline, DEFAULT_SYNC_TIMEOUT_MS); + * the SAME deadline computation and SAME configured timeout the blocking + * paths use, so an async registration observes identical failure timing + * to the blocking path (Req 11.1). The configured sync deadline -- not a + * hardcoded constant -- is the source of truth: the blocking + * persist_commit_barrier borrows p->config.sync_timeout_ms when its + * caller passes timeout_ms == 0 (persist.c), and this async path honors + * the same config here. The sweep WARN below reports the same value. + * A zeroed config (unit fixtures memset()ing the persist_t, no + * persist_init) falls back to DEFAULT_SYNC_TIMEOUT_MS, mirroring + * persist_commit_barrier's timeout_ms == 0 convention. The dedicated + * timeout-sweep thread (commit_sweep_worker) resolves this continuation + * FAILED/-ETIMEDOUT if it is still pending once this deadline passes. */ + { + uint32_t tmo = ce->p->config.sync_timeout_ms; + wait_deadline(&ce->continuations[slot].deadline, + tmo != 0 ? tmo : DEFAULT_SYNC_TIMEOUT_MS); + } /* Wake the sweep thread: a freshly armed deadline may be sooner than * whatever it is currently sleeping toward. */ @@ -1106,6 +1116,16 @@ static void* commit_sweep_worker(void* arg) &next_deadline, &have_next); if (nc > 0 || nf > 0) { + /* Same effective timeout commit_submit_async armed the deadline + * with: the configured sync timeout, DEFAULT only as the + * zeroed-config (unit fixture) fallback. Read under ce->lock — + * persist config is write-once at init in practice, but the + * read is already inside a lock hold that guarantees the + * engine outlives it. */ + uint32_t tmo = ce->p->config.sync_timeout_ms; + if (tmo == 0) { + tmo = DEFAULT_SYNC_TIMEOUT_MS; + } pthread_mutex_unlock(&ce->lock); for (int i = 0; i < nc; i++) { commit_resolve(ce, committed_ready[i], COMMIT_OUTCOME_COMMITTED, 0); @@ -1115,7 +1135,7 @@ static void* commit_sweep_worker(void* arg) "commit (lsn=%lu committed_lsn=%lu writable_lsn=%lu " "timeout_ms=%u)", failed_ready[i], atomic_load(&ce->p->committed_lsn), - ce->writable_lsn, (unsigned)DEFAULT_SYNC_TIMEOUT_MS); + ce->writable_lsn, (unsigned)tmo); commit_resolve(ce, failed_ready[i], COMMIT_OUTCOME_FAILED, -ETIMEDOUT); } diff --git a/src/controller/include/commit.h b/src/controller/include/commit.h index 9857c466..57dbf686 100644 --- a/src/controller/include/commit.h +++ b/src/controller/include/commit.h @@ -51,9 +51,10 @@ typedef struct commit_continuation { /* raft-apply-decoupling task 1.3: absolute CLOCK_REALTIME deadline this * registration must resolve by if majority is never reached -- computed * at registration time via the SAME wait_deadline() helper and the SAME - * DEFAULT_SYNC_TIMEOUT_MS constant commit_submit's blocking WAIT path - * uses, so an async registration observes identical failure timing to - * today's blocking path (Req 11.1). Only meaningful while lsn != 0; + * configured sync timeout (p->config.sync_timeout_ms; + * DEFAULT_SYNC_TIMEOUT_MS as the zeroed-config fallback) the blocking + * paths use, so an async registration observes identical failure timing + * to the blocking path (Req 11.1). Only meaningful while lsn != 0; * overwritten on every fresh registration of a reused slot. Swept by the * commit engine's dedicated timeout-sweep thread (see commit.c). */ struct timespec deadline; @@ -170,8 +171,9 @@ void commit_blocking_release(commit_engine_t* ce); * outcome is known. * * raft-apply-decoupling task 1.3: registration also arms an implicit - * DEFAULT_SYNC_TIMEOUT_MS deadline (identical constant and clock the - * blocking WAIT path uses) -- if lsn has not resolved by then, the commit + * deadline -- the SAME configured sync timeout (p->config.sync_timeout_ms, + * DEFAULT_SYNC_TIMEOUT_MS only as the zeroed-config fallback) and clock the + * blocking WAIT path uses -- if lsn has not resolved by then, the commit * engine's own timeout-sweep thread resolves it FAILED/-ETIMEDOUT with no * caller action required. A genuine majority ack that lands at or after the * deadline still wins: the sweep thread always checks the majority diff --git a/src/frontend/apiserver/controllers/arraystats.go b/src/frontend/apiserver/controllers/arraystats.go index 92162368..31a01b4b 100644 --- a/src/frontend/apiserver/controllers/arraystats.go +++ b/src/frontend/apiserver/controllers/arraystats.go @@ -24,6 +24,15 @@ import ( // arrayAgg accumulates one array's counters while scanning store.Jobs. type arrayAgg struct { info models.ArrayJobInfo + // carrierUID is the element row currently supplying info's identity + // fields. Identity selection must be deterministic: store.Jobs.List() + // walks a Go map, whose order is randomized per call, so "first element + // seen" would make the row -- and the stats identity -- flip run to run + // whenever element rows disagree. The lowest Unique_JobID (the array + // head, element 1) wins instead: every element of one array carries the + // same submit spec in production, so this only matters for the + // disagreement case, and there it always answers the same way. + carrierUID uint64 } // deriveArrayJobStats scans store.Jobs and aggregates every array element row @@ -39,41 +48,30 @@ func deriveArrayJobStats() []models.ArrayJobInfo { base := int64(backend.ExtractArrayBaseID(raw)) agg, ok := aggs[base] if !ok { - agg = &arrayAgg{info: models.ArrayJobInfo{ - JobID: base, - ArrayJobName: job.JobName, - SubmitUser: job.SubmitUser, - QueueName: job.QueueName, - UserGroup: job.UserGroup, - JobGroup: job.JobGroup, - JobDescription: job.JobDescription, - JobName: job.JobName, - ProjectName: job.ProjectName, - LicenseProject: job.LicenseProject, - SLA: job.SLA, - }} + agg = &arrayAgg{info: models.ArrayJobInfo{JobID: base}} aggs[base] = agg } - agg.info.NumJobs++ - // Classify through JobStateStr so the counters always agree with the - // status string every other view shows (suspend states are stored as - // RUN|USUSP / RUN|SSUSP / PEND|PSUSP combinations there). - switch models.JobStateStr(job.JobStatus) { - case "USUSP": - agg.info.NumUSusp++ - case "SSUSP": - agg.info.NumSSusp++ - case "PSUSP": - agg.info.NumPSusp++ - case "RUN": - agg.info.NumRun++ - case "PEND": - agg.info.NumPend++ - case "EXIT": - agg.info.NumExit++ - case "DONE": - agg.info.NumDone++ + if ok && raw >= agg.carrierUID { + // A non-head element: counters only, identity untouched. + agg.info.NumJobs++ + aggCountStatus(agg, job.JobStatus) + continue } + // First-seen (new agg) or a strictly lower-ID element: (re)supply + // the identity from this row. + agg.carrierUID = raw + agg.info.ArrayJobName = job.JobName + agg.info.SubmitUser = job.SubmitUser + agg.info.QueueName = job.QueueName + agg.info.UserGroup = job.UserGroup + agg.info.JobGroup = job.JobGroup + agg.info.JobDescription = job.JobDescription + agg.info.JobName = job.JobName + agg.info.ProjectName = job.ProjectName + agg.info.LicenseProject = job.LicenseProject + agg.info.SLA = job.SLA + agg.info.NumJobs++ + aggCountStatus(agg, job.JobStatus) } out := make([]models.ArrayJobInfo, 0, len(aggs)) @@ -84,6 +82,30 @@ func deriveArrayJobStats() []models.ArrayJobInfo { return out } +// aggCountStatus classifies one element row into the aggregate's counters +// (shared by the identity-carrier and counter-only branches above). +func aggCountStatus(agg *arrayAgg, status int) { + // Classify through JobStateStr so the counters always agree with the + // status string every other view shows (suspend states are stored as + // RUN|USUSP / RUN|SSUSP / PEND|PSUSP combinations there). + switch models.JobStateStr(status) { + case "USUSP": + agg.info.NumUSusp++ + case "SSUSP": + agg.info.NumSSusp++ + case "PSUSP": + agg.info.NumPSusp++ + case "RUN": + agg.info.NumRun++ + case "PEND": + agg.info.NumPend++ + case "EXIT": + agg.info.NumExit++ + case "DONE": + agg.info.NumDone++ + } +} + // arrayStatsToOrderMaps converts derived stats rows to the OrderMap shape // QueryData hands to ExecuteQuery (mirrors store.StoreToOrderMaps). func arrayStatsToOrderMaps(rows []models.ArrayJobInfo) []utils.OrderMap { diff --git a/src/frontend/apiserver/controllers/arraystats_test.go b/src/frontend/apiserver/controllers/arraystats_test.go index c8d75145..24a25145 100644 --- a/src/frontend/apiserver/controllers/arraystats_test.go +++ b/src/frontend/apiserver/controllers/arraystats_test.go @@ -84,23 +84,26 @@ func TestDeriveArrayJobStats_CountsPerStatus(t *testing.T) { func TestDeriveArrayJobStats_IdentityFromFirstElement(t *testing.T) { store.InitStores() - e := mkArrayElement(arrayStatsBase[0], 2, "RUN") - e.UserGroup = "ug1" - e.ProjectName = "proj" - store.Jobs.Put(uint64(e.Unique_JobID), e) - // A later element with different identity fields must not win (elements - // of one array share the submit spec; first-seen supplies the identity). - e2 := mkArrayElement(arrayStatsBase[0], 1, "DONE") - e2.SubmitUser = "bob" + // The identity carrier is the array HEAD (lowest element ID), no matter + // which order the rows were put or List() walks them in (map order is + // randomized). The head (element 1) carries identity fields that must + // win; element 2's different fields must NOT overwrite them. + e2 := mkArrayElement(arrayStatsBase[0], 2, "RUN") + e2.UserGroup = "ug2" + e2.ProjectName = "proj2" store.Jobs.Put(uint64(e2.Unique_JobID), e2) + e := mkArrayElement(arrayStatsBase[0], 1, "DONE") + e.SubmitUser = "bob" + store.Jobs.Put(uint64(e.Unique_JobID), e) rows := deriveArrayJobStats() if len(rows) != 1 { t.Fatalf("expected 1 array row, got %d", len(rows)) } got := rows[0] - if got.SubmitUser != "alice" || got.UserGroup != "ug1" || got.ProjectName != "proj" { - t.Errorf("identity fields: user=%q ug=%q proj=%q, want alice/ug1/proj", + if got.SubmitUser != "bob" || got.UserGroup != "" || got.ProjectName != "" { + t.Errorf("identity fields: user=%q ug=%q proj=%q, want bob/// "+ + "(the head element, not element 2, must supply the identity)", got.SubmitUser, got.UserGroup, got.ProjectName) } } diff --git a/src/include/persist.h b/src/include/persist.h index 9a86830b..d71d9e36 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -758,7 +758,12 @@ wal_commit_mode_t wal_classify(uint32_t wal_type, int detail); * -XCE_ERR_NOT_READY — engine admission cap exhausted (retriable) * timeout_ms = 0 falls back to p->config.sync_timeout_ms. n_peers <= 1 * returns 0 immediately (engine self-quorum, standalone preserved). - * Call only from vtask threads with no entity locks held. */ + * Call only from vtask threads with no entity locks held. + * + * The async sibling persist_submit_job_payloads_async / + * persist_submit_job_array_async arms the same p->config.sync_timeout_ms + * deadline (commit_submit_async) -- the configured value, not a hardcoded + * constant, governs both paths' failure timing. */ int persist_commit_barrier(persist_t* p, uint32_t timeout_ms); /* Force WAL fdatasync (for tests and clean shutdown) */