Module Affect.Action

First-class synchronous actions.

Actions provides a compositional synchronization mechanism for blocking on system calls, asynchronous function completion and synchronization primitives.

Read the basics for a conceptual overview. See a few examples.

Actions

type +'a t

The type for actions whose invocations synchronize with values of type 'a or an exception.

val never : 'a t

never is an action whose invocations are never enabled, they block forever and never synchronize.

val always : 'a -> 'a t

always v is an action whose invocations are always enabled with v, they never block and immediately synchronize with v.

val choose : 'a t list -> 'a t

choose actions is a non-deterministic choice between actions. When choose actions is invoked all actions are invoked and only the first one to be enabled synchronizes. The remaining invocations never synchronize regardless of whether they are enabled or not – however depending on their nature, their invocation may have had an effect on the world for example if they had side-effecting guard.

Warning. For now if multiple actions are enabled at the same time choose is unfair and left-leaning. However this should not be relied on, it may change in the future.

val map : ('a -> 'b) -> 'a t -> 'b t

map f action is an action whose invocations invoke action and applies f to their synchronization result. f is guaranteed to be called only if the invocation synchronizes. If f raises the exception is thrown into the action invocation result.

Warning. If an action invocation ends up being part of a choose in which it does not get selected, the invocation may not, depending on the nature of the action, be immediately garbage collected. So if f is a closure, be wary about what it keeps a reference on – if anything.

Warning. There is no control on the scope in which f ends up being executed. In elaborate scenarios it may even happen outside the scheduler's domains. So it's not wise to perform OCaml effects in them, this includes making asynchronous function calls or synchronizing actions.

val guard : (unit -> 'a t) -> 'a t

guard f is an action that when invoked invokes f () and then invokes the returned action.

If f performs effects on the world these persist even if the resulting action does not eventually synchronize.

This is for example useful for timeouts which need to start counting at the time of invocation. See an example.

Warning. f should not raise exceptions. For now if it does it is thrown into the invoke before any form of synchronization occurs. This may be changed into a panic in the future. One of the problems is that other guards may already have been invoked but the action they returned will never get a chance to have gone through the synchronization protocol, which could break expected variants.

Derived combinators

val either : 'a t -> 'b t -> ('a, 'b) Stdlib.Either.t t

either a0 a1 is choose [map Either.left a0; map Either.right a1].

Invoking

val invoke : 'a t -> 'a

invoke action invokes action and blocks until the invocation synchronizes.

val invoke_poll : 'a t -> 'a option

invoke_poll action invokes action and returns None if the invocation cannot synchronize without blocking in which case the invocation is discarded and never synchronized.

Warning. This function may be removed in a future version of the library. Polling repeatedly is generally not recommended, it may lead to higher computational costs. Besides some invocations will always return None on polling.

  • deprecated May be removed in the future.

Unblockers

type unblocker

The type for action unblockers.

These values are provided by libraries defining actions whose invocations rely on external factors to synchronize. Values of this type should just be given to the invocation of Fun.Async.main so that the scheduler can unblock these action invocations. See for example Affect_unix.Unix.unblocker.

Private

module Private : sig ... end

Private definitions (unstable).

Basics

Action values are like function values, they represent an abstract operation that can be invoked to eventually return a value or an exception. When an action invocation returns we say that it synchronizes. The series of steps action invocations follow to return values is what makes them different from functions and what allows them to abstract synchronization protocols.

Actions are invoked with the Action.invoke function:

val Action.invoke : 'a Action.t -> 'a

This function creates an invocation for the action and blocks the caller until the action synchronizes.

Before being able to synchronize, an action invocation needs be enabled, which means ready to return a value. The condition for the action invocation to be enabled is determined by the nature of the action and can depend on:

Once enabled, an action invocation may eventually synchronize but it also may never do so for one of these reasons:

The fact that an action invocation may be prevented from synchronizing allows to abstract a non-deterministic choice between competing action invocations with the Action.choose combinator:

val Action.choose : 'a Action.t list -> 'a Action.t

This combinator takes a list of actions and returns an action which when invoked invokes each action in the list. It then selects the first one to become enabled, synchronizes it and returns with its value. All the other invocations are prevented from synchronizing (even though depending on their nature, their invocation may have had an effect on the world).

For example the following derived propose' action can be invoked on a port p:

let propose' p v =
  let ours = Action.map (fun () -> v) (Port.offer' p v ()) in
  let theirs = Port.take' p in
  Action.choose [ours; theirs]

The propose' action either offers or receives a value on a port p and synchronizes with either value. If a port p is only known to two functions that both invoke the action propose' with their own value it guarantees they will both synchronize with a consensus on one of the proposed values (before you try, note that a three-way synchronous consensus cannot be achieved with the set of combinators provided by affect, see this paper).

In the example above, the semantics of Action.choose which synchronizes a single enabled action invocation and discards the others ensures that the choice made in propose' between Port.take' and Port.offer' on the same port does not self-synchronize.

Finally, note that invoking the same action more than once can lead to different synchronization behaviours over time. The next section describes a few common behaviours.

Synchronization typology

The synchronization behaviour of invocations of a given action can evolve over time. Here are typical behaviours:

Examples

Some examples need to following preamble:

open Affect
open Affect_unix

Fancy wait readable

The fancy action below waits for a file descriptor to become readable but gives up after 1min30s or if a Sys.sigsur1 signal gets delivered to the process.

let fancy_wait_readable :
  Unix.file_descr -> [ `Readable | `Timeout | `User1_interrupt ] Action.t
=
fun fd ->
  let readable = Unix.wait_readable fd `Readable in
  let timeout = Mtime.wait_for' Mtime.Span.(1*min + 30*s) `Timeout in
  let signal = Unix.Signal.wait' Sys.sigusr1 `User1_interrupt in
  Action.choose [readable; timeout; signal]

Wait readable duration

The action below is like Affect_unix.Unix.wait_readable but it uses a monotonic counter to measure the time it took for the file descriptor to become readable.

let wait_readable_duration : Unix.file_descr -> Mtime.Span.t Action.t =
fun fd ->
  let setup () =
    let counter = Mtime.counter () in
    let readable = Unix.wait_readable fd counter in
    Action.map Mtime.count readable
  in
  Action.guard setup

Say no to speculation

The action below can be used if you have an asynchronous function that returns partial speculated results on cancellation that you don't find useful. It gives you the equivalent of a Fun.Async.get' action that ignores the results in case the asynchronous function was cancelled.

let no_speculation_get : 'a Fun.Async.t -> 'a option Action.t =
fun f ->
  let wait_cancelled = Fun.Async.wait_cancelled f None in
  let get = Action.map Option.some (Fun.Async.get' f) in
  Action.choose [wait_cancelled; get]