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

fix(worker-rejoin-resync): task 4.4 - gate periodic load-indices send on resync_pending

Browse source 
Extends task 4.3's post-join gating to resource_sync_thread()'s
independent periodic reporting pthread, closing the Requirement 3.8 gap
where load-indices sends could interleave with an in-flight job-state
reconciliation report. Also fixes two pre-existing latent bugs
(bug-1756, bug-1757) in resource_sync.c that blocked a truthful RED/GREEN
signal for the new tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019ypUeKgbAqLWUkaSJ79poN
This commit is contained in:
Rongsong Shen 2026-09-02 03:01:56 +00:00
parent 4140ecdbae
commit b6a2f5ec14
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 241 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

60
src/worker/src/resource_sync.c
View file

@ -2,6 +2,7 @@
#include "job_information.h"
#include "node_information.h"
#include "xcecd.h"
#include "xcecd_resync.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <pthread.h>
@ -58,10 +59,23 @@ static void free_resvals(xce_resval_t* vals, int count)
static void init_builtin_load_indices(void)
{
static bool initialized = false;
if (initialized)
return;
/* bug-1756 (companion, worker-rejoin-resync task 4.4): this used to
* guard against re-running registration with its own function-local
* `static bool initialized`, but that guard is a duplicate of -- and
* conflicts with -- collect_load_indices()'s own `g_load_indices ==
* NULL` check, this function's only caller. Once g_load_indices is torn
* down by resource_sync_cleanup_load_indices() (resource_sync_destroy(),
* e.g. a create/start/stop/destroy cycle repeated within one process,
* as the task 4.4 tests below now do for the first time), the caller's
* check correctly wants to re-populate the registry, but the
* function-local `initialized` stayed stuck true from the first run and
* silently no-op'd forever after, leaving g_load_indices permanently
* NULL/empty for the rest of the process ("No load indices to report").
* Removed: the caller's own guard already ensures this body runs
* exactly once per populated-registry window, with no other call site
* (grepped) and no concurrent caller (collect_load_indices() only runs
* on resource_sync_thread()'s single dedicated pthread in production,
* or single-threaded test code). */
g_load_indices_capacity = MAX_LOAD_INDICES;
g_load_indices = (load_index_info_t*)calloc(g_load_indices_capacity,
sizeof(load_index_info_t));
@ -89,8 +103,6 @@ static void init_builtin_load_indices(void)
true);
resource_sync_register_load_index("io", Value_u_fvalue, node_info_get_io,
true);
initialized = true;
}
/**
@ -322,6 +334,30 @@ static void* resource_sync_thread(void* arg)
sync->load_report_counter++;
if (sync->load_report_counter >= LOAD_INDICES_INTERVAL_DEFAULT) {
/* worker-rejoin-resync task 4.4 (Requirement 3.8): this periodic
* load-indices send is worker post-join background activity,
* same as checking_connection()'s rejoin_call() (task 4.3,
* controller_connector.c) -- defer it while a job-state
* reconciliation report is still in flight rather than
* interleaving with it. Unlike the vals==NULL/valptrs==NULL
* skip branches below (genuine local failures, where resetting
* the counter avoids a tight failure-retry loop), this branch
* deliberately does NOT reset load_report_counter: the deferral
* is caused by an external condition outside this loop's
* control, of unknown duration, so the threshold stays crossed
* and the very next tick re-checks immediately once
* xcecd_resync_is_pending() clears -- resuming the send with no
* extra retry/backoff delay, rather than making a resync that
* just finished wait out a fresh full interval on top. */
if (xcecd_resync_is_pending()) {
XCECD_LOG(XCE_LOG_DEBUG,
"resource_sync_thread: deferring load-indices "
"send -- job-state resync report still being "
"sent");
sleep(SYNC_INTERVAL);
continue;
}
vals = collect_load_indices(&vals_count);
if (vals == NULL || vals_count <= 0) {
XCECD_LOG(XCE_LOG_ERROR, "No load indices to report");
@ -381,7 +417,17 @@ void resource_sync_destroy(resource_sync_t* sync)
XCECD_LOG(XCE_LOG_ERROR, "Invalid resource sync module");
return;
}
resource_sync_stop(sync);
/* bug-1756: resource_sync_stop() joins sync->thread but never resets the
* handle afterward, so an unconditional second call here double-joins
* the same pthread_t -- UB, caught by ASAN as "Joining already joined
* thread, aborting." xcecd.c's own shutdown sequence (xcecd_cleanup())
* calls resource_sync_stop() then resource_sync_destroy() back-to-back,
* hitting this exact hazard on every daemon shutdown. Guard the same
* way connection_handler.c's network_handler_destroy() already guards
* its own analogous stop-then-destroy call (`if (handler->running)`). */
if (atomic_load(&sync->running)) {
resource_sync_stop(sync);
}
resource_sync_cleanup_load_indices();
XCECD_LOG(XCE_LOG_DEBUG, "Resource sync module destroyed successfully");
FREEUP(sync);

139
src/worker/testsuite/test_resource_sync.c
View file

