fiber_tests.h

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
#pragma once
 
#include <stdatomic.h>
#include <stdint.h>
#include <string.h>
 
#include "base.h"
#include "serene/rt/fiber.h"
#include "serene/rt/fiber/thread.h"
#include "serene/rt/mm/interface.h"
 
#define FIBER_TESTS(X)                                                                           \
  X("fiber::types", test_fiber_types), X("fiber::switch", test_fiber_switch),                    \
    X("fiber::stack", test_fiber_stack),                                                         \
    X("fiber::thread_stack_bounds", test_fiber_thread_stack_bounds),                             \
    X("fiber::auto_name", test_fiber_auto_name),                                                 \
    X("fiber::make_schedule", test_fiber_make_schedule),                                         \
    X("fiber::sched::auto_workers", test_fiber_sched_auto_workers),                              \
    X("fiber::spawn_copy", test_fiber_spawn_copy), X("fiber::sched::run", test_fiber_sched_run), \
    X("fiber::sched::yield", test_fiber_sched_yield),                                            \
    X("fiber::sched::yield_round_robin", test_fiber_sched_yield_round_robin),                    \
    X("fiber::sched::suspend", test_fiber_sched_suspend),                                        \
    X("fiber::sched::park_abort", test_fiber_sched_park_abort),                                  \
    X("fiber::sched::registry", test_fiber_sched_registry),                                      \
    X("fiber::sched::double_wake", test_fiber_sched_double_wake),                                \
    X("fiber::sched::wait_for", test_fiber_sched_wait_for),                                      \
    X("fiber::sched::wait_for_done", test_fiber_sched_wait_for_done),                            \
    X("fiber::sched::wait_for_many", test_fiber_sched_wait_for_many),                            \
    X("fiber::sched::mt_run", test_fiber_sched_mt_run),                                          \
    X("fiber::sched::mt_yield", test_fiber_sched_mt_yield),                                      \
    X("fiber::sched::stop", test_fiber_sched_stop)
 
// A worker loop fiber must know its own stack bounds from init. A worker
// that only ever resumes stolen fibers never launches a fresh fiber, so
// srn_fiber_on_entry never reports its loop stack, and without the bounds a
// switch back to the loop hands ASan a null stack. Only the ASan build
// records them, the other builds have no reader.
static void test_fiber_thread_stack_bounds() {
  srn_fiber_t loop;
  srn_fiber_init_thread(&loop);
  TEST_CHECK(loop.state == SRN_FIBER_RUNNING);
#if SRN_ASAN
  TEST_CHECK(loop.stack.limit != nullptr);
  TEST_CHECK(loop.stack.start != nullptr);
  // A local's address cannot anchor the check, ASan's fake stack moves
  // address taken locals off the real thread stack. Sanity of the reported
  // range is what is checkable, ordered bounds spanning at least one page.
  TEST_CHECK(srn_fiber_stack_size(loop.stack) >= srn_mm_get_os_page_size());
#else
  TEST_CHECK(loop.stack.limit == nullptr);
#endif
}
 
static srn_fiber_result_t auto_name_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  return nullptr;
}
 
static srn_fiber_t *auto_name_child;
 
static srn_fiber_result_t auto_name_parent_entry(srn_context_t *ctx, void *arg) {
  UNUSED(arg);
  auto_name_child = srn_fiber_spawn(ctx, auto_name_entry, nullptr);
  return nullptr;
}
 
