scheduler.c File Reference

#include <stdatomic.h>
#include "serene/rt/context.h"
#include "serene/rt/engine.h"
#include "serene/rt/fiber.h"
#include "serene/rt/fiber/thread.h"
#include "serene/rt/mm/interface.h"
#include "serene/utils.h"

Include dependency graph for scheduler.c:

Go to the source code of this file.

Data Structures
struct	srn_scheduler_t
struct	srn_worker_t
	The state one os thread uses to run fibers. More...

Macros
#define	SCHED_LOG(FMT, ...)
#define	SCHED_TRACE(...)
	Per-operation deque and queue tracing (push, pop, steal, wake).
#define	SRN_MAX_WORKERS   256
	Upper bound on workers per run.
#define	SRN_FIBER_LOCAL_RING_CAP   256
	Capacity of each worker's local work-stealing deque.

Typedefs
typedef struct srn_worker_t	srn_worker_t
	Defined here, not in fiber.h, which only forward declares srn_scheduler_t.
typedef enum srn_sched_state_t	srn_sched_state_t
	The scheduler's lifecycle as one atomic value.

Enumerations
enum	srn_sched_state_t { SRN_SCHED_IDLE , SRN_SCHED_RUNNING , SRN_SCHED_STOPPING }
	The scheduler's lifecycle as one atomic value. More...

