diff --git a/src/include/dataq.h b/src/include/dataq.h index e16a46ff..47255a04 100644 --- a/src/include/dataq.h +++ b/src/include/dataq.h @@ -9,6 +9,8 @@ typedef struct _data_item { } data_item_t; typedef int (*data_item_match_fn_t)(data_item_t*); +/* like data_item_match_fn_t, plus a caller-owned per-scan context */ +typedef int (*data_item_match_ctx_fn_t)(data_item_t*, void*); typedef void (*data_item_killer_t)(data_item_t*); typedef struct { @@ -22,6 +24,10 @@ typedef struct { extern int dataq_init(dataq_t*, data_item_killer_t); extern data_item_t* dataq_deq(dataq_t*, data_item_match_fn_t); +/* bug-1864: first item matching f, with ctx passed to every f call of this + * one scan (under the queue lock) -- lets a matcher remember what it saw + * earlier in the same scan */ +extern data_item_t* dataq_deq_ctx(dataq_t*, data_item_match_ctx_fn_t, void*); extern int dataq_enq(dataq_t*, data_item_t*); extern void dataq_fini(dataq_t*); diff --git a/src/include/vtaskmgr.h b/src/include/vtaskmgr.h index a3daaf2b..9e48cfb8 100644 --- a/src/include/vtaskmgr.h +++ b/src/include/vtaskmgr.h @@ -99,6 +99,13 @@ extern int vtask_init(vtask_t*, vtask_func_t, void*); extern void vtask_fini(vtask_t*); extern int vtask_enq(vtaskmgr_t*, vtask_t*); + +/* bug-1864: dequeue the first dispatchable task from a vtaskmgr task queue. + * is_blocked reports whether an endpoint already has a task running + * (production: endpoint_is_task_blocked); it is a parameter so a test can + * make an endpoint unblock part-way through one scan. */ +extern vtask_t* vtask_deq_eligible(dataq_t* taskq, + int (*is_blocked)(endpoint_t*)); extern int vtask_handle_msg(vtaskmgr_t*, vtask_func_t, void* arg); /* LivenessAdmission (raft-drive-async task 1.6) — pool-level reservation. diff --git a/src/lib/dataq.c b/src/lib/dataq.c index 2f33a94c..dcfa2650 100644 --- a/src/lib/dataq.c +++ b/src/lib/dataq.c @@ -62,6 +62,47 @@ static data_item_t* dataq_deq_first_match(dataq_t* q, data_item_match_fn_t f) 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; diff --git a/src/lib/meson.build b/src/lib/meson.build index 7eb79a38..5cb9e553 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -226,6 +226,7 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_ctrlstatus.c', 'testsuite/test_vtask_admission.c', 'testsuite/test_vtask_enq_race.c', + 'testsuite/test_vtask_order.c', 'testsuite/test_vtaskmgr_teardown.c', 'testsuite/test_xcealloc.c', 'raft/raft_sim.c', diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index 3a9e1c54..04c21473 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -55,6 +55,7 @@ extern void test_wal_wire(void); extern void test_ctrlstatus(void); extern void test_vtask_admission(void); extern void test_vtask_enq_race(void); +extern void test_vtask_order(void); extern void test_vtaskmgr_teardown(void); extern void test_xcealloc(void); @@ -181,6 +182,7 @@ int main(int argc, char* argv[]) TEST(test_ctrlstatus); TEST(test_vtask_admission); TEST(test_vtask_enq_race); + TEST(test_vtask_order); TEST(test_vtaskmgr_teardown); TEST(test_xcealloc); diff --git a/src/lib/testsuite/test_vtask_order.c b/src/lib/testsuite/test_vtask_order.c new file mode 100644 index 00000000..0a8b9135 --- /dev/null +++ b/src/lib/testsuite/test_vtask_order.c @@ -0,0 +1,177 @@ +/* + * bug-1864: the vtaskmgr dispatcher checked each queued task's endpoint + * "busy" flag one item at a time while the vcpu finishing that endpoint's + * previous message cleared the flag without the queue lock. Within ONE + * scan, task N was skipped (endpoint still busy), the endpoint unblocked, + * and task N+1 for the SAME endpoint was then dequeued first -- a standby + * processed WAL entry seq 258 before 257 and NACKed (log-mismatch storms). + * + * vtask_deq_eligible() takes the busy check as a parameter so these tests + * can make an endpoint unblock part-way through a scan, deterministically. + */ +#include +#include + +#include + +#include "dataq.h" +#include "netio.h" +#include "vtaskmgr.h" + +static endpoint_t* g_ep1; +static endpoint_t* g_ep2; +static int g_ep1_checks; + +static vtask_t* make_io_task(endpoint_t* ep) +{ + vtask_t* t = (vtask_t*)calloc(1, sizeof(*t)); + vtask_data_t* d = (vtask_data_t*)calloc(1, sizeof(*d)); + + TEST_ASSERT_NOT_NULL(t); + TEST_ASSERT_NOT_NULL(d); + d->vtype = VTASK_IO; + d->u.iodata.ep = ep; + t->data = d; + return t; +} + +static vtask_t* make_generic_task(void) +{ + vtask_t* t = (vtask_t*)calloc(1, sizeof(*t)); + vtask_data_t* d = (vtask_data_t*)calloc(1, sizeof(*d)); + + TEST_ASSERT_NOT_NULL(t); + TEST_ASSERT_NOT_NULL(d); + d->vtype = VTASK_GENERIC; + t->data = d; + return t; +} + +static void free_task(vtask_t* t) +{ + if (t != NULL) { + free(t->data); + free(t); + } +} + +/* ep1 is busy on the first check of this scan, then its running task + * finishes and it is free for every later check -- the live interleaving */ +static int ep1_unblocks_mid_scan(endpoint_t* ep) +{ + if (ep == g_ep1) { + return (g_ep1_checks++ == 0); + } + return 0; +} + +static int ep1_busy(endpoint_t* ep) { return (ep == g_ep1); } +static int none_busy(endpoint_t* ep) { (void)ep; return 0; } + +static void order_setup(dataq_t* q) +{ + dataq_init(q, NULL); + g_ep1 = (endpoint_t*)calloc(1, sizeof(endpoint_t)); + g_ep2 = (endpoint_t*)calloc(1, sizeof(endpoint_t)); + g_ep1_checks = 0; +} + +static void order_teardown(dataq_t* q) +{ + data_item_t* it; + + while ((it = dataq_deq(q, NULL)) != NULL) { + free_task(ITPTR_TO_VTASK(it)); + } + free(g_ep1); + free(g_ep2); +} + +/* the bug: a later message must never overtake an earlier one of the same + * endpoint, even if the endpoint frees up while the scan is between them */ +static void test_later_task_of_skipped_endpoint_is_not_dequeued(void) +{ + dataq_t q; + vtask_t* first; + vtask_t* second; + vtask_t* got; + + order_setup(&q); + first = make_io_task(g_ep1); + second = make_io_task(g_ep1); + dataq_enq(&q, &(first->link)); + dataq_enq(&q, &(second->link)); + + got = vtask_deq_eligible(&q, ep1_unblocks_mid_scan); + TEST_ASSERT_NULL_MESSAGE(got, + "task N was skipped as busy, so task N+1 of the same endpoint must " + "not be dequeued ahead of it in the same scan"); + + /* next scan: the endpoint is free, and N comes out first */ + got = vtask_deq_eligible(&q, none_busy); + TEST_ASSERT_EQUAL_PTR(first, got); + free_task(got); + got = vtask_deq_eligible(&q, none_busy); + TEST_ASSERT_EQUAL_PTR(second, got); + free_task(got); + + order_teardown(&q); +} + +/* a busy endpoint must not hold up other endpoints' tasks */ +static void test_other_endpoint_still_dispatched(void) +{ + dataq_t q; + vtask_t* a = NULL; + vtask_t* c = NULL; + vtask_t* got; + + order_setup(&q); + a = make_io_task(g_ep1); + c = make_io_task(g_ep2); + dataq_enq(&q, &(a->link)); + dataq_enq(&q, &(c->link)); + + got = vtask_deq_eligible(&q, ep1_busy); + TEST_ASSERT_EQUAL_PTR(c, got); + free_task(got); + + order_teardown(&q); +} + +/* FIFO when nothing is busy; non-IO tasks are always dispatchable */ +static void test_fifo_and_generic_tasks(void) +{ + dataq_t q; + vtask_t* a; + vtask_t* b; + vtask_t* g; + vtask_t* got; + + order_setup(&q); + a = make_io_task(g_ep1); + b = make_io_task(g_ep1); + dataq_enq(&q, &(a->link)); + dataq_enq(&q, &(b->link)); + got = vtask_deq_eligible(&q, none_busy); + TEST_ASSERT_EQUAL_PTR(a, got); + free_task(got); + order_teardown(&q); + + order_setup(&q); + a = make_io_task(g_ep1); + g = make_generic_task(); + dataq_enq(&q, &(a->link)); + dataq_enq(&q, &(g->link)); + got = vtask_deq_eligible(&q, ep1_busy); + TEST_ASSERT_EQUAL_PTR(g, got); + free_task(got); + order_teardown(&q); +} + +void test_vtask_order(void) +{ + RUN_TEST(test_later_task_of_skipped_endpoint_is_not_dequeued); + RUN_TEST(test_other_endpoint_still_dispatched); + RUN_TEST(test_fifo_and_generic_tasks); +} diff --git a/src/lib/vtaskmgr.c b/src/lib/vtaskmgr.c index de0b56cf..e997c351 100644 --- a/src/lib/vtaskmgr.c +++ b/src/lib/vtaskmgr.c @@ -281,42 +281,79 @@ static vcpu_t* pick_idle_vcpu(vtaskmgr_t* vtaskmgr) return &(vtaskmgr->vcpus[i]); } -static int vtask_eligible(data_item_t* data) -{ - vtask_t* task = ITPTR_TO_VTASK(data); - int eligible = 0; - vtask_data_t* vdata; +/* bug-1864: an endpoint's busy flag is cleared by the vcpu that finished its + * previous message WITHOUT the taskq lock (bug-1282: taking a lock there + * deadlocked), so it can flip from busy to free part-way through one scan. + * Checking each task independently then skips task N (busy) and dequeues + * task N+1 of the SAME endpoint (free by now) -- messages of one connection + * handled out of order (a standby NACKing WAL entry 258 before 257). The + * scan therefore remembers every endpoint it found busy: all later tasks of + * that endpoint stay ineligible for the rest of this scan, so per-endpoint + * FIFO holds. If more distinct busy endpoints turn up than fit, every later + * IO task is treated as ineligible (never reorders; picked up next scan). */ +#define VTASK_SCAN_MAX_SKIPPED 64 + +typedef struct { + int (*is_blocked)(endpoint_t*); + endpoint_t* skipped[VTASK_SCAN_MAX_SKIPPED]; + int nskipped; + int overflow; +} vtask_scan_t; + +static int vtask_eligible_scan(data_item_t* data, void* arg) +{ + vtask_scan_t* scan = (vtask_scan_t*)arg; + vtask_t* task = ITPTR_TO_VTASK(data); + vtask_data_t* vdata = task->data; + endpoint_t* ep; + int i; - vdata = task->data; if ((vdata == NULL) || (vdata->vtype != VTASK_IO)) { return 1; } - if (vdata->u.iodata.ep == NULL) { + ep = vdata->u.iodata.ep; + if (ep == NULL) { WARN("no endpoint for task 0x%p", task); return 0; } - eligible = !(endpoint_is_task_blocked(vdata->u.iodata.ep)); - TRACE("%s vtask %p is %s", __func__, task, - eligible ? "eligible" : "not eligible"); + if (scan->overflow) { + return 0; + } - return eligible; + for (i = 0; i < scan->nskipped; i++) { + if (scan->skipped[i] == ep) { + return 0; + } + } + + if (!scan->is_blocked(ep)) { + return 1; + } + + if (scan->nskipped < VTASK_SCAN_MAX_SKIPPED) { + scan->skipped[scan->nskipped++] = ep; + } else { + scan->overflow = 1; + } + return 0; +} + +vtask_t* vtask_deq_eligible(dataq_t* taskq, int (*is_blocked)(endpoint_t*)) +{ + vtask_scan_t scan = { .is_blocked = is_blocked, .nskipped = 0, + .overflow = 0 }; + data_item_t* data; + + data = dataq_deq_ctx(taskq, vtask_eligible_scan, &scan); + + return (data != NULL) ? ITPTR_TO_VTASK(data) : NULL; } static vtask_t* pick_eligible_task(vtaskmgr_t* vtaskmgr) { - data_item_t* data; - vtask_t* task = NULL; - - data = dataq_deq(&(vtaskmgr->taskq), vtask_eligible); - if (data == NULL) { - return NULL; - } - - task = ITPTR_TO_VTASK(data); - - return task; + return vtask_deq_eligible(&(vtaskmgr->taskq), endpoint_is_task_blocked); } static int bind_vtask(vtask_t* task, vcpu_t* vcpu)