scheduler.c File Reference

#include <inttypes.h>
#include <stdatomic.h>
#include <stdio.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/rt/reactor.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_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_DRAINING , 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.
void	srn_fiber_schedule (srn_fiber_t *fiber)
	Schedule a NEW fiber, 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, size_t 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_sched_drain (srn_scheduler_t *sched)
	Ask a running scheduler to wind down gracefully.
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, waking it when the event it awaited occurs.
srn_fiber_t *	srn_fiber_current (void)
	The fiber currently running on this os thread, or null when the calling thread is not a worker or the worker is running its own loop rather than a fiber.
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.
void	srn_fiber_autoname (srn_engine_t *engine, char *dst, size_t size)
	Write the autogenerated debug name for a new fiber into dst.
srn_worker_id_t	srn_sched_current_worker_id ()
	Return the id of the worker that the calling os thread is running, or SIZE_MAX when the calling thread is not a worker.
bool	srn_sched_accepting_submissions (srn_scheduler_t *sched)
	Whether the scheduler still accepts new IO submissions.
void	srn_sched_wake_worker (srn_scheduler_t *sched, size_t channel)
	Rouse parked workers so the owner of channel consumes its completions.

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 31 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 40 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 219 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 servicing the queues. DRAINING is the graceful wind down set by srn_sched_drain, workers keep running every runnable fiber and let in-flight IO finish, but new IO submissions are fenced (see srn_sched_accepting_submissions) so fibers unwind instead of parking on fresh ops, and the pool converges to quiescence. STOPPING tells the workers to wind down at once, set at natural quiescence or abruptly by srn_sched_stop. IDLE is the resting state before a run starts. Workers read it without the lock, so it is atomic.

The order matters, a state at or past DRAINING no longer accepts new IO, which srn_sched_accepting_submissions relies on.

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 118 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 servicing the queues. DRAINING is the graceful wind down set by srn_sched_drain, workers keep running every runnable fiber and let in-flight IO finish, but new IO submissions are fenced (see srn_sched_accepting_submissions) so fibers unwind instead of parking on fresh ops, and the pool converges to quiescence. STOPPING tells the workers to wind down at once, set at natural quiescence or abruptly by srn_sched_stop. IDLE is the resting state before a run starts. Workers read it without the lock, so it is atomic.

The order matters, a state at or past DRAINING no longer accepts new IO, which srn_sched_accepting_submissions relies on.

Enumerator
SRN_SCHED_IDLE
SRN_SCHED_RUNNING
SRN_SCHED_DRAINING
SRN_SCHED_STOPPING

Definition at line 132 of file scheduler.c.

                               {
  SRN_SCHED_IDLE,
  SRN_SCHED_RUNNING,
  SRN_SCHED_DRAINING,
  SRN_SCHED_STOPPING,
} srn_sched_state_t;

Function Documentation

