Skuld.Fiber (skuld v0.2.3)

View Source

Cooperative fiber primitive for the FiberPool scheduler.

A Fiber wraps a computation that can be run incrementally, suspended when it yields, and resumed with a value. This is the fundamental building block for cooperative concurrency in Skuld.

Lifecycle

  1. Create with new/2 - fiber is :pending
  2. Run with run_until_suspend/1 - fiber becomes :running, then either:
    • Completes: returns {:completed, result, env}
    • Suspends: returns {:suspended, fiber} with fiber in :suspended state
    • Errors: returns {:error, reason, env}
  3. Resume with resume/2 - suspended fiber runs again until completion/suspension
  4. Cancel with cancel/1 - marks fiber as :cancelled

Usage

Fibers are typically managed by a FiberPool, not used directly. The FiberPool scheduler handles running fibers, tracking suspensions, and resuming when results are available.

Internal Details

When a fiber suspends, it stores:

  • suspended_k - the CPS continuation (val, env) -> {result, env}
  • env - the environment at suspension point

Resume calls suspended_k.(value, env) to continue execution.

Summary

Types

status()
t()

Functions

cancel(fiber)

Cancel a fiber.

new(comp, env)

Create a new fiber from a computation.

resume(fiber, value)

Resume a suspended fiber with a value.

run_until_suspend(fiber)

Run a fiber until it completes, suspends, or errors.

terminal?(fiber)

Check if a fiber is in a terminal state (completed, cancelled, or error).

Types

status()

@type status() :: :pending | :running | :suspended | :completed | :cancelled | :error

t()

@type t() :: %Skuld.Fiber{
  computation: Skuld.Comp.Types.computation() | nil,
  env: Skuld.Comp.Types.env() | nil,
  id: reference(),
  status: status(),
  suspended_k: Skuld.Comp.Types.k() | nil
}

Functions

cancel(fiber)

@spec cancel(t()) :: t()

Cancel a fiber.

Marks the fiber as :cancelled. A cancelled fiber cannot be run or resumed. The FiberPool scheduler uses this to clean up fibers when their scope exits.

Example

fiber = Fiber.cancel(fiber)
assert fiber.status == :cancelled

new(comp, env)

@spec new(Skuld.Comp.Types.computation(), Skuld.Comp.Types.env()) :: t()

Create a new fiber from a computation.

The fiber starts in :pending status with the computation stored, ready to be run with run_until_suspend/1.

Parameters

  • comp - The computation to run as a fiber
  • env - The environment to run in (typically inherited from parent)

Example

fiber = Fiber.new(my_comp, env)
assert fiber.status == :pending

resume(fiber, value)

@spec resume(t(), term()) ::
  {:completed, term(), Skuld.Comp.Types.env()}
  | {:suspended, t()}
  | {:internal_suspended, t(), Skuld.Comp.InternalSuspend.t()}
  | {:error, term(), Skuld.Comp.Types.env() | nil}

Resume a suspended fiber with a value.

Takes a :suspended fiber and resumes it by calling the suspended continuation with the provided value. Returns the same result types as run_until_suspend/1.

Parameters

  • fiber - A fiber in :suspended status
  • value - The value to resume with (typically a result from an effect)

Example

{:suspended, fiber} = Fiber.run_until_suspend(fiber)
# ... later, when we have a result ...
case Fiber.resume(fiber, result) do
  {:completed, final, _env} -> final
  {:suspended, fiber} -> # external suspended again
  {:internal_suspended, fiber, suspend} -> # internal suspended
  {:error, reason, _env} -> # errored
end

run_until_suspend(fiber)

@spec run_until_suspend(t()) ::
  {:completed, term(), Skuld.Comp.Types.env()}
  | {:suspended, t()}
  | {:internal_suspended, t(), Skuld.Comp.InternalSuspend.t()}
  | {:error, term(), Skuld.Comp.Types.env() | nil}

Run a fiber until it completes, suspends, or errors.

Takes a :pending fiber and runs its computation. Returns one of:

  • {:completed, result, env} - Fiber completed with result
  • {:suspended, fiber} - External suspension, updated fiber has :suspended_k and :env
  • {:internal_suspended, fiber, internal_suspend} - Internal suspension (batch/channel/await)
  • {:error, reason, env} - Fiber errored

The returned fiber (on suspension) can be resumed with resume/2.

Example

fiber = Fiber.new(my_comp, env)
case Fiber.run_until_suspend(fiber) do
  {:completed, result, _env} -> IO.puts("Done: #{inspect(result)}")
  {:suspended, fiber} -> IO.puts("External suspended, can resume later")
  {:internal_suspended, fiber, suspend} -> IO.puts("Internal suspended: #{inspect(suspend)}")
  {:error, reason, _env} -> IO.puts("Error: #{inspect(reason)}")
end

terminal?(fiber)

@spec terminal?(t()) :: boolean()

Check if a fiber is in a terminal state (completed, cancelled, or error).