package miou

Miou, a simple scheduler for OCaml 5.

Miou is a simple scheduler for OCaml 5 that uses effects. It allows you to launch tasks (functions) concurrently and/or in parallel, as well as offering your application high availability to system events.

Basics.

Effects.

Since OCaml 5, it has been possible to use effects. An effect allows you to suspend the execution of a function and fall into a handler which, depending on the effect, would perform a specific operation that would continue the suspended function with the result of the operation.

For example, a Hello: unit Effect.t effect can suspend using Effect.perform. A pre-installed handler will then retrieve this effect, perform the operation (say, display "Hello"), and unsuspend the function with the result of the operation (here, () : unit).

type _ Effect.t += Hello : unit Effect.t

let handler =
  let retc = Fun.id in
  let exnc = raise in
  let effc
    : type c. c Effect.t -> ((c, 'a) continuation -> 'a) option
    = function
    | Hello ->
      Some (fun k -> continue k (print_endline "Hello"))
    | _ -> None in
  { retc; exnc; effc }

let my_function () =
  Effect.perform Hello;
  print_endline "World"

let () = match_with my_function () handler

To go back to familiar OCaml elements, an effect is like an exception in that it breaks the execution flow. The "handler" is the with ... part of a try ... with ... in OCaml, and its installation corresponds to the try .... Finally, and this is the fundamental difference with exceptions, there is the continuation which allows us to return to the point where the effect was launched.

exception Hello

let my_function () =
  raise Hello;
  print_endline "World"

let () =
  try my_function ()
  with Hello k ->
    print_endline "Hello";
    k ()

Miou defines several effects that allow the user to interact with Miou's "task manager". Miou's effects manager is installed using run. So, if you want to use Miou, you should always start with run:

val my_program : unit -> unit

let () = Miou.run my_program ()

A task manager.

Miou is a task manager. In other words, it manages a list of to-do tasks (which you can add to with call_cc/call) and allows the user to manage these tasks. When a task is created, Miou gives the user a representation of the task: a promise t.

From this promise, the user can:

1. await for the result of the task
2. cancel the task

Here's an example where a list of tasks are initiated and awaited. Interaction (task creation and awaiting) with Miou takes place via effects. Miou manages the execution order of these tasks and attempts to finish them all in order to terminate your program.

let digest filename =
  Miou.call_cc @@ fun () ->
  (filename, Digest.file filename)

let my_program filenames =
  (* 1) we create a list of tasks *)
  let prms = List.map digest filenames in
  (* 2) Miou manages the execution of these tasks *)
  (* 3) we wait these tasks *)
  let results = List.map Miou.await_exn prms in
  (* 4) we print results *)
  List.iter (fun (filename, hash) ->
    Format.printf "%s: %s\n%!" filename (Digest.to_hex hash))
    results

let () = Miou.run @@ fun () ->
  my_program ["file01.ml"; "file02.ml"]

Miou suggests a little exercise to implement a task manager with effects. It explains the role of promises, creation and awaiting for a task (it does not, however, describe cancellation).

Domains.

Since OCaml 5, it has been possible to run functions in parallel. These functions run on a domain that has its own minor heap - so allocating small amounts of data doesn't require synchronization with other domains. Miou provides a pool of domains to which the user can assign tasks to run in parallel.

# let dom = Stdlib.Domain.spawn my_parallel_computation ;;
# Stdlib.Domain.join dom ;;

For more details on parallelism and garbage collection, we recommend reading the OCaml manual.

Miou prepares the allocation of a certain number of domains in advance. These will be waiting for tasks. The call function is used to assign a new task to one of these domains. The user can specify the number of domains Miou can allocate via the domains argument to the run function. We recommend using Domain.recommended_domain_count () - 1 domains.

Design.

After this brief introduction to the basics of Miou (i.e. the use of effects and domains and the definition of a task manager), Miou stands out from other schedulers through its design, which we'll describe here.

However, we need to define 3 terms that will be used throughout this description:

