lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/testsuite/test_dispatch_commit.c
Rongsong Shen f8851353cd
test(controller): cover WAL_JOB_DISPATCH allocation persistence (bug-1678) 
test_dispatch_commit.c's test_dc_success_appends_wal_job_dispatch_durably
(already drove the real dispatch_a_job_() production path through a
real WAL-write + persist_recover() restart cycle, previously only
checking recovered state) now also asserts the recovered job.allocs
matches what was dispatched -- host, resource name, value. This is
the primary regression test for the user's exact acceptance
criterion: "after recover the job should get the same as it's state
before shutdown."

Independently reproduced RED->GREEN (not just narrated): reverted the
production fix with the test extension kept, rebuilt, confirmed FAIL
at exactly this assertion; restored, rebuilt, confirmed PASS.

test_wal_job_dispatch_fact.c: added 3 new tests exercising the richer
payload directly on both apply_committed_entry and apply_wal_entry,
plus a full restart-replay variant. All 3 pre-existing tests (bare
wal_jobstate_t payloads, no trailing bytes) pass completely unchanged,
proving backward compatibility with any WAL_JOB_DISPATCH entry written
before this fix.

Full controller_unit_test suite independently re-run: 1136/1136 pass,
zero regressions.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
2026-08-25 03:12:58 +00:00

671 lines
26 KiB
C
Raw Blame History