announce_work()

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 427 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()

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 669 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 (size_t 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.
      size_t index         = (w->id + i) % sched->nworkers;
      srn_worker_t *victim = &sched->workers[index];
      fiber                = local_steal(victim);
 
      if (fiber != nullptr) {
        FIBER_TRACEPOINT(sched_steal, w->id, victim->id, (void *)fiber);
        SCHED_TRACE("worker %zu stole fiber %p from worker %zu", 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()

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 559 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()

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 585 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()

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 480 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 %zu popped the last fiber %p", w->id, (void *)fiber);
 
      } else {
        SCHED_TRACE("worker %zu 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()

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 451 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 %zu local-push fiber %p", w->id, (void *)fiber);
  return true;
}

Here is the caller graph for this function:

local_steal()

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 522 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()

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 606 of file scheduler.c.

                                                                   {
  srn_worker_t *w = current_worker;
 
  // The publish (local_push) happens before announce_work bumps `runnable`,
  // so a thief that takes the fiber first transiently drives `runnable` to
  // SIZE_MAX. Every check of the counter is `== 0`, which the wrap cannot
  // satisfy, so this is benign; do not add `> 0` style or signed comparisons
  // on `runnable` without fixing the ordering here.
 
  // 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;
    }
 
    FIBER_TRACEPOINT(sched_overflow, w->id, (void *)fiber);
    SCHED_TRACE("worker %zu 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()

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 658 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()

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 296 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()

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 309 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_autoname()

void srn_fiber_autoname	(	srn_engine_t *	engine,
		char *	dst,
		size_t	size )

Write the autogenerated debug name for a new fiber into dst.

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. The worker part is provenance, not placement, a stolen fiber runs anywhere. srn_fiber_make uses this when given no name.

Definition at line 1141 of file scheduler.c.

                                                                      {
  PANIC_IF_NULL(engine);
  PANIC_IF_NULL(dst);
 
  srn_worker_t *w = current_worker;
  if (w != nullptr) {
    // The tag names the creating worker and its spawn count. Provenance, not
    // placement, the fiber may be stolen and run anywhere.
    (void)snprintf(dst, size, "f#%zu:%" PRIu64, (size_t)w->id, ++w->spawned);
    return;
  }
  // Off the pool there is no worker id. The tag is `m` and the engine wide
  // object id keeps names unique across every creating thread.
  (void)snprintf(dst, size, "f#m:%" PRIu64, srn_allocate_object_id(engine));
}

Here is the call graph for this function:

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, or null when the calling thread is not a worker or the worker is running its own loop rather than a fiber.

Definition at line 1097 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, waking it when the event it awaited occurs.

Callable from a fiber, a worker, or the reactor. The reactor is the only legitimate waker outside the worker pool.

Definition at line 1083 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 the
  // reactor -- the one legitimate waker outside the worker pool, since
  // quiescence accounts for its in-flight ops -- can wake it too. An
  // unrelated os thread must not, its pending wake is invisible to
  // quiescence, so the run can end before the wake arrives.
  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_schedule()

void srn_fiber_schedule

(

srn_fiber_t *

fiber

)

Schedule a NEW fiber, making it eligible to run.

A fiber is scheduled exactly once, scheduling one that is not NEW panics, and waking a suspended fiber is srn_fiber_ready's job instead.

Definition at line 638 of file scheduler.c.

                                            {
  PANIC_IF_NULL(fiber);
 
  // The NEW to READY flip admits exactly one scheduler of this fiber, the
  // same guard ready_fiber uses for SUSPENDED, so a double schedule panics
  // at the losing call site instead of double enqueuing.
  srn_fiber_state_t expected = SRN_FIBER_NEW;
  PANIC_IF(
    !atomic_compare_exchange_strong(&fiber->state, &expected, SRN_FIBER_READY),
    "srn_fiber_schedule needs a NEW fiber. A fiber is scheduled exactly once, "
    "and a suspended one is woken with srn_fiber_ready"
  );
  push_ready(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 1063 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.

If the target is already done, returns immediately.

Definition at line 1130 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 1101 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 1039 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;
  PANIC_IF_NULL(self);
  srn_fiber_switch(self, &worker->loop);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_sched_accepting_submissions()

bool srn_sched_accepting_submissions

(

srn_scheduler_t *

sched

)

Whether the scheduler still accepts new IO submissions.

True only while RUNNING; false once srn_sched_drain or srn_sched_stop has begun winding the pool down. The IO bridge reads this to fence submissions during a wind-down, so a fiber's IO call comes back cancelled instead of parking on an op the wind-down would have to wait out.

Definition at line 1163 of file scheduler.c.

                                                             {
  PANIC_IF_NULL(sched);
  // Only a RUNNING scheduler takes new IO. Once DRAINING or STOPPING, the IO
  // bridge fences submissions so fibers unwind instead of parking on ops the
  // wind-down would have to wait out.
  return atomic_load(&sched->state) == SRN_SCHED_RUNNING;
}

Here is the caller graph for this function:

srn_sched_current_worker_id()

srn_worker_id_t srn_sched_current_worker_id

(

void

)

Return the id of the worker that the calling os thread is running, or SIZE_MAX when the calling thread is not a worker.

Definition at line 1157 of file scheduler.c.

                                              {
  // We use the SIZE_MAX as an idicator that there is no current
  // worker for the running os thread. (size_t)-1 == SIZE_MAX
  return current_worker == nullptr ? (srn_worker_id_t)-1 : current_worker->id;
}

Here is the caller graph for this function:

srn_sched_drain()

void srn_sched_drain

(

srn_scheduler_t *

sched

)

Ask a running scheduler to wind down gracefully.

Unlike srn_sched_stop, the workers keep running every runnable fiber and let in-flight IO complete; only new IO submissions are fenced, so a fiber's next IO call returns -ECANCELED and the fiber unwinds rather than parking on a fresh op. The pool then reaches the same quiescence as a natural finish and srn_sched_run returns. An in-flight op that never completes (an idle recv, a long sleep) stalls the wind-down until it finishes; bounding that needs op cancellation and is not provided yet. Does not wait. Safe to call from any os thread or a fiber. A no-op if the scheduler is not running.

Definition at line 1002 of file scheduler.c.

                                             {
  PANIC_IF_NULL(sched);
  // Begin a graceful winddown, only a RUNNING scheduler can enter DRAINING.
  // Already draining or stopping, or not running at all, leaves the state as
  // is.
  srn_sched_state_t expected = SRN_SCHED_RUNNING;
 
  if (!atomic_compare_exchange_strong(&sched->state, &expected, SRN_SCHED_DRAINING)) {
    return;
  }
 
  // From here `srn_sched_accepting_submissions` returns false, so the next IO a
  // fiber attempts is fenced into a cancelled completion and the fiber unwinds
  // rather than parking on a fresh op. Workers do NOT break on DRAINING, so
  // every runnable fiber still runs and every in-flight op still completes; the
  // pool converges to the same quiescence as a natural finish, which then moves
  // the state to STOPPING. A never-completing in-flight op (an idle recv, a
  // long sleep) stalls this until it finishes -- bounding that needs op CANCEL
  // (E1) and is out of scope here.
  //
  // The notify wakes any os thread already parked so it re-checks state. A
  // worker parked on outstanding IO simply re-parks (DRAINING keeps it
  // parking), which is harmless.
  srn_mutex_lock(&sched->lock);
  srn_cond_notify_all(&sched->work);
  srn_mutex_unlock(&sched->lock);
  SCHED_LOG("drain requested");
}

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 630 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:

srn_sched_init()

srn_scheduler_t * srn_sched_init ( srn_engine_t * engine )

nodiscard

Definition at line 263 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 324 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,
		size_t	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 875 of file scheduler.c.

                                                            {
  PANIC_IF_NULL(sched);
  PANIC_IF(sched->destroyed, "srn_sched_run called on a scheduler that was already shut down");
 
  // Claiming run_active up front turns an overlapping run into a clean panic
  // instead of two runs clobbering the worker arrays under each other.
  PANIC_IF(
    atomic_exchange(&sched->run_active, true),
    "srn_sched_run called while another run is active on this scheduler"
  );
 
  // A finished run leaves its worker arrays behind for shutdown to join and
  // free. A second run would replace them while stragglers from the first may
  // still be winding down, reviving those stragglers against the new run's
  // state, and the reactor cannot be activated twice either. A scheduler
  // therefore runs once; re-run support requires reactor reactivation.
  PANIC_IF(
    sched->workers != nullptr,
    "srn_sched_run called on a scheduler that has already run; re-run is not "
    "supported (the reactor cannot be reactivated)"
  );
 
  // A caller that does not pick a count gets the configured one, and every
  // request is clamped to the configured ceiling. SRN_MAX_WORKERS stays the
  // absolute ceiling above whatever the configuration asks for.
  const srn_configuration_t *config = &sched->engine->config;
  nworkers                          = nworkers == 0 ? config->fiber.workers : nworkers;
 
  if (nworkers == 0) {
    // A configured count of zero delegates to the machine, one worker per
    // CPU the process may run on.
    nworkers = srn_thread_cpu_count();
  }
 
  if (nworkers > config->fiber.max_workers) {
    nworkers = config->fiber.max_workers;
  }
 
  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_malloc(sched->engine->mm, nworkers * sizeof(srn_worker_t));
  PANIC_IF_NULL(sched->workers);
 
  sched->os_threads = srn_mm_malloc(sched->engine->mm, nworkers * sizeof(srn_thread_t));
  PANIC_IF_NULL(sched->os_threads);
 
  for (size_t i = 0; i < nworkers; i++) {
    srn_worker_t *w = &sched->workers[i];
    w->sched        = sched;
    w->id           = i;
    w->current      = nullptr;
    w->spawned      = 0;
    // 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` was
  // claimed at the top of this call; shutdown reads it to see a run in flight.
  sched->idle     = 0;
  sched->nworkers = nworkers;
  atomic_store(&sched->state, SRN_SCHED_RUNNING);
 
  // Bring the reactor up with one channel per worker before any worker starts,
  // so a fiber's first IO has a channel to submit on. The notify seam wakes the
  // worker that owns the channel a completion lands on.
  srn_reactor_activate(sched->engine->reactor, nworkers, srn_sched_wake_worker);
 
  FIBER_TRACEPOINT(sched_run, nworkers);
 
  // 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 (size_t 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");
    }
    FIBER_TRACEPOINT(sched_thread_spawn, i);
  }
 
  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 333 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 (size_t 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' (never scheduled, or suspended with no waker?)",
      fiber->name
    );
    // TODO(lxsameer): Free up the ring here as well
    FIBER_TRACEPOINT(fiber_stack_free, (void *)fiber, fiber->name);
    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 or a fiber, but NOT from a signal handler. It takes the scheduler mutex and signals the condition variable, neither of which is async signal safe.

Definition at line 978 of file scheduler.c.

                                            {
  PANIC_IF_NULL(sched);
  // Flip RUNNING or DRAINING to STOPPING once. A drain stalled on an op that
  // never completes must remain abortable, so stop escalates a drain rather
  // than deferring to it. 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)) {
    expected = SRN_SCHED_DRAINING;
    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:

srn_sched_wake_worker()

void srn_sched_wake_worker	(	srn_scheduler_t *	sched,
		size_t	channel )

Rouse parked workers so the owner of channel consumes its completions.

The reactor's notify hook. The wake is currently a broadcast, channel names the worker that must look, but every parked worker is notified and the ones with nothing to do re-park.

Definition at line 1171 of file scheduler.c.

                                                                   {
  // TODO(lxsameer): Wake up the worker in charge of the given channel. instead
  // of waking all.
  UNUSED(channel);
  srn_mutex_lock(&sched->lock);
  srn_cond_notify_all(&sched->work);
  srn_mutex_unlock(&sched->lock);
}

Here is the call graph for this function:

Here is the caller graph for this function:

wait_for_park()

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 1112 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()

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 866 of file scheduler.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

worker_run()

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) touches only this worker's own lock free deque. The global lock is reached to park, for the global queue, and on every yield, a yielded fiber goes to the global tail so its peers get a turn, which is the point of yielding.

Definition at line 707 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;
    }
 
    // Drain this worker's reactor completions before looking for fibers. Runs
    // on the worker, so srn_fiber_ready pushes the woken fibers onto its own
    // deque, keeping them local.
    srn_reactor_consume(sched->engine->reactor, worker->id);
 
    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 &&
        (int)srn_reactor_idle(sched->engine->reactor)
      ) {
 
        // All the os threads are idle. Time to stop
        atomic_store(&sched->state, SRN_SCHED_STOPPING);
        FIBER_TRACEPOINT(sched_quiesce, worker->id);
        srn_cond_notify_all(&sched->work);
      }
 
      FIBER_TRACEPOINT(sched_worker_park, worker->id);
      while (atomic_load(&sched->runnable) == 0 &&
             atomic_load(&sched->state) != SRN_SCHED_STOPPING &&
             !srn_reactor_channel_has_completions(sched->engine->reactor, worker->id)) {
        // No runnable fiber around and we are not stopping. A draining pool
        // parks here too, waiting for its in-flight ops to complete and ready
        // their fibers, so they can unwind. Only STOPPING ends the park.
        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);
      FIBER_TRACEPOINT(sched_worker_unpark, worker->id);
      bool stop = atomic_load(&sched->state) == SRN_SCHED_STOPPING;
      srn_mutex_unlock(&sched->lock);
 
      if (stop) {
        break;
      }
 
      continue;
    }
 
    worker->current = fiber;
    // Read here because the switch may reap the fiber, after which only the address value survives,
    // not the name.
    FIBER_TRACEPOINT(sched_fiber_running, worker->id, (void *)fiber, fiber->name);
 
    // 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);
 
    const char *slice_reason;
 
    // `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) {
      slice_reason = "suspend";
      // 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;
      FIBER_TRACEPOINT(sched_fiber_suspended, worker->id, (void *)fiber, fiber->name);
      // `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) {
      slice_reason = "done";
      FIBER_TRACEPOINT(sched_fiber_done, worker->id, (void *)fiber, fiber->name);
      // 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
      FIBER_TRACEPOINT(fiber_stack_free, (void *)fiber, fiber->name);
      srn_fiber_stack_free(fiber->stack);
      srn_fiber_on_reap(fiber);
    } else {
      slice_reason = "yield";
      // 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. The fiber goes to the global queue tail, not the local deque:
      // the worker pops its deque LIFO, so a local push would run the same
      // fiber again immediately and starve its peers, making yield a no-op.
      fiber->state = SRN_FIBER_READY;
      FIBER_TRACEPOINT(sched_fiber_ready, worker->id, (void *)fiber, fiber->name);
      global_enqueue(sched, fiber);
    }
    // Slice end, once the reason is known. Pairs with worke_run_end by the
    // fiber address. Passes only the pointer value, never a dereference, so it
    // is safe even after the done branch reaped the fiber's stack.
    FIBER_TRACEPOINT(sched_fiber_finished, worker->id, (void *)fiber, slice_reason);
 
    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 257 of file scheduler.c.