picos 0.6.0 · OCaml Package

A cancelable computation.

A computation basically holds the status, i.e.

running,
returned, or
canceled,

of some sort of computation and most importantly allows anyone to be notified when the status of the computation changes from running to completed.

A hopefully enlightening analogy is that a computation is a kind of single-shot atomic event.

Another hopefully helpful analogy is that a computation is basically like a cancelable promise and a basic non-cancelable promise can be implemented trivially on top of a computation.

To define a computation, one first creates it and then arranges for the computation to be completed by returning a value through it or by canceling it with an exception at some point in the future. There are no restrictions on what it means for a computation to be running. The cancelation status of a computation can be polled or checked explicitly. Observers can also attach triggers to a computation to get a signal when the computation is completed or await the computation.

Here is an example:

run begin fun () ->
  Flock.join_after @@ fun () ->

  let computation =
    Computation.create ()
  in

  let canceler = Flock.fork_as_promise @@ fun () ->
    Fiber.sleep ~seconds:1.0;

    Computation.cancel computation
      Exit (Printexc.get_callstack 0)
  in

  Flock.fork begin fun () ->
    let rec fib i =
      Computation.check
        computation;
      if i <= 1 then
        i
      else
        fib (i - 1) + fib (i - 2)
    in
    Computation.capture computation
      fib 10;

    Promise.terminate canceler
  end;

  Computation.await computation
end

A fiber is always associated with at least a single computation. However, it is possible for multiple fibers to share a single computation and it is also possible for a single fiber to perform multiple computations. Furthermore, the computation associated with a fiber can be changed by the fiber.

Computations are not hierarchical. In other words, computations do not directly implement structured concurrency. However, it is possible to propagate cancelation to implement structured concurrency on top of computations.

Operations on computations are either wait-free or lock-free and designed to avoid starvation and complete in amortized constant time. The properties of operations to complete a computation depend on the properties of actions attached to the triggers.

Interface for creating

type !'a t

Represents a cancelable computation. A computation is either running or has been completed either with a return value or with canceling exception with a backtrace.

ℹ️ Once a computation becomes completed it no longer changes state.

🏎️ A computation that has been completed is a small object that only holds onto the return value or the canceling exception with a backtrace.

⚠️ In the running state a computation may refer to any number of triggers and it is important to make sure that any triggers attached to a computation are detached when they are no longer needed unless the computation has been completed.

val create : ?mode:[ `FIFO | `LIFO ] -> unit -> 'a t

create () creates a new computation in the running state.

The optional mode specifies the order in which triggers attached to the computation will be signaled after the computation has been completed. `FIFO ordering may reduce latency of IO bound computations and is the default. `LIFO may improve thruput of CPU bound computations and be preferable on a work-stealing scheduler, for example.

ℹ️ Typically the creator of a computation object arranges for the computation to be completed by using the capture helper, for example. However, it is possible and safe to race multiple threads of execution to complete a computation.

val returned : 'a -> 'a t

returned value returns a constant computation that has returned the given value.

val finished : unit t

finished is a constant finished computation.

val try_return : 'a t -> 'a -> bool

try_return computation value attempts to complete the computation with the specified value and returns true on success. Otherwise returns false, which means that the computation had already been completed before.

val return : 'a t -> 'a -> unit

return computation value is equivalent to try_return computation value |> ignore.

val try_finish : unit t -> bool

try_finish computation is equivalent to try_return computation ().

val finish : unit t -> unit

finish computation is equivalent to try_finish computation |> ignore.

val try_capture : 'a t -> ('b -> 'a) -> 'b -> bool

try_capture computation fn x calls fn x and tries to complete the computation with the value returned or the exception raised by the call and returns true on success. Otherwise returns false, which means that the computation had already been completed before.

val capture : 'a t -> ('b -> 'a) -> 'b -> unit

capture computation fn x is equivalent to try_capture computation fn x |> ignore.

module Tx : sig ... end

Transactional interface for atomically completing multiple computations.

Interface for canceling

type packed =

| Packed : 'a t -> packed

An existential wrapper for computations.

val try_cancel : 'a t -> exn -> Printexc.raw_backtrace -> bool

try_cancel computation exn bt attempts to mark the computation as canceled with the specified exception and backtrace and returns true on success. Otherwise returns false, which means that the computation had already been completed before.

val cancel : 'a t -> exn -> Printexc.raw_backtrace -> unit

cancel computation exn bt is equivalent to try_cancel computation exn bt |> ignore.

Interface for timeouts

val cancel_after : 
  'a t ->
  seconds:float ->
  exn ->
  Printexc.raw_backtrace ->
  unit

cancel_after ~seconds computation exn bt arranges to cancel the computation after the specified time with the specified exception and backtrace. Completion of the computation before the specified time effectively cancels the timeout.

ℹ️ The behavior is that cancel_after first checks that seconds is not negative, and then

on OCaml 5, cancel_after will perform the Cancel_after effect, and
on OCaml 4, cancel_after will call the cancel_after operation of the current handler.

raises Invalid_argument
if seconds is negative or too large as determined by the scheduler.

Interface for polling

val is_running : 'a t -> bool

is_running computation determines whether the computation is in the running state meaning that it has not yet been completed.

val is_canceled : 'a t -> bool

is_canceled computation determines whether the computation is in the canceled state.

val canceled : 'a t -> (exn * Printexc.raw_backtrace) option

canceled computation returns the exception that the computation has been canceled with or returns None in case the computation has not been canceled.