@ -28,11 +28,13 @@
*/
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#include "xcecd.h"
#include "xcecd_resync.h"
/* Real globals this white-box TU's resource_sync.c references
* (xcecd_ctx.virtual_name) -- not linking xcecd.c itself (its own
@ -62,20 +64,41 @@ char* node_info_get_mem(void) { return stub_getter(); }
char* node_info_get_tmp(void) { return stub_getter(); }
char* node_info_get_io(void) { return stub_getter(); }
/* Linker stub for controller_connector.c's real implementation -- only
* referenced by resource_sync_thread() (the background polling loop),
* never invoked by this test (which calls collect_load_indices()
* directly). */
/* Linker stub for controller_connector.c's real implementation, spied on by
* the worker-rejoin-resync task 4.4 tests below (Requirement 3.8) to observe
* whether resource_sync_thread()'s periodic loop actually attempted a send
* on a given tick. */
static int g_send_load_indices_call_count = 0;
bool xcecd_send_load_indices(xce_resval_t** vals, int count)
{
(void)vals;
(void)count;
g_send_load_indices_call_count++;
return true;
}
/* worker-rejoin-resync task 4.4 (Requirement 3.8): link-satisfying stub for
* xcecd_resync.c's public gate query -- this file's white-box target is
* resource_sync.c only, it does not #include xcecd_resync.c (that module's
* own resync_pending lifecycle is covered against the real production code
* by test_xcecd_resync.c). g_test_resync_pending lets the tests below
* control what resource_sync_thread()'s new gate check reads, same
* convention as test_controller_connector.c's own g_test_resync_pending
* double for the same real function. */
static bool g_test_resync_pending = false;
bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
#include "../src/resource_sync.c"
void setUp(void) { return; }
void setUp(void)
{
memset(&xcecd_ctx, 0, sizeof(xcecd_ctx));
xcecd_ctx.virtual_name = (char*)"pc13v";
g_test_resync_pending = false;
g_send_load_indices_call_count = 0;
}
void tearDown(void) { return; }
static void
@ -129,10 +152,116 @@ test_collect_load_indices_falls_back_to_gethostname_when_virtual_name_unset(
"virtual_name_unset: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* worker-rejoin-resync task 4.4 (Requirement 3.8): resource_sync_thread()'s
* periodic load-indices send is worker post-join background activity
* distinct from checking_connection()'s rejoin_call() (task 4.3, a separate
* pthread -- controller_connector.c's connector_to_controller() vs. this
* file's resource_sync_start()) -- it must also defer to an in-flight
* job-state reconciliation report rather than interleaving with it.
*
* resource_sync_thread() is a real pthread with its own SYNC_INTERVAL=1s
* sleep between checks; load_report_counter is seeded to
* LOAD_INDICES_INTERVAL_DEFAULT - 1 before starting the thread so the very
* first tick already crosses the reporting threshold, keeping these tests
* fast instead of waiting a full 15s interval.
* ════════════════════════════════════════════════════════════════════════ */
static void
test_resource_sync_thread_skips_load_indices_send_while_resync_pending(void)
{
resource_sync_t* sync = resource_sync_create();
TEST_ASSERT_NOT_NULL(sync);
g_test_resync_pending = true;
sync->load_report_counter = LOAD_INDICES_INTERVAL_DEFAULT - 1;
TEST_ASSERT_TRUE(resource_sync_start(sync));
/* Several ticks, resync_pending held true throughout -- proves the skip
* is not a one-shot fluke of the first check. */
sleep(2);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_send_load_indices_call_count,
"a load-indices send must not fire while resync_pending is true "
"(Requirement 3.8)");
resource_sync_stop(sync);
resource_sync_destroy(sync);
printf("test_resource_sync_thread_skips_load_indices_send_while_resync_"
"pending: PASS\n");
}
static void
test_resource_sync_thread_resumes_load_indices_send_once_resync_clears(void)
{
resource_sync_t* sync = resource_sync_create();
TEST_ASSERT_NOT_NULL(sync);
g_test_resync_pending = true;
sync->load_report_counter = LOAD_INDICES_INTERVAL_DEFAULT - 1;
TEST_ASSERT_TRUE(resource_sync_start(sync));
sleep(2);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_send_load_indices_call_count,
"sanity: no send should have fired yet while resync_pending was "
"true");
/* Simulates xcecd_resync_handle_query_page() finishing the report and
* clearing the gate (xcecd_resync.c, mirroring task 4.3's own
* resync_pending lifecycle). load_report_counter was never reset by
* the skip path above, so the very next tick's check should fire a
* send immediately -- no extra retry/backoff delay. */
g_test_resync_pending = false;
sleep(2);
TEST_ASSERT_TRUE_MESSAGE(
g_send_load_indices_call_count >= 1,
"the load-indices send must resume on the next periodic check once "
"resync_pending clears (Requirement 3.8)");
resource_sync_stop(sync);
resource_sync_destroy(sync);
printf("test_resource_sync_thread_resumes_load_indices_send_once_"
"resync_clears: PASS\n");
}
static void
test_resource_sync_thread_sends_load_indices_normally_when_not_pending(void)
{
resource_sync_t* sync = resource_sync_create();
TEST_ASSERT_NOT_NULL(sync);
g_test_resync_pending = false;
sync->load_report_counter = LOAD_INDICES_INTERVAL_DEFAULT - 1;
TEST_ASSERT_TRUE(resource_sync_start(sync));
sleep(2);
TEST_ASSERT_TRUE_MESSAGE(
g_send_load_indices_call_count >= 1,
"normal (non-reconciling) periodic load-indices reporting must be "
"unchanged by the new resync_pending gate");
resource_sync_stop(sync);
resource_sync_destroy(sync);
printf("test_resource_sync_thread_sends_load_indices_normally_when_not_"
"pending: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_collect_load_indices_tags_host_with_virtual_name_not_gethostname);
RUN_TEST(test_collect_load_indices_falls_back_to_gethostname_when_virtual_name_unset);
RUN_TEST(test_resource_sync_thread_skips_load_indices_send_while_resync_pending);
RUN_TEST(test_resource_sync_thread_resumes_load_indices_send_once_resync_clears);
RUN_TEST(test_resource_sync_thread_sends_load_indices_normally_when_not_pending);
return UNITY_END();
}