// Every fiber gets an autogenerated debug name at make time. Created on a
// worker the tag is f#<worker>:<n>, the creating worker's id and its spawn
// count. Created off the pool it is f#m:<id> from the engine wide object id
// counter. Provenance, not placement, a stolen fiber runs elsewhere.
static void test_fiber_auto_name() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
 
  srn_fiber_t *a  = srn_fiber_spawn(ctx, auto_name_entry, nullptr);
  srn_fiber_t *b  = srn_fiber_spawn(ctx, auto_name_entry, nullptr);
  auto_name_child = nullptr;
  (void)srn_fiber_spawn(ctx, auto_name_parent_entry, nullptr);
 
  // Made off the pool, the creator tag is `m`.
  TEST_CHECK(strncmp(a->name, "f#m:", 4) == 0);
  TEST_CHECK(strncmp(b->name, "f#m:", 4) == 0);
  TEST_CHECK(strcmp(a->name, b->name) != 0);
 
  // Run the fibers so the registry is empty before the context goes away.
  srn_sched_run(sched, 1);
 
  // The child was made by the parent running on the single worker, id 0,
  // and it is that worker's first spawn.
  TEST_ASSERT(auto_name_child != nullptr);
  TEST_CHECK(strcmp(auto_name_child->name, "f#0:1") == 0);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// srn_fiber_make leaves the fiber NEW and registered, srn_fiber_schedule
// starts it exactly once. An unscheduled fiber never runs, does not block
// quiescence, and is still reaped at shutdown.
static void test_fiber_make_schedule() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
 
  srn_fiber_t *stay = srn_fiber_make(ctx, sched, "left-new", auto_name_entry, nullptr, 0);
  srn_fiber_t *go   = srn_fiber_make(ctx, sched, "scheduled", auto_name_entry, nullptr, 0);
 
  TEST_CHECK(stay->state == SRN_FIBER_NEW);
  TEST_CHECK(strcmp(go->name, "scheduled") == 0);
  TEST_CHECK(strcmp(stay->name, "left-new") == 0);
 
  srn_fiber_schedule(go);
  srn_sched_run(sched, 1);
 
  TEST_CHECK(go->state == SRN_FIBER_DONE);
  TEST_CHECK(stay->state == SRN_FIBER_NEW);
 
  // Reap the never scheduled fiber while its struct (in the context block)
  // is still alive, before releasing the context.
  srn_sched_shutdown(sched);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
typedef struct {
  int value;
} spawn_copy_args_t;
 
static int spawn_copy_seen;
 
static srn_fiber_result_t spawn_copy_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  spawn_copy_seen = ((spawn_copy_args_t *)arg)->value;
  return nullptr;
}
 
// The copying spawn owns its argument, so the caller's stack frame can die
// before the fiber runs. Passing the address of the scoped struct directly
// would be a stack use after scope (the 12_http_server bug shape).
static void test_fiber_spawn_copy() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
 
  spawn_copy_seen = 0;
  {
    spawn_copy_args_t args = {.value = 4242};
    (void)SRN_FIBER_SPAWN_COPY(ctx, spawn_copy_entry, args);
  }
 
  srn_sched_run(sched, 1);
  TEST_CHECK(spawn_copy_seen == 4242);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
static atomic_int auto_workers_ran;
 
static srn_fiber_result_t auto_workers_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  atomic_fetch_add(&auto_workers_ran, 1);
  return nullptr;
}
 
