myxce/src/lib/dataq.c
Rongsong Shen 0d3bb965e5
fix(lib): keep per-endpoint message order in the vtask dispatcher (bug-1864)
The dispatcher checked each queued task's endpoint busy flag one item at a
time, while the vcpu that finished the endpoint's previous message cleared
the flag without the taskq lock (a lock there deadlocked, bug-1282). Within
one scan, task N was skipped (endpoint still busy), the endpoint freed up,
and task N+1 of the SAME endpoint was dequeued first. Standby controllers
then handled WAL catch-up entry 258 before 257 and NACKed (log mismatch):
26 reorders -> ~576 NACKs in a 45-min production-like run, both standbys
at once at 21:39:24, primary lost its write quorum and answered
jobdispatch/join with E_XCE_NOT_READY, the scheduler client failed over and
rejoined, and the rejoin JOBS_PUSH carried lagging committed state
("refusing resync state ... existing job is RUN" WARNs).

dataq gains dataq_deq_ctx() (a per-scan matcher context). The scan is now
vtask_deq_eligible(taskq, is_blocked): it remembers every endpoint found
busy and keeps all later tasks of that endpoint ineligible for the rest of
the scan (past 64 distinct busy endpoints, every later IO task waits for
the next scan) -- per-endpoint FIFO without any new lock.

Test lib/testsuite/test_vtask_order.c injects is_blocked to free the
endpoint mid-scan: task N+1 was dequeued before the fix, none after; other
endpoints and FIFO/generic tasks unaffected. Suites: lib 1/1, controller
5/5, worker 10/10, scheduler 87/87, harness 5/5. Chaotic HA soak 2/2 PASS
with 0 WAL reorders (2 and 5 per run before), no ASAN.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-25 14:44:14 +00:00

157 lines
3.6 KiB
C

#include <stdlib.h>
#include "dataq.h"
int dataq_init(dataq_t* q, data_item_killer_t killer)
{
q->head = q->tail = NULL;
q->killer = killer;
mylock_init(&(q->lock));
return 0;
}
static data_item_t* dataq_deq_head(dataq_t* q)
{
if (q->head != NULL) {
data_item_t* b = NULL;
b = q->head;
q->head = b->next;
b->next = NULL;
if (q->head == NULL) {
q->tail = NULL;
}
return b;
}
return NULL;
}
static data_item_t* dataq_deq_first_match(dataq_t* q, data_item_match_fn_t f)
{
data_item_t* prev = NULL;
data_item_t* curr = q->head;
while (curr != NULL) {
if ((*f)(curr)) { // Found the item to remove
if (prev == NULL) {
// Removing the head item
q->head = curr->next;
if (q->head == NULL) {
// Queue became empty after removing head
q->tail = NULL;
}
// else: tail remains unchanged (it wasn't the head)
} else {
// Removing an item after prev (middle or tail)
prev->next = curr->next;
if (curr == q->tail) { // Check if we are removing the tail
// We removed the tail item, prev is the new tail
q->tail = prev;
}
// else: tail remains unchanged (it was further down the list)
}
curr->next = NULL; // Isolate the removed item
return curr; // Return the removed item
}
// Move to the next item
prev = curr;
curr = curr->next;
}
return NULL;
}
static data_item_t* dataq_deq_first_match_ctx(dataq_t* q,
data_item_match_ctx_fn_t f,
void* ctx)
{
data_item_t* prev = NULL;
data_item_t* curr = q->head;
while (curr != NULL) {
if ((*f)(curr, ctx)) {
if (prev == NULL) {
q->head = curr->next;
if (q->head == NULL) {
q->tail = NULL;
}
} else {
prev->next = curr->next;
if (curr == q->tail) {
q->tail = prev;
}
}
curr->next = NULL;
return curr;
}
prev = curr;
curr = curr->next;
}
return NULL;
}
data_item_t* dataq_deq_ctx(dataq_t* q, data_item_match_ctx_fn_t f, void* ctx)
{
data_item_t* b;
mylock_lock(&(q->lock));
b = dataq_deq_first_match_ctx(q, f, ctx);
mylock_unlock(&(q->lock));
return b;
}
data_item_t* dataq_deq(dataq_t* q, data_item_match_fn_t f)
{
data_item_t* b = NULL;
mylock_lock(&(q->lock));
if (f == NULL) {
b = dataq_deq_head(q);
} else {
b = dataq_deq_first_match(q, f);
}
mylock_unlock(&(q->lock));
return b;
}
int dataq_enq(dataq_t* q, data_item_t* b)
{
int rc = 0;
b->next = NULL;
mylock_lock(&(q->lock));
if (q->tail != NULL) {
q->tail->next = b;
}
if (q->head == NULL) {
q->head = b;
}
q->tail = b;
mylock_unlock(&(q->lock));
return rc;
}
void dataq_fini(dataq_t* q)
{
data_item_t* it;
while ((it = dataq_deq_head(q)) != NULL) {
if (q->killer != NULL) {
(q->killer)(it);
}
// Note: dataq_deq_head already isolates 'it' (it->next = NULL)
}
q->head = q->tail = NULL; // Explicitly set just in case.
q->killer = NULL;
return;
}