feat(job-array-one-based-index): classify every job submission first (task 2.1)
handle_jobsubmit classifies (first_jobid, array_size) before the field checks, the before-event hook and any registry access: a tagged array starting at index 0 is answered with the new E_XCE_ARRAY_START0 and an ERROR log asking for the client to be upgraded (7.1, wins per 7.9); size 0, tags 2/3, untagged multi-element submissions and arrays whose last element exceeds 32767 (checked without addition) get -EINVAL. Nothing is stored for a rejected submission. This also closes a crash: a size-0 submission used to reach payload building and overflow calloc. Existing array tests move to tagged block|1 first IDs. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
ef728435d4
commit
e6c484a0e7
GPG key ID:
80533536B0EB651C
11 changed files with 574 additions and 19 deletions
|
|
@ -25,7 +25,7 @@
|
|||
- _Boundary: IdgenArray_
|
||||
|
||||
- [ ] 2. Controller: array admission and collision detection
|
||||
- [ ] 2.1 Classify every submission before any other check
|
||||
- [x] 2.1 Classify every submission before any other check
|
||||
- Add the distinct "array numbering starts at 1" error status and its message text to the shared error table
|
||||
- Classify a submission from its first ID and 64-bit size only: start-at-0 array first; then size 0, an unknown marking, an unmarked multi-element submission, and a last element above 32767 (computed without addition so no size can wrap)
|
||||
- Run the classification as the first step after the reply is initialised, before the existing field checks and the before-event hook; answer start-at-0 with the new status plus an ERROR log saying numbering starts at 1 and the client must be upgraded, and the other classes with invalid-argument; store nothing
|
||||
|
|
|
|||
|
|
@ -107,6 +107,20 @@ extern int handle_join(cluster_t*, endpoint_t*, xce_request_t*);
|
|||
extern int handle_reply(endpoint_t*, struct Request*, cluster_t*);
|
||||
extern int handle_jobdispatch(cluster_t*, endpoint_t*, xce_request_t*);
|
||||
extern int handle_jobsubmit(cluster_t*, endpoint_t*, xce_request_t*);
|
||||
|
||||
/* job-array-one-based-index (Req 7.1-7.5, 7.9): classification of a job
|
||||
* submission from its first ID and its wire (UInt64) element count only.
|
||||
* Pure, no locks. INDEX0: an array (tag TAG_JOB_ARRAY) whose first element
|
||||
* number is 0 (7.1, checked first). INVALID: size 0 (7.4), a tag other than
|
||||
* 0 or TAG_JOB_ARRAY (7.5), an unmarked multi-element submission (7.3), or
|
||||
* an array whose last element number would exceed MAX_ARRAY_IDX (7.2). */
|
||||
typedef enum {
|
||||
JOBSUBMIT_OK,
|
||||
JOBSUBMIT_INDEX0,
|
||||
JOBSUBMIT_INVALID
|
||||
} jobsubmit_class_t;
|
||||
extern jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid,
|
||||
uint64_t size);
|
||||
#ifdef XCE_TESTING
|
||||
/* bug-1419: clears job.c's process-lifetime in-flight-jobid registry.
|
||||
* Tests that deliberately leave a submission's commit continuation
|
||||
|
|
|
|||
|
|
@ -2431,6 +2431,31 @@ static int handle_jobsubmit_array(cluster_t* c, endpoint_t* ep,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* job-array-one-based-index (Req 7.1-7.5, 7.9; design "ControllerValidate"):
|
||||
* classify a submission from its first ID and its wire element count only.
|
||||
* 7.1 is checked first, so an index-0 array is answered as such whatever
|
||||
* else is wrong with it (7.9); then 7.4, 7.5, 7.3 and 7.2, all INVALID.
|
||||
* 7.2 is evaluated as size > 32768 - idx (idx >= 1 here), never as
|
||||
* idx + size - 1 > 32767, so no 64-bit size can wrap past it. */
|
||||
jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid, uint64_t size)
|
||||
{
|
||||
uint64_t tag = EXTRACT_ID_TAG(first_jobid);
|
||||
int is_array = is_array_tag(first_jobid);
|
||||
uint64_t idx = array_idx(first_jobid);
|
||||
|
||||
if (is_array && idx == 0) /* 7.1 */
|
||||
return JOBSUBMIT_INDEX0;
|
||||
if (size == 0) /* 7.4 */
|
||||
return JOBSUBMIT_INVALID;
|
||||
if (tag != 0 && !is_array) /* 7.5 */
|
||||
return JOBSUBMIT_INVALID;
|
||||
if (!is_array && size > 1) /* 7.3 */
|
||||
return JOBSUBMIT_INVALID;
|
||||
if (is_array && size > (MAX_ARRAY_IDX + 1) - idx) /* 7.2 */
|
||||
return JOBSUBMIT_INVALID;
|
||||
return JOBSUBMIT_OK;
|
||||
}
|
||||
|
||||
int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
|
||||
{
|
||||
uint64_t jobid;
|
||||
|
|
@ -2444,6 +2469,38 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
|
|||
|
||||
init_reply_by_req(&reply, request);
|
||||
|
||||
/* job-array-one-based-index (Req 7.1-7.6, 7.9): classify before the
|
||||
* field checks, the BEFORE_EVENT hook and any registry/table access, so
|
||||
* a rejected submission is answered with its class's status alone and
|
||||
* nothing is stored. */
|
||||
switch (jobsubmit_classify(submit->jobid, submit->array_size)) {
|
||||
case JOBSUBMIT_INDEX0: {
|
||||
char idstr[64];
|
||||
|
||||
fmt_jobid(idstr, sizeof(idstr), submit->jobid);
|
||||
ERROR("jobsubmit: rejected array submission %s (%lu elements) from "
|
||||
"%s: array element numbers start at 1, this submission starts "
|
||||
"at 0; the submitting client must be upgraded",
|
||||
idstr, (unsigned long)submit->array_size,
|
||||
(request->from != NULL && request->from->name != NULL)
|
||||
? request->from->name
|
||||
: "unknown");
|
||||
reply.status = -E_XCE_ARRAY_START0;
|
||||
reply.data_which = RequestReply_data_nodata;
|
||||
|
||||
xce_reply(ep, c->myhost, &reply);
|
||||
return -E_XCE_ARRAY_START0;
|
||||
}
|
||||
case JOBSUBMIT_INVALID:
|
||||
reply.status = -EINVAL;
|
||||
reply.data_which = RequestReply_data_nodata;
|
||||
|
||||
xce_reply(ep, c->myhost, &reply);
|
||||
return -EINVAL;
|
||||
case JOBSUBMIT_OK:
|
||||
break;
|
||||
}
|
||||
|
||||
if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue)
|
||||
|| (submit->submit->start_info == NULL)
|
||||
|| NULLSTR(submit->submit->start_info->cmdline)) {
|
||||
|
|
|
|||
|
|
@ -110,6 +110,7 @@ controller_unit_test_src = [
|
|||
'testsuite/test_commit_engine.c',
|
||||
'testsuite/test_commit_barrier_rollout.c',
|
||||
'testsuite/test_commit_barrier_reply.c',
|
||||
'testsuite/test_jobsubmit_validate.c',
|
||||
'testsuite/test_egress_filter.c',
|
||||
'testsuite/test_demotion_convergence.c',
|
||||
'testsuite/test_convergence_verification.c',
|
||||
|
|
|
|||
|
|
@ -113,6 +113,7 @@ extern void test_dispatch_fail_retry(void);
|
|||
extern void test_jobdispatch_no_double_notify(void);
|
||||
extern void test_worker_resync(void);
|
||||
extern void test_standby_state_peer_idx(void);
|
||||
extern void test_jobsubmit_validate(void);
|
||||
|
||||
void setUp(void) { return; }
|
||||
|
||||
|
|
@ -303,6 +304,7 @@ int main(int argc, char* argv[])
|
|||
TEST(test_jobdispatch_no_double_notify);
|
||||
TEST(test_worker_resync);
|
||||
TEST(test_standby_state_peer_idx);
|
||||
TEST(test_jobsubmit_validate);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1636,7 +1636,12 @@ static void test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs(v
|
|||
* nothing") and is pinned below.) */
|
||||
static void test_jobsubmit_array_timeout_appends_exactly_one_wal_entry_and_replies_once(void)
|
||||
{
|
||||
const uint64_t base_jobid = 9140;
|
||||
/* job-array-one-based-index task 2.1 (Req 7.1-7.3): an array
|
||||
* submission must carry the array tag and start at element 1 of an
|
||||
* aligned block; untagged multi-element or index-0 IDs are rejected by
|
||||
* jobsubmit_classify before reaching the commit path under test. */
|
||||
const uint64_t base_jobid = MAKE_JOBID(
|
||||
0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)1 << N_ARRAY_IDX_BITS) | 1);
|
||||
const uint64_t array_size = 3;
|
||||
uint64_t lsn_before, watermark;
|
||||
int create_rc, join_rc;
|
||||
|
|
@ -1871,7 +1876,9 @@ static void test_jobsubmit_success_reply_after_apply_carries_jobid(void)
|
|||
static void test_jobsubmit_array_success_appends_one_entry_and_replies_once_with_base_jobid(
|
||||
void)
|
||||
{
|
||||
const uint64_t base_jobid = 9210;
|
||||
/* tagged block|1 first ID (job-array-one-based-index task 2.1) */
|
||||
const uint64_t base_jobid = MAKE_JOBID(
|
||||
0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)2 << N_ARRAY_IDX_BITS) | 1);
|
||||
const uint64_t array_size = 3;
|
||||
uint64_t lsn_before, watermark;
|
||||
cbrp_reply_obs_t final = { 0 };
|
||||
|
|
|
|||
|
|
@ -2324,16 +2324,24 @@ static void test_job_array_submit_one_entry_one_continuation_across_sizes(void)
|
|||
* test_job_array_submit_payload_size_independent_of_array_size above
|
||||
* compares an arbitrary "large" sample (10000) against a small one. This
|
||||
* test additionally exercises the REAL structural ceiling design.md relies
|
||||
* on to satisfy Requirement 7 without a chunking policy: idgen() rejects
|
||||
* arraySize > MAX_ARRAY_IDX + 1, so MAX_ARRAY_IDX + 1 (32768, config.h) is
|
||||
* the largest array_size that can ever legally reach this function in
|
||||
* production. Confirms the encoded payload at that exact ceiling is still
|
||||
* identical in size to a tiny array's payload AND stays well under a sane
|
||||
* absolute byte bound -- Requirement 7.1's "no single record exceeds a
|
||||
* bounded size" is an absolute claim, not only a relative one. */
|
||||
* on to satisfy Requirement 7 without a chunking policy: since
|
||||
* job-array-one-based-index task 2.1, handle_jobsubmit's jobsubmit_classify
|
||||
* admits only tagged arrays numbered from 1 whose last element is at most
|
||||
* MAX_ARRAY_IDX, so MAX_ARRAY_IDX (32767, config.h) elements starting at a
|
||||
* tagged block|1 first ID is the largest array that can ever legally reach
|
||||
* this function in production. Confirms the encoded payload at that exact
|
||||
* ceiling is still identical in size to a tiny array's payload AND stays
|
||||
* well under a sane absolute byte bound -- Requirement 7.1's "no single
|
||||
* record exceeds a bounded size" is an absolute claim, not only a relative
|
||||
* one. */
|
||||
static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void)
|
||||
{
|
||||
const uint64_t ceiling_array_size = MAX_ARRAY_IDX + 1; /* 32768 */
|
||||
const uint64_t ceiling_array_size = MAX_ARRAY_IDX; /* 32767 */
|
||||
/* tagged block|1 first IDs (job-array-one-based-index task 2.1) */
|
||||
const uint64_t small_first = MAKE_JOBID(
|
||||
0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)3 << N_ARRAY_IDX_BITS) | 1);
|
||||
const uint64_t ceiling_first = MAKE_JOBID(
|
||||
0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)4 << N_ARRAY_IDX_BITS) | 1);
|
||||
wal_entry_hdr_t hdr_small, hdr_ceiling;
|
||||
uint64_t out_lsn = 0;
|
||||
int rc;
|
||||
|
|
@ -2344,7 +2352,7 @@ static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void)
|
|||
jap_notify_reset();
|
||||
|
||||
off_small = (off_t)g_jap.seg_offset;
|
||||
rc = persist_submit_job_array_async(&g_jap, 40000, 2, &g_jap_spec,
|
||||
rc = persist_submit_job_array_async(&g_jap, small_first, 2, &g_jap_spec,
|
||||
&out_lsn, jap_notify_fn, &g_jap_obs);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
|
|
@ -2352,17 +2360,18 @@ static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void)
|
|||
|
||||
jap_notify_reset();
|
||||
off_ceiling = (off_t)g_jap.seg_offset;
|
||||
rc = persist_submit_job_array_async(&g_jap, 50000, ceiling_array_size,
|
||||
&g_jap_spec, &out_lsn,
|
||||
jap_notify_fn, &g_jap_obs);
|
||||
|
||||
rc = persist_submit_job_array_async(&g_jap, ceiling_first,
|
||||
ceiling_array_size, &g_jap_spec,
|
||||
&out_lsn, jap_notify_fn, &g_jap_obs);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
jap_read_wal_hdr_at(&g_jap, off_ceiling, &hdr_ceiling));
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len,
|
||||
hdr_ceiling.payload_len,
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(
|
||||
hdr_small.payload_len, hdr_ceiling.payload_len,
|
||||
"the payload size at the actual upstream array_size ceiling "
|
||||
"(MAX_ARRAY_IDX + 1 = 32768) must be identical to a tiny array's "
|
||||
"(MAX_ARRAY_IDX = 32767) must be identical to a tiny array's "
|
||||
"payload size -- proving Requirement 7's bound holds at the real "
|
||||
"cap, not just at an arbitrary large sample");
|
||||
TEST_ASSERT_TRUE_MESSAGE(hdr_ceiling.payload_len < 4096,
|
||||
|
|
|
|||
444
src/controller/testsuite/test_jobsubmit_validate.c
Normal file
444
src/controller/testsuite/test_jobsubmit_validate.c
Normal file
|
|
@ -0,0 +1,444 @@
|
|||
/*
|
||||
* 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()).
|
||||
*/
|
||||
|
||||
/* 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);
|
||||
}
|
||||
|
||||
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 */
|
||||
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();
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
|
@ -23,6 +23,8 @@ enum {
|
|||
E_XCE_REDIRECT, /* Standby is redirecting client to primary */
|
||||
E_XCE_NOT_READY, /* Controller in CTRL_INIT or CTRL_ELECTING state */
|
||||
E_XCE_STANDBY, /* Controller in CTRL_STANDBY; op is primary-only */
|
||||
E_XCE_ARRAY_START0, /* Array submission numbered from 0; numbering starts
|
||||
* at 1, the submitting client must be upgraded */
|
||||
E_XCE_MAX
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "utils.h"
|
||||
|
|
@ -76,6 +77,20 @@ static void test_strerror_out_of_range(void)
|
|||
TEST_ASSERT_TRUE(strcmp(buf, "unknown error") == 0);
|
||||
}
|
||||
|
||||
/* job-array-one-based-index (Req 7.1): the distinct array-numbering status
|
||||
* sits right after E_XCE_STANDBY and its text says numbering starts at 1
|
||||
* and the client must be upgraded. */
|
||||
static void test_strerror_array_start0(void)
|
||||
{
|
||||
char buf[128];
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(E_XCE_STANDBY + 1, E_XCE_ARRAY_START0);
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
xce_strerror_r(E_XCE_ARRAY_START0, buf, sizeof(buf)));
|
||||
TEST_ASSERT_NOT_NULL(strstr(buf, "start at 1"));
|
||||
TEST_ASSERT_NOT_NULL(strstr(buf, "upgrade"));
|
||||
}
|
||||
|
||||
void test_error(void)
|
||||
{
|
||||
printf("\n>> error test\n\n");
|
||||
|
|
@ -83,6 +98,7 @@ void test_error(void)
|
|||
RUN_TEST(test_error_setget);
|
||||
RUN_TEST(test_strerror_all_xce_codes);
|
||||
RUN_TEST(test_strerror_out_of_range);
|
||||
RUN_TEST(test_strerror_array_start0);
|
||||
#if 0
|
||||
RUN_TEST(test_strerror_r);
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -438,7 +438,10 @@ static char* xce_err_str[] = { "invalid job", /* E_XCE_BAD_JOB */
|
|||
"no known primary", /* E_XCE_NOMASTER */
|
||||
"redirect to primary", /* E_XCE_REDIRECT */
|
||||
"controller not ready", /* E_XCE_NOT_READY */
|
||||
"controller is standby" };/* E_XCE_STANDBY */
|
||||
"controller is standby", /* E_XCE_STANDBY */
|
||||
"array element numbers start at 1; "
|
||||
"upgrade the submitting client" };
|
||||
/* E_XCE_ARRAY_START0 */
|
||||
|
||||
_Static_assert(sizeof(xce_err_str) / sizeof(xce_err_str[0])
|
||||
== (size_t)(E_XCE_MAX - XCE_ERRNO_BASE),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue