Add an actor model implementation example

polytypic · polytypic · commit 673b05e2d4e0 · 2024-09-12T17:08:26.000+03:00
diff --git a/lib/picos/bootstrap/picos_bootstrap.ml b/lib/picos/bootstrap/picos_bootstrap.ml
@@ -499,6 +499,10 @@ module Fiber = struct
         r.fls <- fls;
         Array.unsafe_set fls key
           (Sys.opaque_identity (Obj.magic value : non_float))
+
+    let remove (type a) (Fiber r : fiber) (key : a t) =
+      let fls = r.fls in
+      if key < Array.length fls then Array.unsafe_set fls key unique
   end
 end
 
diff --git a/lib/picos/picos.mli b/lib/picos/picos.mli
@@ -994,6 +994,9 @@ module Fiber : sig
 
         ⚠️ It is only safe to call [set] from the fiber itself or when the fiber
         is known not to be running. *)
+
+    val remove : fiber -> 'a t -> unit
+    (** *)
   end
 
   (** {2 Interface for spawning} *)
diff --git a/test/lib/hoot/dune b/test/lib/hoot/dune
@@ -0,0 +1,3 @@
+(library
+ (name hoot)
+ (libraries backoff multicore-magic picos))
diff --git a/test/lib/hoot/hoot.ml b/test/lib/hoot/hoot.ml
@@ -0,0 +1,167 @@
+open Picos
+
+let[@inline never] impossible () = failwith "impossible"
+let[@inline never] not_in_a_hoot () = raise (Sys_error "Not in a hoot")
+let[@inline never] already_in_a_hoot () = raise (Sys_error "Already in a hoot")
+
+module Message = struct
+  type t = ..
+end
+
+type _ tdt =
+  | Nil : [> `Nil ] tdt
+  | Message : {
+      message : Message.t;
+      next : [ `Nil | `Message ] tdt;
+    }
+      -> [> `Message ] tdt
+  | Wait : Trigger.t -> [> `Wait ] tdt
+
+type incoming = In : [< `Nil | `Message | `Wait ] tdt -> incoming [@@unboxed]
+
+type pid = {
+  computation : unit Computation.t;
+  terminated : unit Computation.t;
+  incoming : incoming Atomic.t;
+  mutable received : [ `Nil | `Message ] tdt;
+}
+
+let pid_key = Fiber.FLS.create ()
+
+let self () =
+  let fiber = Fiber.current () in
+  try Fiber.FLS.get_exn fiber pid_key
+  with Fiber.FLS.Not_set -> not_in_a_hoot ()
+
+module Pid = struct
+  type t = pid
+end
+
+let run main =
+  let fiber = Fiber.current () in
+  match Fiber.FLS.get_exn fiber pid_key with
+  | _ -> already_in_a_hoot ()
+  | exception Fiber.FLS.Not_set ->
+      let computation = Computation.create ~mode:`LIFO () in
+      let inner = Computation.Packed computation in
+      let t =
+        {
+          computation;
+          terminated = Computation.create ~mode:`LIFO ();
+          incoming = Atomic.make (In Nil) |> Multicore_magic.copy_as_padded;
+          received = Nil;
+        }
+      in
+      Fiber.FLS.set fiber pid_key t;
+      let (Packed parent as outer) = Fiber.get_computation fiber in
+      let canceler =
+        Computation.attach_canceler ~from:parent ~into:computation
+      in
+      Fiber.set_computation fiber inner;
+      Computation.capture t.terminated main ();
+      Computation.finish t.computation;
+      Fiber.set_computation fiber outer;
+      Computation.detach parent canceler;
+      Fiber.FLS.remove fiber pid_key
+
+let wait t = Computation.await t.terminated
+
+let spawn main =
+  let t =
+    {
+      computation = Computation.create ~mode:`LIFO ();
+      terminated = Computation.create ~mode:`LIFO ();
+      incoming = Atomic.make (In Nil);
+      received = Nil;
+    }
+  in
+  let fiber = Fiber.create ~forbid:false t.computation in
+  Fiber.FLS.set fiber pid_key t;
+  begin
+    Fiber.spawn fiber @@ fun fiber ->
+    let t = Fiber.FLS.get_exn fiber pid_key in
+    Computation.capture t.terminated main ();
+    Computation.finish t.computation
+  end;
+  t
+
+let rec rev_to (Message _ as ms : [ `Message ] tdt) :
+    [ `Nil | `Message ] tdt -> _ = function
+  | Nil -> ms
+  | Message r -> rev_to (Message { message = r.message; next = ms }) r.next
+
+let rev (Message r : [ `Message ] tdt) =
+  rev_to (Message { message = r.message; next = Nil }) r.next
+
+let rec receive t =
+  match Atomic.get t.incoming with
+  | In Nil as before ->
+      let trigger = Trigger.create () in
+      let after = In (Wait trigger) in
+      if Atomic.compare_and_set t.incoming before after then begin
+        match Trigger.await trigger with
+        | None -> ()
+        | Some (exn, bt) -> Printexc.raise_with_backtrace exn bt
+      end;
+      receive t
+  | _ -> begin
+      match Atomic.exchange t.incoming (In Nil) with
+      | In (Wait _ | Nil) -> impossible ()
+      | In (Message _ as ms) ->
+          let (Message r : [ `Message ] tdt) = rev ms in
+          t.received <- r.next;
+          r.message
+    end
+
+let receive () =
+  let fiber = Fiber.current () in
+  match Fiber.FLS.get_exn fiber pid_key with
+  | t -> begin
+      match t.received with
+      | Message r ->
+          t.received <- r.next;
+          r.message
+      | Nil -> receive t
+    end
+  | exception Fiber.FLS.Not_set -> not_in_a_hoot ()
+
+let rec send t message backoff =
+  match Atomic.get t.incoming with
+  | In ((Nil | Message _) as before) ->
+      let after = Message { message; next = before } in
+      if not (Atomic.compare_and_set t.incoming (In before) (In after)) then
+        send t message (Backoff.once backoff)
+  | In (Wait trigger as before) ->
+      let after = Message { message; next = Nil } in
+      if Atomic.compare_and_set t.incoming (In before) (In after) then
+        Trigger.signal trigger
+      else send t message (Backoff.once backoff)
+
+let send t message = send t message Backoff.default
+
+type Message.t += Terminated of Pid.t
+
+let monitor ~at ~the =
+  let[@alert "-handler"] trigger =
+    Trigger.from_action at the @@ fun _ at the -> send at (Terminated the)
+  in
+  if not (Computation.try_attach the.terminated trigger) then
+    send at (Terminated the)
+
+exception Terminate
+
+let empty_bt = Printexc.get_callstack 0
+
+let link t1 t2 =
+  let[@alert "-handler"] trigger =
+    Trigger.from_action t1 t2 @@ fun _ t1 t2 ->
+    Computation.cancel t1.computation Terminate empty_bt;
+    Computation.cancel t2.computation Terminate empty_bt
+  in
+  if
+    (not (Computation.try_attach t1.terminated trigger))
+    || not (Computation.try_attach t2.terminated trigger)
+  then begin
+    Computation.cancel t1.computation Terminate empty_bt;
+    Computation.cancel t2.computation Terminate empty_bt
+  end
diff --git a/test/lib/hoot/hoot.mli b/test/lib/hoot/hoot.mli
@@ -0,0 +1,42 @@
+val run : (unit -> unit) -> unit
+(** *)
+
+module Pid : sig
+  (** *)
+
+  type t
+  (** *)
+end
+
+val spawn : (unit -> unit) -> Pid.t
+(** *)
+
+val self : unit -> Pid.t
+(** *)
+
+val wait : Pid.t -> unit
+(** *)
+
+module Message : sig
+  (** *)
+
+  type t = ..
+  (** *)
+end
+
+val receive : unit -> Message.t
+(** *)
+
+val send : Pid.t -> Message.t -> unit
+(** *)
+
+type Message.t += Terminated of Pid.t  (** *)
+
+val monitor : at:Pid.t -> the:Pid.t -> unit
+(** *)
+
+exception Terminate
+(** *)
+
+val link : Pid.t -> Pid.t -> unit
+(** *)

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+(library`
	`2`	`+ (name hoot)`
	`3`	`+ (libraries backoff multicore-magic picos))`