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

feat(scheduler-core): implement builtin_fcfs lifecycle and private state (task 2.1)

Browse source 
New src/scheduler/src/plugins/builtin_fcfs.c implements
builtin_fcfs_init()/fini() and the private fcfs_plugin_data_t (ordq_t
queues, per-queue ordq_t pendJobs) that all later builtin_fcfs hooks
build on. fcfs.c (today's live job_elect_fn-based picker) is untouched
-- still production, until task 2.4 flips the registration.

picker == xce_sched_plugin_t* calling convention (private state stored
in plugin->data) confirmed consistent with task 1.2's legacy wrapper.

Requirements: 3.1 (context), 3.3
This commit is contained in:
Rongsong Shen 2026-08-03 06:03:49 +00:00
parent 0d63a789a1
commit f7921b961a
4 changed files with 476 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

240
src/scheduler/src/plugins/builtin_fcfs.c Normal file
View file

@ -0,0 +1,240 @@
/*
* builtin_fcfs.c - default xce_job_picker_iface_t implementation
* (scheduler-core spec, Req 3: Builtin default picker).
*
* This is the eventual replacement for plugins/fcfs.c's job_elect_fn-based
* picker (design.md's File Structure Plan: "plugins/fcfs.c # rewritten
* (Phase 1) to xce_job_picker_iface_t, drops job_elect_fn"). That rewrite
* is delivered incrementally across tasks 2.1-2.4, and plugins/fcfs.c is
* still the LIVE, production picker (registered as "xce_sched_fcfs",
* driving real scheduling decisions) until task 2.4 flips the switch. This
* file is therefore kept SEPARATE and UNREGISTERED for now:
* - separate: so nothing in this task can regress fcfs.c's current
* behavior -- fcfs.c is not touched at all by task 2.1.
* - unregistered: builtin_fcfs is not added to any plugin table here;
* that wiring is task 2.4's "Wire builtin_fcfs as the registered
* default" (tasks.md). Until then this file is exercised only by its
* own testsuite (test_picker_fcfs.c, white-box #include).
* A later task (2.4, per design.md's stated end state) is expected to fold
* this file's content over plugins/fcfs.c (or otherwise retire the old
* file) once every hook here is complete and the switch is flipped -- that
* consolidation decision belongs to that task, not this one.
*
* Task 2.1 scope: init()/fini() and the private ordered-queue state
* container only (Req 3 AC 1, AC 2's init/fini rows, AC 3). Every other
* hook (on_queue_update, on_job_push, next_job, ...) is NULL/unimplemented
* here and lands in tasks 2.2/2.3 -- see tasks.md.
*
* "picker" calling convention: every xce_job_picker_iface_t hook receives
* `void* picker` as its first argument. The already-implemented job_elect_fn
* compatibility wrapper (plugin.c's legacy_picker_wrapper_iface, task 1.2)
* requires that argument to be the owning xce_sched_plugin_t* itself --
* legacy_picker_next_job() casts `picker` straight to xce_sched_plugin_t*
* to reach plugin->job_elect_fn, and test_plugin_picker_registration.c
* confirms this by calling `plugin.picker_iface->next_job(&plugin)`. Since
* a single, picker-implementation-agnostic call site (the future picker
* thread loop, tasks.md group 2.4/2.5) must drive both legacy-wrapped and
* native pickers identically, "picker" == xce_sched_plugin_t* for EVERY
* picker, native ones included. builtin_fcfs therefore follows the exact
* convention plugins/fcfs.c's own (old-style) fcfs_init/fcfs_fini already
* use: allocate the private fcfs_plugin_data_t and store it in
* plugin->data (the existing plugin_data_t field on xce_sched_plugin_t),
* not in a new field of its own.
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include "ordq.h"
#include "plugin.h"
#include "sched/types.h"
#include "utils.h"
/* Private per-queue candidate state (Req 3 AC 2 "init" row: "one
* job-ordering structure per queue"). Populated by on_queue_update (task
* 2.2) -- init()/fini() only manage it through fcfs_plugin_data_t.queues's
* ordq_data_killer_t callback below, they never allocate a fcfs_queue_t
* themselves. */
typedef struct {
xce_sched_queue_t* queue;
ordq_t pendJobs;
} fcfs_queue_t;
/* Private per-job candidate wrapper stored inside a queue's pendJobs
* (populated by on_job_push, task 2.2). */
typedef struct {
xce_sched_job_t* job;
} fcfs_job_t;
/* builtin_fcfs's entire private state (Req 3 AC 3: "maintain its own
* private state, not query the core's sched->queues/sched->jobs during
* next_job()"). Allocated by builtin_fcfs_init(), freed by
* builtin_fcfs_fini() -- see below. */
typedef struct {
ordq_t queues; /* ordered by queue->priority desc, see
* fcfs_queue_cmp */
ordq_item_t* cursor; /* iteration cursor over queues; NULL until
* begin_session (task 2.3) sets it -- init() only
* zeroes it here, next_job() (task 2.3) is what
* walks it */
} fcfs_plugin_data_t;
/* ordq_data_killer_t for a queue's pendJobs: frees the fcfs_job_t wrapper.
* Does not touch job->job (the xce_sched_job_t itself) -- that record is
* owned by the scheduler core, not the picker (Req 3 AC 3). */
static void fcfs_job_killer(void* arg)
{
fcfs_job_t* j = (fcfs_job_t*)arg;
if (j == NULL) {
return;
}
FREEUP(j);
}
/* ordq_data_killer_t for fcfs_plugin_data_t.queues: recursively finalizes
* the queue's own pendJobs ordq_t (invoking fcfs_job_killer for every
* fcfs_job_t still queued) before freeing the fcfs_queue_t wrapper itself.
* This is the cascade Req 3 AC 2's "fini" row requires: "Free all per-queue
* ordq_t pendJobs, the ordq_t queues, and plugin data." Does not touch
* q->queue (the xce_sched_queue_t itself) -- owned by the scheduler core. */
static void fcfs_queue_killer(void* arg)
{
fcfs_queue_t* q = (fcfs_queue_t*)arg;
if (q == NULL) {
return;
}
ordq_fini(&(q->pendJobs));
q->queue = NULL;
FREEUP(q);
}
/* ordq_cmp_fn_t for fcfs_plugin_data_t.queues: descending queue->priority
* (Req 3 AC 2 "init" row, Req 3 AC 4 "inter-queue by descending
* queue->priority"). Matches plugins/fcfs.c's fcfs_queue_cmp exactly --
* this is the ordering-equivalence contract task 2.1 sets up for task 2.3's
* next_job() to walk. */
static int fcfs_queue_cmp(void* ctx, void* arg1, void* arg2)
{
fcfs_queue_t* q1 = (fcfs_queue_t*)arg1;
fcfs_queue_t* q2 = (fcfs_queue_t*)arg2;
(void)ctx;
if (q1->queue->priority > q2->queue->priority) {
return -1;
} else if (q1->queue->priority < q2->queue->priority) {
return 1;
}
return 0;
}
/* ordq_cmp_fn_t for a queue's pendJobs: priority desc, queue_time asc (Req
* 3 AC 2 "on_job_push" row, Req 3 AC 4 "intra-queue by priority desc,
* queue_time asc"). Not exercised by task 2.1's own hooks (no job is ever
* pushed by init()), but the private-state container this task builds
* needs a real comparator ready for task 2.2's on_job_push to hand each new
* queue's pendJobs to ordq_init with -- defining it here keeps the
* ordq_init call sites next to the struct they order. */
static int fcfs_job_cmp(void* ctx, void* arg1, void* arg2)
{
fcfs_job_t* j1 = (fcfs_job_t*)arg1;
fcfs_job_t* j2 = (fcfs_job_t*)arg2;
(void)ctx;
if (j1->job->priority > j2->job->priority) {
return -1;
} else if (j1->job->priority < j2->job->priority) {
return 1;
} else if (j1->job->queue_time < j2->job->queue_time) {
return -1;
} else if (j1->job->queue_time > j2->job->queue_time) {
return 1;
}
return 0;
}
/*
* builtin_fcfs_init - xce_job_picker_iface_t.init (Req 3 AC 2 "init" row).
*
* Allocates fcfs_plugin_data_t, initializes its ordq_t queues (ordered by
* queue->priority desc via fcfs_queue_cmp), and sets the iteration cursor
* to NULL. Stores the result in ((xce_sched_plugin_t*)picker)->data -- see
* the file header comment for why "picker" is the owning
* xce_sched_plugin_t*, not the private struct itself.
*
* `core` is accepted per the interface signature but deliberately never
* dereferenced: xce_sched_core_t is still only forward-declared (task
* 1.1), nothing is defined on it until group 4's core extraction.
*
* Returns 0 on success, a negative errno (-EINVAL, -ENOMEM) on failure --
* on failure, plugin->data is left untouched (NULL, if this is the first
* init) so a caller cannot observe a half-initialized picker.
*/
int builtin_fcfs_init(void* picker, xce_sched_core_t* core)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
fcfs_plugin_data_t* priv = NULL;
int rc = 0;
(void)core;
if (plugin == NULL) {
return -EINVAL;
}
priv = (fcfs_plugin_data_t*)calloc(1, sizeof(*priv));
if (priv == NULL) {
return -ENOMEM;
}
rc = ordq_init(&(priv->queues), (ordq_data_killer_t)fcfs_queue_killer,
(ordq_cmp_fn_t)fcfs_queue_cmp);
if (rc < 0) {
FREEUP(priv);
return rc;
}
priv->cursor = NULL;
plugin->data = (plugin_data_t)priv;
return 0;
}
/*
* builtin_fcfs_fini - xce_job_picker_iface_t.fini (Req 3 AC 2 "fini" row).
*
* Frees all per-queue ordq_t pendJobs (via fcfs_queue_killer, invoked by
* ordq_fini's killer callback for every fcfs_queue_t still in
* priv->queues), the ordq_t queues itself, and the plugin data struct.
* Safe to call with picker == NULL or with a picker that was never (or
* already) init()'d -- same defensive-NULL contract plugins/fcfs.c's own
* fcfs_fini follows.
*/
void builtin_fcfs_fini(void* picker)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
fcfs_plugin_data_t* priv = NULL;
if (plugin == NULL) {
return;
}
priv = (fcfs_plugin_data_t*)plugin->data;
if (priv == NULL) {
return;
}
ordq_fini(&(priv->queues));
FREEUP(priv);
plugin->data = NULL;
}