// A zero worker request with the delegating zero default resolves to the
// CPU count at run time. The pool comes up, drains every fiber, and reaches
// quiescence with however many workers the machine provided.
static void test_fiber_sched_auto_workers() {
  TEST_CHECK(srn_thread_cpu_count() >= 1);
 
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
 
  atomic_store(&auto_workers_ran, 0);
  for (int i = 0; i < 32; i++) {
    (void)srn_fiber_spawn(ctx, auto_workers_entry, nullptr);
  }
 
  srn_sched_run(sched, 0);
  TEST_CHECK(atomic_load(&auto_workers_ran) == 32);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
static void test_fiber_types() {
  // Smoke test, the public types exist and the build/test wiring works.
  // The saved context is a single stack-pointer word.
  TEST_CHECK(sizeof(srn_fiber_ctx_t) == sizeof(void *));
  // The lifecycle states run from creation to completion in order.
  TEST_CHECK(SRN_FIBER_NEW == 0);
  TEST_CHECK(SRN_FIBER_NEW < SRN_FIBER_READY);
  TEST_CHECK(SRN_FIBER_READY < SRN_FIBER_RUNNING);
  TEST_CHECK(SRN_FIBER_RUNNING < SRN_FIBER_SUSPENDED);
  TEST_CHECK(SRN_FIBER_SUSPENDED < SRN_FIBER_DONE);
}
 
// The POSIX provider hands back a guard-paged region, `guard` is the mmap base
// (the PROT_NONE page), `limit` is the low end of the usable area one page
// above it, and `start` is the high end where the stack pointer begins.
static void test_fiber_stack() {
  size_t page = srn_mm_get_os_page_size();
 
  srn_fiber_stack_t s = srn_fiber_stack_alloc(0);
 
  TEST_CHECK(s.guard != nullptr);
  TEST_CHECK(s.limit != nullptr);
  TEST_CHECK(s.start != nullptr);
  // The guard sits exactly one page below the usable region.
  TEST_CHECK((char *)s.limit == (char *)s.guard + page);
  // The usable region is non-empty, page-aligned, and at least the default.
  TEST_CHECK((char *)s.start > (char *)s.limit);
  TEST_CHECK(((uintptr_t)s.limit & (page - 1)) == 0);
  TEST_CHECK(srn_fiber_stack_size(s) >= SRN_FIBER_DEFAULT_STACK_SIZE);
 
  // The whole usable region is writable end to end. The guard is left alone.
  ((char *)s.limit)[0]  = (char)0x5A;
  ((char *)s.start)[-1] = (char)0x5A;
 
  srn_fiber_stack_free(s);
}
 
// Ping-pong, a worker fiber and the calling thread hand control back and forth.
// This exercises the whole switch path -- srn_fiber_ctx_make seeds the worker,
// the first srn_fiber_switch starts it via the trampoline, each yield resumes
// the thread, and the worker finishes with srn_fiber_switch_final. The worker
// runs on a real guard-paged stack from the provider.
static srn_fiber_t fiber_pp_thread;
static srn_fiber_t fiber_pp_worker;
static int fiber_pp_ticks;
static int fiber_pp_rounds;
 
static void fiber_pp_entry(void *arg) {
  UNUSED(arg);
  srn_fiber_on_entry(&fiber_pp_thread);
  for (int i = 0; i < fiber_pp_rounds; i++) {
    fiber_pp_ticks++;
    srn_fiber_switch(&fiber_pp_worker, &fiber_pp_thread);
  }
  srn_fiber_switch_final(&fiber_pp_thread);
}
 
static void test_fiber_switch() {
  srn_fiber_init_thread(&fiber_pp_thread);
 
  srn_fiber_stack_t worker_stack = srn_fiber_stack_alloc(0);
 
  fiber_pp_worker       = (srn_fiber_t){};
  fiber_pp_worker.stack = worker_stack;
  fiber_pp_worker.state = SRN_FIBER_NEW;
  srn_fiber_ctx_make(&fiber_pp_worker.fiber_ctx, worker_stack, fiber_pp_entry, nullptr);
 
  fiber_pp_ticks  = 0;
  fiber_pp_rounds = 5;
 
  // `rounds` yields, plus one more switch to let the worker run to completion.
  for (int i = 0; i <= fiber_pp_rounds; i++) {
    srn_fiber_switch(&fiber_pp_thread, &fiber_pp_worker);
  }
 
  TEST_CHECK(fiber_pp_ticks == fiber_pp_rounds);
  TEST_CHECK(fiber_pp_thread.state == SRN_FIBER_RUNNING);
 
  srn_fiber_stack_free(worker_stack);
}
 
// A shared return value, these tests do not inspect the payload, only that a
// fiber ran and finished (a null result is legal too).
static int fiber_ok_value;
 
// fiber::sched::run -- spawn N fibers, each bumps a counter and returns. After
// srn_sched_run drains the queue every fiber must have run once and reached
// DONE, with its result recorded.
static int fiber_run_counter;
 
static srn_fiber_result_t fiber_run_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  fiber_run_counter++;
  return &fiber_ok_value;
}
 
