Picos — Interoperable effects based concurrency

https://github.com/ocaml-multicore/picos

What is Picos?

It is an interface:

  Libraries & Applications          '' Schedulers ''

                                      +-- Async
                                      |
     Sync and Comm  --+               +-- Domainslib
                      |               |
           Async IO --+               +-- Eio
                      |               |
          Conn pool --+               +-- Fuseau
                      |               |
      Concurrent DS --+-->  Picos  <--+-- Lwt
                      |               |
 Structuring models --+               +-- Miou
                      |               |
     Parallel algos --+               +-- Moonpool
                      |               |
                STM --+               +-- Oslo
                      |               |
                      .               +-- Riot
                      .               .
                      .               .

    "Implemented in Picos"         "Picos compatible"

Schedulers in scare quotes?

What is a scheduler?
- Is it async IO integration?
- Is it a concurrent programming model?
- Is it a way of life?
- Is it a loop that uses a queue to manage fibers?
In OCaml, traditionally, a scheduler is a way of life. It consumes your application and you have to yield to it periodically. Schedulers are written by gods. 😅
In Picos, a scheduler can be just a loop that manages fibers with a queue of some kind. You can realistically write one in an hour or two.
- All the elements of concurrent programming, including schedulers, are highly modular pieces of code
- Can be mixed and matched
  - Everything "Implemented in Picos" can run on any "Picos compatible" scheduler
  - Every "Picos compatible" scheduler can run anything "Implemented in Picos"
- Instead of every '' scheduler '' implementing its own copy of everything, all the bits and pieces can be shared
- Much like with ordinary synchronous code
With Picos, schedulers might become "curated concurrent programming models"
- Want actors? Use Riot
- Want capabilities? Use Eio
- Want work-stealing? Use Moonpool or Domainslib
- Want more traditional concurrency? Use Fuseau or Miou
- ...
- The internals and libraries could be largely shared
- A single application could use libraries written internally using different programming models
  - Would not recommend mixing multiple models within a single library
  - Would recommend preferring "conservative" APIs for libraries
    - But you can often write a shallow layer on top to change model

Picos is all about cancelation

Mutex.protect mutex begin fun () ->
  while true do
    Condition.wait condition mutex
  done
end

Picos API

Concepts:

Trigger
Computation
Fiber

Effects:

Await
Cancel_after
Current
Yield
Spawn

Signature extract

module Trigger : sig
  type t

  val create : unit -> t
  val await : t -> Exn_bt.t option
  val signal : t -> unit

  val on_signal : t -> 'x -> 'y -> (t -> 'x -> 'y -> unit) -> bool
  type _ Effect.t += private Await : t -> exn_bt option Effect.t
end

module Computation : sig
  type !'a t

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

  val return : 'a t -> 'a -> unit
  val cancel : 'a t -> Exn_bt.t -> unit

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

  val await : 'a t -> 'a

  val cancel_after : 'a t -> seconds:float -> Exn_bt.t -> unit
  type _ Effect.t +=
    private
    | Cancel_after : {
        seconds : float;
        exn_bt : exn_bt;
        computation : 'a t;
      }
        -> unit Effect.t
end

