lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/testsuite/test_jobsubmit_validate.c
Rongsong Shen fe5ddc9854
feat(job-array-one-based-index): collision checks after registration (task 2.3) 
handle_jobsubmit now registers the submission first and checks the job
table second; a hit is answered -EEXIST and released through the atomic
reply-and-release step. For array submissions, array_identity_exists()
does one read-locked scan of the 32768 tagged element keys of the
array's block (element 0 included for pre-upgrade arrays), covering
both the element range and the array identity; unmarked submissions
keep jobid_range_exists. An existing X|3 now blocks a new X|5.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-01 10:14:47 +00:00

918 lines
35 KiB
C
Raw Blame History

/*
* test_jobsubmit_validate.c -- controller array admission
* (job-array-one-based-index task 2.1, Requirements 7.1-7.6, 7.9).
*
* Two layers:
* - jobsubmit_classify(first_jobid, size) over every 7.1-7.5 class, the
* 7.9 precedence (an index-0 array wins over every other break), and
* the no-wrap 7.2 bound (sizes of 2^32+2 and UINT64_MAX with index 2,
* neither truncated nor wrapped).
* - handle_jobsubmit end to end for rejected submissions: the reply
* status, the 7.1 controller ERROR line, and that no job is stored
* (7.6). Classification runs before the field checks, the hook and any
* registry/table/persist access, so a rejected submission needs only a
* reply endpoint, a job table to prove nothing was stored, and an
* initialised hook system (never reached once classification exists).
*
* Reply observation: as in test_commit_barrier_reply.c, the requester
* endpoint is a zeroed endpoint_t with no iomgr, so xce_reply() leaves the
* encoded packet on ep->outq; it is drained with endpoint_deq() and parsed
* with the raw capnp reader.
*
* 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.
*
* Task 2.3 (Requirements 7.10, 7.11): the job-table checks run after
* registration. An existing element of the same array block (X|3 blocks a
* new X|5; element 0 and element 32767 included) or of the element range
* is answered -EEXIST through the atomic reply-and-release step, observed
* with the same reply hook; array_identity_exists() is also driven directly.
*/
/* rtti.h must come first -- it defines 'var' and 'typeinfo_t' used by kvmgr.h
*/
#include <rtti.h>
#include <errno.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <capnp_c.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
#include "cluster.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "netio.h"
#include "request.h"
#include "utils.h"
/* ── Job identities ──────────────────────────────────────────────────── */
/* An array block (sequence aligned to 32768) at a fixed cluster/apiserver/
* timestamp; jsv_elem(k) is element k of it. */
#define JSV_CLUSTER 3
#define JSV_APISERVER 2
#define JSV_TS 1700000000ULL
#define JSV_BLOCK ((uint64_t)5 << N_ARRAY_IDX_BITS)
static uint64_t jsv_id(uint32_t tag, uint64_t seq)
{
return MAKE_JOBID(JSV_CLUSTER, JSV_APISERVER, JSV_TS, tag, seq);
}
static uint64_t jsv_elem(uint64_t k)
{
return jsv_id(TAG_JOB_ARRAY, JSV_BLOCK | k);
}
/* ── jobsubmit_classify ──────────────────────────────────────────────── */
typedef struct {
const char* what;
uint64_t first;
uint64_t size;
jobsubmit_class_t expect;
} jsv_case_t;
static void test_jobsubmit_classify_table(void)
{
const jsv_case_t cases[] = {
/* 7.1: tagged array starting at element 0 */
{ "7.1 index-0 array of 3", jsv_elem(0), 3, JOBSUBMIT_INDEX0 },
{ "7.1 index-0 array of 1", jsv_elem(0), 1, JOBSUBMIT_INDEX0 },
{ "7.1 index-0 array of 32768", jsv_elem(0), 32768, JOBSUBMIT_INDEX0 },
/* 7.9: 7.1 wins over every 7.2-7.5 break */
{ "7.9 index-0 + size 0 (7.4)", jsv_elem(0), 0, JOBSUBMIT_INDEX0 },
{ "7.9 index-0 + last > 32767 (7.2)", jsv_elem(0), 40000,
JOBSUBMIT_INDEX0 },
{ "7.9 index-0 + UINT64_MAX size", jsv_elem(0), UINT64_MAX,
JOBSUBMIT_INDEX0 },
/* 7.4: size 0, any tag */
{ "7.4 ordinary size 0", jsv_id(0, 77), 0, JOBSUBMIT_INVALID },
{ "7.4 array size 0", jsv_elem(1), 0, JOBSUBMIT_INVALID },
/* 7.5: tags 2 and 3 (index 0 under them is not an array, so not
* 7.1) */
{ "7.5 tag 2", jsv_id(2, JSV_BLOCK | 1), 1, JOBSUBMIT_INVALID },
{ "7.5 tag 3", jsv_id(3, JSV_BLOCK | 1), 1, JOBSUBMIT_INVALID },
{ "7.5 tag 2 index 0", jsv_id(2, JSV_BLOCK), 3, JOBSUBMIT_INVALID },
{ "7.5 tag 3 index 0 size 0", jsv_id(3, JSV_BLOCK), 0,
JOBSUBMIT_INVALID },
/* 7.3: unmarked multi-element */
{ "7.3 ordinary size 2", jsv_id(0, 77), 2, JOBSUBMIT_INVALID },
{ "7.3 ordinary size 3", jsv_id(0, 9140), 3, JOBSUBMIT_INVALID },
{ "7.3 ordinary UINT64_MAX", jsv_id(0, 77), UINT64_MAX,
JOBSUBMIT_INVALID },
/* 7.2: last element above 32767, computed without addition */
{ "7.2 idx 1 size 32768", jsv_elem(1), 32768, JOBSUBMIT_INVALID },
{ "7.2 idx 2 size 32767", jsv_elem(2), 32767, JOBSUBMIT_INVALID },
{ "7.2 idx 32767 size 2", jsv_elem(32767), 2, JOBSUBMIT_INVALID },
{ "7.2 idx 2 size 2^32+2", jsv_elem(2), (1ULL << 32) + 2,
JOBSUBMIT_INVALID },
{ "7.2 idx 1 size 2^32+1", jsv_elem(1), (1ULL << 32) + 1,
JOBSUBMIT_INVALID },
{ "7.2 idx 2 size UINT64_MAX", jsv_elem(2), UINT64_MAX,
JOBSUBMIT_INVALID },
/* accepted */
{ "ok ordinary single", jsv_id(0, 77), 1, JOBSUBMIT_OK },
{ "ok ordinary single seq 0", jsv_id(0, 0), 1, JOBSUBMIT_OK },
{ "ok one-element array", jsv_elem(1), 1, JOBSUBMIT_OK },
{ "ok array 1..3", jsv_elem(1), 3, JOBSUBMIT_OK },
{ "ok array 1..32767", jsv_elem(1), 32767, JOBSUBMIT_OK },
{ "ok array 2..32767", jsv_elem(2), 32766, JOBSUBMIT_OK },
{ "ok array 32767..32767", jsv_elem(32767), 1, JOBSUBMIT_OK },
};
size_t i;
char msg[160];
for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
snprintf(msg, sizeof(msg), "case %zu: %s", i, cases[i].what);
TEST_ASSERT_EQUAL_INT_MESSAGE(
cases[i].expect, jobsubmit_classify(cases[i].first, cases[i].size),
msg);
}
}
/* ── handle_jobsubmit fixture ────────────────────────────────────────── */
static cluster_t g_jsv_c;
static endpoint_t g_jsv_ep;
static ep_private_t g_jsv_priv;
static job_startinfo_t g_jsv_startinfo;
static job_submit_data_t g_jsv_data;
static xce_job_submit_t g_jsv_msg;
static xce_request_t g_jsv_req;
static xce_peerinfo_t g_jsv_from;
typedef struct {
int n_replies;
int32_t last_status;
int32_t last_data_which;
uint64_t last_seq;
} jsv_reply_obs_t;
static void jsv_setup(void)
{
kvmgr_params_t int_params
= { .key_type = KVMGR_KEY_INT, .nslots = 64, .delta = 16 };
memset(&g_jsv_c, 0, sizeof(g_jsv_c));
memset(&g_jsv_ep, 0, sizeof(g_jsv_ep));
memset(&g_jsv_priv, 0, sizeof(g_jsv_priv));
g_jsv_ep.fd = -1;
g_jsv_ep.data = &g_jsv_priv;
g_jsv_c.myhost = &g_jsv_c.myhost_info;
g_jsv_c.myhost_info.name = (char*)"ctrl-test";
g_jsv_c.myhost_info.key = (char*)"";
TEST_ASSERT_EQUAL_INT(
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);
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);
xce_hook_fini(&g_jsv_c.hooksys);
}
/* A submission with every field the existing field checks require, unless
* empty_queue is set (then the queue is "" and the field check would reject
* it with -EINVAL). */
static void jsv_arrange(uint64_t first, uint64_t size, bool empty_queue)
{
memset(&g_jsv_startinfo, 0, sizeof(g_jsv_startinfo));
g_jsv_startinfo.cmdline = (char*)"/bin/true";
memset(&g_jsv_data, 0, sizeof(g_jsv_data));
g_jsv_data.queue = (char*)(empty_queue ? "" : "normal");
g_jsv_data.user = (char*)"tester";
g_jsv_data.start_info = &g_jsv_startinfo;
memset(&g_jsv_msg, 0, sizeof(g_jsv_msg));
g_jsv_msg.jobid = first;
g_jsv_msg.array_size = size;
g_jsv_msg.submit = &g_jsv_data;
g_jsv_from.name = (char*)"requester";
g_jsv_from.key = (char*)"";
memset(&g_jsv_req, 0, sizeof(g_jsv_req));
g_jsv_req.opcode = OPCode_jobsub;
g_jsv_req.seq = 21;
g_jsv_req.from = &g_jsv_from;
g_jsv_req.body_which = Request_body_jobsubmsg;
g_jsv_req.body.jobsubmsg = &g_jsv_msg;
}
static void jsv_drain(jsv_reply_obs_t* obs)
{
data_item_t* it;
while ((it = endpoint_deq(&g_jsv_ep)) != NULL) {
io_packet_t* pkt = ITPTR_TO_PACKET(it);
struct capn c;
struct Request rq;
Request_ptr rp;
capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0);
rp.p = capn_getp(capn_root(&c), 0, 1);
read_Request(&rq, rp);
if (rq.body_which == Request_body_replymsg) {
struct RequestReply rr;
read_RequestReply(&rr, rq.body.replymsg);
obs->n_replies += 1;
obs->last_status = rr.status;
obs->last_data_which = rr.data_which;
obs->last_seq = rr.seq;
}
capn_free(&c);
free_packet(pkt);
free(pkt);
}
}
static bool jsv_job_exists(uint64_t jobid)
{
job_t* j = NULL;
int rc;
lock(g_jsv_c.jobmgr, RTTI_LOCK_READ);
rc = key_find(g_jsv_c.jobmgr, (void*)jobid, (void**)&j);
unlock(g_jsv_c.jobmgr);
return rc > 0 && j != NULL;
}
/* Run handle_jobsubmit with STDOUT_FILENO redirected to a temp file; returns
* the handler's rc and the captured output (caller frees). */
static int jsv_submit_capture(char** out)
{
char tmp[] = "/tmp/xce_jsv_logcap_XXXXXX";
int fd = mkstemp(tmp);
int saved, rc;
off_t len;
char* buf;
TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "fixture: mkstemp must succeed");
fflush(stdout);
saved = dup(STDOUT_FILENO);
TEST_ASSERT_TRUE_MESSAGE(saved >= 0, "fixture: dup(stdout) must succeed");
TEST_ASSERT_TRUE_MESSAGE(dup2(fd, STDOUT_FILENO) >= 0,
"fixture: dup2 redirect must succeed");
rc = handle_jobsubmit(&g_jsv_c, &g_jsv_ep, &g_jsv_req);
fflush(stdout);
dup2(saved, STDOUT_FILENO);
close(saved);
len = lseek(fd, 0, SEEK_CUR);
TEST_ASSERT_TRUE(len >= 0);
buf = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(buf);
if (len > 0) {
TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0);
TEST_ASSERT_TRUE(read(fd, buf, (size_t)len) == len);
}
close(fd);
unlink(tmp);
*out = buf;
return rc;
}
/* Assert one reply with `status`, the handler returned it, and none of the
* first min(size, 8) element identities is in the job table (7.6). */
static void jsv_expect_rejected(int rc, int32_t status, uint64_t first,
uint64_t size, const char* what)
{
jsv_reply_obs_t obs = { 0 };
char msg[200];
uint64_t i, n = size < 8 ? size : 8;
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: reply status", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(status, obs.last_status, msg);
snprintf(msg, sizeof(msg), "%s: reply carries no data", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_nodata, obs.last_data_which,
msg);
snprintf(msg, sizeof(msg), "%s: reply echoes the request seq", what);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(21, obs.last_seq, msg);
snprintf(msg, sizeof(msg), "%s: handler return value", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(status, rc, msg);
if (n == 0)
n = 1;
for (i = 0; i < n; i++) {
snprintf(msg, sizeof(msg), "%s: no job stored at element %llu", what,
(unsigned long long)i);
TEST_ASSERT_FALSE_MESSAGE(jsv_job_exists(first + i), msg);
}
}
/* 7.1 + 7.9: an index-0 array with an empty queue is answered with the
* distinct status (not the field check's -EINVAL) and the ERROR line. */
static void test_jobsubmit_index0_empty_queue_start0_and_log(void)
{
const uint64_t first = jsv_elem(0);
char* out = NULL;
char idstr[64];
int rc;
jsv_setup();
jsv_arrange(first, 3, true);
rc = jsv_submit_capture(&out);
jsv_expect_rejected(rc, -E_XCE_ARRAY_START0, first, 3,
"index-0 array, empty queue");
fmt_jobid(idstr, sizeof(idstr), first);
TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "ERROR"),
"7.1: an ERROR line must be logged");
TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, idstr),
"7.1: the log line must name the job");
TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "start at 1"),
"7.1: the log line must say array element "
"numbers start at 1");
TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "must be upgraded"),
"7.1: the log line must say the submitting "
"client must be upgraded");
free(out);
jsv_teardown();
}
/* 7.1 with every field valid: the same status and log, nothing stored. */
static void test_jobsubmit_index0_valid_fields_start0(void)
{
const uint64_t first = jsv_elem(0);
char* out = NULL;
int rc;
jsv_setup();
jsv_arrange(first, 3, false);
rc = jsv_submit_capture(&out);
jsv_expect_rejected(rc, -E_XCE_ARRAY_START0, first, 3,
"index-0 array, valid fields");
TEST_ASSERT_NOT_NULL(strstr(out, "must be upgraded"));
free(out);
jsv_teardown();
}
/* 7.9: index 0 with size 0 -- 7.1 alone answers. */
static void test_jobsubmit_index0_size0_start0(void)
{
const uint64_t first = jsv_elem(0);
char* out = NULL;
int rc;
jsv_setup();
jsv_arrange(first, 0, false);
rc = jsv_submit_capture(&out);
jsv_expect_rejected(rc, -E_XCE_ARRAY_START0, first, 0,
"index-0 array, size 0");
free(out);
jsv_teardown();
}
/* 7.2-7.5: answered -EINVAL before any field check, hook or registry; no
* 7.1 log line; nothing stored; the controller keeps serving (the next case
* runs against the same handler). */
static void test_jobsubmit_invalid_classes_einval(void)
{
const struct {
const char* what;
uint64_t first;
uint64_t size;
bool empty_queue;
} cases[] = {
{ "7.4 size 0", jsv_id(0, 9300), 0, false },
{ "7.4 size 0, empty queue", jsv_id(0, 9301), 0, true },
{ "7.5 tag 2", jsv_id(2, JSV_BLOCK | 1), 1, false },
{ "7.5 tag 3", jsv_id(3, JSV_BLOCK | 1), 1, false },
{ "7.3 unmarked size 3", jsv_id(0, 9310), 3, false },
{ "7.2 idx 1 size 32768", jsv_elem(1), 32768, false },
{ "7.2 idx 2 size 2^32+2", jsv_elem(2), (1ULL << 32) + 2, false },
{ "7.2 idx 2 size UINT64_MAX", jsv_elem(2), UINT64_MAX, false },
};
size_t i;
jsv_setup();
for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
char* out = NULL;
int rc;
jsv_arrange(cases[i].first, cases[i].size, cases[i].empty_queue);
rc = jsv_submit_capture(&out);
jsv_expect_rejected(rc, -EINVAL, cases[i].first, cases[i].size,
cases[i].what);
TEST_ASSERT_NULL_MESSAGE(strstr(out, "must be upgraded"),
"7.2-7.5 rejections must not log the 7.1 "
"line");
free(out);
}
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();
}
/* ── Task 2.3: job-table collisions after registration ───────────────── */
static job_submit_data_t g_jsv_seed_submit;
/* Store an existing job `jobid` in the fixture's job table. */
static void jsv_seed_job(uint64_t jobid)
{
xce_job_t jdef;
memset(&g_jsv_seed_submit, 0, sizeof(g_jsv_seed_submit));
g_jsv_seed_submit.queue = (char*)"normal";
g_jsv_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = &g_jsv_seed_submit;
TEST_ASSERT_NOT_NULL_MESSAGE(makeJob(g_jsv_c.jobmgr, &jdef),
"fixture: seeding an existing job");
TEST_ASSERT_TRUE(jsv_job_exists(jobid));
}
/* Submit (first, size) against a table holding `existing` and expect the
* duplicate-submission reply: answered through inflight_reply_and_remove
* while the submission is registered (so the table check ran after
* registration), its entry released afterwards, and nothing new stored. */
static void jsv_expect_table_collision(uint64_t existing, uint64_t first,
uint64_t size, const char* what)
{
char* out = NULL;
char msg[200];
int is_array = is_array_tag(first) ? 1 : 0;
uint64_t array_id = is_array ? array_jobid(first) : 0;
uint64_t i;
int rc;
jsv_setup();
jsv_seed_job(existing);
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);
{
jsv_reply_obs_t obs = { 0 };
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: duplicate-submission reply", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, obs.last_status, msg);
snprintf(msg, sizeof(msg), "%s: reply echoes the request seq", what);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(21, obs.last_seq, msg);
snprintf(msg, sizeof(msg), "%s: handler return value", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc, msg);
}
snprintf(msg, sizeof(msg),
"%s: the table hit is answered through the atomic "
"reply-and-release step (registered first)",
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: entry registered while answered", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_hook.entry_present, msg);
snprintf(msg, sizeof(msg), "%s: entry released after the reply", what);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0,
jobsubmit_inflight_overlaps_for_test(first, size, is_array, array_id),
msg);
snprintf(msg, sizeof(msg), "%s: the existing job is unchanged", what);
TEST_ASSERT_TRUE_MESSAGE(jsv_job_exists(existing), msg);
for (i = 0; i < size && i < 8; i++) {
if (first + i == existing)
continue;
snprintf(msg, sizeof(msg), "%s: no job stored at offset %llu", what,
(unsigned long long)i);
TEST_ASSERT_FALSE_MESSAGE(jsv_job_exists(first + i), msg);
}
jsv_teardown();
}
/* 7.10: an existing element X|3 blocks a new X|5 (same array identity,
* disjoint elements) -- one-element and multi-element submissions. */
static void test_jobsubmit_existing_element_blocks_same_identity(void)
{
jsv_expect_table_collision(jsv_elem(3), jsv_elem(5), 1,
"existing X|3, new X|5 (1 element)");
jsv_expect_table_collision(jsv_elem(3), jsv_elem(5), 4,
"existing X|3, new X|5..X|8");
/* the identity scan covers the whole block, both ends */
jsv_expect_table_collision(jsv_elem(MAX_ARRAY_IDX), jsv_elem(1), 2,
"existing X|32767, new X|1..X|2");
/* a pre-upgrade element 0 holds the identity too */
jsv_expect_table_collision(jsv_elem(0), jsv_elem(1), 3,
"existing pre-upgrade X|0, new X|1..X|3");
}
/* 7.10: the element-range check -- an existing element inside the range. */
static void test_jobsubmit_existing_element_in_range_blocks(void)
{
jsv_expect_table_collision(jsv_elem(2), jsv_elem(1), 3,
"existing X|2, new X|1..X|3");
jsv_expect_table_collision(jsv_elem(1), jsv_elem(1), 1,
"existing X|1, new X|1");
}
/* Unmarked submissions keep the range check, now also after registration. */
static void test_jobsubmit_existing_ordinary_job_blocks(void)
{
jsv_expect_table_collision(jsv_id(0, 9500), jsv_id(0, 9500), 1,
"existing ordinary job, same ID");
}
/* No false collisions: another block's element, or an ordinary job whose
* sequence lies in X's block, leaves X's identity free -- the submission is
* admitted (it reaches the WAL and fails there with -EIO, not -EEXIST). */
static void test_jobsubmit_no_collision_admitted(void)
{
const uint64_t other_elem
= jsv_id(TAG_JOB_ARRAY, ((uint64_t)9 << N_ARRAY_IDX_BITS) | 3);
const uint64_t ordinary_in_block = jsv_id(0, JSV_BLOCK | 3);
const struct {
const char* what;
uint64_t existing;
uint64_t first;
uint64_t size;
} cases[] = {
{ "other block's element", other_elem, jsv_elem(5), 2 },
{ "ordinary job in X's block", ordinary_in_block, jsv_elem(5), 1 },
{ "array element, new ordinary job", jsv_elem(3), jsv_id(0, 9510), 1 },
};
size_t i;
for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
jsv_reply_obs_t obs = { 0 };
char* out = NULL;
jsv_setup();
jsv_seed_job(cases[i].existing);
jsv_arrange(cases[i].first, cases[i].size, false);
(void)jsv_submit_capture(&out);
free(out);
jsv_drain(&obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, cases[i].what);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EIO, obs.last_status, cases[i].what);
jsv_teardown();
}
}
/* array_identity_exists: one scan over the block's 32768 element keys. */
static void test_array_identity_exists_scan(void)
{
const uint64_t other_block = (uint64_t)9 << N_ARRAY_IDX_BITS;
jsv_setup();
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
array_identity_exists(&g_jsv_c, JSV_ARRAY_ID),
"empty table: identity free");
jsv_seed_job(jsv_id(0, JSV_BLOCK | 3));
jsv_seed_job(jsv_id(TAG_JOB_ARRAY, other_block | 1));
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
array_identity_exists(&g_jsv_c, JSV_ARRAY_ID),
"an ordinary job or another block's "
"element does not hold X's identity");
jsv_seed_job(jsv_elem(MAX_ARRAY_IDX));
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
array_identity_exists(&g_jsv_c, JSV_ARRAY_ID),
"element 32767 holds X's identity");
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, array_identity_exists(&g_jsv_c, jsv_id(0, other_block)),
"the other block's identity is held too");
jsv_teardown();
jsv_setup();
jsv_seed_job(jsv_elem(0));
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
array_identity_exists(&g_jsv_c, JSV_ARRAY_ID),
"a pre-upgrade element 0 holds the identity");
jsv_teardown();
}
void test_jobsubmit_validate(void)
{
UnitySetTestFile(__FILE__);
RUN_TEST(test_jobsubmit_classify_table);
RUN_TEST(test_jobsubmit_index0_empty_queue_start0_and_log);
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);
RUN_TEST(test_jobsubmit_existing_element_blocks_same_identity);
RUN_TEST(test_jobsubmit_existing_element_in_range_blocks);
RUN_TEST(test_jobsubmit_existing_ordinary_job_blocks);
RUN_TEST(test_jobsubmit_no_collision_admitted);
RUN_TEST(test_array_identity_exists_scan);
}
Reference in a new issue View git blame Copy permalink