13
src/scheduler/testsuite/meson.build
View file

@ -93,6 +93,17 @@ sched_picker_config_fields_test = executable('sched_picker_config_fields_test',
dependencies: [scheduler_test_deps],
install: false)
# White-box test for builtin_fcfs's xce_job_picker_iface_t lifecycle hooks
# (scheduler-core spec, task 2.1). It #includes ../src/plugins/builtin_fcfs.c
# directly (same white-box pattern as sched_picker_registration_test) so it
# exercises the real builtin_fcfs_init()/builtin_fcfs_fini() and the real
# private fcfs_plugin_data_t layout, not a re-implementation of either.
sched_picker_fcfs_test = executable('sched_picker_fcfs_test',
['test_picker_fcfs.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Register as meson tests so `meson test --suite scheduler` runs them all.
# detect_leaks=0 matches the suite convention (fixtures are not torn down).
test('scheduler test', scheduler_test, suite: 'scheduler',
@ -111,3 +122,5 @@ test('scheduler picker registration test', sched_picker_registration_test, suite
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler picker config fields test', sched_picker_config_fields_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler picker fcfs test', sched_picker_fcfs_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

222
src/scheduler/testsuite/test_picker_fcfs.c Normal file
View file

@ -0,0 +1,222 @@
/*
* test_picker_fcfs.c - white-box tests for builtin_fcfs's
* xce_job_picker_iface_t hooks (scheduler-core spec, Req 3/4).
*
* Task 2.1 (this file's first content): lifecycle hooks only --
* builtin_fcfs_init()/builtin_fcfs_fini() and the private ordered-queue
* state they allocate/free.
* RED: before task 2.1, plugins/builtin_fcfs.c did not exist -- this
* file failed to compile (no such file/symbols).
* GREEN: builtin_fcfs_init()/builtin_fcfs_fini() allocate/free the
* picker's private ordered-queue state exactly as Req 3 AC 2's
* init/fini hook-table rows specify, with no core-table queries
* (Req 3 AC 3).
*
* This file is expected to grow across tasks 2.2/2.3 as the remaining
* hooks (on_queue_update, on_job_push, next_job, ...) land -- matching
* design.md's File Structure Plan, which names a single test_picker_fcfs.c
* for all of Req 3/4's hook-by-hook and ordering-equivalence coverage.
*
* #include's the real plugins/builtin_fcfs.c (white-box, same convention
* as test_plugin_picker_registration.c's #include "../src/plugin.c") so
* this test exercises the real allocator/destructor and the real private
* struct layout, not a re-implementation of either.
*/
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "plugin.h"
#include "sched/types.h"
#include "../src/plugins/builtin_fcfs.c"
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture: a bare xce_sched_plugin_t -- "picker" is passed to every hook
* as this struct's address, the same convention the job_elect_fn
* compatibility wrapper requires (plugin.c's legacy_picker_next_job casts
* its `picker` argument straight to xce_sched_plugin_t*, confirmed live by
* test_plugin_picker_registration.c's
* `plugin.picker_iface->next_job(&plugin)`) and the only convention that
* lets one uniform call site drive both legacy-wrapped and native pickers
* without knowing which kind it holds. builtin_fcfs stores its private
* fcfs_plugin_data_t* in plugin.data, mirroring today's
* fcfs_init/fcfs_fini (plugins/fcfs.c) exactly. Nothing is torn down
* between tests (suite convention, see meson.build: detect_leaks=0 for the
* normal run; the leak-checked run is separate, see task 2.1's Validation).
* ---------------------------------------------------------------------- */
static void fixture_init(xce_sched_plugin_t* plugin)
{
memset(plugin, 0, sizeof(*plugin));
}
/* Req 3 AC 2 "init" row + AC 3: init() allocates fcfs_plugin_data_t,
* initializes an EMPTY ordq_t queues, and sets the iteration cursor to
* NULL -- the picker holds an empty private candidate set immediately
* after init(), before any state-change hook has ever run. */
static void test_builtin_fcfs_init_allocates_empty_private_state(void)
{
xce_sched_plugin_t plugin;
fcfs_plugin_data_t* priv;
int rc;
fixture_init(&plugin);
rc = builtin_fcfs_init(&plugin, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(plugin.data);
priv = (fcfs_plugin_data_t*)plugin.data;
TEST_ASSERT_NULL(priv->queues.head); /* empty ordq_t queues */
TEST_ASSERT_NULL(priv->cursor); /* cursor reset to NULL */
TEST_ASSERT_EQUAL_PTR(fcfs_queue_cmp, priv->queues.cmp);
TEST_ASSERT_EQUAL_PTR(fcfs_queue_killer, priv->queues.killer);
/* Paired fini() -- every init()-calling test in this file balances it
* with fini() so the whole binary is leak-clean under ASAN's default
* leak detector (task 2.1 Validation: the leak-checked run proves
* fini() releases picker-owned memory, which only means something if
* every init() this suite performs is actually matched by a fini()). */
builtin_fcfs_fini(&plugin);
}
/* init() never dereferences its xce_sched_core_t* core argument (task 2.1
* scope note: xce_sched_core_t is still only forward-declared -- nothing is
* defined on it until group 4). Passing an obviously-invalid non-NULL
* pointer and getting a clean, correct init proves init() doesn't touch
* it. */
static void test_builtin_fcfs_init_does_not_dereference_core(void)
{
xce_sched_plugin_t plugin;
int rc;
fixture_init(&plugin);
rc = builtin_fcfs_init(&plugin, (xce_sched_core_t*)0x1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(plugin.data);
builtin_fcfs_fini(&plugin); /* see paired-fini note above */
}
/* init() rejects a NULL picker rather than crashing -- every other
* lifecycle hook in this codebase (fcfs_fini, plugin_register) validates
* its handle before touching it. */
static void test_builtin_fcfs_init_rejects_null_picker(void)
{
int rc = builtin_fcfs_init(NULL, NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
/* Req 3 AC 2 "fini" row: fini() frees plugin data even when the private
* ordq_t queues is empty (the common case immediately after init(), before
* any queue has ever been pushed), and clears plugin->data so a
* use-after-fini is observable rather than silently reusing freed
* memory. */
static void test_builtin_fcfs_fini_releases_empty_state(void)
{
xce_sched_plugin_t plugin;
fixture_init(&plugin);
TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
builtin_fcfs_fini(&plugin);
TEST_ASSERT_NULL(plugin.data);
}
/* fini(NULL) and fini() on a plugin with no private state must not crash --
* defensive-null contract every fini_fn in this codebase follows (see
* plugins/fcfs.c's fcfs_fini). */
static void test_builtin_fcfs_fini_handles_null_and_unset_picker(void)
{
xce_sched_plugin_t plugin;
builtin_fcfs_fini(NULL);
fixture_init(&plugin); /* plugin.data == NULL, never init()'d */
builtin_fcfs_fini(&plugin);
TEST_ASSERT_NULL(plugin.data);
}
/* Req 3 AC 2 "fini" row, full cascade: "Free all per-queue ordq_t
* pendJobs, the ordq_t queues, and plugin data." Manually populates the
* private queues/pendJobs structures the same way a future
* on_queue_update/on_job_push (task 2.2) would, then relies on fini()'s
* ordq_fini cascade (via fcfs_queue_killer -> ordq_fini(&q->pendJobs) ->
* fcfs_job_killer per entry) to release every layer. This test's own
* assertions only confirm the call completes and plugin.data is cleared;
* the actual "all picker-owned memory is released" proof for this
* Observable comes from running this binary under ASAN's default leak
* detector (task 2.1 Validation) -- not something a Unity assertion alone
* can express. */
static void test_builtin_fcfs_fini_releases_populated_queues_and_jobs(void)
{
xce_sched_plugin_t plugin;
fcfs_plugin_data_t* priv;
xce_sched_queue_t q1 = { .name = "q1", .priority = 5 };
xce_sched_queue_t q2 = { .name = "q2", .priority = 1 };
xce_sched_job_t j1 = { .jobid = 1, .priority = 3 };
xce_sched_job_t j2a = { .jobid = 2, .priority = 7 };
xce_sched_job_t j2b = { .jobid = 3, .priority = 4 };
fcfs_queue_t* fq1;
fcfs_queue_t* fq2;
fcfs_job_t* fj1;
fcfs_job_t* fj2a;
fcfs_job_t* fj2b;
fixture_init(&plugin);
TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
priv = (fcfs_plugin_data_t*)plugin.data;
/* queue 1: one pending job */
fq1 = (fcfs_queue_t*)calloc(1, sizeof(*fq1));
fq1->queue = &q1;
TEST_ASSERT_EQUAL_INT(0, ordq_init(&(fq1->pendJobs),
(ordq_data_killer_t)fcfs_job_killer,
(ordq_cmp_fn_t)fcfs_job_cmp));
fj1 = (fcfs_job_t*)calloc(1, sizeof(*fj1));
fj1->job = &j1;
TEST_ASSERT_EQUAL_INT(0, ordq_put(&(fq1->pendJobs), fj1));
TEST_ASSERT_EQUAL_INT(0, ordq_put(&(priv->queues), fq1));
/* queue 2: two pending jobs */
fq2 = (fcfs_queue_t*)calloc(1, sizeof(*fq2));
fq2->queue = &q2;
TEST_ASSERT_EQUAL_INT(0, ordq_init(&(fq2->pendJobs),
(ordq_data_killer_t)fcfs_job_killer,
(ordq_cmp_fn_t)fcfs_job_cmp));
fj2a = (fcfs_job_t*)calloc(1, sizeof(*fj2a));
fj2a->job = &j2a;
fj2b = (fcfs_job_t*)calloc(1, sizeof(*fj2b));
fj2b->job = &j2b;
TEST_ASSERT_EQUAL_INT(0, ordq_put(&(fq2->pendJobs), fj2a));
TEST_ASSERT_EQUAL_INT(0, ordq_put(&(fq2->pendJobs), fj2b));
TEST_ASSERT_EQUAL_INT(0, ordq_put(&(priv->queues), fq2));
builtin_fcfs_fini(&plugin);
TEST_ASSERT_NULL(plugin.data);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_builtin_fcfs_init_allocates_empty_private_state);
RUN_TEST(test_builtin_fcfs_init_does_not_dereference_core);
RUN_TEST(test_builtin_fcfs_init_rejects_null_picker);
RUN_TEST(test_builtin_fcfs_fini_releases_empty_state);
RUN_TEST(test_builtin_fcfs_fini_handles_null_and_unset_picker);
RUN_TEST(test_builtin_fcfs_fini_releases_populated_queues_and_jobs);
return UNITY_END();
}