val check : 'a t -> unit

check computation is equivalent to Option.iter (fun (exn, bt) -> Printexc.raise_with_backtrace exn bt) (canceled computation).

val peek : 'a t -> ('a, exn * Printexc.raw_backtrace) result option

peek computation returns the result of the computation or None in case the computation has not completed.

exception Running

Exception raised by peek_exn when it's used on a still running computation. This should never be surfaced to the user.

val peek_exn : 'a t -> 'a

peek_exn computation returns the result of the computation or raises an exception. It is important to catch the exception. If the computation was cancelled with exception exn then exn is re-raised with its original backtrace.

raises Running
if the computation has not completed.

Interface for awaiting

val try_attach : 'a t -> Trigger.t -> bool

try_attach computation trigger tries to attach the trigger to be signaled on completion of the computation and returns true on success. Otherwise returns false, which means that the computation has already been completed or the trigger has already been signaled.

⚠️ Always detach a trigger after it is no longer needed unless the computation is known to have been completed.

val detach : 'a t -> Trigger.t -> unit

detach computation trigger signals the trigger and detaches it from the computation.

🏎️ The try_attach and detach operations essentially implement a lock-free bag. While not formally wait-free, the implementation is designed to avoid starvation by making sure that any potentially expensive operations are performed cooperatively.

val await : 'a t -> 'a

await computation waits for the computation to complete and either returns the value of the completed computation or raises the exception the computation was canceled with.

ℹ️ If the computation has already completed, then await returns or raises immediately without performing any effects.

val wait : _ t -> unit

wait computation waits for the computation to complete.

Interface for propagating cancelation

val canceler : from:_ t -> into:_ t -> Trigger.t

canceler ~from ~into creates a trigger that propagates cancelation from one computation into another on signal. The returned trigger is not attached to any computation.

The caller usually attaches the returned trigger to the computation from which cancelation is to be propagated and the trigger should usually also be detached after it is no longer needed. See also attach_canceler.

The intended use case of canceler is as a low level building block of structured concurrency mechanisms. Picos does not require concurrent programming models to be hierarchical or structured.

⚠️ The returned trigger will be in the awaiting state, which means that it is an error to call Trigger.await or Trigger.on_signal on it.

val attach_canceler : from:_ t -> into:_ t -> Trigger.t

attach_canceler ~from ~into tries to attach a canceler to the computation from to propagate cancelation to the computation into and returns the canceler when successful. If the computation from has already been canceled, the exception that from was canceled with will be raised.

raises Invalid_argument
if the from computation has already returned.

Interface for schedulers

type Effect.t += private

| Cancel_after : {
1. seconds : float;
  (*
  Guaranteed to be non-negative.
  *)
2. exn : exn;
3. bt : Printexc.raw_backtrace;
4. computation : 'a t;
} -> unit Effect.t

Schedulers must handle the Cancel_after effect to implement the behavior of cancel_after.

The scheduler should typically attach a trigger to the computation passed with the effect and arrange the timeout to be canceled upon signal to avoid space leaks.

The scheduler should measure time using a monotonic clock.

In case the fiber permits propagation of cancelation and the computation associated with the fiber has been canceled the scheduler is free to discontinue the fiber before setting up the timeout.

If the fiber is continued normally, i.e. without raising an exception, the scheduler should guarantee that the cancelation will be delivered eventually.

The scheduler is free to choose which ready fiber to resume next.

val with_action : 
  ?mode:[ `FIFO | `LIFO ] ->
  'x ->
  'y ->
  (Trigger.t -> 'x -> 'y -> unit) ->
  'a t

with_action x y resume is equivalent to

let computation = create () in
let trigger =
  Trigger.from_action x y resume in
let _ : bool =
  try_attach computation trigger in
computation

⚠️ The same warnings as with Trigger.from_action apply.

alert handler This is an escape hatch for experts implementing schedulers or structured concurrency mechanisms. If you know what you are doing, use [@alert "-handler"].

Design rationale

The main feature of the computation concept is that anyone can efficiently attach triggers to a computation to get notified when the status of the computation changes from running to completed and can also efficiently detach such triggers in case getting a notification is no longer necessary. This allows the status change to be propagated omnidirectionally and is what makes computations able to support a variety of purposes such as the implementation of structured concurrency.

The computation concept can be seen as a kind of single-shot atomic event that is a generalization of both a cancelation Context or token and of a promise. Unlike a typical promise mechanism, a computation can be canceled. Unlike a typical cancelation mechanism, a computation can and should also be completed in case it is not canceled. This promotes proper scoping of computations and resource cleanup at completion, which is how the design evolved from a more traditional cancelation context design.

Every fiber is associated with a computation. Being able to return a value through the computation means that no separate promise is necessarily required to hold the result of a fiber. On the other hand, multiple fibers may share a single computation. This allows multiple fibers to be canceled efficiently through a single atomic update. In other words, the design allows various higher level patterns to be implemented efficiently.

Instead of directly implementing a hierarchy of computations, the design allows attaching triggers to computations and a special trigger constructor is provided for propagating cancelation. This helps to keep the implementation lean, i.e. not substantially heavier than a typical promise implementation.

Finally, just like with Trigger.Await, a key idea is that the handler of Computation.Cancel_after does not need to run arbitrary user defined code. The action of any trigger attached to a computation either comes from some scheduler calling Trigger.on_signal or from Computation.canceler.

package picos