static void test_fiber_sched_run() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_run_counter = 0;
  enum { N = 4 };
  srn_fiber_t *fibers[N];
  for (int i = 0; i < N; i++) {
    fibers[i] = srn_fiber_spawn(ctx, fiber_run_entry, nullptr);
    ASSERT_NOT_NULL(fibers[i]);
  }
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_run_counter == N);
  for (int i = 0; i < N; i++) {
    // The struct lives in the context block after reaping. Only the stack is
    // released, so state and result are still readable here.
    TEST_CHECK(fibers[i]->state == SRN_FIBER_DONE);
    TEST_CHECK(fibers[i]->result == &fiber_ok_value);
  }
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::yield -- two fibers each log their id twice, yielding between.
// Run on one worker. Per-worker queues do not guarantee a strict interleaving,
// so the invariant checked is that both fibers ran to completion, each id
// appears exactly twice, four entries in all.
static int fiber_yield_log[4];
static int fiber_yield_n;
 
static srn_fiber_result_t fiber_yield_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  int id = (int)(intptr_t)arg;
  for (int i = 0; i < 2; i++) {
    fiber_yield_log[fiber_yield_n++] = id;
    srn_fiber_yield();
  }
  return &fiber_ok_value;
}
 
static void test_fiber_sched_yield() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_yield_n = 0;
  (void)srn_fiber_spawn(ctx, fiber_yield_entry, (void *)(intptr_t)1);
  (void)srn_fiber_spawn(ctx, fiber_yield_entry, (void *)(intptr_t)2);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_yield_n == 4);
  int ones = 0;
  int twos = 0;
  for (int i = 0; i < 4; i++) {
    if (fiber_yield_log[i] == 1) {
      ones++;
    } else if (fiber_yield_log[i] == 2) {
      twos++;
    }
  }
  TEST_CHECK(ones == 2);
  TEST_CHECK(twos == 2);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::yield_round_robin -- yield moves the fiber to the back of the
// queue, so two yielding fibers on one worker interleave step by step. A
// yield that re-runs its own fiber produces runs of the same id instead.
static int fiber_rr_log[6];
static int fiber_rr_n;
 
static srn_fiber_result_t fiber_rr_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  int id = (int)(intptr_t)arg;
  for (int i = 0; i < 3; i++) {
    if (fiber_rr_n < 6) {
      fiber_rr_log[fiber_rr_n++] = id;
    }
    srn_fiber_yield();
  }
  return &fiber_ok_value;
}
 
static void test_fiber_sched_yield_round_robin() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_rr_n = 0;
  (void)srn_fiber_spawn(ctx, fiber_rr_entry, (void *)(intptr_t)1);
  (void)srn_fiber_spawn(ctx, fiber_rr_entry, (void *)(intptr_t)2);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_rr_n == 6);
  // Strict alternation, a yielding fiber never runs two steps back to back
  // while a peer is ready.
  for (int i = 1; i < 6; i++) {
    TEST_CHECK(fiber_rr_log[i] != fiber_rr_log[i - 1]);
    TEST_MSG(
      "order: %d %d %d %d %d %d", fiber_rr_log[0], fiber_rr_log[1], fiber_rr_log[2],
      fiber_rr_log[3], fiber_rr_log[4], fiber_rr_log[5]
    );
  }
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::suspend -- a one-slot mailbox handoff. The consumer runs first
// and parks. Its commit registers it as the waiter (the slot is empty). The
// producer fills the slot and readies the consumer, which resumes and reads it.
typedef struct {
  int value;
  bool has_value;
  srn_fiber_t *waiter;
} fiber_mbox_t;
 
