lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(job-array-one-based-index): atomic reply-and-release of in-flight submissions (task 2.2)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-01 09:59:45 +00:00
parent 36c2c243d6
commit 81bfaf1f50
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 413 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

145
src/controller/job.c
View file

@ -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;