• a task is the smallest sequence of programmed instructions that can be managed by Miou. In concrete terms, it's an OCaml function.
• a domain (see Domain is a resource representing a processor available to execute a task
• a fiber is a task that will run on a domain in cooperation with other tasks

Preemption and cooperation

OCaml offers only a fundamentally cooperative model for executing these tasks. Indeed, there are no mechanisms in OCaml to force the suspension of a given task. However, a given task can suspend itself in order to cooperate with other tasks on a limited resource such as a particular domain.

Miou offers a way of creating tasks (see call_cc) that are more precisely called fibers. These fibers must cooperate with each other to share the domain on which they run. This means that a fiber should not have exclusive domain control when other fibers are waiting to be executed.

# Miou.run @@ fun () ->
  let rec pr str n =
    if n >= 0 then
    begin Miou.yield (); print_endline str; pr str (pred n) end in
  let prm0 = Miou.call_cc @@ fun () -> pr "Hello" 1 in
  let prm1 = Miou.call_cc @@ fun () -> pr "World" 1 in
  Miou.await_exn prm0;
  Miou.await_exn prm1 ;;
Hello
World
Hello
World
- : unit = ()

This task cooperation is achieved by yield, which interrupts the current task to leave the domain to another waiting task.

The problem with cooperation is that it does not take into account the irruption of external elements such as system events. Miou's objective is to be able to interrupt your application as soon as these events occur: in other words, to preempt the interruption of your tasks when these events occur.

Availability.

If it's important for us to interrupt your application as soon as these events are received, it's to increase the availability of your application to handle these events.

Let's take the example of a task making a long calculation. If an event such as the arrival of a TCP/IP connection were to occur at the same time as the calculation, we would consider the latter to be more important than the completion of the calculation. So we'd like to interrupt the calculation so that your application can handle this event as a priority.

let _, _ = Miou.run @@ fun () ->
  let rec server () =
    let socket = accept () in
    let _ = Miou.call (handler socket) in
    server () in
  let prm0 = Miou.call_cc server in
  let prm1 = Miou.call_cc my_long_computation in
  Miou.both prm0 prm1 ;;

(* [my_long_computation] should have multiple cooperative points to let
   the other task (our [server]) to accept incoming TCP/IP connexions. *)

In other words, your application is "more" available to handle events than to perform the calculations requested. This approach becomes interesting for services (such as an HTTP server) where the availability to handle such events is more important than prioritizing the calculation requested by a client.

Cooperation and effects.

And therein lies the crux of the problem: how do you preempt in a fundamentally cooperative system?

If we want to be in the best position to manage system events, we need to increase the points of cooperation that the fibers can emit. This is how Miou came up with a fundamental rule: an effect yields.

All effects (those defined by Miou as well as those defined by the user) reorder task execution. During this reordering, Miou can collect the system events that have just occurred. The objective is to do this as often as possible!

Performance and events.

At this point, we need to make clear to our future users a crucial choice we made for Miou: we prefer a scheduler that's available for system events, rather than one that performs well in calculations.

Indeed, on the cooperation points presented above, Miou will systematically ask whether any system events have occurred. However, if your ambition is to do nothing but calculations, the latter will be "polluted" by these unnecessary points of cooperation. So, by default, Miou is less efficient than other schedulers.

Miou and the system.

One of Miou's objectives is to be used in a unikernel. As far as the latter is concerned, possible interactions can be very limited. So we decided to separate the scheduler (Miou) from interactions with the underlying system (Miou_unix for a UNIX system).

There is a way to do this:

1. define suspensions that will only be released when certain events occur
2. Miou can collect these events and we can tell it which suspension should be "unblocked" according to the collection.

Suspension points.

Miou offers a way of creating what we call a suspension point. This can be created from a value syscall (with a unique identifier) that the user can keep. This suspension point can be "unblocked" if the select function given to Miou (via run) informs it of the "continuation" of this suspension.

Let's take read() as an example. This function, which interacts with the system, can "block" (for example, if you try to read from a socket). To avoid blocking (and leave it to the system), you can ask Miou to suspend just beforehand so that it can do other tasks, and inform Miou as soon as you know that this read() will not block.

let global = Hashtbl.create 0x100

let miou_read fd buf off len =
  let syscall = Miou.make (Fun.const ()) in
  Hashtbl.add global fd syscall;
  Miou.suspend syscall;
  Unix.read fd buf off len

Here, we use a global table to remind us that the file-descriptor we're using is associated with a syscall we've just created. The next objective is to define a select function that will observe whether the added file-descriptor is ready to be read.

The select() function.

Miou lets you inject a function to observe system events. This should appear each time tasks are rescheduled, as explained above with regard to application availability. This function should return any suspension points that can be unblocked.

let select ~poll:_ _cancelled_points =
  let fds = Hashtbl.to_seq_keys global |> List.of_seq in
  match Unix.select fds [] [] 0.1 with
  | fds ->
    let continuations = List.map (fun fd ->
      let syscall = Hashtbl.find global fd in
      Miou.continue_with syscall (fun () -> Hashtbl.remove global fd))
      fds in
    continuations

let run fn =
  let events _domain =
    { Miou.select; interrupt= Fun.const () } in
  Miou.run ~events fn

As you can see, the next step is to produce a run function that uses our select. This is what Miou_unix proposes for the example. However, there are a number of unresolved issues:

1. in particular, the cancellation of a task and its suspension point
2. or the ability to wait only for system events
3. or the ability to interrupt such an observation

We recommend reading the chapter on system events and its tutorial on sleepers.

User task management.

Users can manipulate their tasks via their promises (see t). A promise is an OCaml value representing the task. It can be used to await for or cancel a task. However, certain rules apply to its use.

Rule 1, await for all your tasks.

It is forbidden to forget your children. The creation of a task necessarily implies that the developer awaits or cancels the task afterwards:

# Miou.run @@ fun () -> Miou.call_cc (Fun.const ()) ;;
Exception: Miou.Still_has_children.

Rule 2, only await for direct children.

You can only await for your direct children. Transferring a promise to another task so that it can await for it is illegal:

# Miou.run @@ fun () ->
  let p = Miou.call_cc (Fun.const ()) in
  let q = Miou.call_cc (fun () -> Miou.await_exn p) in
  Miou.await_all [ p; q ] |> ignore
Exception: Miou.Not_a_child.

Task relationship clarifies what is involved in managing tasks and what they should transmit to each other. To answer this question, users will have to find their own mechanisms (mutex, condition, ipc, etc.) to share results between tasks that are not directly related.

Rule 3, a task can only be awaited or cancelled.

Miou only allows you to await for or cancel a task. It is also impossible to detach a task. For more information on this subject, we recommend reading the Daemon and orphan tasks. section and our following rule: background tasks.

Rule 4, a task only finishes after its children have finished.

By extension, as soon as a task is finished, all its children are finished too. The same applies to cancellation. If you cancel a task, you also cancel its children.

Rule 5, background tasks.

There is, however, a pattern in which we'd like to put a task aside: in other words, forget about it for a while. Miou offers a specific API for this pattern, described here.

Rule 6, a parallel task will never appear in the main domain.

There may be a contention problem if you involve dom0 in the tasks to be run in parallel. There may in fact be a situation where dom0 is awaiting for dom1, which is awaiting for dom0.

Miou does not allow dom0 to be assigned a parallel task. These assertions in the code below are true all the time.

# Miou.run @@ fun () ->
  let prm1 = Miou.call @@ fun () ->
    let prm2 = Miou.call @@ fun () ->
      Miou.Domain.self () in
    Miou.await_exn prm2, Miou.Domain.self () in
  let u, v = Miou.await_exn prm1 in
  assert (Miou.Domain.Uid.to_int u <> 0);
  assert (Miou.Domain.Uid.to_int v <> 0);
  assert (u <> v);;
- : unit = ()

However, you can involve dom0 in the calculations with call_cc.

let () = Miou.run @@ fun () ->
  let prm = Miou.call_cc server in
  Miou.parallel server (List.init (Miou.Domain.count ()) (Fun.const ()))
  |> List.iter (function Ok () -> () | Error exn -> raise exn);
  Miou.await_exn prm

The above rule also limits the use of call if you only have (or want) less than 3 domains. In fact, if you only have one domain, call cannot launch tasks in parallel. In the event that you only have 2 domains, it is possible to launch a task in parallel from dom0 but it is impossible to launch a task in parallel from this dom1.

In both cases and in such a situation, an exception is thrown: No_domain_available.

Rule 7, suspension points are local to the domain.

A suspension point is local to the domain. This means that only the domain in which it was created can unlock it. The Event value is created for each domain created by Miou. The unique identifier is also local to the domain.

# Miou.run @@ fun () ->
  let fn () = let s = Miou.make (Fun.const ()) in Miou.uid s in
  let uids = Miou.parallel fn (List.init 2 (Fun.const ())) in
  let[@warning "-8"] [ a; b ] = List.map Result.get_ok uids in
  assert (a = b)
;;
- : unit = ()

This code shows that two suspension points are created in two different domains and their unique ids are the same.

The advantage of making suspension points local to domains is that the domain is solely responsible for these points and there are no inter-domain transfer mechanisms for managing system events. For the example, Domain.DLS can be used for a table of current events in each domain.

let get, set =
  let make () = Hashtbl.create () in
  let dom = Stdlib.Domain.DLS.new_key make in
  let get () = Stdlib.Domain.DLS.get dom in
  let set value = Stdlib.Domain.DLS.set dom value in
  get, set

let miou_read fd buf off len =
  let syscall = Miou.make (Fun.const ()) in
  let tbl = get () in
  Hashtbl.add tbl fd syscall;
  set tbl;
  Miou.suspend syscall;
  Unix.read fd buf off len

Daemon and orphan tasks.

The prerogative of absolutely awaiting all of its direct children limits the user to considering certain anti-patterns. The best known is the background task: it consists of running a task that we would like to detach from the main task so that it can continue its life in autonomy. For OCaml/lwt aficionados, this corresponds to Lwt.async:

val detach : (unit -> unit t) -> unit

Not that we want to impose an authoritarian family approach between parent and children, but the fact remains that these orphaned tasks have resources that we need to manage and free-up (even in an abnormal situation). We consider detachment to be an anti-pattern, since it requires the developer to take particular care (compared to other promises) not to 'forget' resources that could lead to memory leaks.

Instead of letting the developer commit to using a function that might be problematic, Miou offers a completely different interface that consists of assisting the developer in a coherent (and consistent) approach to responding to a particular design that is not all that systematic.

So a promise can be associated with an orphans. The latter will then collect the results of the associated promise tasks and give you back the promises (via care) in a 'non-blocking' mode: applying await to them will give you the results directly.

In this way, by creating promises associated with this orphans value, we can at the same time "clean up" these background tasks, as this code shows:

let rec clean_up orphans =
  match Miou.care orphans with
  | None -> ()
  | Some prm -> Miou.await_exn prm; clean_up orphans

let rec server orphans =
  clean_up orphans;
  ignore (Miou.call ~orphans handler);
  server orphans

let () = Miou.run @@ fun () -> server (Miou.orphans ())

There is a step-by-step tutorial on how to create an echo server and how to create a daemon with Miou.

Launch a promise.

# Miou.run @@ fun () ->
  let r = Miou.Ownership.own ~finally:Fun.id () in
  let p = Miou.call ~give:[ r ] @@ fun () -> Miou.Ownership.disown r in
  Miou.await_exn p;;
Exception: Miou.Resource_leak.

let () = Miou.run @@ fun () ->
  let p = Miou.call @@ fun () ->
    let u = Miou.Domain.self () in
    let q = Miou.call @@ fun () -> Miou.Domain.self () in
    (u, Miou.await_exn q) in
  let u, v = Miou.await_exn p in
  assert (v <> u) ;;

let () = Miou.run @@ fun () ->
  let p = Miou.call @@ fun () -> Miou.Domain.self () in
  let q = Miou.call @@ fun () -> Miou.Domain.self () in
  let u = Miou.await_exn p in
  let v = Miou.await_exn q in
  assert (u = v);

let sort ~compare (arr, lo, hi) =
  if hi - lo >= 2 then begin
    let mi = (lo + hi) / 2 in
    sort ~compare (arr, lo, mi);
    sort ~compare (arr, mi, hi);
    merge ~compare arr lo mi hi
  end

let sort ~compare (arr, lo, hi) =
  if hi - lo >= 2 then begin
    let mi = (lo + hi) / 2 in
    ignore (Miou.parallel (sort ~compare)
      [ (arr, lo, mi); (arr, mi, hi) ]);
    merge ~compare arr lo mi hi
  end

val server : unit -> unit

let () = Miou.run @@ fun () ->
  let p = Miou.call_cc server in
  Miou.parallel server (List.init (Miou.Domain.count ()) (Fun.const ())
  |> List.iter (function Ok () -> () | Error exn -> raise exn);
  Miou.await_exn p

System events.

Miou does not monitor system events. We arbitrarily leave this event monitoring to the user (so that Miou only requires OCaml to run). The advantage is that you can inject an event monitor from a specific system (such as a unikernel) if you want. However, Miou_unix is available if you want to do input/output.

To facilitate the integration of an event monitor, Miou offers an API for creating "suspension points" (see suspend). In other words, points where execution will be blocked for as long as you wish. These points can be unblocked as soon as the monitor gives Miou a "continuation" to these points with continue_with.

The user must specify a select function (via the run function and the events type), which must correspond to system event monitoring (probably using Unix.select). From these events, the monitor can decide which suspension point (thanks to its uid) should be released. Miou will then call this function for each quanta consumed. This gives Miou a high degree of availability to consume and process system events.

Domains, suspension and monitoring.

Each domain has its own monitor so that the suspension and its continuation given by the monitor is always local to the domain (the domain managing the suspension is the only one allowed to execute the continuation). One events is allocated per domain - it is given on which domain the event value is assigned. In this way, the values (such as a table of active file-descriptors) required to monitor system events need not be thread-safe.

Sleep state.

Sometimes, Miou only has suspension points. In other words, only system events are required to execute tasks (typically waiting for a TCP/IP connection). We say we're in a sleep state. In this case, Miou informs the monitor select that it can wait indefinitely (with poll:true).

Cancellation.

It can happen that a task executed by one domain is cancelled by another domain (if the first was created by call). This cancellation of a task can also mean the cancellation of existing suspension points into the task. Miou must therefore be able to interrupt a domain (especially if the latter is in a sleep state).

Thus, the user must have a mechanism for stopping event monitoring, which must be given to Miou via the interrupt field (see events).

Finally, Miou informs the monitor of any points that have been cancelled, so that the associated events can no longer be monitored (this could involve cleaning up the table of active file-descritpors).

Tutorial.

To help you understand all these related elements, the distribution offers a short tutorial on how to implement functions that can block a given time (such as Unix.sleep): sleepers.

# Miou.run @@ fun () ->
  let p () =
    let s0 = Miou.make (Fun.const ()) in
    let s1 = Miou.make (Fun.const ()) in
    assert (Miou.uid s0 <> Miou.uid s1);
    Miou.uid s0 in
  Miou.parallel p (List.init 2 (Fun.const ()))
  |> function
  | [ Ok p0s; Ok p1s ] -> assert (p0s = p1s)
  | _ -> assert false 
;;
- : unit = ()

Await a promise.

# Miou_unix.run @@ fun () ->
  let p = Miou.call_cc @@ fun () ->
    let child_of_p = Miou.call_cc @@ fun () -> Miou_unix.sleep 10. in
    failwith "p";
    Miou.await_exn child_of_p in
  Miou.await p ;;
- (unit, exn) result = Error (Failure "p")
# (* [child_of_p] was cancelled and you don't await 10s. *)

# Miou.run @@ fun () ->
  ignore (Miou.call_cc (Fun.const ())) ;;
Exception: Miou.Still_has_children.

# exception Timeout ;;
# Miou_unix.run @@ fun () ->
  let p0 = Miou.call_cc (Fun.const ()) in
  let p1 = Miou.call_cc @@ fun () -> Miou_unix.sleep 2.; raise Timeout in
  Miou.await_first [ p0; p1 ] ;;
- : (unit, exn) result = Ok ()

# Miou.run @@ fun () ->
  Miou.await_one
    [ Miou.call_cc (Fun.const 1)
    ; Miou.call_cc (Fun.const 2) ] ;;
Exception: Miou.Still_has_children

let both prm0 prm1 =
  let a = Miou.await prm0 in
  let b = Miou.await prm1 in
  (a, b)

# Miou.run @@ fun () ->
  let p = Miou.call_cc @@ fun () -> print_endline "Hello" in
  print_endline "World";
  Miou.await_exn p ;;
World
Hello
- : unit = ()
# Miou.run @@ fun () ->
  let p = Miou.call_cc @@ fun () -> print_endline "Hello" in
  Miou.yield ();
  print_endline "World";
  Miou.await_exn p
Hello
World
- : unit = ()

Cancellation.

# Miou.run @@ fun () ->
  ignore (Miou.cancel (Miou.call (Fun.const ()))) ;;
- : unit = ()
# Miou.run @@ fun () ->
  ignore (Miou.call (Fun.const ())) ;;
Exception: Miou.Still_has_children

# Miou.run @@ fun () ->
  let p = Miou.call_cc (Fun.const ()) in
  Miou.await_exn p;
  Miou.cancel p;
  Miou.await_exn p ;;
Exception: Miou.Cancelled.

# Miou.run @@ fun () ->
  let p = Miou.call_cc @@ fun () -> print_endline "Resolved!" in
  Miou.yield ();
  Miou.cancel p;
  Miou.await_exn p ;;
Resolved!
Exception: Miou.Cancelled.

Composition.

It is possible to compose Miou with a library that also generates effects. The user can compose in 2 ways:

• simply apply his/her effect manager with its function in Miou Miou.call{,_cc} @@ fun () -> handler fn ()
• inform Miou of an effect handler that should comply with the quanta rule

Remember that Miou suspends a task as soon as it emits an effect. The second case can be interesting in order to always ensure the availability of the application regardless effect handlers. Here's a basic example of how to compose.

type _ Effect.t += Foo : unit Effect.t

let handler fn v =
  let open Effect.Deep in
  let retc = Fun.id and exnc = raise in
  let effc : type c. c Effect.t -> ((c, 'a) continuation -> 'b) option =
    function Foo -> Some (fun k -> continue k ())
           | _ -> None in
  match_with fn v { retc; exnc; effc; }

let () = Miou.run ~handler:{ Miou.handler } @@ fun () ->
  let prm = Miou.call @@ fun () -> Effect.perform Foo in
  Miou.await_exn prm

The user can also compose several effects managers:

# let compose { Miou.handler= a } { Miou.handler= b } =
  { Miou.handler= fun fn v -> (a (b fn)) v } ;;
val compose : Miou.handler -> Miou.handler -> Miou.handler = <fun>

NOTE: We want to reiterate that such a composition implies that the effect will not be executed immediately: the task will be suspended and the effect will be produced only as soon as the said task has its execution slot. One type of composition therefore becomes impossible if we want the production of an effect to produce an immediate action.

The example below shows that Effect.perform suspends and gives _p the opportunity to execute its task before the Unhandled exception is raised.

# try
    Miou.run @@ fun () ->
    let _p = Miou.call_cc @@ fun () -> print_endline "Hello" in
    Effect.perform Foo
  with Stdlib.Effect.Unhandled Foo -> print_endline "World"
;;
Hello
World
- : unit = ()