Module Eio.Fiber

A fiber is a light-weight thread.

Within a domain, only one fiber can be running at a time. A fiber runs until it performs an IO operation (directly or indirectly). At that point, it may be suspended and the next fiber on the run queue runs.

val both : (unit -> unit) -> (unit -> unit) -> unit

both f g runs f () and g () concurrently.

They run in a new cancellation sub-context, and if either raises an exception, the other is cancelled. both waits for both functions to finish even if one raises (it will then re-raise the original exception).

f runs immediately, without switching to any other thread. g is inserted at the head of the run-queue, so it runs next even if other threads are already enqueued. You can get other scheduling orders by adding calls to yield in various places. e.g. to append both fibers to the end of the run-queue, yield immediately before calling both.

If both fibers fail, Exn.combine is used to combine the exceptions.

val pair : (unit -> 'a) -> (unit -> 'b) -> 'a * 'b

pair f g is like both, but returns the two results.

val all : (unit -> unit) list -> unit

all fs is like both, but for any number of fibers. all [] returns immediately.

val first : ?combine:('a -> 'a -> 'a) -> (unit -> 'a) -> (unit -> 'a) -> 'a

first f g runs f () and g () concurrently.

They run in a new cancellation sub-context, and when one finishes the other is cancelled. If one raises, the other is cancelled and the exception is reported.

As with both, f runs immediately and g is scheduled next, ahead of any other queued work.

If both fibers fail, Exn.combine is used to combine the exceptions.

Warning: it is always possible that both operations will succeed. This is because there is a period of time after the first operation succeeds when it is waiting in the run-queue to resume during which the other operation may also succeed.

If both fibers succeed, combine a b is used to combine the results (where a is the result of the first fiber to return and b is the second result). The default is fun a _ -> a, which discards the later result.

val any : ?combine:('a -> 'a -> 'a) -> (unit -> 'a) list -> 'a

any fs is like first, but for any number of fibers.

any [] just waits forever (or until cancelled).

val n_any : (unit -> 'a) list -> 'a list

n_any fs is like any, expect that if multiple fibers return values then they are all returned, in the order in which the fibers finished.

val await_cancel : unit -> 'a

await_cancel () waits until cancelled.

val fork : sw:Switch.t -> (unit -> unit) -> unit

fork ~sw fn runs fn () in a new fiber, but does not wait for it to complete.

The new fiber is attached to sw (which can't finish until the fiber ends).

The new fiber inherits sw's cancellation context. If the fiber raises an exception, Switch.fail sw is called. If sw is already off then fn fails immediately, but the calling thread continues.

fn runs immediately, without switching to any other fiber first. The calling fiber is placed at the head of the run queue, ahead of any previous items.

val fork_promise : sw:Switch.t -> (unit -> 'a) -> 'a Promise.or_exn

fork_promise ~sw fn schedules fn () to run in a new fiber and returns a promise for its result.

This is just a convenience wrapper around fork. If fn raises an exception then the promise is resolved to the error, but sw is not failed.

val fork_seq : sw:Switch.t -> (('a -> unit) -> unit) -> 'a Seq.t

fork_seq ~sw fn creates (but does not start) a new fiber to run fn yield.

Requesting the next item from the returned sequence resumes the fiber until it calls yield x, using x value as the next item in the sequence. If fn returns without producing a value then the result is Seq.node.Nil (end-of-sequence).

The returned sequence can be consumed safely from another domain. fn itself always runs in the domain that called fork_seq.

Example:

Switch.run @@ fun sw ->
let seq = Fiber.fork_seq ~sw (fun yield ->
    for i = 1 to 3 do
      traceln "Yielding %d" i;
      yield i
    done
  ) in
Seq.iter (traceln "Got: %d") seq

If fn raises an exception then the consumer receives it. If the consumer cancels while awaiting a value, the producer is cancelled when it next calls yield. It is an error to request two items at once, or to request items out of sequence.

  • parameter sw

    When the switch finishes, the fiber is cancelled (if still running). Attempting to read from the sequence after this raises an exception.

val fork_daemon : sw:Switch.t -> (unit -> [ `Stop_daemon ]) -> unit

fork_daemon is like fork except that instead of waiting for the fiber to finish, the switch will cancel it once all non-daemon fibers are done.

The switch will still wait for the daemon fiber to finish cancelling.

The return type of [`Stop_daemon] instead of unit is just to catch mistakes, as daemons normally aren't expected to return.

val check : unit -> unit

check () checks that the fiber's context hasn't been cancelled. Many operations automatically check this before starting.

val is_cancelled : unit -> bool

is_cancelled () is true iff check would raise an exception.

val yield : unit -> unit

yield () asks the scheduler to switch to the next runnable task. The current task remains runnable, but goes to the back of the queue. Automatically calls check just before resuming.

module List : sig ... end

Concurrent list operations.

Fiber-local variables

Each fiber maintains a map of additional variables associated with it, which can be used to store fiber-related state or context. This map is propagated to any forked fibers.

While fiber-local variables can be useful, they can also make code much harder to reason about, as they effectively act as another form of global state. When possible, prefer passing arguments around explicitly.

Fiber-local variables are particularly useful for attaching extra information for debugging, such as a request ID that the log system can include in all logged messages.

type 'a key

'a key is a fiber-local variable of type 'a.

Since the key is required to get or set a variable, a library can keep its key private to control how the variable can be accessed.

val create_key : unit -> 'a key

create_key () creates a new fiber-local variable.

val get : 'a key -> 'a option

get key reads key from the map of fiber local variables, returning its value or None if it has not been bound.

val with_binding : 'a key -> 'a -> (unit -> 'b) -> 'b

with_binding key value fn runs fn with key bound to the provided value.

Whilst this binding only exists for the duration of this function on this fiber, it will be propagated to any forked fibers. If fn creates fibers using an external switch, the bound value may be continue to be used after this function returns.

val without_binding : 'a key -> (unit -> 'b) -> 'b

with_binding key value fn runs fn with any binding for key removed.