module Fiber : sig
  val spawn : forbid:bool -> 'a Computation.t -> (unit -> unit) list -> unit
  type _ Effect.t +=
    private
    | Spawn : {
        forbid : bool;
        computation : 'a computation;
        mains : (unit -> unit) list;
      }
        -> unit Effect.t

  val create : forbid:bool -> 'a Computation.t -> t

  type t

  val current : unit -> t
  type _ Effect.t += private Current : t Effect.t

  val forbid : t -> (unit -> 'a) -> 'a
  val get_computation : t -> Computation.packed
  val set_computation : t -> Computation.packed -> unit

  val yield : unit -> unit
  type _ Effect.t += private Yield : unit Effect.t
end

Samples

https://ocaml-multicore.github.io/picos/doc/picos/index.html#libraries

Schedulers:

Picos_fifos
Picos_lwt
Picos_treaded

Scheduler agnostic libs:

Picos_structured
Picos_sync
Picos_stdio
Picos_select

Aux:

Picos_mpsc_queue
Picos_htbl
...

MirageOS

Original:

module Main
    (Time : Mirage_time.S)
    (PClock : Mirage_clock.PCLOCK)
    (MClock : Mirage_clock.MCLOCK) =
struct
  let str_of_time (posix_time, timezone) =
    Format.asprintf "%a" (Ptime.pp_human ?tz_offset_s:timezone ()) posix_time

  let start _time pclock mclock =
    let rec speak pclock mclock () =
      let current_time = PClock.now_d_ps pclock |> Ptime.v in
      let tz = PClock.current_tz_offset_s pclock in
      let str =
        Printf.sprintf
          "%Lu nanoseconds have elapsed. \n\
          \ At the stroke, the time will be %s \x07 *BEEP*"
          (MClock.elapsed_ns mclock)
        @@ str_of_time (current_time, tz)
      in
      Log.info (fun f -> f "%s" str);
      Time.sleep_ns 1_000_000_000L >>= fun () -> speak pclock mclock ()
    in
    speak pclock mclock ()
end

Direct-style with Picos_lwt:

module Main
    (Time : Mirage_time.S)
    (PClock : Mirage_clock.PCLOCK)
    (MClock : Mirage_clock.MCLOCK) =
struct
  open Picos_structured

  module Lwt_ds = Picos_lwt.Make (struct
    let sleep seconds =
      Time.sleep_ns (Int64.of_float (seconds *. 1_000_000_000.0))
  end)

  let str_of_time (posix_time, timezone) =
    Format.asprintf "%a" (Ptime.pp_human ?tz_offset_s:timezone ()) posix_time

  let start _time pclock mclock =
    Lwt_ds.run ~forbid:false @@ fun () ->
    while true do
      let current_time = PClock.now_d_ps pclock |> Ptime.v in
      let tz = PClock.current_tz_offset_s pclock in
      let str =
        Printf.sprintf
          "%Lu nanoseconds have elapsed. \n\
          \ At the stroke, the time will be %s \x07 *BEEP*"
          (MClock.elapsed_ns mclock)
        @@ str_of_time (current_time, tz)
      in
      Log.info (fun f -> f "%s" str);
      Control.sleep ~seconds:1.0
      (* OR *)
      Lwt_ds.await (fun () -> Time.sleep_ns 1_000_000_000L)
    done
end

Defunctorized:

module Main
    (Time : Mirage_time.S)
    (PClock : Mirage_clock.PCLOCK)
    (MClock : Mirage_clock.MCLOCK) =
struct
  open Picos_structured

  let sleep seconds =
    Time.sleep_ns (Int64.of_float (seconds *. 1_000_000_000.0))

  let str_of_time (posix_time, timezone) =
    Format.asprintf "%a" (Ptime.pp_human ?tz_offset_s:timezone ()) posix_time

  let start _time pclock mclock =
    Picos_lwt.run ~sleep @@ fun () ->
    while true do
      let current_time = PClock.now_d_ps pclock |> Ptime.v in
      let tz = PClock.current_tz_offset_s pclock in
      let str =
        Printf.sprintf
          "%Lu nanoseconds have elapsed. \n\
          \ At the stroke, the time will be %s \x07 *BEEP*"
          (MClock.elapsed_ns mclock)
        @@ str_of_time (current_time, tz)
      in
      Log.info (fun f -> f "%s" str);
      Control.sleep ~seconds:1.0
      (* OR *)
      Picos_lwt.await (fun () -> Time.sleep_ns 1_000_000_000L)
    done
end

https://github.com/ocaml-multicore/picos/blob/main/lib/picos_lwt/picos_lwt.ml

Performance

Stdlib:

Eio (without Picos support):

Conclusion:

It would seem that it is possible to write reasonably efficient Picos compatible schedulers and implement reasonably efficient concurrent programming primitives in Picos.

polytypic/article.md

Picos — Interoperable effects based concurrency

What is Picos?

Schedulers in scare quotes?

Picos is all about cancelation

Picos API

Signature extract

Samples

MirageOS

Performance