/* test_dispatch_commit.c — scheduler-dispatch-commit task 3.1
*
* Proves dispatch_a_job_()'s (job.c) restructured commit: the job's
* resource allocation and its transition to a running state are committed
* together, ONLY once the dispatch request to the worker actually succeeds
* (Req 2.1-2.4, 5.1) -- gating a commit that used to run unconditionally
* regardless of xce_startjob_req()'s outcome. Also proves the consecutive
* retryable-failure counter (job_t.dispatch_fail_count, task 1.3) resets to
* zero on that same successful outcome (Req 4.3), and that the new durable
* WAL_JOB_DISPATCH fact (task 1.2) is appended as part of it (Req 2.4).
*
* dispatch_a_job_() is static/file-local (its only caller, handle_
* jobdispatch(), is untouched by this task); reached here through
* dispatch_a_job_test(), an XCE_TESTING-only wrapper this task adds --
* mirroring xce_classify_dispatch_fail_test()'s existing precedent in the
* same file (test_job_dispatch.c, task 1.1).
*
* No live peer / synthetic reply thread is needed to make a dispatch
* "succeed": xce_startjob_req() (request.c) is a genuinely non-blocking,
* fire-and-forget send -- xce_request() only capnp-encodes the request and
* endpoint_enq()s it onto the endpoint's outq, it never waits for a reply.
* endpoint_enq() succeeds unconditionally against a zeroed endpoint_t with
* no iomgr registered: dataq_enq() (dataq.c) has no failure path, and
* endpoint_update_epoll() (netio.c) is a pure no-op whenever ep->iomgr ==
* NULL. This exact "zeroed, no-iomgr endpoint" pattern is already
* established and proven live by test_commit_barrier_reply.c's
* cbrp_arrange_jobdispatch()/g_wkr_disp_ep fixture (see that file's own
* header comment on its reply-observation harness) -- this file's fixtures
* mirror that queue/job/worker/alloc construction, trimmed to what
* dispatch_a_job_() itself reads (no HA commit-engine/standby-registry
* machinery, which is only needed there for that file's own, unrelated
* reply-ordering assertions).
*
* Two fixture tiers, mirroring test_wal_job_dispatch_fact.c's own split:
* - dc_setup/dc_teardown: cluster_t (jobmgr/workermgr/qmgr/hooksys) + a
* real disk-backed persist_t initialized with cluster == NULL, so
* wal_hook_append_classified() (persist.c) takes the direct
* persist_wal_async() path -- no raft/commit-engine machinery needed
* for a plain RETAIN-mode fact append.
* - the durability test builds its own two short-lived cluster_t/
* persist_t pairs directly, mirroring test_wal_job_dispatch_fact.c's
* wjd_make_fresh_persist/restart-replay pattern exactly.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "dataq.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "netio.h"
#include "persist.h"
#include "qobj.h"
#include "wal_types.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xstring.h"
/* dispatch_a_job_test: task 3.1's XCE_TESTING-only wrapper (job.c), thinly
* exposing the static dispatch_a_job_() for direct testing -- see this
* file's header and job.c's own comment beside the wrapper. */
extern int dispatch_a_job_test(cluster_t* c, endpoint_t* ep,
xce_job_alloc_t* alloc);
/* jobj.c's job type table and the process-global xstring table are shared
* across every test file in the controller_unit_test binary -- guard
* idempotently, mirroring every sibling fixture in this suite
* (test_wal_job_dispatch_fact.c's g_wjd_jobobj_ready). */
static int g_dc_jobobj_ready;
static cluster_t g_dc;
static char g_dc_tmpdir[256];
/* Worker endpoint: zeroed, no iomgr -- xce_startjob_req()'s enqueue always
* succeeds against it (see file header). File-static, not stack-local, so
* it stays valid for the whole fixture's lifetime regardless of which
* function last touched worker_t.ep (mirrors test_commit_barrier_reply.c's
* g_wkr_disp_ep). */
static endpoint_t g_dc_worker_ep;
static void dc_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32, .delta = 16 };
persist_config_t cfg;
if (makeString((char*)"dc-probe") == NULL)
xstring_table_init(64);
if (!g_dc_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_dc_jobobj_ready = 1;
}
memset(&g_dc, 0, sizeof(g_dc));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_dc.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_dc.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_dc.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_dc.qmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_dc.jobmgr);
TEST_ASSERT_NOT_NULL(g_dc.workermgr);
TEST_ASSERT_NOT_NULL(g_dc.qmgr);
/* Real disk-backed persist, cluster == NULL: wal_hook_append_
* classified() takes the direct persist_wal_async() path -- no raft/
* commit-engine machinery needed for a plain fact append (this exact
* routing is what lets a bare cluster_t exercise dispatch_a_job_()'s
* real persist_submit_job_fact() call without a full HA harness). */
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
strcpy(g_dc_tmpdir, "/tmp/xce_dc31_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_dc_tmpdir));
persist_set_workdir(&g_dc.persist, g_dc_tmpdir, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_dc.persist, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_dc.persist));
memset(&g_dc_worker_ep, 0, sizeof(g_dc_worker_ep));
g_dc_worker_ep.fd = -1;
}
static void dc_teardown(void)
{
char cmd[512];
persist_stop(&g_dc.persist);
persist_fini(&g_dc.persist);
if (g_dc.workermgr) dereference(g_dc.workermgr);
if (g_dc.qmgr) dereference(g_dc.qmgr);
if (g_dc.jobmgr) dereference(g_dc.jobmgr);
xce_hook_fini(&g_dc.hooksys);
snprintf(cmd, sizeof(cmd), "rm -rf %s", g_dc_tmpdir);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", g_dc_tmpdir);
}
/* Frees every packet left queued on an endpoint's outq (the startjob
* request xce_startjob_req() enqueued) -- mirrors test_commit_barrier_
* reply.c's cbrp_scan_replies, trimmed: this file never needs to decode the
* packets, only reclaim them. */
static void dc_drain_ep(endpoint_t* ep)
{
data_item_t* it;
while ((it = endpoint_deq(ep)) != NULL) {
io_packet_t* pkt = ITPTR_TO_PACKET(it);
free_packet(pkt);
free(pkt);
}
}
/* Seeds a queue, a PEND job with a real submit/start_info (deep-copied by
* makeJob via makeJobsub -- dispatch_a_job_() reads submit->queue and
* submit->start_info), and a ready worker whose ep is g_dc_worker_ep.
* Mirrors test_commit_barrier_reply.c's cbrp_arrange_jobdispatch fixture,
* built directly against g_dc rather than through handle_jobdispatch().
* Returns the seeded job_t* (borrowed -- g_dc.jobmgr owns it). */
static job_t* dc_seed_ready(uint64_t jobid, uint32_t start_fail_count,
const char* qname, const char* wname)
{
xce_queue_t qdef;
xce_worker_t wdef;
hostattrs_t* attrs;
job_startinfo_t startinfo;
job_submit_data_t submit;
xce_job_t jdef;
job_t* j = NULL;
worker_t* w = NULL;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char*)qname;
TEST_ASSERT_NOT_NULL(makeQueue(g_dc.qmgr, &qdef));
memset(&startinfo, 0, sizeof(startinfo));
startinfo.cmdline = (char*)"/bin/true";
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)qname;
submit.user = (char*)"tester";
submit.start_info = &startinfo;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
j = makeJob(g_dc.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j);
j->dispatch_fail_count = start_fail_count;
attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)wname;
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(g_dc.workermgr, &wdef));
lock(g_dc.workermgr, RTTI_LOCK_READ);
key_find(g_dc.workermgr, (void*)wname, (void**)&w);
unlock(g_dc.workermgr);
TEST_ASSERT_NOT_NULL(w);
w->ep = &g_dc_worker_ep; /* host.status defaults to 0 (ready) via calloc */
return j;
}
#define DC_JOBID_SUCCESS 91001ULL
/* THE core proof (Req 2.1, 2.3, 4.3): a successful dispatch commits
* job.allocs and transitions job.state to RUN together, and resets
* dispatch_fail_count to 0 from a nonzero starting value -- all in one
* operation, with no dependence on any later worker confirmation. */
static void test_dc_success_commits_alloc_state_and_resets_counter(void)
{
job_t* j;
xce_job_alloc_t* alloc;
int rc;
dc_setup();
j = dc_seed_ready(DC_JOBID_SUCCESS, 3 /* nonzero starting fail count */,
"dq", "wkr-dc");
alloc = make_job_alloc(DC_JOBID_SUCCESS, 1,
make_host_allocs(1,
make_host_alloc((char*)"wkr-dc", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
rc = dispatch_a_job_test(&g_dc, NULL, alloc);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0,
"a dispatch to a live, no-iomgr worker endpoint must succeed -- "
"xce_startjob_req() is a fire-and-forget enqueue, not a blocking "
"round trip");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
"the job must transition to RUN in the same operation that commits "
"the allocation, unaffected by whether the worker has separately "
"confirmed job start yet (Req 2.1, 2.3)");
TEST_ASSERT_NOT_NULL(j->job.allocs);
TEST_ASSERT_EQUAL_UINT64(DC_JOBID_SUCCESS, j->job.allocs->jobid);
TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
TEST_ASSERT_NOT_NULL(j->job.allocs->allocs);
TEST_ASSERT_EQUAL_STRING("wkr-dc", j->job.allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, j->dispatch_fail_count,
"a successful dispatch must reset the consecutive retryable-"
"failure counter to zero (Req 4.3)");
free_JobAlloc_ptr(&alloc); /* allocs already MOVE_PTR'd out -- top-level
* struct only */
dc_drain_ep(&g_dc_worker_ep);
dc_teardown();
printf("test_dc_success_commits_alloc_state_and_resets_counter: PASS\n");
}
#define DC_JOBID_NOTFOUND 91002ULL
/* Regression safety net (untouched failure branch): a dispatch whose
* target worker cannot be found must leave the job's ALLOCATION exactly
* as it was before the attempt -- still PEND, allocs still uncommitted --
* and must append no WAL entry. Confirms this task's restructuring did
* not disturb the five failure branches ahead of xce_startjob_req().
*
* scheduler-dispatch-commit task 3.3 update: dispatch_fail_count is no
* longer asserted unchanged here -- task 3.3 wired this exact failure
* branch (whichever of XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP/WORKER_NULL
* the runtime takes for an unregistered worker key) to increment the
* counter by exactly 1 as part of its retryable resolution (Req 4.3),
* which this test's own original comment explicitly anticipated
* ("incrementing it is task 3.3's job, not 3.1's"). The counter's own
* increment-by-1/bound behavior is covered in depth by
* test_dispatch_fail_retry.c; this test only re-confirms the seeded
* starting value (2) plus this task's own increment (+1) lands at 3,
* still within the bound (3), so the job stays PEND -- not a regression
* in this test's own subject (allocation-gating), just an update to an
* assertion that depended on timing task 3.3 had not landed yet. */
static void test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted(void)
{
job_t* j;
xce_job_alloc_t* alloc;
int rc;
uint64_t lsn_before, lsn_after;
dc_setup();
/* Seed only the job/queue -- deliberately no matching worker in
* g_dc.workermgr, so key_find() finds nothing (w stays NULL) and
* dispatch_a_job_() takes its WORKER_NULL failure branch. */
{
job_startinfo_t startinfo;
job_submit_data_t submit;
xce_job_t jdef;
xce_queue_t qdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char*)"dq2";
TEST_ASSERT_NOT_NULL(makeQueue(g_dc.qmgr, &qdef));
memset(&startinfo, 0, sizeof(startinfo));
startinfo.cmdline = (char*)"/bin/true";
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"dq2";
submit.user = (char*)"tester";
submit.start_info = &startinfo;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = DC_JOBID_NOTFOUND;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
j = makeJob(g_dc.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j);
j->dispatch_fail_count = 2;
}
alloc = make_job_alloc(DC_JOBID_NOTFOUND, 1,
make_host_allocs(1,
make_host_alloc((char*)"no-such-worker", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
pthread_mutex_lock(&g_dc.persist.lock);
lsn_before = g_dc.persist.current_lsn;
pthread_mutex_unlock(&g_dc.persist.lock);
rc = dispatch_a_job_test(&g_dc, NULL, alloc);
TEST_ASSERT_TRUE_MESSAGE(rc < 0, "dispatch to a nonexistent worker must "
"fail");
pthread_mutex_lock(&g_dc.persist.lock);
lsn_after = g_dc.persist.current_lsn;
pthread_mutex_unlock(&g_dc.persist.lock);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"a dispatch failure ahead of xce_startjob_req() must leave the job "
"state untouched (still PEND) -- this task only gates the SUCCESS "
"commit, it does not change any failure branch's behavior");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"job.allocs must remain uncommitted on this failure path");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(3, j->dispatch_fail_count,
"task 3.3: this retryable failure path increments the counter by "
"exactly 1 from its seeded starting value (2 -> 3), still within "
"the bound (3)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a failed dispatch must append no WAL entry");
free_JobAlloc_ptr(&alloc);
dc_teardown();
printf("test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted: "
"PASS\n");
}
/* ── Durability: WAL_JOB_DISPATCH survives a restart replay ─────────────
*
* Mirrors test_wal_job_dispatch_fact.c's own restart-replay test exactly,
* except the WAL_JOB_DISPATCH fact is produced by a REAL successful
* dispatch_a_job_() call (this task's actual new code), not a hand-built
* wal_jobstate_t -- proving the fact this task's restructuring appends is
* durable and independently sufficient to recover the job as RUNNING, with
* no WAL_JOB_START anywhere in the log. */
static persist_t* dc_make_fresh_persist(const char* tmpdir)
{
persist_t* p = calloc(1, sizeof(persist_t));
persist_config_t cfg;
if (!p) return NULL;
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
persist_set_workdir(p, tmpdir, "ctrl-test");
if (persist_init(p, &cfg, NULL) != 0) {
free(p);
return NULL;
}
if (persist_start(p) != 0) {
persist_fini(p);
free(p);
return NULL;
}
return p;
}
static void dc_teardown_persist(persist_t* p)
{
persist_stop(p);
persist_fini(p);
free(p);
}
/* Encode helper mirroring test_wal_job_dispatch_fact.c's wjd_encode_job
* (capnp_encode_job is file-static in persist.c, so a direct WAL_JOB_SUBMIT
* seed needs its own encoder built from the exported encode_Job/new_Job/
* write_Job codec chain). */
static int dc_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len)
{
struct capn c;
struct capn_segment* cs;
struct Job cj;
Job_ptr jp;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_Job(cs, &cj, (xce_job_t*)j);
jp = new_Job(cs);
write_Job(&cj, jp);
capn_setp(capn_root(&c), 0, jp.p);
*out_len = capn_size(&c);
*out = malloc(*out_len);
if (!*out) { capn_free(&c); return -ENOMEM; }
capn_write_mem(&c, *out, *out_len, 0);
capn_free(&c);
return 0;
}
#define DC_JOBID_REPLAY 91003ULL
static void test_dc_success_appends_wal_job_dispatch_durably(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_dc31_replay_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: real cluster + real disk persist. Seed WAL_JOB_SUBMIT
* directly (so replay can reconstruct the job), then drive an actual
* successful dispatch_a_job_() call and let IT append WAL_JOB_DISPATCH
* -- this is the production code path under test, not a hand-built
* WAL entry. */
{
cluster_t c;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32, .delta = 16 };
persist_config_t cfg;
endpoint_t wep;
job_t* j;
worker_t* w = NULL;
xce_job_alloc_t* alloc;
int rc;
if (makeString((char*)"dc-probe") == NULL)
xstring_table_init(64);
if (!g_dc_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_dc_jobobj_ready = 1;
}
memset(&c, 0, sizeof(c));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX));
c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
c.workermgr = makeInstance(&xkvmgr_type, &str_params);
c.qmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(c.jobmgr);
TEST_ASSERT_NOT_NULL(c.workermgr);
TEST_ASSERT_NOT_NULL(c.qmgr);
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
persist_set_workdir(&c.persist, tmp, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist));
/* Durable WAL_JOB_SUBMIT seed -- no WAL_JOB_START anywhere in this
* log; the recovered RUN state must come from WAL_JOB_DISPATCH
* alone. */
{
job_submit_data_t wal_submit;
xce_job_t wal_jdef;
uint8_t* buf = NULL;
size_t sz = 0;
memset(&wal_submit, 0, sizeof(wal_submit));
wal_submit.queue = (char*)"dq";
wal_submit.user = (char*)"tester";
memset(&wal_jdef, 0, sizeof(wal_jdef));
wal_jdef.jobid = DC_JOBID_REPLAY;
wal_jdef.state = JOB_STAT_PEND;
wal_jdef.submit = &wal_submit;
TEST_ASSERT_EQUAL_INT(0, dc_encode_job(&wal_jdef, &buf, &sz));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(&c.persist, WAL_JOB_SUBMIT, buf,
(uint32_t)sz));
free(buf);
}
/* dc_seed_ready() operates against the module-global g_dc fixture,
* not usable here (this test's cluster_t is a local `c`, and
* dc_setup()/dc_teardown() are deliberately not called in this
* test) -- so the queue/job/worker are seeded directly below. */
{
xce_queue_t qdef;
job_startinfo_t startinfo;
job_submit_data_t submit;
xce_job_t jdef;
hostattrs_t* attrs;
xce_worker_t wdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char*)"dq";
TEST_ASSERT_NOT_NULL(makeQueue(c.qmgr, &qdef));
memset(&startinfo, 0, sizeof(startinfo));
startinfo.cmdline = (char*)"/bin/true";
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"dq";
submit.user = (char*)"tester";
submit.start_info = &startinfo;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = DC_JOBID_REPLAY;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
j = makeJob(c.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j);
attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)"wkr-replay";
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(c.workermgr, &wdef));
}
memset(&wep, 0, sizeof(wep));
wep.fd = -1;
lock(c.workermgr, RTTI_LOCK_READ);
key_find(c.workermgr, (void*)"wkr-replay", (void**)&w);
unlock(c.workermgr);
TEST_ASSERT_NOT_NULL(w);
w->ep = &wep;
alloc = make_job_alloc(DC_JOBID_REPLAY, 1,
make_host_allocs(1,
make_host_alloc((char*)"wkr-replay", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
rc = dispatch_a_job_test(&c, NULL, alloc);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist));
free_JobAlloc_ptr(&alloc);
dc_drain_ep(&wep);
persist_stop(&c.persist);
persist_fini(&c.persist);
dereference(c.workermgr);
dereference(c.qmgr);
dereference(c.jobmgr);
xce_hook_fini(&c.hooksys);
}
/* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only --
* apply_committed_entry()'s WAL_JOB_SUBMIT/WAL_JOB_DISPATCH cases need
* nothing else), recover from disk -- simulates a controller restart. */
{
persist_t* p2 = dc_make_fresh_persist(tmp);
cluster_t c2;
job_t* j2 = NULL;
int rc;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
TEST_ASSERT_NOT_NULL(p2);
memset(&c2, 0, sizeof(c2));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX));
c2.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(c2.jobmgr);
p2->cluster = &c2;
rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
lock(c2.jobmgr, RTTI_LOCK_READ);
key_find(c2.jobmgr, (void*)DC_JOBID_REPLAY, (void**)&j2);
unlock(c2.jobmgr);
TEST_ASSERT_NOT_NULL_MESSAGE(j2,
"restart replay must reconstruct the job via WAL_JOB_SUBMIT");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j2->job.state,
"restart replay must recover the job as RUNNING from the "
"WAL_JOB_DISPATCH fact appended by dispatch_a_job_()'s real "
"success path -- no WAL_JOB_START anywhere in this log");
/* bug-1678: restart replay must recover not just STATE but the
* resource allocation itself, or EXEC_HOST is permanently blank for
* a job that was genuinely RUNNING before shutdown -- the user's
* exact acceptance criterion ("after recover the job should get the
* same as it's state before shutdown"). Mirrors Phase 1's real
* dispatch: host "wkr-replay", one resource "slots" value 1.0. */
TEST_ASSERT_NOT_NULL_MESSAGE(j2->job.allocs,
"restart replay must recover the dispatch-time resource "
"allocation (bug-1678), not just the RUN state");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, j2->job.allocs->nhosts,
"recovered allocation must have exactly 1 host, matching what "
"was dispatched in Phase 1");
TEST_ASSERT_NOT_NULL(j2->job.allocs->allocs);
TEST_ASSERT_NOT_NULL(j2->job.allocs->allocs[0]);
TEST_ASSERT_EQUAL_STRING_MESSAGE("wkr-replay",
j2->job.allocs->allocs[0]->host,
"recovered allocation's host must match the dispatched worker");
TEST_ASSERT_EQUAL_INT(1, j2->job.allocs->allocs[0]->nallocs);
TEST_ASSERT_NOT_NULL(j2->job.allocs->allocs[0]->allocs);
TEST_ASSERT_NOT_NULL(j2->job.allocs->allocs[0]->allocs[0]);
TEST_ASSERT_EQUAL_STRING("slots",
j2->job.allocs->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_FLOAT(1.0f,
j2->job.allocs->allocs[0]->allocs[0]->value);
xce_hook_fini(&c2.hooksys);
if (c2.jobmgr) dereference(c2.jobmgr);
dc_teardown_persist(p2);
}
{
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);
}
printf("test_dc_success_appends_wal_job_dispatch_durably: PASS\n");
}
void test_dispatch_commit(void)
{
UnitySetTestFile(__FILE__);
printf("\n>> dispatch_a_job_() commit-gating tests (Req 2.1-2.4, 4.3, "
"5.1)\n\n");
RUN_TEST(test_dc_success_commits_alloc_state_and_resets_counter);
RUN_TEST(test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted);
RUN_TEST(test_dc_success_appends_wal_job_dispatch_durably);
}
Reference in a new issue View git blame Copy permalink