From 81bfaf1f50c675368b1121f6e5203957fba382b1 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 1 Oct 2026 09:59:45 +0000 Subject: [PATCH] feat(job-array-one-based-index): atomic reply-and-release of in-flight submissions (task 2.2) In-flight registry entries carry is_array and array_id; two array entries with the same array identity overlap. inflight_reply_and_remove() sends the reply and unlinks the entry under the registry lock (xce_reply is non-blocking) and is used on every post-registration path, inside the async one-shot reply guard; jobsubmit_async_ctx_release no longer touches the registry, so a retry registered after the reply cannot be deleted by the original's release. Test-only seams under XCE_TESTING make the atomicity observable deterministically. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../specs/job-array-one-based-index/tasks.md | 3 +- src/controller/include/cluster.h | 24 ++ src/controller/job.c | 145 ++++++++-- .../testsuite/test_jobsubmit_validate.c | 273 ++++++++++++++++++ 4 files changed, 413 insertions(+), 32 deletions(-) diff --git a/.kiro/specs/job-array-one-based-index/tasks.md b/.kiro/specs/job-array-one-based-index/tasks.md index dda5242d..db6b2218 100644 --- a/.kiro/specs/job-array-one-based-index/tasks.md +++ b/.kiro/specs/job-array-one-based-index/tasks.md @@ -35,7 +35,7 @@ - _Requirements: 7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 7.9_ - _Depends: 1.1_ -- [ ] 2.2 Make answering a registered submission and releasing it one atomic step +- [x] 2.2 Make answering a registered submission and releasing it one atomic step - Extend in-flight registry entries with an array flag and array identity; two array entries with the same array identity overlap - Add one reply-and-remove operation that sends the reply and unlinks the entry while holding the registry lock (the reply is non-blocking), and use it on every path after registration: the payload-build and allocation-failure early-outs, commit failure or timeout, and apply success (inside the existing one-shot reply guard) - Stop the async-context release from removing registry entries, so it is the only removal point @@ -240,3 +240,4 @@ ## Implementation Notes - 1.2: `REGEX_JOBID`, `REGEX_JOBARRAY_ID` and `N_MAX_FIELDS_JOBID` (`src/include/config.h`) are now unused, and comments still describe regex-based `parse_jobid` in `src/scheduler/src/repl.c:143`, `src/harness/testsuite/test_test_cli.c:2324-2326` and `src/frontend/xcecli/common/format/jobid.go:86-93`; clean these up in 4.3 / 8.1 / 9.x where those files are touched. A direct `parse_jobid("...[0]")` accept test is a cheap addition when next touching `test_jobsel.c`. - 1.3: plain `idgen()` still has no sequence-overflow guard; after `idgen_array` takes block 31 the context sequence is 2^20, so a plain `idgen()` in the same second would spill into the timestamp (pre-existing class). Add an `-EOVERFLOW` check to plain `idgen()` in 9.1. +- 2.2: two extra XCE_TESTING seams exist (`jobsubmit_test_release_hook`, `jobsubmit_inflight_try_add_for_test`); the controller test fixture has no WAL, so async commit-failure runs synchronously inside handle_jobsubmit — reuse it in 2.3. Stale comment at `src/controller/testsuite/test_commit_barrier_reply.c:1458` (claims release via jobsubmit_async_ctx_release) — fix in 2.3. diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 0085db66..72ea4a83 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -129,6 +129,30 @@ extern jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid, * teardown so it doesn't outlive the test and spuriously reject an * unrelated later test reusing the same jobid range. */ extern void inflight_submits_reset_for_test(void); + +/* job-array-one-based-index (Req 7.7, 7.10, 7.11): seams on the in-flight + * submission registry. jobsubmit_test_reply_hook is called inside + * inflight_reply_and_remove after xce_reply and before the entry is + * unlinked, with the registry lock held; jobsubmit_test_release_hook is + * called on entry to the async-context release (after the reply). The + * overlap accessors apply the registry's own overlap rule (element ranges + * intersect, or both entries are arrays with the same array identity); + * the _locked variant expects the caller to hold the registry lock (i.e. be + * inside jobsubmit_test_reply_hook). jobsubmit_inflight_trylock_for_test + * returns pthread_mutex_trylock's result, unlocking on success. + * jobsubmit_inflight_try_add_for_test registers (base, size) exactly as + * handle_jobsubmit does; returns 1 if registered, 0 on overlap. */ +extern void (*jobsubmit_test_reply_hook)(void); +extern void (*jobsubmit_test_release_hook)(void); +extern int jobsubmit_inflight_overlaps_for_test(uint64_t base, uint64_t size, + int is_array, + uint64_t array_id); +extern int jobsubmit_inflight_overlaps_locked_for_test(uint64_t base, + uint64_t size, + int is_array, + uint64_t array_id); +extern int jobsubmit_inflight_trylock_for_test(void); +extern int jobsubmit_inflight_try_add_for_test(uint64_t base, uint64_t size); #endif extern int handle_worker_leave(cluster_t*, int); extern int mark_worker_unavail(cluster_t*, worker_t*); diff --git a/src/controller/job.c b/src/controller/job.c index 0c2e6971..697c94f7 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -102,26 +102,47 @@ static job_t* create_job(cluster_t* c, xce_job_t* j) * c->jobmgr hit — both mean "this exact write already happened or is * already in flight," which is precisely what a retried write must see as * a duplicate. Populated in handle_jobsubmit() right after this check - * passes (before any WAL work); cleared in jobsubmit_async_ctx_release() - * (the single choke point already reached on every resolution path: - * commit failure, apply completion, and OOM/alloc-failure early-outs). */ + * passes (before any WAL work). + * + * job-array-one-based-index (Req 7.7, 7.10, 7.11): an entry also carries + * whether it is an array and its array identity (tag cleared, low 15 bits + * cleared); two entries overlap if their element ranges intersect, or if + * both are arrays with the same array identity. An entry is removed only by + * inflight_reply_and_remove(), which answers the submission and unlinks its + * entry under one hold of s_inflight_submits_lock -- so another + * submission's registration (same lock) sees the entry exactly while the + * submission is unanswered. jobsubmit_async_ctx_release() no longer touches + * the registry: a removal by exact (base, size) after the reply could + * delete a retry of the same submission that registered in between. */ typedef struct inflight_submit { uint64_t base_jobid; uint64_t array_size; + int is_array; + uint64_t array_id; /* 0 unless is_array */ struct inflight_submit* next; } inflight_submit_t; static pthread_mutex_t s_inflight_submits_lock = PTHREAD_MUTEX_INITIALIZER; static inflight_submit_t* s_inflight_submits = NULL; -static bool inflight_range_overlaps_locked(uint64_t base_jobid, - uint64_t array_size) +/* Array identity of a classified submission's first ID: 0 for an ordinary + * job; for an array, the ID with its tag and element number cleared. */ +static uint64_t inflight_array_id(uint64_t base_jobid, int* is_array) +{ + *is_array = is_array_tag(base_jobid) ? 1 : 0; + return *is_array ? array_jobid(base_jobid) : 0; +} + +static bool inflight_overlaps_locked(uint64_t base_jobid, uint64_t array_size, + int is_array, uint64_t array_id) { for (inflight_submit_t* n = s_inflight_submits; n != NULL; n = n->next) { uint64_t lo = base_jobid, hi = base_jobid + array_size; uint64_t nlo = n->base_jobid, nhi = n->base_jobid + n->array_size; if (lo < nhi && nlo < hi) return true; + if (is_array && n->is_array && n->array_id == array_id) + return true; } return false; } @@ -136,14 +157,18 @@ static bool inflight_range_overlaps_locked(uint64_t base_jobid, * exactly as it already does for a c->jobmgr hit. */ static bool inflight_range_try_add(uint64_t base_jobid, uint64_t array_size) { - bool ok = false; + bool ok = false; + int is_array; + uint64_t array_id = inflight_array_id(base_jobid, &is_array); pthread_mutex_lock(&s_inflight_submits_lock); - if (!inflight_range_overlaps_locked(base_jobid, array_size)) { + if (!inflight_overlaps_locked(base_jobid, array_size, is_array, array_id)) { inflight_submit_t* n = calloc(1, sizeof(*n)); if (n != NULL) { n->base_jobid = base_jobid; n->array_size = array_size; + n->is_array = is_array; + n->array_id = array_id; n->next = s_inflight_submits; s_inflight_submits = n; ok = true; @@ -157,9 +182,27 @@ static bool inflight_range_try_add(uint64_t base_jobid, uint64_t array_size) return ok; } -static void inflight_range_remove(uint64_t base_jobid, uint64_t array_size) +#ifdef XCE_TESTING +void (*jobsubmit_test_reply_hook)(void) = NULL; +void (*jobsubmit_test_release_hook)(void) = NULL; +#endif + +/* Answer a registered submission and unlink its registry entry as one step + * (job-array-one-based-index, Req 7.7, 7.10, 7.11): xce_reply() only queues + * the packet on the endpoint (never blocks), so it may run under + * s_inflight_submits_lock (ha-async-io.md). Every path after a successful + * inflight_range_try_add() answers through here, exactly once; it is the + * only place an entry is removed (by exact (base, size)). */ +static void inflight_reply_and_remove(endpoint_t* ep, xce_peerinfo_t* from, + xce_reply_t* reply, uint64_t base_jobid, + uint64_t array_size) { pthread_mutex_lock(&s_inflight_submits_lock); + xce_reply(ep, from, reply); +#ifdef XCE_TESTING + if (jobsubmit_test_reply_hook != NULL) + jobsubmit_test_reply_hook(); +#endif inflight_submit_t** pp = &s_inflight_submits; while (*pp != NULL) { if ((*pp)->base_jobid == base_jobid && (*pp)->array_size == array_size) { @@ -196,6 +239,38 @@ void inflight_submits_reset_for_test(void) s_inflight_submits = NULL; pthread_mutex_unlock(&s_inflight_submits_lock); } + +int jobsubmit_inflight_overlaps_locked_for_test(uint64_t base, uint64_t size, + int is_array, uint64_t array_id) +{ + return inflight_overlaps_locked(base, size, is_array, array_id) ? 1 : 0; +} + +int jobsubmit_inflight_overlaps_for_test(uint64_t base, uint64_t size, + int is_array, uint64_t array_id) +{ + int hit; + + pthread_mutex_lock(&s_inflight_submits_lock); + hit = jobsubmit_inflight_overlaps_locked_for_test(base, size, is_array, + array_id); + pthread_mutex_unlock(&s_inflight_submits_lock); + return hit; +} + +int jobsubmit_inflight_trylock_for_test(void) +{ + int rc = pthread_mutex_trylock(&s_inflight_submits_lock); + + if (rc == 0) + pthread_mutex_unlock(&s_inflight_submits_lock); + return rc; +} + +int jobsubmit_inflight_try_add_for_test(uint64_t base, uint64_t size) +{ + return inflight_range_try_add(base, size) ? 1 : 0; +} #endif /* True if any jobid in [base_jobid, base_jobid + array_size) already exists @@ -2205,23 +2280,25 @@ typedef struct { * 6f1fa97), which always reported the BASE id, * never a per-element id */ uint64_t array_size; /* bug-1419: paired with jobid to remove the - * matching inflight_range_add() entry on - * release -- see jobid_range_exists' doc - * comment for the full contract */ + * matching in-flight registry entry when + * the batch is answered + * (inflight_reply_and_remove) */ pthread_mutex_t lock; /* guards pending/replied below */ int pending; /* entries not yet FULLY resolved (see above) */ int replied; /* 1 once a reply has been sent for this batch */ } jobsubmit_async_ctx_t; +/* Frees the async context only. The in-flight registry entry was already + * removed together with the batch's reply (inflight_reply_and_remove, from + * jobsubmit_reply_*_once); removing it here, after the reply, could delete + * a retry of the same (base, size) registered in between + * (job-array-one-based-index, Req 7.7). */ static void jobsubmit_async_ctx_release(jobsubmit_async_ctx_t* actx) { - /* bug-1419: this is the single choke point every resolution path - * (commit failure, apply completion, and the OOM/alloc-failure - * early-outs in handle_jobsubmit below) already reaches -- clearing - * the in-flight registration here means a retry is only ever - * dedup-blocked while this exact submission's outcome is genuinely - * still unresolved, never after. */ - inflight_range_remove(actx->jobid, actx->array_size); +#ifdef XCE_TESTING + if (jobsubmit_test_release_hook != NULL) + jobsubmit_test_release_hook(); +#endif endpoint_unref(actx->ep); pthread_mutex_destroy(&actx->lock); free(actx); @@ -2252,7 +2329,10 @@ static void jobsubmit_reply_failure_once(jobsubmit_async_ctx_t* actx, int rc) reply.status = rc; reply.data_which = RequestReply_data_nodata; - xce_reply(actx->ep, actx->c->myhost, &reply); + /* Only the guard's winner gets here: reply and registry release happen + * together, exactly once per batch. */ + inflight_reply_and_remove(actx->ep, actx->c->myhost, &reply, actx->jobid, + actx->array_size); } /* Sends exactly one success reply for the whole batch, carrying the base @@ -2283,7 +2363,10 @@ static void jobsubmit_reply_success_once(jobsubmit_async_ctx_t* actx) reply.data_which = RequestReply_data_jobid; reply.data.jobid = actx->jobid; - xce_reply(actx->ep, actx->c->myhost, &reply); + /* Only the guard's winner gets here: reply and registry release happen + * together, exactly once per batch. */ + inflight_reply_and_remove(actx->ep, actx->c->myhost, &reply, actx->jobid, + actx->array_size); } /* raft-apply-decoupling task 6 hand-off target: invoked by the ordered apply @@ -2398,12 +2481,11 @@ static int handle_jobsubmit_array(cluster_t* c, endpoint_t* ep, actx = calloc(1, sizeof(*actx)); if (actx == NULL) { - inflight_range_remove(jobid, submit->array_size); - reply.status = -ENOMEM; reply.data_which = RequestReply_data_nodata; - xce_reply(ep, c->myhost, &reply); + inflight_reply_and_remove(ep, c->myhost, &reply, jobid, + submit->array_size); return -ENOMEM; } actx->c = c; @@ -2532,10 +2614,11 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) * concurrent collision, or -- the actual motivating case -- this exact * request's own retried resend racing its still-in-flight original * attempt) already holds this range; reject exactly as the - * jobid_range_exists branch above does. Released in - * jobsubmit_async_ctx_release() once this submission's outcome is - * fully resolved; the two error returns below (which occur BEFORE - * actx exists to carry that release) remove it directly. */ + * jobid_range_exists branch above does. From here on every answer + * goes through inflight_reply_and_remove(), which sends the reply and + * releases this registration as one step (job-array-one-based-index, + * Req 7.7, 7.10, 7.11) -- the early-outs below directly, the commit + * outcome via jobsubmit_reply_*_once. */ if (!inflight_range_try_add(jobid, submit->array_size)) { errno = EEXIST; reply.status = -errno; @@ -2567,12 +2650,12 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) } if (rc < 0) { free_job_submit_payloads(&plist); - inflight_range_remove(jobid, submit->array_size); reply.status = rc; reply.data_which = RequestReply_data_nodata; - xce_reply(ep, c->myhost, &reply); + inflight_reply_and_remove(ep, c->myhost, &reply, jobid, + submit->array_size); return rc; } @@ -2606,12 +2689,12 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) if (actx == NULL) { free_job_submit_payloads(&plist); free(lsns); - inflight_range_remove(jobid, submit->array_size); reply.status = -ENOMEM; reply.data_which = RequestReply_data_nodata; - xce_reply(ep, c->myhost, &reply); + inflight_reply_and_remove(ep, c->myhost, &reply, jobid, + submit->array_size); return -ENOMEM; } actx->c = c; diff --git a/src/controller/testsuite/test_jobsubmit_validate.c b/src/controller/testsuite/test_jobsubmit_validate.c index b4178050..32c7ca5a 100644 --- a/src/controller/testsuite/test_jobsubmit_validate.c +++ b/src/controller/testsuite/test_jobsubmit_validate.c @@ -22,6 +22,17 @@ * Log observation: this test binary never calls tlog_init(), so ERROR() * falls through tlog's early-print path straight to STDOUT_FILENO (the * established precedent of test_worker_join.c's capture_stdout_during()). + * + * Task 2.2 (Requirements 7.7, 7.10, 7.11): answering a registered + * submission and removing its in-flight registry entry are one step under + * the registry lock. The fixture's persist has no WAL (wal_fd -1 and a + * wal_path that cannot be created), so an admitted submission's local WAL + * append fails synchronously and its commit continuation runs the async + * failure path (reply once, then release the async context) inside + * handle_jobsubmit -- deterministic, no thread. jobsubmit_test_reply_hook + * observes the state between the reply and the unlink; + * jobsubmit_test_release_hook registers a retry between the reply and the + * context release. */ /* rtti.h must come first -- it defines 'var' and 'typeinfo_t' used by kvmgr.h @@ -176,6 +187,12 @@ static void jsv_setup(void) 0, xce_hook_init(&g_jsv_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_jsv_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_jsv_c.jobmgr); + + /* No WAL: an admitted submission's local append fails with -EIO (see + * the file header). */ + g_jsv_c.persist.wal_fd = -1; + snprintf(g_jsv_c.persist.wal_path, sizeof(g_jsv_c.persist.wal_path), + "/nonexistent-xce-jsv/wal"); } static void jsv_drain(jsv_reply_obs_t* obs); @@ -185,6 +202,8 @@ static void jsv_teardown(void) jsv_reply_obs_t drain = { 0 }; jsv_drain(&drain); /* free any leftover packets */ + jobsubmit_test_reply_hook = NULL; + jobsubmit_test_release_hook = NULL; inflight_submits_reset_for_test(); if (g_jsv_c.jobmgr) dereference(g_jsv_c.jobmgr); @@ -433,6 +452,253 @@ static void test_jobsubmit_invalid_classes_einval(void) jsv_teardown(); } +/* ── Task 2.2: reply-and-remove atomicity ────────────────────────────── */ + +/* The array identity of jsv_elem(k): tag cleared, low 15 bits cleared. */ +#define JSV_ARRAY_ID jsv_id(0, JSV_BLOCK) + +typedef struct { + int calls; + int trylock_rc; + bool reply_queued; + int entry_present; + uint64_t base; + uint64_t size; + int is_array; + uint64_t array_id; +} jsv_hook_obs_t; + +static jsv_hook_obs_t g_jsv_hook; + +/* Runs inside inflight_reply_and_remove, after xce_reply, before the unlink, + * with the registry lock held: records (1) that the lock is held, (2) that + * the reply is already on the endpoint's outq, (3) that the entry is still + * registered (lock-free accessor, since the lock is held). */ +static void jsv_reply_hook(void) +{ + g_jsv_hook.calls += 1; + g_jsv_hook.trylock_rc = jobsubmit_inflight_trylock_for_test(); + g_jsv_hook.reply_queued = (g_jsv_ep.outq.head != NULL); + g_jsv_hook.entry_present = jobsubmit_inflight_overlaps_locked_for_test( + g_jsv_hook.base, g_jsv_hook.size, g_jsv_hook.is_array, + g_jsv_hook.array_id); +} + +static void jsv_expect_atomic_reply(uint64_t first, uint64_t size, int is_array, + uint64_t array_id, const char* what) +{ + jsv_reply_obs_t obs = { 0 }; + char* out = NULL; + char msg[200]; + int rc; + + jsv_setup(); + memset(&g_jsv_hook, 0, sizeof(g_jsv_hook)); + g_jsv_hook.base = first; + g_jsv_hook.size = size; + g_jsv_hook.is_array = is_array; + g_jsv_hook.array_id = array_id; + jobsubmit_test_reply_hook = jsv_reply_hook; + + jsv_arrange(first, size, false); + rc = jsv_submit_capture(&out); + free(out); + + snprintf(msg, sizeof(msg), "%s: the reply goes through the hook once", + what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_hook.calls, msg); + snprintf(msg, sizeof(msg), "%s: registry lock held across the reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(EBUSY, g_jsv_hook.trylock_rc, msg); + snprintf(msg, sizeof(msg), "%s: reply already queued inside the hook", + what); + TEST_ASSERT_TRUE_MESSAGE(g_jsv_hook.reply_queued, msg); + snprintf(msg, sizeof(msg), "%s: entry still registered inside the hook", + what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_hook.entry_present, msg); + snprintf(msg, sizeof(msg), "%s: entry gone after the reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, + jobsubmit_inflight_overlaps_for_test(first, size, is_array, array_id), + msg); + + (void)rc; /* the async failure path returns 0 from the handler */ + jsv_drain(&obs); + snprintf(msg, sizeof(msg), "%s: exactly one reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, msg); + snprintf(msg, sizeof(msg), "%s: commit-failure status", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EIO, obs.last_status, msg); + TEST_ASSERT_FALSE(jsv_job_exists(first)); + jsv_teardown(); +} + +/* Single-job path: commit failure answered and released atomically. */ +static void test_jobsubmit_reply_release_atomic_single(void) +{ + jsv_expect_atomic_reply(jsv_id(0, 9400), 1, 0, 0, "single job"); +} + +/* Array path (handle_jobsubmit_array): the same, for an array of 3. */ +static void test_jobsubmit_reply_release_atomic_array(void) +{ + jsv_expect_atomic_reply(jsv_elem(1), 3, 1, JSV_ARRAY_ID, "array of 3"); +} + +/* One-element array (the single-job WAL path with a tagged ID). */ +static void test_jobsubmit_reply_release_atomic_one_element_array(void) +{ + jsv_expect_atomic_reply(jsv_elem(1), 1, 1, JSV_ARRAY_ID, + "one-element array"); +} + +static uint64_t g_jsv_retry_base, g_jsv_retry_size; +static int g_jsv_retry_added; + +/* Runs when the original's async context is released, i.e. after its reply: + * a retry of the same (base, size) registers here, as it may once the + * original is answered (7.7). */ +static void jsv_release_hook_register_retry(void) +{ + if (g_jsv_retry_added == 0) + g_jsv_retry_added = jobsubmit_inflight_try_add_for_test( + g_jsv_retry_base, g_jsv_retry_size) + ? 1 + : -1; +} + +static void jsv_expect_retry_survives(uint64_t first, uint64_t size, + int is_array, uint64_t array_id, + const char* what) +{ + char* out = NULL; + char msg[200]; + + jsv_setup(); + g_jsv_retry_base = first; + g_jsv_retry_size = size; + g_jsv_retry_added = 0; + jobsubmit_test_release_hook = jsv_release_hook_register_retry; + + jsv_arrange(first, size, false); + (void)jsv_submit_capture(&out); + free(out); + + snprintf(msg, sizeof(msg), + "%s: the retry registers after the original's reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_retry_added, msg); + snprintf(msg, sizeof(msg), + "%s: the original's context release leaves the retry " + "registered", + what); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, + jobsubmit_inflight_overlaps_for_test(first, size, is_array, array_id), + msg); + jsv_teardown(); +} + +static void test_jobsubmit_retry_survives_release_single(void) +{ + jsv_expect_retry_survives(jsv_id(0, 9410), 1, 0, 0, "single job"); +} + +static void test_jobsubmit_retry_survives_release_array(void) +{ + jsv_expect_retry_survives(jsv_elem(1), 3, 1, JSV_ARRAY_ID, "array of 3"); +} + +/* Registry overlap rule: element ranges intersect, or both entries are + * arrays with the same array identity. */ +static void test_jobsubmit_inflight_overlap_rule(void) +{ + const uint64_t other_block = (uint64_t)9 << N_ARRAY_IDX_BITS; + const uint64_t other_array = jsv_id(0, other_block); + + jsv_setup(); + /* X|1..X|2 registered as an array */ + TEST_ASSERT_EQUAL_INT(1, + jobsubmit_inflight_try_add_for_test(jsv_elem(1), 2)); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, + jobsubmit_inflight_overlaps_for_test(jsv_elem(2), 1, 1, JSV_ARRAY_ID), + "intersecting element range overlaps"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, + jobsubmit_inflight_overlaps_for_test(jsv_elem(5), 1, 1, JSV_ARRAY_ID), + "same array identity, disjoint elements: overlaps (7.10, 7.11)"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, + jobsubmit_inflight_overlaps_for_test( + jsv_id(TAG_JOB_ARRAY, other_block | 1), 3, 1, other_array), + "a different array identity does not overlap"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, jobsubmit_inflight_overlaps_for_test(jsv_id(0, 9420), 1, 0, 0), + "an ordinary job outside the range does not overlap"); + + /* the registry refuses a second same-identity array */ + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, jobsubmit_inflight_try_add_for_test(jsv_elem(5), 1), + "a same-identity array cannot register while the first is"); + /* an ordinary entry is compared by range only */ + TEST_ASSERT_EQUAL_INT( + 1, jobsubmit_inflight_try_add_for_test(jsv_id(0, 9421), 1)); + TEST_ASSERT_EQUAL_INT( + 0, jobsubmit_inflight_overlaps_for_test(jsv_id(0, 9422), 1, 0, 0)); + jsv_teardown(); +} + +/* 7.11: while X|1..X|2 is registered and unanswered, a submission X|5 with + * the same array identity is answered duplicate-submission (-EEXIST) and + * the first stays registered; once the first is answered (with a failure) + * a new submission with that identity is admitted (7.7). */ +static void test_jobsubmit_same_array_identity_while_registered(void) +{ + jsv_reply_obs_t obs = { 0 }; + char* out = NULL; + int rc; + + jsv_setup(); + memset(&g_jsv_hook, 0, sizeof(g_jsv_hook)); + jobsubmit_test_reply_hook = jsv_reply_hook; + + TEST_ASSERT_EQUAL_INT(1, + jobsubmit_inflight_try_add_for_test(jsv_elem(1), 2)); + + jsv_arrange(jsv_elem(5), 1, false); + rc = jsv_submit_capture(&out); + free(out); + jsv_expect_rejected(rc, -EEXIST, jsv_elem(5), 1, + "same array identity while registered"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jsv_hook.calls, + "a submission refused at registration " + "has no entry to release"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, + jobsubmit_inflight_overlaps_for_test(jsv_elem(1), 2, 1, JSV_ARRAY_ID), + "the first submission stays registered"); + jsv_teardown(); + + /* answered: A fails its commit, then B with the same identity is + * admitted (it reaches the WAL and fails there too, not -EEXIST) */ + jsv_setup(); + jsv_arrange(jsv_elem(1), 2, false); + (void)jsv_submit_capture(&out); + free(out); + jsv_drain(&obs); + TEST_ASSERT_EQUAL_INT(-EIO, obs.last_status); + + memset(&obs, 0, sizeof(obs)); + jsv_arrange(jsv_elem(5), 1, false); + (void)jsv_submit_capture(&out); + free(out); + jsv_drain(&obs); + TEST_ASSERT_EQUAL_INT(1, obs.n_replies); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EIO, obs.last_status, + "after the first is answered, a same-" + "identity submission is admitted"); + jsv_teardown(); +} + void test_jobsubmit_validate(void) { UnitySetTestFile(__FILE__); @@ -441,4 +707,11 @@ void test_jobsubmit_validate(void) RUN_TEST(test_jobsubmit_index0_valid_fields_start0); RUN_TEST(test_jobsubmit_index0_size0_start0); RUN_TEST(test_jobsubmit_invalid_classes_einval); + RUN_TEST(test_jobsubmit_reply_release_atomic_single); + RUN_TEST(test_jobsubmit_reply_release_atomic_array); + RUN_TEST(test_jobsubmit_reply_release_atomic_one_element_array); + RUN_TEST(test_jobsubmit_retry_survives_release_single); + RUN_TEST(test_jobsubmit_retry_survives_release_array); + RUN_TEST(test_jobsubmit_inflight_overlap_rule); + RUN_TEST(test_jobsubmit_same_array_identity_while_registered); }