Functions
srn_scheduler_t *	srn_sched_init (srn_engine_t *engine)
static void	registry_add (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Insert at the head of the registry. Caller must hold sched->lock.
static void	registry_remove (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Unlink from the registry.
void	srn_sched_register (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Record a fiber in the scheduler's registry of live fibers, where it stays until it is reaped.
void	srn_sched_shutdown (srn_scheduler_t *sched)
	The one stop tear down of the fiber subsystem, should be called once srn_sched_run has returned.
static void	announce_work (srn_scheduler_t *sched)
	Wake the os thread of one parked worker after a fiber has joined a queue.
static bool	local_push (srn_worker_t *w, srn_fiber_t *fiber)
	This operation is only for the owner of the ring.
static srn_fiber_t *	local_pop (srn_worker_t *w)
	Owner only.
static srn_fiber_t *	local_steal (srn_worker_t *victim)
	Thief side.
static void	global_enqueue (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Append a fiber to the global/overflow queue.
static srn_fiber_t *	global_take (srn_scheduler_t *sched)
	Pop the head of the global queue, or null when empty.
static void	push_ready (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Put a runnable fiber on a queue, with its state already set to READY.
void	srn_sched_enqueue (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Place a fiber on a scheduler's ready queue, making it eligible to run.
static void	ready_fiber (srn_scheduler_t *sched, srn_fiber_t *fiber)
	Wake a parked fiber by flipping SUSPENDED to READY and enqueuing it.
static srn_fiber_t *	find_work (srn_worker_t *w)
	Find a fiber to run: the worker's own deque first, then the global queue, then a steal of one fiber from each peer in turn.
static void	worker_run (srn_worker_t *worker)
	Run the worker routine over worker on the calling os thread.
static void	worker_main (void *arg)
	The entry an os thread starts in.
void	srn_sched_run (srn_scheduler_t *sched, int nworkers)
	Run the scheduler with nworkers os threads draining it, returning once the pool goes quiescent (every os thread parked on an empty queue) or a stop is requested with srn_sched_stop.
void	srn_sched_stop (srn_scheduler_t *sched)
	Ask a running scheduler to stop.
void	srn_fiber_yield (void)
	Yield cooperatively: re-enqueue the running fiber and run the next ready one.
void	srn_fiber_suspend (srn_fiber_park_fn commit, void *arg)
	A suspended fiber is on no scheduler queue, and the scheduler does not track what it waits on – whoever wakes it does.
void	srn_fiber_ready (srn_fiber_t *fiber)
	Mark a suspended fiber runnable again.
srn_fiber_t *	srn_fiber_current (void)
	The fiber currently running on this os thread (the bootstrap fiber if none).
srn_fiber_t *	srn_fiber_worker_loop (void)
	The worker's loop of the worker running on the calling os thread.
static bool	wait_for_park (srn_fiber_t *self, void *arg)
	Add the calling fiber to the target's waiter list and stay parked, unless the target has already finished, in which case decline to park so the caller resumes at once.
srn_fiber_result_t	srn_fiber_wait_for (srn_fiber_t *target)
	Block the calling fiber until target finishes, then return its result.

Variables
static _Thread_local srn_worker_t *	current_worker = nullptr
	The worker the calling os thread is running, or null when this os thread is not running the worker routine (srn_sched_run is not active on it).

Macro Definition Documentation

SCHED_LOG

#define SCHED_LOG	(		FMT,
			... )

Value:

DBG("SCHED", FMT __VA_OPT__(, ) __VA_ARGS__)

Definition at line 28 of file scheduler.c.

SCHED_TRACE

#define SCHED_TRACE

(

...

)

Value:

  do {                                                                                             \
  } while (0)

Per-operation deque and queue tracing (push, pop, steal, wake).

This fires on the hot path, so it floods a debug build and is off unless SRN_SCHED_TRACE is defined. SCHED_LOG (scheduler lifecycle: stop, shutdown reap) stays on in a debug build. Both are silent in release, since DBG is.

Definition at line 37 of file scheduler.c.

#define SCHED_TRACE(...)                                                                           \
  do {                                                                                             \
  } while (0)

SRN_FIBER_LOCAL_RING_CAP

#define SRN_FIBER_LOCAL_RING_CAP   256

Capacity of each worker's local work-stealing deque.

Must be a power of two: the live slot for a deque index is index & (cap - 1). 256 matches the common choice (Go, Tokio). A fiber that does not fit overflows to the global queue. This is the single source of truth for the size, so a configuration layer can later drive it.

Definition at line 209 of file scheduler.c.

SRN_MAX_WORKERS

#define SRN_MAX_WORKERS   256

Upper bound on workers per run.

Clamps the worker count srn_sched_run accepts.

Definition at line 44 of file scheduler.c.

Typedef Documentation

srn_sched_state_t

typedef enum srn_sched_state_t srn_sched_state_t

The scheduler's lifecycle as one atomic value.

RUNNING means a run is draining the queues. STOPPING tells the workers to wind down, set at natural quiescence or by srn_sched_stop. IDLE is the resting state before a run starts. Workers read it without the lock, so it is atomic.

srn_worker_t

typedef struct srn_worker_t srn_worker_t

Defined here, not in fiber.h, which only forward declares srn_scheduler_t.

Consumers hold a srn_scheduler_t * and never see the layout. Two reasons:

1. The layout can evolve without recompiling or disturbing consumers. Going M:N this struct grows per-thread local queues, a worker array, a reactor handle, steal state – none of which should ripple into every translation unit that includes fiber.h.
2. It makes the decide/execute boundary physical. The scheduler owns the ready queue and the picking policy. The worker routine and fibers must reach it only through enqueue/yield/ready. Keeping the fields private means no caller can poke the queue directly – the encapsulation is the contract, enforced by the compiler rather than by convention.

Definition at line 119 of file scheduler.c.

Enumeration Type Documentation

srn_sched_state_t

enum srn_sched_state_t

The scheduler's lifecycle as one atomic value.

RUNNING means a run is draining the queues. STOPPING tells the workers to wind down, set at natural quiescence or by srn_sched_stop. IDLE is the resting state before a run starts. Workers read it without the lock, so it is atomic.

Enumerator
SRN_SCHED_IDLE
SRN_SCHED_RUNNING
SRN_SCHED_STOPPING

Definition at line 125 of file scheduler.c.

                               {
  SRN_SCHED_IDLE,
  SRN_SCHED_RUNNING,
  SRN_SCHED_STOPPING,
} srn_sched_state_t;

Function Documentation

announce_work()

static void announce_work ( srn_scheduler_t * sched )

static

Wake the os thread of one parked worker after a fiber has joined a queue.

"Parked" means that os thread is asleep in srn_cond_wait because it found no runnable fiber anywhere. This is not a fiber suspending. It is the whole os thread blocked, and the notify wakes it so it looks again.

runnable is bumped first, then idle is read. Paired against the park path, which bumps idle then reads runnable, this ordering means the two sides can never both miss, so a wakeup is never lost. The notify takes the global lock (the condition's lock) but only when an os thread is actually parked, so the common busy case never touches it.

WARNING: this runs with NO lock around the runnable++ and the idle read, so its only tie to the park path is the seq_cst total order. Both must stay seq_cst. RELAX EITHER AND THE WAKEUP CAN BE LOST (an os thread parked with a runnable fiber queued). See the srn_scheduler_t coordination comment for the full reasoning.

Definition at line 402 of file scheduler.c.

                                                  {
  PANIC_IF_NULL(sched);
 
  atomic_fetch_add(&sched->runnable, 1);
 
  if (atomic_load(&sched->idle) > 0) {
    // We have idle os threads. Wake them up
    srn_mutex_lock(&sched->lock);
    SCHED_TRACE("waking a parked os thread (runnable=%ld)", (long)atomic_load(&sched->runnable));
    srn_cond_notify_one(&sched->work);
    srn_mutex_unlock(&sched->lock);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

find_work()

static srn_fiber_t * find_work ( srn_worker_t * w )

static

Find a fiber to run: the worker's own deque first, then the global queue, then a steal of one fiber from each peer in turn.

Null when nothing is runnable anywhere this worker can reach. Decrements runnable for whatever it takes.

Definition at line 616 of file scheduler.c.

                                               {
  srn_scheduler_t *sched = w->sched;
 
  srn_fiber_t *fiber = local_pop(w);
  if (fiber == nullptr) {
    fiber = global_take(sched);
  }
 
  if (fiber == nullptr) {
    for (int i = 1; i < sched->nworkers; i++) {
      // We start form the right side neighbour and with `i` growing we will
      // eventually loop back to the left side neighbour in the workers array.
      int index = (w->id + i) % sched->nworkers;
      srn_worker_t *victim = &sched->workers[index];
      fiber = local_steal(victim);
 
      if (fiber != nullptr) {
        SCHED_TRACE("worker %d stole fiber %p from worker %d", w->id, (void *)fiber, victim->id);
        break;
      }
    }
  }
 
  if (fiber != nullptr) {
    atomic_fetch_sub(&sched->runnable, 1);
  }
 
  return fiber;
}

Here is the call graph for this function:

Here is the caller graph for this function:

global_enqueue()

static void global_enqueue ( srn_scheduler_t * sched, srn_fiber_t * fiber )

static

Append a fiber to the global/overflow queue.

The caller has set its state. The push, the runnable bump, and the wake all run under the global lock, so this path is trivially serialized against the park path and needs no separate ordering argument.

Put a fiber on the global queue and wake a parked os thread if any. Unlike announce_work, the runnable++, the idle read, and the notify all happen under the lock, so this path is safe by mutual exclusion and does not lean on the seq_cst ordering the lockless path does.

Definition at line 530 of file scheduler.c.

                                                                       {
  srn_mutex_lock(&sched->lock);
 
  fiber->link = nullptr;
 
  if (sched->ready_tail == nullptr) {
    sched->ready_head = fiber;
  } else {
    sched->ready_tail->link = fiber;
  }
  sched->ready_tail = fiber;
 
  atomic_fetch_add(&sched->runnable, 1);
 
  SCHED_TRACE("global-push fiber %p (runnable=%ld)", (void *)fiber,
              (long)atomic_load(&sched->runnable));
 
  if (atomic_load(&sched->idle) > 0) {
    srn_cond_notify_one(&sched->work);
  }
  srn_mutex_unlock(&sched->lock);
}

Here is the call graph for this function:

Here is the caller graph for this function:

global_take()

static srn_fiber_t * global_take ( srn_scheduler_t * sched )

static

Pop the head of the global queue, or null when empty.

The runnable adjustment is left to find_work, the only taker.

Definition at line 555 of file scheduler.c.

                                                        {
  srn_mutex_lock(&sched->lock);
  srn_fiber_t *fiber = sched->ready_head;
 
  if (fiber != nullptr) {
    sched->ready_head = fiber->link;
    if (sched->ready_head == nullptr) {
      sched->ready_tail = nullptr;
    }
    fiber->link = nullptr;
  }
 
  srn_mutex_unlock(&sched->lock);
  return fiber;
}

Here is the call graph for this function:

Here is the caller graph for this function:

local_pop()

static srn_fiber_t * local_pop ( srn_worker_t * w )

static

Owner only.

Pop a fiber from the bottom, or null when empty. The seq_cst fence and the compare-and-swap settle the race with a thief over the last element.

Definition at line 455 of file scheduler.c.

                                               {
  PANIC_IF_NULL(w);
 
  // Since bottom is local to the owner, there is no other writer competing to
  // write to it. So a load/store is enough here no need for `atomic_fetch_sub`.
  intptr_t b = atomic_load_explicit(&w->bottom, memory_order_relaxed) - 1;
  atomic_store_explicit(&w->bottom, b, memory_order_relaxed);
 
  atomic_thread_fence(memory_order_seq_cst);
  intptr_t t = atomic_load_explicit(&w->top, memory_order_relaxed);
 
  srn_fiber_t *fiber = nullptr;
  if (t <= b) {
    // Non-empty.
    fiber =
        atomic_load_explicit(&w->ring[b & (SRN_FIBER_LOCAL_RING_CAP - 1)], memory_order_relaxed);
    if (t == b) {
      // Last element. The owner and a thief can race for it, so settle it with
      // the CAS on `top`. Exactly one of them wins.
      if (atomic_compare_exchange_strong_explicit(&w->top, &t, t + 1, memory_order_seq_cst,
                                                  memory_order_relaxed)) {
        SCHED_TRACE("worker %d popped the last fiber %p", w->id, (void *)fiber);
 
      } else {
        SCHED_TRACE("worker %d lost the last fiber %p to a thief", w->id, (void *)fiber);
        fiber = nullptr; // the thief won
      }
      atomic_store_explicit(&w->bottom, b + 1, memory_order_relaxed);
    }
  } else {
    // Empty. Restore bottom.
    atomic_store_explicit(&w->bottom, b + 1, memory_order_relaxed);
  }
  return fiber;
}

Here is the caller graph for this function:

local_push()

static bool local_push ( srn_worker_t * w, srn_fiber_t * fiber )

static

This operation is only for the owner of the ring.

Push a fiber on the bottom. Returns false when the deque is full, so the caller can overflow it to the global queue. The caller has set state.

Definition at line 426 of file scheduler.c.

                                                            {
  PANIC_IF_NULL(w);
  PANIC_IF_NULL(fiber);
 
  intptr_t b = atomic_load_explicit(&w->bottom, memory_order_relaxed);
  intptr_t t = atomic_load_explicit(&w->top, memory_order_acquire);
  if (b - t >= (intptr_t)SRN_FIBER_LOCAL_RING_CAP) {
    return false; // full
  }
 
  atomic_store_explicit(&w->ring[b & (SRN_FIBER_LOCAL_RING_CAP - 1)], fiber, memory_order_relaxed);
  // After ^^^, the slot isn't published to thieves yet, because they decide
  // what's live by reading bottom, which we haven't bumped
 
  // Publish the slot. write before the bottom store that exposes it to a thief.
  // This is the key barrier. It orders the slot write before the bottom bump
  // that follows. Paired with a thief's `acquire-load` of bottom in
  // `local_steal`, it guarantees: if a thief sees the new bottom, it also sees
  // the fiber we just wrote, never a stale/garbage slot.
  atomic_thread_fence(memory_order_release);
  atomic_store_explicit(&w->bottom, b + 1, memory_order_relaxed);
 
  SCHED_TRACE("worker %d local-push fiber %p", w->id, (void *)fiber);
  return true;
}

Here is the caller graph for this function:

local_steal()

static srn_fiber_t * local_steal ( srn_worker_t * victim )

static

Thief side.

Take a fiber from victim's top, or null when the deque is empty or a concurrent take won the race – the caller then just moves to the next victim.

Definition at line 494 of file scheduler.c.

                                                      {
  PANIC_IF_NULL(victim);
 
  intptr_t t = atomic_load_explicit(&victim->top, memory_order_acquire);
  // Pairs with the `seq_cst` fence in `local_pop`. The two fences force a
  // single total order in which the owner (lowering bottom, fence, reading top)
  // and this thief (reading top, fence, reading bottom) cannot both decide they
  // got the last element.
  // Basically this fence handles the steal race against local_pop.
  // Note: Don't mixup `memory_order_seq_cst` with `memory_order_acquire` that
  // we use for loading victim's bottom the next line.
  atomic_thread_fence(memory_order_seq_cst);
  // This acquire on bottom handles slot visibility against `local_push`
  intptr_t b = atomic_load_explicit(&victim->bottom, memory_order_acquire);
 
  srn_fiber_t *fiber = nullptr;
  if (t < b) {
    fiber = atomic_load_explicit(&victim->ring[t & (SRN_FIBER_LOCAL_RING_CAP - 1)],
                                 memory_order_relaxed);
    if (!atomic_compare_exchange_strong_explicit(&victim->top, &t, t + 1, memory_order_seq_cst,
                                                 memory_order_relaxed)) {
      fiber = nullptr; // lost the race
    }
  }
  return fiber;
}

Here is the caller graph for this function:

push_ready()

static void push_ready ( srn_scheduler_t * sched, srn_fiber_t * fiber )

static

Put a runnable fiber on a queue, with its state already set to READY.

A fiber enqueued while running on a worker goes onto that worker's local deque, keeping its work local. One enqueued from off a worker (the initial fibers made before the run, or an external waker), or one that does not fit a full local deque, goes to the global queue.

Definition at line 576 of file scheduler.c.

                                                                   {
  srn_worker_t *w = current_worker;
 
  // On a worker: try its own deque first, falling through on a full deque.
  if (w != nullptr) {
    if (local_push(w, fiber)) {
      announce_work(sched);
      return;
    }
 
    SCHED_TRACE("worker %d local deque full, overflow fiber %p to global", w->id, (void *)fiber);
  }
 
  // Off a worker, or the deque was full: the global queue takes it.
  global_enqueue(sched, fiber);
}

Here is the call graph for this function:

Here is the caller graph for this function:

ready_fiber()

static void ready_fiber ( srn_scheduler_t * sched, srn_fiber_t * fiber )

static

Wake a parked fiber by flipping SUSPENDED to READY and enqueuing it.

Only the flip's winner enqueues, so racing wakers cannot double-enqueue it, and a fiber that is not parked is left untouched. The scheduler does not check the awaited condition. A fiber woken early resumes, re-checks, and parks again.

Definition at line 605 of file scheduler.c.

                                                                    {
  srn_fiber_state_t expected = SRN_FIBER_SUSPENDED;
  if (atomic_compare_exchange_strong(&fiber->state, &expected, SRN_FIBER_READY)) {
    push_ready(sched, fiber);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

registry_add()

static void registry_add ( srn_scheduler_t * sched, srn_fiber_t * fiber )

static

Insert at the head of the registry. Caller must hold sched->lock.

Definition at line 277 of file scheduler.c.

                                                                     {
  fiber->reg_prev = nullptr;
  fiber->reg_next = sched->registry;
 
  if (sched->registry != nullptr) {
    sched->registry->reg_prev = fiber;
  }
 
  sched->registry = fiber;
}

Here is the caller graph for this function:

registry_remove()

static void registry_remove ( srn_scheduler_t * sched, srn_fiber_t * fiber )

static

Unlink from the registry.

O(1), thanks to the back pointer. Caller must hold sched->lock.

Definition at line 290 of file scheduler.c.

                                                                        {
  if (fiber->reg_prev != nullptr) {
    fiber->reg_prev->reg_next = fiber->reg_next;
  } else {
    sched->registry = fiber->reg_next;
  }
 
  if (fiber->reg_next != nullptr) {
    fiber->reg_next->reg_prev = fiber->reg_prev;
  }
 
  fiber->reg_prev = nullptr;
  fiber->reg_next = nullptr;
}

Here is the caller graph for this function:

srn_fiber_current()

srn_fiber_t * srn_fiber_current

(

void

)

The fiber currently running on this os thread (the bootstrap fiber if none).

Definition at line 930 of file scheduler.c.

                                     {
  return current_worker != nullptr ? current_worker->current : nullptr;
}

Here is the caller graph for this function:

srn_fiber_ready()

void srn_fiber_ready

(

srn_fiber_t *

fiber

)

Mark a suspended fiber runnable again.

This is the seam an IO reactor, timer, or peer fiber uses to wake a fiber when the event it awaited occurs.

Definition at line 918 of file scheduler.c.

                                         {
  PANIC_IF_NULL(fiber);
 
  // Wake a suspended fiber. The flip in `ready_fiber` lets exactly one of
  // several racing wakers enqueue it (an IO completion and a timeout firing on
  // it, say), while the rest find it no longer `SUSPENDED` and do nothing. The
  // scheduler is resolved from the fiber, not the calling os thread, so a waker
  // that is not itself running a worker (the IO reactor, a peer os thread) can
  // wake it.
  ready_fiber(srn_fiber_get_scheduler_m(fiber), fiber);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_suspend()

void srn_fiber_suspend	(	srn_fiber_park_fn	commit,
		void *	arg )

A suspended fiber is on no scheduler queue, and the scheduler does not track what it waits on – whoever wakes it does.

Park the running fiber until a party calls srn_fiber_ready.

The commit callback runs on the worker's loop side once the fiber has switched out. It hands the fiber's pointer to the event source it blocks on (a peer fiber, a lock's waiter list, the IO reactor's fd table), so that party can call srn_fiber_ready when the awaited event occurs. Running commit only after the suspend completes is what makes the hand-off race free, a waker can never observe a half-suspended fiber. If commit registers the fiber nowhere, it is genuinely lost – a deadlock, like an os thread blocking on a condition nobody signals.

Definition at line 898 of file scheduler.c.

                                                            {
  PANIC_IF_NULL(commit);
 
  srn_worker_t *worker = current_worker;
  PANIC_IF_NULL(worker);
 
  srn_fiber_t *self = worker->current;
  PANIC_IF_NULL(self);
 
  // The fiber carries its own commit. The worker routine runs it after we
  // switch out -- the one safe point to publish a fully suspended fiber to its
  // waker. The routine also stamps the `SUSPENDED` state once the switch
  // completes, so the `state` never marks a fiber that is still suspending.
  // This call leaves the `state` as `RUNNING` and lets the switch carry the
  // fiber off the os thread.
  self->park_commit = commit;
  self->park_arg = arg;
  srn_fiber_switch(self, &worker->loop);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_wait_for()

srn_fiber_result_t srn_fiber_wait_for

(

srn_fiber_t *

target

)

Block the calling fiber until target finishes, then return its result.

Definition at line 963 of file scheduler.c.

                                                           {
  PANIC_IF_NULL(target);
  PANIC_IF(target == srn_fiber_current(), "srn_fiber_wait_for: a fiber cannot wait for itself");
 
  // Suspend until the target finishes (wait_for_park registers us on its waiter
  // list). The target's DONE handling in the worker routine wakes us. The
  // result is read from the struct, which survives the target's reap.
  srn_fiber_suspend(wait_for_park, target);
  return target->result;
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_worker_loop()

srn_fiber_t * srn_fiber_worker_loop

(

void

)

The worker's loop of the worker running on the calling os thread.

The fiber that resumes when the current fiber yields, suspends, or finishes, the resumer a fiber's launcher hands control back to. Each worker has its own, so this is valid only on an os thread that is currently running the worker routine.

Definition at line 934 of file scheduler.c.

                                         {
  PANIC_IF_NULL(current_worker);
  return &current_worker->loop;
}

Here is the caller graph for this function:

srn_fiber_yield()

void srn_fiber_yield

(

void

)

Yield cooperatively: re-enqueue the running fiber and run the next ready one.

Acts on the fiber currently running on this thread.

Definition at line 876 of file scheduler.c.

                           {
  srn_worker_t *worker = current_worker;
  PANIC_IF_NULL(worker);
 
  // Switch to the worker's loop without enqueuing first. The worker routine
  // puts this fiber back on the ready queue once the switch has saved its
  // context. Enqueuing here, before the switch, would let another os thread
  // dequeue and resume the fiber while this os thread is still saving its
  // context -- two os threads on one fiber stack, which corrupts the switch.
  srn_fiber_t *self = worker->current;
  srn_fiber_switch(self, &worker->loop);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_enqueue()

void srn_sched_enqueue	(	srn_scheduler_t *	sched,
		srn_fiber_t *	fiber )

Place a fiber on a scheduler's ready queue, making it eligible to run.

Definition at line 593 of file scheduler.c.

                                                                   {
  PANIC_IF_NULL(sched);
  PANIC_IF_NULL(fiber);
 
  fiber->state = SRN_FIBER_READY;
  push_ready(sched, fiber);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_init()

srn_scheduler_t * srn_sched_init ( srn_engine_t * engine )

nodiscard

Definition at line 246 of file scheduler.c.

                                                      {
  PANIC_IF_NULL(engine);
  // The scheduler outlives every context and fiber, so it is allocated from the
  // immortal region rather than a releasable block.
  srn_scheduler_t *sched = srn_mm_immortal_allocate(engine->mm, srn_scheduler_t);
  PANIC_IF_NULL(sched);
 
  sched->engine = engine;
  sched->ready_head = nullptr;
  sched->ready_tail = nullptr;
  sched->registry = nullptr;
  sched->idle = 0;
  sched->runnable = 0;
  sched->nworkers = 0;
  sched->state = SRN_SCHED_IDLE;
  sched->workers = nullptr;
  sched->os_threads = nullptr;
  sched->destroyed = false;
  atomic_init(&sched->run_active, false);
 
  PANIC_IF(srn_mutex_init(&sched->lock) != SRN_THREAD_OK,
           "failed to initialise the scheduler lock");
 
  PANIC_IF(srn_cond_init(&sched->work) != SRN_THREAD_OK,
           "failed to initialise the scheduler condition");
 
  // srn_engine_make will store this scheduler in the engine
  return sched;
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_register()

void srn_sched_register	(	srn_scheduler_t *	sched,
		srn_fiber_t *	fiber )

Record a fiber in the scheduler's registry of live fibers, where it stays until it is reaped.

Definition at line 305 of file scheduler.c.

                                                                    {
  PANIC_IF_NULL(sched);
  PANIC_IF_NULL(fiber);
 
  srn_mutex_lock(&sched->lock);
  registry_add(sched, fiber);
  srn_mutex_unlock(&sched->lock);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_run()

void srn_sched_run	(	srn_scheduler_t *	sched,
		int	nworkers )

Run the scheduler with nworkers os threads draining it, returning once the pool goes quiescent (every os thread parked on an empty queue) or a stop is requested with srn_sched_stop.

The calling os thread becomes worker 0, so it does not return until then. nworkers is clamped to at least 1; with 1 it is the calling os thread alone, which keeps execution single-threaded and cooperatively ordered. The spawned os threads are not joined here – srn_sched_shutdown joins them as part of tearing the subsystem down.

Definition at line 784 of file scheduler.c.

                                                         {
  PANIC_IF_NULL(sched);
  PANIC_IF(sched->destroyed, "srn_sched_run called on a scheduler that was already shut down");
 
  if (nworkers < 1) {
    nworkers = 1;
  }
 
  if (nworkers > SRN_MAX_WORKERS) {
    nworkers = SRN_MAX_WORKERS;
  }
 
  // Allocate `workers` and `os_threads` on the scheduler so
  // `srn_sched_shutdown` can join the threads and free them later. `workers`
  // has one entry per worker. `os_threads` has one per spawned thread, with
  // slot 0 left empty because the caller runs worker 0 inline (see the struct
  // comment). `runnable` is left alone, it already counts the fibers queued
  // before the run.
  sched->workers = srn_mm_allocate(sched->engine->mm, (size_t)nworkers * sizeof(srn_worker_t));
  PANIC_IF_NULL(sched->workers);
 
  sched->os_threads = srn_mm_allocate(sched->engine->mm, (size_t)nworkers * sizeof(srn_thread_t));
  PANIC_IF_NULL(sched->os_threads);
 
  for (int i = 0; i < nworkers; i++) {
    srn_worker_t *w = &sched->workers[i];
    w->sched = sched;
    w->id = i;
    w->current = nullptr;
    // The deque indices start empty. Its ring slots are written before they are
    // read, and the worker's loop is set up by worker_main on its own os
    // thread.
    atomic_init(&w->top, 0);
    atomic_init(&w->bottom, 0);
  }
 
  // Publish the coordination state before any os thread starts. `nworkers` must
  // be set first so the quiescence check counts the right total, and the state
  // must be RUNNING before an os thread can observe it. `run_active` lets
  // shutdown see that a run is in flight.
  sched->idle = 0;
  sched->nworkers = nworkers;
  atomic_store(&sched->state, SRN_SCHED_RUNNING);
  atomic_store(&sched->run_active, true);
 
  // Spawn nworkers - 1 os threads. The calling os thread runs worker 0 inline.
  // A spawn failure at startup is fatal, a partial pool would never reach `idle
  // == nworkers` and so never quiesce.
  for (int i = 1; i < nworkers; i++) {
    if (srn_thread_spawn(&sched->os_threads[i], worker_main, &sched->workers[i]) != SRN_THREAD_OK) {
      PANIC("failed to spawn an os thread");
    }
  }
 
  worker_main(&sched->workers[0]);
 
  // Worker 0 has stopped, so the run is over from the caller's point of view.
  // The spawned os threads may still be winding down, so they are NOT joined
  // here. `srn_sched_shutdown` joins them (the `os_threads` live on the
  // scheduler) as part of tearing the subsystem down. Clearing `run_active`
  // lets shutdown proceed. The state stays STOPPING, which keeps any os thread
  // still looping on its way out.
  atomic_store(&sched->run_active, false);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_shutdown()

void srn_sched_shutdown

(

srn_scheduler_t *

sched

)

The one stop tear down of the fiber subsystem, should be called once srn_sched_run has returned.

It joins the os threads, then releases the stack of every fiber still registered (any left unreaped, such as one suspended with no party to wake it, or one left queued when the run stopped early), and frees the scheduler's own resources. The fiber structs themselves live in context blocks and are reclaimed with those blocks, not here.

The scheduler is NOT usable after this. A later srn_sched_run panics, and a repeated srn_sched_shutdown is a no-op (so the engine can call it unconditionally even after a caller already did).

Must be called from outside the pool, never from an os thread that is running a worker nor from a fiber, and only after srn_sched_run has returned. Calling it on a live pool panics – stop the pool with srn_sched_stop, let srn_sched_run return, then shut down.

Definition at line 314 of file scheduler.c.

                                                {
  PANIC_IF_NULL(sched);
 
  if (sched->destroyed) {
    return;
  }
 
  // It must run on a thread outside the pool: a worker, or a fiber (which runs
  // on a worker), would be tearing down the scheduler it is itself running on.
  // `current_worker` is null only off a worker, so it is the test for that.
  PANIC_IF(current_worker != nullptr,
           "srn_sched_shutdown must be called from outside the worker pool");
 
  // And it must run after `srn_sched_run` has returned, not while a run is in
  // flight. Stop a running pool with `srn_sched_stop` and let `srn_sched_run`
  // return first.
  PANIC_IF(atomic_load(&sched->run_active),
           "srn_sched_shutdown called while srn_sched_run is active; call "
           "srn_sched_stop and let the run return first");
 
  // The run has returned, so worker 0 has stopped. The spawned os threads may
  // still be winding down (`srn_sched_run` does not join them), so join them
  // now. `os_threads` slot 0 is the inline worker 0, never spawned, so the
  // spawned os threads to join are 1..nworkers-1.
  for (int i = 1; i < sched->nworkers; i++) {
    (void)srn_thread_join(&sched->os_threads[i]);
  }
 
  // Every worker is gone, so this runs single threaded now.
  //
  // Any fiber still in the registry never finished: it was left parked in
  // SRN_FIBER_SUSPENDED with no party able to wake it (a deadlock), or was left
  // queued when the run stopped early. Release its stack so it does not leak.
  // The fiber structs themselves live in context blocks and are reclaimed with
  // those blocks, not here. The scheduler is immortal-allocated, so it is not
  // freed either.
  //
  // Unlike the reap path, this unmaps for good: shutdown runs after the workers
  // are gone, so the per-thread stack ring from the fiber.h TODO no longer
  // exists and there is nothing to recycle into. (Draining that ring, when it
  // exists, also belongs here.)
  srn_fiber_t *fiber = sched->registry;
  while (fiber != nullptr) {
    srn_fiber_t *next = fiber->reg_next;
    SCHED_LOG("shutdown reaping unfinished fiber '%s' (suspended with no waker?)",
              fiber->name != nullptr ? fiber->name : "<unnamed>");
    // TODO(lxsameer): Free up the ring here as well
    srn_fiber_stack_free(fiber->stack);
    srn_fiber_on_reap(fiber);
    fiber->reg_prev = nullptr;
    fiber->reg_next = nullptr;
    fiber = next;
  }
  sched->registry = nullptr;
 
  // Release the run-scoped storage (srn_mm_free tolerates null, so a scheduler
  // that never ran is fine).
  srn_mm_free(sched->engine->mm, sched->os_threads);
  srn_mm_free(sched->engine->mm, sched->workers);
  sched->os_threads = nullptr;
  sched->workers = nullptr;
  sched->nworkers = 0;
 
  // Destroy the synchronisation primitives. The scheduler is not usable after
  // this, so they are not re-initialised. No worker holds or waits on them now,
  // the join above made sure of that.
  (void)srn_cond_destroy(&sched->work);
  (void)srn_mutex_destroy(&sched->lock);
 
  sched->destroyed = true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_stop()

void srn_sched_stop

(

srn_scheduler_t *

sched

)

Ask a running scheduler to stop.

Each worker routine checks the request at the top of each turn and stops once the fiber it is running yields or finishes, so a slice in flight is never cut mid-execution. srn_sched_run then returns. Fibers left queued stay unrun and are reclaimed by srn_sched_shutdown. Does not wait. Safe to call from any os thread, including a signal-driven context or a fiber. A no-op if the scheduler is not running.

Definition at line 849 of file scheduler.c.

                                            {
  PANIC_IF_NULL(sched);
  // Flip RUNNING to STOPPING once. If the scheduler is not running, or is
  // already stopping, there is nothing to do.
  srn_sched_state_t expected = SRN_SCHED_RUNNING;
 
  if (!atomic_compare_exchange_strong(&sched->state, &expected, SRN_SCHED_STOPPING)) {
    return;
  }
 
  // Running os threads see STOPPING at the top of their next turn. Parked os
  // threads are roused to observe it. The notify is under the lock, paired with
  // the park path, so no wakeup is lost.
  srn_mutex_lock(&sched->lock);
  srn_cond_notify_all(&sched->work);
  srn_mutex_unlock(&sched->lock);
  SCHED_LOG("stop requested");
}

Here is the call graph for this function:

Here is the caller graph for this function:

wait_for_park()

static bool wait_for_park ( srn_fiber_t * self, void * arg )

static

Add the calling fiber to the target's waiter list and stay parked, unless the target has already finished, in which case decline to park so the caller resumes at once.

The DONE check and the list insert run together under the global lock, which also guards the list against the DONE handler that drains it. So this either sees the target finished and declines, or joins the list before the drain and is woken by it, never lost in between.

Definition at line 945 of file scheduler.c.

                                                        {
  srn_fiber_t *target = arg;
  srn_scheduler_t *sched = srn_fiber_get_scheduler_m(self);
 
  srn_mutex_lock(&sched->lock);
 
  if (target->state == SRN_FIBER_DONE) {
    srn_mutex_unlock(&sched->lock);
    return false;
  }
 
  self->link = target->waiters;
  target->waiters = self;
 
  srn_mutex_unlock(&sched->lock);
  return true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

worker_main()

static void worker_main ( void * arg )

static

The entry an os thread starts in.

It sets up its worker's loop – on its own os thread, so the sanitizer captures the right stack bounds – then runs the worker routine until the pool is quiescent. arg is the worker.

Definition at line 778 of file scheduler.c.

                                   {
  srn_worker_t *worker = arg;
  srn_fiber_init_thread(&worker->loop);
  worker_run(worker);
}

Here is the call graph for this function:

Here is the caller graph for this function:

worker_run()

static void worker_run ( srn_worker_t * worker )

static

Run the worker routine over worker on the calling os thread.

Find a fiber, run it, handle how it gave up control, and park when nothing is runnable, until the pool is quiescent. Owns the current_worker thread-local for its duration. The hot path (find_work hitting the local deque, run, re-enqueue on yield) touches only this worker's own lock free deque. The global lock is reached only to park or for the global queue.

Definition at line 652 of file scheduler.c.

                                             {
  srn_scheduler_t *sched = worker->sched;
  current_worker = worker;
 
  for (;;) {
    // We check for termination here, between fibers, so an os thread stops at a
    // clean boundary even while its worker's deque still holds work. Whatever
    // is left unrun is reclaimed by `srn_sched_shutdown` through the registry.
    if (atomic_load(&sched->state) == SRN_SCHED_STOPPING) {
      break;
    }
 
    srn_fiber_t *fiber = find_work(worker);
    if (fiber == nullptr) {
      // Nothing runnable here or in any peer to steal from, so park this os
      // thread. Parking while every other os thread is already parked and
      // nothing is queued means the pool is quiescent. Nothing running can
      // produce more work, so the run ends. Move to STOPPING and wake every os
      // thread to exit. The `runnable` re-check in the loop closes the race
      // with a fiber enqueued between find_work and taking the lock, and
      // absorbs spurious wakeups.
      srn_mutex_lock(&sched->lock);
      atomic_fetch_add(&sched->idle, 1);
 
      if (atomic_load(&sched->idle) == sched->nworkers && atomic_load(&sched->runnable) == 0) {
        // All the os threads are idle. Time to stop
        atomic_store(&sched->state, SRN_SCHED_STOPPING);
        srn_cond_notify_all(&sched->work);
      }
 
      while (atomic_load(&sched->runnable) == 0 &&
             atomic_load(&sched->state) == SRN_SCHED_RUNNING) {
        // No runnable fiber around, and the scheduler still running. Going to
        // sleep.
        srn_cond_wait(&sched->work, &sched->lock);
      }
 
      // it has woken. This os thread is no longer parked. Snapshot whether
      // we're stopping, then drop the lock.
      atomic_fetch_sub(&sched->idle, 1);
      bool stop = atomic_load(&sched->state) == SRN_SCHED_STOPPING;
      srn_mutex_unlock(&sched->lock);
 
      if (stop) {
        break;
      }
 
      continue;
    }
 
    worker->current = fiber;
    // The worker routine is the single owner of the RUNNING transition, for a
    // fiber's first run and every resume after a yield.
    fiber->state = SRN_FIBER_RUNNING;
    srn_fiber_switch(&worker->loop, fiber);
 
    // `fiber` has switched back, and is not on any queue. How it gave up the
    // CPU is read from the fiber. A parked fiber left a commit on itself
    // (`park_commit` set), a finished one is `DONE`, and a yielded one is still
    // `RUNNING`.
    if (fiber->park_commit != nullptr) {
      // Parked. It is fully off the CPU now, with its context saved, so this is
      // the first moment it is safe to wake (by others). Stamp `SUSPENDED`
      // here, not in `srn_fiber_suspend` before the switch, so the label only
      // ever marks a fiber that is parked and safe to resume. A waker flipping
      // `SUSPENDED` to `READY` can therefore never catch a fiber still parking.
      fiber->state = SRN_FIBER_SUSPENDED;
 
      // `park_commit`/`park_arg` are one-shot, carrying the commit across the
      // switch. Clearing them now loses nothing, since the next suspend sets
      // them again and the fiber never reads them on resume.
      srn_fiber_park_fn commit = fiber->park_commit;
      void *park_arg = fiber->park_arg;
      fiber->park_commit = nullptr;
      fiber->park_arg = nullptr;
 
      // Run the commit now that the fiber is parked. It hands the fiber to its
      // waker (a waiter list, the reactor, and so on), which reschedules it
      // later. A true return means stay parked. A false return means the
      // condition already held, so wake it back up.
      if (!commit(fiber, park_arg)) {
        ready_fiber(sched, fiber);
      }
    } else if (fiber->state == SRN_FIBER_DONE) {
      // Detach the waiter list (fibers blocked in `srn_fiber_wait_for`) and
      // drop the fiber from the registry under the global lock, which also
      // guards the waiter list against `wait_for_park`. Then wake the waiters
      // and free the stack outside the lock. Each waiter reads this fiber's
      // result, which outlives the reap since only the stack is freed, not the
      // struct.
      srn_mutex_lock(&sched->lock);
      srn_fiber_t *waiters = fiber->waiters;
      fiber->waiters = nullptr;
      registry_remove(sched, fiber);
      srn_mutex_unlock(&sched->lock);
 
      while (waiters != nullptr) {
        srn_fiber_t *waiter = waiters;
        // Advance before the wake reuses `link`
        waiters = waiter->link;
        ready_fiber(sched, waiter);
      }
 
      // TODO(lxsameer): Instead of freeing the stack, return it to the ring
      // pool
      srn_fiber_stack_free(fiber->stack);
      srn_fiber_on_reap(fiber);
    } else {
      // Yielded. It is fully off the CPU now, with its context saved, so this
      // is the first moment it is safe to put back on a queue, where another
      // worker may take it at once. `srn_fiber_yield` does not enqueue before
      // switching, which would expose a context still being saved to a resuming
      // worker.
      fiber->state = SRN_FIBER_READY;
      push_ready(sched, fiber);
    }
 
    worker->current = nullptr;
  }
 
  current_worker = nullptr;
}

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

current_worker

_Thread_local srn_worker_t* current_worker = nullptr

static

The worker the calling os thread is running, or null when this os thread is not running the worker routine (srn_sched_run is not active on it).

This thread-local is the seam that resolves the resumer, the current fiber, and "are we in a fiber?" – all per os thread state that cannot live in the single shared scheduler.

Definition at line 240 of file scheduler.c.