static fiber_mbox_t fiber_mbox;
static int fiber_mbox_got;
 
// Park commit, register as the mailbox waiter, unless a value is already there
// (in which case decline to park so the consumer resumes at once).
static bool fiber_mbox_park(srn_fiber_t *self, void *arg) {
  fiber_mbox_t *mb = arg;
  if (mb->has_value) {
    return false;
  }
  mb->waiter = self;
  return true;
}
 
static srn_fiber_result_t fiber_consumer_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  srn_fiber_suspend(fiber_mbox_park, &fiber_mbox);
  fiber_mbox_got = fiber_mbox.value;
  return &fiber_ok_value;
}
 
static srn_fiber_result_t fiber_producer_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  fiber_mbox.value     = 42;
  fiber_mbox.has_value = true;
  if (fiber_mbox.waiter != nullptr) {
    srn_fiber_t *w    = fiber_mbox.waiter;
    fiber_mbox.waiter = nullptr;
    srn_fiber_ready(w);
  }
  return &fiber_ok_value;
}
 
static void test_fiber_sched_suspend() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_mbox     = (fiber_mbox_t){0};
  fiber_mbox_got = 0;
 
  // Order matters, the consumer must be enqueued first so it runs and suspends
  // before the producer fills the slot.
  (void)srn_fiber_spawn(ctx, fiber_consumer_entry, nullptr);
  (void)srn_fiber_spawn(ctx, fiber_producer_entry, nullptr);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_mbox_got == 42);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::park_abort -- the commit declines to park. The value is already
// in the slot when the consumer runs, so fiber_mbox_park returns false and the
// consumer resumes immediately without ever parking or registering a waiter.
static void test_fiber_sched_park_abort() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_mbox           = (fiber_mbox_t){0};
  fiber_mbox.value     = 7;
  fiber_mbox.has_value = true; // value present before the consumer runs
  fiber_mbox_got       = 0;
 
  (void)srn_fiber_spawn(ctx, fiber_consumer_entry, nullptr);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_mbox_got == 7);
  TEST_CHECK(fiber_mbox.waiter == nullptr); // never registered as a waiter
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::registry -- a fiber that suspends with no one to wake it. The
// run loop drains the ready queue and returns, leaving the fiber parked and its
// stack mapped. The scheduler still tracks it through the registry, so shutdown
// reclaims the stack instead of leaking it.
// Park commit that registers the fiber nowhere, so nothing can ever wake it.
static bool fiber_stuck_park(srn_fiber_t *self, void *arg) {
  UNUSED(self);
  UNUSED(arg);
  return true; // stay parked, with no waker
}
 
static srn_fiber_result_t fiber_stuck_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  srn_fiber_suspend(fiber_stuck_park, nullptr); // never readied
  return &fiber_ok_value;
}
 
static void test_fiber_sched_registry() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  srn_fiber_t *stuck = srn_fiber_spawn(ctx, fiber_stuck_entry, nullptr);
  ASSERT_NOT_NULL(stuck);
 
  // Drains the ready queue. The stuck fiber suspends and is abandoned.
  srn_sched_run(sched, 1);
  TEST_CHECK(stuck->state == SRN_FIBER_SUSPENDED);
 
  // The registry lets shutdown find and free the abandoned fiber's stack.
  srn_sched_shutdown(sched);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::double_wake -- two wakers ready the same suspended fiber (as an
// IO event and a timeout might race). The wake must be idempotent, the fiber
// runs exactly once, not enqueued twice. A double enqueue would resume an
// already-reaped fiber on the second pass and trip the sanitizer.
static srn_fiber_t *fiber_dwake_waiter;
static bool fiber_dwake_signalled;
static int fiber_dwake_runs;
 
static bool fiber_dwake_park(srn_fiber_t *self, void *arg) {
  UNUSED(arg);
  if (fiber_dwake_signalled) {
    return false;
  }
  fiber_dwake_waiter = self;
  return true;
}
 
static srn_fiber_result_t fiber_dwake_consumer_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  srn_fiber_suspend(fiber_dwake_park, nullptr);
  fiber_dwake_runs++;
  return &fiber_ok_value;
}
 
