scheduler: time-window dispatch reason (PEND-reason #17/E11)

Add queue dispatch-window support so a queue whose window is closed
(queues.yaml dispatch_window, e.g. 19:00-08:30) is treated as
non-electable and reports PREASON_TIME_WINDOW.

- lib/window.c: new xce_timewindow_is_open() — NULL/empty always open,
  same-day and overnight (close < open) segments, day-scoped (0=every day,
  1-7=Mon..Sun).
- scheduler: make_sched_queue duplicates dispatchwindow_config onto
  xce_sched_queue_t.dispatchwindow (freed by kill_sched_queue);
  sched_queue_is_ready() checks it; sched_queue_status_reason() now takes
  the queue (was a bare status bitmask) and maps a closed window to
  PREASON_TIME_WINDOW.
- fcfs: keep the windowed-out queue skip; do NOT attach the queue-status
  reason to job->schedobj from the picker thread (bug-1521: races the
  allocator thread's MOVE_PTR(obj->reason,...) and corrupts the heap).
- tests: test_window.c (5 cases) covers no-window/daily/overnight/
  day-scoped/multi-seg; full suite 295/295 pass.

Cluster-verified: with a 09:00-17:00 window on the xce queue (currently
outside the window), fcfs reports "queue xce ... ready=0 reason=8"
(PREASON_TIME_WINDOW).

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-08-15 06:50:51 +00:00
parent 53c0083410
commit ee791791bd
10 changed files with 243 additions and 34 deletions

View file

@ -124,6 +124,7 @@ extern int copy_int_list(int**, int*, int);
extern int copy_strlist(xce_strlist_t*, xce_strlist_t);
extern void free_timewindow_config(xce_timewindow_config_t*);
extern xce_timewindow_config_t* dup_timewindow_config(xce_timewindow_config_t*);
extern int xce_timewindow_is_open(xce_timewindow_config_t*, time_t);
extern void free_resource_thresholds(xce_res_threshold_t**, int);
extern xce_res_threshold_t** dup_resource_thresholds(xce_res_threshold_t**,
int);

View file

@ -152,6 +152,7 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_delay.c',
'testsuite/test_ordq.c',
'testsuite/test_preason.c',
'testsuite/test_window.c',
'testsuite/test_kvmgr.c',
'testsuite/test_group.c',
'testsuite/test_xlist.c',

View file

@ -14,6 +14,7 @@ extern void test_utils(void);
extern void test_config(void);
extern void test_idgen(void);
extern void test_preason(void);
extern void test_window(void);
extern void test_ordq(void);
extern void test_delay(void);
extern void test_kvmgr(void);
@ -133,6 +134,7 @@ int main(int argc, char* argv[])
TEST(test_error);
TEST(test_utils);
TEST(test_preason);
TEST(test_window);
TEST(test_ordq);
TEST(test_delay);
TEST(test_kvmgr);

View file

@ -0,0 +1,133 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "unity.h"
#include "utils.h"
#include "xce/types.h"
/* 2026-08-15 is a Saturday (tm_wday=6 → our day index 6), 2026-08-16 Sunday. */
static time_t at(int day_of_month, int hour, int min)
{
struct tm tm;
memset(&tm, 0, sizeof(tm));
tm.tm_year = 2026 - 1900;
tm.tm_mon = 8 - 1;
tm.tm_mday = day_of_month;
tm.tm_hour = hour;
tm.tm_min = min;
tm.tm_isdst = -1;
return mktime(&tm);
}
static xce_timewindow_config_t* make_window(int nseg)
{
xce_timewindow_config_t* w = (xce_timewindow_config_t*)calloc(1, sizeof(*w));
w->nsegs = nseg;
w->segs = (xce_timewindow_seg_config_t**)calloc(
nseg, sizeof(xce_timewindow_seg_config_t*));
return w;
}
static void add_seg(xce_timewindow_config_t* w, int idx, int day, int oh,
int om, int ch, int cm)
{
xce_timewindow_seg_config_t* s
= (xce_timewindow_seg_config_t*)calloc(1, sizeof(*s));
s->open = (xce_timewindow_time_t*)calloc(1, sizeof(xce_timewindow_time_t));
s->close = (xce_timewindow_time_t*)calloc(1, sizeof(xce_timewindow_time_t));
s->open->day = day;
s->open->hour = oh;
s->open->min = om;
s->close->day = day;
s->close->hour = ch;
s->close->min = cm;
w->segs[idx] = s;
}
static void test_no_window_always_open(void)
{
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(NULL, at(15, 10, 0)));
xce_timewindow_config_t* empty = make_window(0);
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(empty, at(15, 10, 0)));
free_timewindow_config(empty);
}
static void test_daily_window(void)
{
xce_timewindow_config_t* w = make_window(1);
/* every day 09:00–17:00 */
add_seg(w, 0, 0, 9, 0, 17, 0);
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 10, 0)));
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 9, 0)));
TEST_ASSERT_EQUAL_INT(0, xce_timewindow_is_open(w, at(15, 17, 0)));
TEST_ASSERT_EQUAL_INT(0, xce_timewindow_is_open(w, at(15, 18, 0)));
TEST_ASSERT_EQUAL_INT(0, xce_timewindow_is_open(w, at(15, 8, 59)));
free_timewindow_config(w);
}
static void test_overnight_window(void)
{
xce_timewindow_config_t* w = make_window(1);
/* every day 19:00–07:00 */
add_seg(w, 0, 0, 19, 0, 7, 0);
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 20, 0)));
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 23, 59)));
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(16, 3, 0)));
TEST_ASSERT_EQUAL_INT(0, xce_timewindow_is_open(w, at(16, 12, 0)));
free_timewindow_config(w);
}
static void test_day_scoped_window(void)
{
xce_timewindow_config_t* w = make_window(1);
/* Saturday (day 6) 09:00–17:00 only */
add_seg(w, 0, 6, 9, 0, 17, 0);
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 10, 0))); /* Sat */
TEST_ASSERT_EQUAL_INT(0, xce_timewindow_is_open(w, at(16, 10, 0))); /* Sun */
free_timewindow_config(w);
}
static void test_multi_seg_any_open(void)
{
xce_timewindow_config_t* w = make_window(2);
/* 09:00–12:00 and 14:00–18:00 */
add_seg(w, 0, 0, 9, 0, 12, 0);
add_seg(w, 1, 0, 14, 0, 18, 0);
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 10, 0)));
TEST_ASSERT_EQUAL_INT(1, xce_timewindow_is_open(w, at(15, 15, 0)));
TEST_ASSERT_EQUAL_INT(0, xce_timewindow_is_open(w, at(15, 13, 0)));
free_timewindow_config(w);
}
void test_window(void)
{
printf("\n>> window test\n\n");
RUN_TEST(test_no_window_always_open);
RUN_TEST(test_daily_window);
RUN_TEST(test_overnight_window);
RUN_TEST(test_day_scoped_window);
RUN_TEST(test_multi_seg_any_open);
}

