test_mt.c (16256B)
1 /* 2 * test_mt.c — exercises the XCO_MT extension (compile with -DXCO_MT). 3 * 4 * Layered like the extension itself: 5 * - Inbox mechanics on one OS thread: post/drain FIFO, wake-edge 6 * dedup, try_park handshake. xco_thread_t is freestanding, so two 7 * logical threads on one OS thread exercise all routing paths 8 * deterministically. 9 * - The fire seam: home routing, mid-flight retarget re-route, the 10 * deferred waker path (event set outside xco_rt_run routes the 11 * waker through the runtime's own inbox). 12 * - Real pthreads: the two-leg job round-trip from XCO_MT.md's "Op 13 * round-trip" section — coroutines await ops, a worker executes 14 * leg 1 on its thread, completion runs on the runtime's thread — 15 * and a multi-producer MPSC stress with per-producer FIFO checks. 16 * 17 * Host bindings here are pthread mutex/condvar pairs; xco itself never 18 * sees them — only the wakeup fn pointer and the try_park handshake. 19 */ 20 21 #include "xco.h" 22 23 #include <assert.h> 24 #include <pthread.h> 25 #include <stdio.h> 26 #include <stdlib.h> 27 #include <string.h> 28 29 #define CONTAINER_OF(ptr, type, member) \ 30 ((type *)((char *)(ptr) - offsetof(type, member))) 31 32 #define STACK_BYTES (64 * 1024) 33 34 /* ---- Host binding: a thread that parks on a condvar ---------------- */ 35 36 typedef struct { 37 xco_thread_t t; /* first member: wake hook casts back */ 38 pthread_mutex_t mu; 39 pthread_cond_t cv; 40 int wakes; /* wake-hook invocations (hook-side) */ 41 } host_thread_t; 42 43 static void host_wake(xco_thread_t *t) { 44 host_thread_t *h = (host_thread_t *)t; 45 pthread_mutex_lock(&h->mu); 46 h->wakes++; 47 pthread_cond_signal(&h->cv); 48 pthread_mutex_unlock(&h->mu); 49 } 50 51 static void host_thread_init(host_thread_t *h) { 52 xco_thread_init(&h->t, host_wake, NULL); 53 pthread_mutex_init(&h->mu, NULL); 54 pthread_cond_init(&h->cv, NULL); 55 h->wakes = 0; 56 } 57 58 /* Park until the inbox has work. The mutex serializes the wake hook 59 * against the park handshake, closing the signal-before-wait window. */ 60 static void host_block(host_thread_t *h) { 61 pthread_mutex_lock(&h->mu); 62 while (xco_thread_try_park(&h->t)) 63 pthread_cond_wait(&h->cv, &h->mu); 64 pthread_mutex_unlock(&h->mu); 65 } 66 67 /* ---- Recording waiter ---------------------------------------------- */ 68 69 typedef struct { 70 xco_waiter_t w; 71 int fired; 72 uintptr_t value; 73 xco_thread_t *seen_thread; /* xco__thread_current at fire time */ 74 int order; /* global fire sequence, from *order_ctr */ 75 int *order_ctr; 76 } rec_waiter_t; 77 78 static void rec_fire(xco_waiter_t *w, uintptr_t value) { 79 rec_waiter_t *r = CONTAINER_OF(w, rec_waiter_t, w); 80 r->fired++; 81 r->value = value; 82 r->seen_thread = xco__thread_current; 83 if (r->order_ctr) r->order = (*r->order_ctr)++; 84 } 85 86 static void rec_init(rec_waiter_t *r, xco_thread_t *home, int *order_ctr) { 87 xco_waiter_init(&r->w, rec_fire); 88 r->w.home = home; 89 r->fired = 0; 90 r->value = 0; 91 r->seen_thread = NULL; 92 r->order = -1; 93 r->order_ctr = order_ctr; 94 } 95 96 /* ---- Post/drain FIFO, single OS thread ------------------------------ */ 97 98 static void test_post_drain_fifo(void) { 99 host_thread_t h; 100 host_thread_init(&h); 101 102 int order = 0; 103 rec_waiter_t r[5]; 104 for (int i = 0; i < 5; i++) { 105 rec_init(&r[i], &h.t, &order); 106 r[i].w.value = (uintptr_t)(100 + i); /* as if stashed by a router */ 107 xco_thread_post(&h.t, &r[i].w); 108 } 109 for (int i = 0; i < 5; i++) assert(!r[i].fired); 110 111 xco_thread_drain(&h.t); 112 for (int i = 0; i < 5; i++) { 113 assert(r[i].fired == 1); 114 assert(r[i].value == (uintptr_t)(100 + i)); 115 assert(r[i].order == i); /* FIFO */ 116 assert(r[i].seen_thread == &h.t); /* TLS installed by drain */ 117 } 118 assert(xco__thread_current == NULL); /* restored */ 119 assert(xco_thread_try_park(&h.t)); /* drained: parkable */ 120 } 121 122 /* ---- Wake-edge dedup + try_park handshake --------------------------- */ 123 124 static void test_wake_dedup(void) { 125 host_thread_t h; 126 host_thread_init(&h); 127 128 rec_waiter_t r[3]; 129 for (int i = 0; i < 3; i++) rec_init(&r[i], &h.t, NULL); 130 131 /* Only the empty -> non-empty edge wakes. */ 132 xco_thread_post(&h.t, &r[0].w); 133 xco_thread_post(&h.t, &r[1].w); 134 assert(h.wakes == 1); 135 136 /* Non-empty: try_park refuses and re-arms the pending flag, so the 137 * still-queued items don't re-wake either. */ 138 assert(!xco_thread_try_park(&h.t)); 139 xco_thread_post(&h.t, &r[2].w); 140 assert(h.wakes == 1); 141 142 xco_thread_drain(&h.t); 143 assert(r[0].fired && r[1].fired && r[2].fired); 144 145 /* Parked again: the next post is a fresh edge. */ 146 assert(xco_thread_try_park(&h.t)); 147 rec_init(&r[0], &h.t, NULL); 148 xco_thread_post(&h.t, &r[0].w); 149 assert(h.wakes == 2); 150 xco_thread_drain(&h.t); 151 assert(r[0].fired); 152 } 153 154 /* ---- Fire seam: routing + mid-flight retarget ----------------------- */ 155 156 static void test_fire_routing(void) { 157 host_thread_t a, b; 158 host_thread_init(&a); 159 host_thread_init(&b); 160 161 /* home == NULL: fires inline on the calling thread. */ 162 rec_waiter_t inl; 163 rec_init(&inl, NULL, NULL); 164 xco_waiter_fire(&inl.w, 7); 165 assert(inl.fired == 1 && inl.value == 7 && inl.seen_thread == NULL); 166 167 /* home == a, fired from outside any drain: becomes the message. */ 168 rec_waiter_t ra; 169 rec_init(&ra, &a.t, NULL); 170 xco_waiter_fire(&ra.w, 42); 171 assert(!ra.fired && a.wakes == 1); 172 xco_thread_drain(&a.t); 173 assert(ra.fired == 1 && ra.value == 42 && ra.seen_thread == &a.t); 174 175 /* Mid-flight retarget: posted to a, but home says b — a's drain 176 * re-routes instead of firing. */ 177 rec_waiter_t rb; 178 rec_init(&rb, &b.t, NULL); 179 xco_thread_post(&a.t, &rb.w); 180 rb.w.value = 9; /* payload rides the inbox */ 181 xco_thread_drain(&a.t); 182 assert(!rb.fired && b.wakes == 1); 183 xco_thread_drain(&b.t); 184 assert(rb.fired == 1 && rb.value == 9 && rb.seen_thread == &b.t); 185 186 /* Same-home fire during drain runs inline (no self re-post). */ 187 assert(xco_thread_try_park(&a.t)); 188 } 189 190 /* ---- Deferred waker: event set outside xco_rt_run ------------------- */ 191 192 typedef struct { 193 xco_runtime_t *rt; 194 xco_latch_t *latch; 195 } await_ctx_t; 196 197 static uintptr_t await_latch_fn(uintptr_t arg) { 198 await_ctx_t *c = (await_ctx_t *)arg; 199 return xco_await(c->rt, &c->latch->base); 200 } 201 202 static void test_deferred_waker(void) { 203 host_thread_t h; 204 host_thread_init(&h); 205 xco_runtime_t rt; 206 xco_rt_init(&rt); 207 xco_rt_attach_thread(&rt, &h.t); 208 assert(h.t.rt == &rt); 209 210 xco_latch_t latch; 211 xco_latch_init(&latch); 212 213 static unsigned char stack[STACK_BYTES] __attribute__((aligned(XCO_STACK_ALIGN))); 214 xco_coro_t co; 215 await_ctx_t ctx = { .rt = &rt, .latch = &latch }; 216 xco_spawn(&co, await_latch_fn, stack, sizeof stack, (uintptr_t)&ctx); 217 assert(xco_mach_status(&co.base) == XCO_STEP_SUSPENDED); 218 219 /* Set the latch on the owner thread but outside xco_rt_run: TLS is 220 * NULL, the waker's home is h.t, so it rides the runtime's own 221 * inbox rather than enqueueing inline. */ 222 xco_latch_set(&latch, 1234); 223 assert(xco_mach_status(&co.base) == XCO_STEP_SUSPENDED); 224 assert(h.wakes == 1); 225 226 xco_rt_run(&rt, 0); /* drains inbox -> enqueue -> step */ 227 assert(xco_mach_status(&co.base) == XCO_STEP_DEAD); 228 229 /* Inside xco_rt_run the same fire is inline: re-run with a second 230 * co whose latch is set by the first co's body — covered implicitly 231 * by the op round-trip test below; here just confirm quiescence. */ 232 assert(xco_thread_try_park(&h.t)); 233 } 234 235 /* ---- Two-leg job round-trip over real pthreads ---------------------- */ 236 237 /* Worker: a bare xco_thread_t (no runtime) on its own pthread. Jobs are 238 * waiters with home = the worker; a poison job flips `stop`. */ 239 typedef struct { 240 host_thread_t ht; 241 pthread_t pt; 242 int stop; /* set by the poison job, on this thread */ 243 pthread_t tid; /* filled at thread start */ 244 } worker_t; 245 246 static void *worker_main(void *arg) { 247 worker_t *wk = (worker_t *)arg; 248 wk->tid = pthread_self(); 249 for (;;) { 250 xco_thread_drain(&wk->ht.t); 251 if (wk->stop) break; 252 host_block(&wk->ht); 253 } 254 return NULL; 255 } 256 257 typedef struct { 258 xco_waiter_t w; 259 worker_t *wk; 260 } poison_t; 261 262 static void poison_fire(xco_waiter_t *w, uintptr_t v) { 263 (void)v; 264 CONTAINER_OF(w, poison_t, w)->wk->stop = 1; 265 } 266 267 enum { OP_DOUBLE = 1 }; 268 269 typedef struct { 270 xco_op_t base; 271 xco_waiter_t job; /* leg 1: worker inbox; leg 2: home inbox */ 272 xco_thread_t *home; /* where complete must run */ 273 uintptr_t input; 274 uintptr_t result; /* filled by the worker */ 275 pthread_t exec_tid; /* thread leg 1 ran on */ 276 pthread_t complete_tid; /* thread leg 2 ran on */ 277 } double_op_t; 278 279 /* Leg 2 — home thread (via drain): completion is an owner-thread call. */ 280 static void double_op_completed(xco_waiter_t *w, uintptr_t v) { 281 (void)v; 282 double_op_t *op = CONTAINER_OF(w, double_op_t, job); 283 op->complete_tid = pthread_self(); 284 xco_op_complete(&op->base, XCO_OP_COMPLETED); 285 } 286 287 /* Leg 1 — worker thread (via drain): execute, then ride the same waiter 288 * home. */ 289 static void double_op_execute(xco_waiter_t *w, uintptr_t v) { 290 (void)v; 291 double_op_t *op = CONTAINER_OF(w, double_op_t, job); 292 op->result = op->input * 2; 293 op->exec_tid = pthread_self(); 294 w->fire = double_op_completed; 295 w->home = op->home; 296 xco_waiter_fire(w, 0); /* routes home; completion runs there */ 297 } 298 299 enum { N_COS = 8, OPS_PER_CO = 4, N_WORKERS = 2 }; 300 301 typedef struct { 302 xco_runtime_t *rt; 303 xco_thread_t *home; 304 double_op_t ops[OPS_PER_CO]; 305 uintptr_t seed; 306 } round_trip_ctx_t; 307 308 static uintptr_t round_trip_fn(xco_task_t *task, uintptr_t arg) { 309 (void)task; 310 round_trip_ctx_t *c = (round_trip_ctx_t *)arg; 311 uintptr_t sum = 0; 312 for (int i = 0; i < OPS_PER_CO; i++) { 313 double_op_t *op = &c->ops[i]; 314 op->base.kind = OP_DOUBLE; 315 op->home = c->home; 316 op->input = c->seed + (uintptr_t)i; 317 op->result = 0; 318 xco_op_submit(c->rt, &op->base); 319 uintptr_t st = xco_await(c->rt, &op->base.done.base); 320 assert(st == XCO_OP_COMPLETED); 321 assert(op->result == op->input * 2); 322 sum += op->result; 323 } 324 return sum; 325 } 326 327 static void test_job_round_trip(void) { 328 host_thread_t hm; /* the "main"/runtime thread */ 329 host_thread_init(&hm); 330 xco_runtime_t rt; 331 xco_rt_init(&rt); 332 xco_rt_attach_thread(&rt, &hm.t); 333 334 worker_t workers[N_WORKERS]; 335 for (int i = 0; i < N_WORKERS; i++) { 336 host_thread_init(&workers[i].ht); 337 workers[i].stop = 0; 338 int rc = pthread_create(&workers[i].pt, NULL, worker_main, &workers[i]); 339 assert(rc == 0); 340 } 341 342 static round_trip_ctx_t ctxs[N_COS]; 343 static xco_cotask_t tasks[N_COS]; 344 static unsigned char stacks[N_COS][STACK_BYTES] 345 __attribute__((aligned(XCO_STACK_ALIGN))); 346 uintptr_t expect = 0; 347 for (int i = 0; i < N_COS; i++) { 348 ctxs[i].rt = &rt; 349 ctxs[i].home = &hm.t; 350 ctxs[i].seed = (uintptr_t)(i * 100 + 1); 351 for (int k = 0; k < OPS_PER_CO; k++) 352 expect += (ctxs[i].seed + (uintptr_t)k) * 2; 353 xco_cotask_spawn(&tasks[i], round_trip_fn, 354 stacks[i], STACK_BYTES, (uintptr_t)&ctxs[i]); 355 } 356 357 /* Host main loop: run to quiescence, dispatch the batch round-robin, 358 * park until completions ride home. */ 359 int rr = 0; 360 for (;;) { 361 xco_rt_run(&rt, 0); 362 363 int finished = 0; 364 for (int i = 0; i < N_COS; i++) 365 if (xco_task_finished(&tasks[i].task)) finished++; 366 if (finished == N_COS) break; 367 368 xco_op_t *batch = xco_rt_take_ops(&rt, NULL); 369 while (batch) { 370 xco_op_t *next = batch->next; /* posting clobbers links */ 371 double_op_t *op = (double_op_t *)batch; 372 assert(op->base.kind == OP_DOUBLE); 373 op->job.fire = double_op_execute; 374 op->job.home = &workers[rr].ht.t; 375 rr = (rr + 1) % N_WORKERS; 376 xco_waiter_fire(&op->job, 0); /* routes to the worker */ 377 batch = next; 378 } 379 380 host_block(&hm); 381 } 382 383 uintptr_t total = 0; 384 for (int i = 0; i < N_COS; i++) total += tasks[i].task.done.value; 385 assert(total == expect); 386 387 /* Every leg ran on the right thread. */ 388 pthread_t self = pthread_self(); 389 for (int i = 0; i < N_COS; i++) { 390 for (int k = 0; k < OPS_PER_CO; k++) { 391 double_op_t *op = &ctxs[i].ops[k]; 392 assert(pthread_equal(op->complete_tid, self)); 393 int on_worker = 0; 394 for (int wkr = 0; wkr < N_WORKERS; wkr++) 395 if (pthread_equal(op->exec_tid, workers[wkr].tid)) on_worker = 1; 396 assert(on_worker); 397 } 398 } 399 400 /* Shut the workers down via poison jobs. */ 401 poison_t poison[N_WORKERS]; 402 for (int i = 0; i < N_WORKERS; i++) { 403 xco_waiter_init(&poison[i].w, poison_fire); 404 poison[i].wk = &workers[i]; 405 xco_thread_post(&workers[i].ht.t, &poison[i].w); 406 pthread_join(workers[i].pt, NULL); 407 } 408 } 409 410 /* ---- MPSC stress: N producers, per-producer FIFO -------------------- */ 411 412 enum { N_PRODUCERS = 4, ITEMS_PER_PRODUCER = 10000 }; 413 414 typedef struct { 415 xco_waiter_t w; 416 int producer; 417 int seq; 418 } stress_item_t; 419 420 static struct { 421 host_thread_t consumer; 422 int received; 423 int next_seq[N_PRODUCERS]; 424 } g_stress; 425 426 static void stress_fire(xco_waiter_t *w, uintptr_t v) { 427 (void)v; 428 stress_item_t *it = CONTAINER_OF(w, stress_item_t, w); 429 assert(xco__thread_current == &g_stress.consumer.t); 430 assert(g_stress.next_seq[it->producer] == it->seq); /* per-producer FIFO */ 431 g_stress.next_seq[it->producer]++; 432 g_stress.received++; 433 } 434 435 typedef struct { 436 stress_item_t *items; 437 int producer; 438 pthread_t pt; 439 } producer_t; 440 441 static void *producer_main(void *arg) { 442 producer_t *p = (producer_t *)arg; 443 for (int i = 0; i < ITEMS_PER_PRODUCER; i++) { 444 stress_item_t *it = &p->items[i]; 445 xco_waiter_init(&it->w, stress_fire); 446 it->w.home = &g_stress.consumer.t; 447 it->producer = p->producer; 448 it->seq = i; 449 xco_waiter_fire(&it->w, 0); /* off-home: becomes the message */ 450 } 451 return NULL; 452 } 453 454 static void test_mpsc_stress(void) { 455 host_thread_init(&g_stress.consumer); 456 g_stress.received = 0; 457 memset(g_stress.next_seq, 0, sizeof g_stress.next_seq); 458 459 producer_t producers[N_PRODUCERS]; 460 for (int i = 0; i < N_PRODUCERS; i++) { 461 producers[i].items = malloc(sizeof(stress_item_t) * ITEMS_PER_PRODUCER); 462 assert(producers[i].items); 463 producers[i].producer = i; 464 int rc = pthread_create(&producers[i].pt, NULL, producer_main, &producers[i]); 465 assert(rc == 0); 466 } 467 468 const int total = N_PRODUCERS * ITEMS_PER_PRODUCER; 469 while (g_stress.received < total) { 470 xco_thread_drain(&g_stress.consumer.t); 471 if (g_stress.received < total) host_block(&g_stress.consumer); 472 } 473 assert(g_stress.received == total); 474 for (int i = 0; i < N_PRODUCERS; i++) 475 assert(g_stress.next_seq[i] == ITEMS_PER_PRODUCER); 476 477 for (int i = 0; i < N_PRODUCERS; i++) { 478 pthread_join(producers[i].pt, NULL); 479 free(producers[i].items); 480 } 481 } 482 483 /* -------------------------------------------------------------------- */ 484 485 int main(void) { 486 test_post_drain_fifo(); 487 printf("test_post_drain_fifo OK\n"); 488 test_wake_dedup(); 489 printf("test_wake_dedup OK\n"); 490 test_fire_routing(); 491 printf("test_fire_routing OK\n"); 492 test_deferred_waker(); 493 printf("test_deferred_waker OK\n"); 494 test_job_round_trip(); 495 printf("test_job_round_trip OK\n"); 496 test_mpsc_stress(); 497 printf("test_mpsc_stress OK\n"); 498 printf("test_mt: all OK\n"); 499 return 0; 500 }