static srn_fiber_result_t fiber_dwake_producer_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  fiber_dwake_signalled = true;
  if (fiber_dwake_waiter != nullptr) {
    srn_fiber_t *w     = fiber_dwake_waiter;
    fiber_dwake_waiter = nullptr;
    srn_fiber_ready(w); // first waker: SUSPENDED -> READY, enqueued
    srn_fiber_ready(w); // second waker: already READY -> no-op
  }
  return &fiber_ok_value;
}
 
static void test_fiber_sched_double_wake() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_dwake_waiter    = nullptr;
  fiber_dwake_signalled = false;
  fiber_dwake_runs      = 0;
 
  // Consumer first, so it runs and suspends before the producer wakes it.
  (void)srn_fiber_spawn(ctx, fiber_dwake_consumer_entry, nullptr);
  (void)srn_fiber_spawn(ctx, fiber_dwake_producer_entry, nullptr);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_dwake_runs == 1);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::wait_for -- the caller blocks until the target finishes and
// reads its result. The waiter runs first and parks. The target runs later,
// finishes, and the DONE handling wakes the waiter.
static int fiber_wf_result;              // the target's result
static srn_fiber_t *fiber_wf_target;     // set before the run
static srn_fiber_result_t fiber_wf_seen; // what the waiter read
 
static srn_fiber_result_t fiber_wf_target_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  return &fiber_wf_result;
}
 
static srn_fiber_result_t fiber_wf_waiter_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  fiber_wf_seen = srn_fiber_wait_for(fiber_wf_target);
  return &fiber_ok_value;
}
 
static void test_fiber_sched_wait_for() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_wf_seen = nullptr;
  // Waiter made first, so it runs and parks before the target finishes.
  (void)srn_fiber_spawn(ctx, fiber_wf_waiter_entry, nullptr);
  fiber_wf_target = srn_fiber_spawn(ctx, fiber_wf_target_entry, nullptr);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_wf_seen == &fiber_wf_result);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::wait_for_done -- waiting for an already-finished target returns
// its result at once. The commit sees DONE and declines to park.
static void test_fiber_sched_wait_for_done() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_wf_seen = nullptr;
  // Target made first, so it runs and finishes before the waiter runs.
  fiber_wf_target = srn_fiber_spawn(ctx, fiber_wf_target_entry, nullptr);
  (void)srn_fiber_spawn(ctx, fiber_wf_waiter_entry, nullptr);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_wf_seen == &fiber_wf_result);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::wait_for_many -- several fibers wait for one target. All of
// them wake and read its result once it finishes.
static int fiber_wf_woke;
 
static srn_fiber_result_t fiber_wf_counting_waiter(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  if (srn_fiber_wait_for(fiber_wf_target) == &fiber_wf_result) {
    fiber_wf_woke++;
  }
  return &fiber_ok_value;
}
 
static void test_fiber_sched_wait_for_many() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_wf_woke = 0;
  // Three waiters made first so they all park, then the target finishes.
  (void)srn_fiber_spawn(ctx, fiber_wf_counting_waiter, nullptr);
  (void)srn_fiber_spawn(ctx, fiber_wf_counting_waiter, nullptr);
  (void)srn_fiber_spawn(ctx, fiber_wf_counting_waiter, nullptr);
  fiber_wf_target = srn_fiber_spawn(ctx, fiber_wf_target_entry, nullptr);
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(fiber_wf_woke == 3);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::mt_run -- many fibers across several worker threads, each doing
// one atomic increment. With true parallelism the only safe shared state is the
// atomic counter. After the run every increment must have landed (no lost
// updates) and the pool must have reached quiescence, which is what makes
// srn_sched_run return.
static atomic_int fiber_mt_counter;
 