View file

@ -1,5 +1,6 @@
#include <errno.h>
#include <stdlib.h>
#include <time.h>
#include "utils.h"
#include "xce/errnox.h"
@ -46,6 +47,66 @@ void free_timewindow_config(xce_timewindow_config_t* w)
kill_timewindow_config(w);
}
/*
* xce_timewindow_is_open — is `now` inside the dispatch/run window?
*
* A NULL config (or one with no segments) is always open. A segment is open
* on the given day (seg->open->day: 0 = every day, 1-7 = Mon..Sun) between
* its open and close wall-clock times; a close time earlier than the open
* time denotes an overnight window (e.g. 19:00–07:00), which is open from
* open-time today until midnight and again from midnight until close-time.
*/
int xce_timewindow_is_open(xce_timewindow_config_t* cfg, time_t now)
{
struct tm tm_now;
int day;
int cur_min;
int i;
if ((cfg == NULL) || (cfg->nsegs <= 0) || (cfg->segs == NULL)) {
return 1;
}
localtime_r(&now, &tm_now);
day = (tm_now.tm_wday == 0) ? 7 : tm_now.tm_wday; /* 1=Mon..7=Sun */
cur_min = tm_now.tm_hour * 60 + tm_now.tm_min;
for (i = 0; i < cfg->nsegs; i++) {
xce_timewindow_seg_config_t* seg = cfg->segs[i];
int open_min;
int close_min;
if ((seg == NULL) || (seg->open == NULL) || (seg->close == NULL)) {
continue;
}
/* day-scoped segment: must match today (0 = every day) */
if ((seg->open->day != 0) && (seg->open->day != day)) {
continue;
}
if ((seg->close->day != 0) && (seg->close->day != day)) {
continue;
}
open_min = seg->open->hour * 60 + seg->open->min;
close_min = seg->close->hour * 60 + seg->close->min;
if (open_min <= close_min) {
/* same-day window [open, close) */
if ((cur_min >= open_min) && (cur_min < close_min)) {
return 1;
}
} else {
/* overnight window: from open today until 24:00, or before close */
if ((cur_min >= open_min) || (cur_min < close_min)) {
return 1;
}
}
}
return 0;
}
static xce_timewindow_seg_config_t*
dup_timewindow_seg_config(xce_timewindow_seg_config_t* src)
{

View file

@ -75,6 +75,9 @@ typedef struct {
int32_t accept_interval;
int32_t sched_delay;
char* resreq;
/* dispatch window (queues.yaml start_time/end_time); NULL = always open.
* Duplicated from the controller's queue config; freed by kill_sched_queue. */
xce_timewindow_config_t* dispatchwindow;
} xce_sched_queue_t;
typedef struct xce_sched_job_ {

View file

@ -148,7 +148,7 @@ extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
extern int sched_job_is_ready(xce_sched_job_t*);
extern int sched_queue_is_ready(xce_sched_queue_t*);
extern int sched_queue_status_reason(uint32_t status);
extern int sched_queue_status_reason(xce_sched_queue_t*);
extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);

View file

@ -233,31 +233,13 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin)
pendJobs = &(q->pendJobs);
if (!sched_queue_is_ready(q->queue)) {
/* Attach the queue-state reason (closed/draining/windowed-out)
* to every PEND job in this queue so a skipped queue explains
* itself (PEND-reason P1). */
int qreason = sched_queue_status_reason(q->queue->status);
if (qreason != 0) {
ordq_item_t* it3;
for (it3 = pendJobs->head; it3 != NULL; it3 = it3->next) {
fcfs_job_t* j = (fcfs_job_t*)it3->data;
xce_sched_obj_t* obj;
if (j == NULL || j->job == NULL) {
continue;
}
obj = j->job->schedobj;
if (obj == NULL) {
continue;
}
if (obj->reason == NULL) {
obj->reason = make_pendreason(0);
}
if (obj->reason != NULL) {
add_job_preason(obj->reason->reasons, qreason,
q->queue->name, NULL);
}
}
}
/* bug-1521: attaching the queue-state reason to job->schedobj
* here races the allocator thread's MOVE_PTR(obj->reason, ...)
* (sched_make_*_allocation) — same job, two threads, no lock →
* heap corruption (crash stacks land in random capnp/limit
* code). The queue-status reason is reported through the
* scheduler's pending-reason chain instead; do NOT touch
* schedobj->reason from the picker thread. */
continue;
}
for (it2 = pendJobs->head; it2 != NULL; it2 = it2->next) {

View file

@ -212,9 +212,14 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
void kill_sched_queue(void* arg)
{
if (arg == NULL) {
xce_sched_queue_t* q = (xce_sched_queue_t*)arg;
if (q == NULL) {
return;
}
free_timewindow_config(q->dispatchwindow);
q->dispatchwindow = NULL;
}
xce_sched_queue_t* make_sched_queue(void* arg, xce_queue_t* q)
@ -238,6 +243,16 @@ xce_sched_queue_t* make_sched_queue(void* arg, xce_queue_t* q)
queue->accept_interval = q->accept_interval;
queue->sched_delay = q->sched_delay;
if (q->dispatchwindow_config != NULL) {
queue->dispatchwindow = dup_timewindow_config(q->dispatchwindow_config);
if (queue->dispatchwindow == NULL) {
/* treat a failed dup as "always open" rather than failing the
* queue — the window is advisory for PEND reasons. */
WARN("%s: dup dispatch window for queue %s failed, treating as "
"always open", __func__, queue->name);
}
}
return queue;
}

View file

@ -1451,8 +1451,9 @@ int sched_job_is_ready(xce_sched_job_t* j)
/* A queue is electable only while it accepts new jobs. The controller
* maintains q->status (queue.c: closed / inactivated-by-admin); the dispatch
* window state (QUEUE_STAT_INACT_WINDOW) is not yet computed anywhere, so it
* is not checked here (PEND-reason P1 note). */
* window (start_time/end_time from queues.yaml, retained on the queue by
* make_sched_queue) is evaluated here because the controller does not compute
* QUEUE_STAT_INACT_WINDOW. */
int sched_queue_is_ready(xce_sched_queue_t* q)
{
if (q == NULL) {
@ -1462,19 +1463,29 @@ int sched_queue_is_ready(xce_sched_queue_t* q)
& (QUEUE_STAT_CLOSED | QUEUE_STAT_INACT_ADMIN | QUEUE_STAT_INACT_WINDOW)) {
return 0;
}
if (!xce_timewindow_is_open(q->dispatchwindow, time(NULL))) {
return 0;
}
return 1;
}
/* Map a queue status bitmask to the PEND reason it should report. */
int sched_queue_status_reason(uint32_t status)
/* Map a queue to the PEND reason it should report: status flags take
* precedence, then the dispatch window. */
int sched_queue_status_reason(xce_sched_queue_t* q)
{
if (status & QUEUE_STAT_CLOSED) {
if (q == NULL) {
return 0;
}
if (q->status & QUEUE_STAT_CLOSED) {
return PREASON_QUEUE_CLOSED;
}
if (status & QUEUE_STAT_INACT_ADMIN) {
if (q->status & QUEUE_STAT_INACT_ADMIN) {
return PREASON_QUEUE_DRAINING;
}
if (status & QUEUE_STAT_INACT_WINDOW) {
if (q->status & QUEUE_STAT_INACT_WINDOW) {
return PREASON_TIME_WINDOW;
}
if (!xce_timewindow_is_open(q->dispatchwindow, time(NULL))) {
return PREASON_TIME_WINDOW;
}
return 0;