static srn_fiber_result_t fiber_mt_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  atomic_fetch_add(&fiber_mt_counter, 1);
  return &fiber_ok_value;
}
 
static void test_fiber_sched_mt_run() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  atomic_store(&fiber_mt_counter, 0);
  enum { N = 256 };
  for (int i = 0; i < N; i++) {
    (void)srn_fiber_spawn(ctx, fiber_mt_entry, nullptr);
  }
 
  srn_sched_run(sched, 4);
 
  TEST_CHECK(atomic_load(&fiber_mt_counter) == N);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::mt_yield -- fibers that yield repeatedly across worker threads,
// so each is re-enqueued between rounds and may resume on a different thread
// than it last ran on. Exercises the cross-thread re-enqueue and the park/wake
// path under contention. The counter totals every round of every fiber.
static atomic_int fiber_mt_yield_counter;
 
static srn_fiber_result_t fiber_mt_yield_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  int rounds = (int)(intptr_t)arg;
  for (int i = 0; i < rounds; i++) {
    atomic_fetch_add(&fiber_mt_yield_counter, 1);
    srn_fiber_yield();
  }
  return &fiber_ok_value;
}
 
static void test_fiber_sched_mt_yield() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  atomic_store(&fiber_mt_yield_counter, 0);
  enum { N = 64, ROUNDS = 8 };
  for (int i = 0; i < N; i++) {
    (void)srn_fiber_spawn(ctx, fiber_mt_yield_entry, (void *)(intptr_t)ROUNDS);
  }
 
  srn_sched_run(sched, 4);
 
  TEST_CHECK(atomic_load(&fiber_mt_yield_counter) == N * ROUNDS);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// fiber::sched::stop -- a fiber stops the scheduler before all the work is
// done. The stopper runs a few rounds then calls srn_sched_stop, so
// srn_sched_run returns instead of draining the queue. The pending fibers
// behind it are left unrun, and srn_sched_shutdown reaps their stacks. Checks
// the run returns rather than hangs, only the stopper ran, and the leftover
// teardown is clean.
static atomic_int fiber_stop_rounds;
 
static srn_fiber_result_t fiber_stopper_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  srn_scheduler_t *sched = arg;
  // No yield in this loop, a yielded fiber goes to the back of the queue, so
  // yielding would hand the pending fibers a turn before the stop lands.
  for (int i = 0; i < 3; i++) {
    atomic_fetch_add(&fiber_stop_rounds, 1);
  }
  srn_sched_stop(sched);
  return &fiber_ok_value;
}
 
static srn_fiber_result_t fiber_pending_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  // Never reached, the stopper ends the run first. It adds a large amount, so
  // any accidental execution is visible in the count.
  atomic_fetch_add(&fiber_stop_rounds, 1000);
  return &fiber_ok_value;
}
 
static void test_fiber_sched_stop() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  atomic_store(&fiber_stop_rounds, 0);
  // Stopper first so it reaches the single worker before the pending fibers.
  (void)srn_fiber_spawn(ctx, fiber_stopper_entry, sched);
  enum { PENDING = 3 };
  for (int i = 0; i < PENDING; i++) {
    (void)srn_fiber_spawn(ctx, fiber_pending_entry, nullptr);
  }
 
  // Returns when the stopper calls srn_sched_stop, with the pending fibers
  // still queued.
  srn_sched_run(sched, 1);
 
  // Only the stopper ran (three rounds); no pending fiber ran.
  TEST_CHECK(atomic_load(&fiber_stop_rounds) == 3);
 
  // Reap the pending fibers while their structs (in the context block) are
  // still alive, before releasing the context. SHUTDOWN_ENGINE calls shutdown
  // again, which is a no-op on an already torn-down scheduler.
  srn_sched_shutdown(sched);
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}