Make meio-runtime-events a library

dune-project

@@ -1,2 +1,2 @@
 (lang dune 2.9)
-(name ec)
+(name meio)

example/zero.ml

@@ -8,9 +8,9 @@ let main clock =
       (* Eio.Switch.run ~name:"forked context" @@ fun sw ->
       traceln "in another context I do this" *)) *)
   (* for i = 0 to 10 do
-    Eio.Switch.run ~name:(Fmt.str "switch %d" i) @@ fun sw ->
-    traceln "welcome %d" i
-  done; *)
+       Eio.Switch.run ~name:(Fmt.str "switch %d" i) @@ fun sw ->
+       traceln "welcome %d" i
+     done; *)
   Eio.Time.sleep clock 100000.0
 
 let () =

meio-runtime-events.opam

@@ -0,0 +1,31 @@
+# This file is generated by dune, edit dune-project instead
+opam-version: "2.0"
+synopsis: "Monitor live Eio programs"
+description:
+  "Eio-console provides an executable that allows you to monitor OCaml programs using Eventring."
+maintainer: ["[email protected]"]
+authors: ["Patrick Ferris"]
+license: "MIT"
+homepage: "https://github.com/patricoferris/eio-console"
+bug-reports: "https://github.com/patricoferris/eio-console/issues"
+depends: [
+  "dune" {>= "2.9"}
+  "odoc" {with-doc}
+]
+build: [
+  ["dune" "subst"] {dev}
+  [
+    "dune"
+    "build"
+    "-p"
+    name
+    "-j"
+    jobs
+    "--promote-install-files=false"
+    "@install"
+    "@runtest" {with-test}
+    "@doc" {with-doc}
+  ]
+  ["dune" "install" "-p" name "--create-install-files" name]
+]
+dev-repo: "git+https://github.com/patricoferris/eio-console.git"

src/bin/dune

@@ -1,4 +1,5 @@
 (executable
  (name meio)
  (public_name meio)
+ (package meio)
  (libraries meio))

src/bin/meio.ml

@@ -1,4 +1,4 @@
-let run _stdenv exec_args =
+let run exec_args =
   let argsl = String.split_on_char ' ' exec_args in
   let executable_filename = List.hd argsl in
   let argsl = Array.of_list argsl in
@@ -12,7 +12,9 @@ let run _stdenv exec_args =
       |]
       (Unix.environment ())
   in
-  let dev_null = Unix.openfile Filename.null [ Unix.O_WRONLY; Unix.O_KEEPEXEC ] 0o666 in
+  let dev_null =
+    Unix.openfile Filename.null [ Unix.O_WRONLY; Unix.O_KEEPEXEC ] 0o666
+  in
   let child_pid =
     Unix.create_process_env executable_filename argsl env Unix.stdin dev_null
       dev_null
@@ -27,4 +29,4 @@ let run _stdenv exec_args =
   in
   Unix.unlink ring_file
 
-let () = Eio_main.run @@ fun stdenv -> run stdenv Sys.argv.(1)
+let () = run Sys.argv.(1)

src/lib/console.ml

@@ -147,7 +147,7 @@ let render_task sort now ~depth ~filtered
   let kind =
     W.string ~attr
       (match kind with
-      | Cancellation_context _ -> "cc"
+      | Meio_runtime_events.Cancellation_context _ -> "cc"
       | Task -> "task"
       | _ -> "??")
   in

src/lib/dune

@@ -4,4 +4,11 @@
  (foreign_stubs
   (language c)
   (names meio_console_stubs))
- (libraries runtime_events ptime logs fmt nottui hdr_histogram))
+ (libraries
+  runtime_events
+  ptime
+  logs
+  fmt
+  nottui
+  hdr_histogram
+  meio-runtime-events))

src/lib/help.ml

@@ -22,7 +22,7 @@ let footer sort screen =
       space;
       key_help ~attr:(attr `Task) 'e' "Fiber info";
       space;
-      key_help ~attr:(attr `Logs) 'l' ("Logs");
+      key_help ~attr:(attr `Logs) 'l' "Logs";
       space;
       key_help 's' ("Sort " ^ Sort.to_string sort);
       space;

src/lib/lf_queue.ml

@@ -0,0 +1,140 @@
+(* Copyright (C) 2021 Anil Madhavapeddy
+      Copyright (C) 2022 Thomas Leonard
+
+   Permission to use, copy, modify, and distribute this software for any
+   purpose with or without fee is hereby granted, provided that the above
+   copyright notice and this permission notice appear in all copies.
+
+   THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+   WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+   MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+   ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+   WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+   ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+   OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. *)
+
+(* A lock-free multi-producer, single-consumer, thread-safe queue without support for cancellation.
+   This makes a good data structure for a scheduler's run queue.
+
+   See: "Implementing lock-free queues"
+   https://people.cs.pitt.edu/~jacklange/teaching/cs2510-f12/papers/implementing_lock_free.pdf
+
+   It is simplified slightly because we don't need multiple consumers.
+   Therefore [head] is not atomic. *)
+
+exception Closed
+
+module Node : sig
+  type 'a t = { next : 'a opt Atomic.t; mutable value : 'a }
+  and +'a opt
+
+  val make : next:'a opt -> 'a -> 'a t
+
+  val none : 'a opt
+  (** [t.next = none] means that [t] is currently the last node. *)
+
+  val closed : 'a opt
+  (** [t.next = closed] means that [t] will always be the last node. *)
+
+  val some : 'a t -> 'a opt
+  val fold : 'a opt -> none:(unit -> 'b) -> some:('a t -> 'b) -> 'b
+end = struct
+  (* https://github.com/ocaml/RFCs/pull/14 should remove the need for magic here *)
+
+  type +'a opt (* special | 'a t *)
+  type 'a t = { next : 'a opt Atomic.t; mutable value : 'a }
+  type special = Nothing | Closed
+
+  let none : 'a. 'a opt = Obj.magic Nothing
+  let closed : 'a. 'a opt = Obj.magic Closed
+  let some (t : 'a t) : 'a opt = Obj.magic t
+
+  let fold (opt : 'a opt) ~none:n ~some =
+    if opt == none then n ()
+    else if opt == closed then raise Closed
+    else some (Obj.magic opt : 'a t)
+
+  let make ~next value = { value; next = Atomic.make next }
+end
+
+type 'a t = { tail : 'a Node.t Atomic.t; mutable head : 'a Node.t }
+(* [head] is the last node dequeued (or a dummy node, initially).
+   [head.next] gives the real first node, if not [Node.none].
+   If [tail.next] is [none] then it is the last node in the queue.
+   Otherwise, [tail.next] is a node that is closer to the tail. *)
+
+let push t x =
+  let node = Node.(make ~next:none) x in
+  let rec aux () =
+    let p = Atomic.get t.tail in
+    (* While [p.next == none], [p] is the last node in the queue. *)
+    if Atomic.compare_and_set p.next Node.none (Node.some node) then
+      (* [node] has now been added to the queue (and possibly even consumed).
+         Update [tail], unless someone else already did it for us. *)
+      ignore (Atomic.compare_and_set t.tail p node : bool)
+    else
+      (* Someone else added a different node first ([p.next] is not [none]).
+         Make [t.tail] more up-to-date, if it hasn't already changed, and try again. *)
+      Node.fold (Atomic.get p.next)
+        ~none:(fun () -> assert false)
+        ~some:(fun p_next ->
+          ignore (Atomic.compare_and_set t.tail p p_next : bool);
+          aux ())
+  in
+  aux ()
+
+let rec push_head t x =
+  let p = t.head in
+  let next = Atomic.get p.next in
+  if next == Node.closed then raise Closed;
+  let node = Node.make ~next x in
+  if Atomic.compare_and_set p.next next (Node.some node) then
+    if
+      (* We don't want to let [tail] get too far behind, so if the queue was empty, move it to the new node. *)
+      next == Node.none
+    then ignore (Atomic.compare_and_set t.tail p node : bool)
+    else
+      ( (* If the queue wasn't empty, there's nothing to do.
+           Either tail isn't at head or there is some [push] thread working to update it.
+           Either [push] will update it directly to the new tail, or will update it to [node]
+           and then retry. Either way, it ends up at the real tail. *) )
+  else (
+    (* Someone else changed it first. This can only happen if the queue was empty. *)
+    assert (next == Node.none);
+    push_head t x)
+
+let rec close (t : 'a t) =
+  (* Mark the tail node as final. *)
+  let p = Atomic.get t.tail in
+  if not (Atomic.compare_and_set p.next Node.none Node.closed) then
+    (* CAS failed because [p] is no longer the tail (or is already closed). *)
+    Node.fold (Atomic.get p.next)
+      ~none:(fun () -> assert false)
+        (* Can't switch from another state to [none] *)
+      ~some:(fun p_next ->
+        (* Make [tail] more up-to-date if it hasn't changed already *)
+        ignore (Atomic.compare_and_set t.tail p p_next : bool);
+        (* Retry *)
+        close t)
+
+let pop t =
+  let p = t.head in
+  (* [p] is the previously-popped item. *)
+  let node = Atomic.get p.next in
+  Node.fold node
+    ~none:(fun () -> None)
+    ~some:(fun node ->
+      t.head <- node;
+      let v = node.value in
+      node.value <- Obj.magic ();
+      (* So it can be GC'd *)
+      Some v)
+
+let is_empty t =
+  Node.fold (Atomic.get t.head.next)
+    ~none:(fun () -> true)
+    ~some:(fun _ -> false)
+
+let create () =
+  let dummy = { Node.value = Obj.magic (); next = Atomic.make Node.none } in
+  { tail = Atomic.make dummy; head = dummy }

src/lib/logging.ml

@@ -1,4 +1,4 @@
-module Queue = Eio_utils.Lf_queue
+module Queue = Lf_queue
 
 let table = Lwd_table.make ()
 let waiting = Queue.create ()

src/lib/meio.ml

@@ -1,10 +1,10 @@
-module Ctf = Eio.Private.Ctf
+module Ctf = Meio_runtime_events
 
 let add_callback = Runtime_events.Callbacks.add_user_event
 
 let task_events ~latency_begin ~latency_end q =
   let current_id = ref (-1) in
-  let module Queue = Eio_utils.Lf_queue in
+  let module Queue = Lf_queue in
   let evs =
     Runtime_events.Callbacks.create ~runtime_begin:latency_begin
       ~runtime_end:latency_end
@@ -14,7 +14,9 @@ let task_events ~latency_begin ~latency_end q =
   in
   let id_event_callback d ts c ((i : Ctf.id), v) =
     match (Runtime_events.User.tag c, v) with
-    | Ctf.Created, (Ctf.Task | Ctf.Cancellation_context _) ->
+    | ( Meio_runtime_events.Created,
+        (Meio_runtime_events.Task | Meio_runtime_events.Cancellation_context _)
+      ) ->
         Queue.push q (`Created ((i :> int), !current_id, d, ts, v))
     | _ -> ()
   in
@@ -82,7 +84,7 @@ let screens duration hist sort =
 let min_thresh = 1e-6
 let sleepf f = if f > min_thresh then Unix.sleepf f
 
-module Queue = Eio_utils.Lf_queue
+module Queue = Lf_queue
 
 let runtime_event_loop ~child_pid ~q ~stop ~cursor ~callbacks =
   let max_sleep = 0.01 in

src/lib/state.ml

@@ -2,11 +2,13 @@
 let tasks = Task_tree.make ()
 
 let set_parent ~child ~parent ts =
-  Task_tree.set_parent tasks ~child:(Task.Id.eio_of_int child)
-    ~parent:(Task.Id.eio_of_int parent)
+  Task_tree.set_parent tasks
+    ~child:(Task.Id.extern_of_int child)
+    ~parent:(Task.Id.extern_of_int parent)
     (Runtime_events.Timestamp.to_int64 ts)
 
 let add_tasks (id, parent_id, domain, ts, kind) =
+  Logs.info (fun f -> f "Created %i" id);
   let task =
     Task.create ~id ~domain ~parent_id
       (Runtime_events.Timestamp.to_int64 ts)
@@ -15,23 +17,24 @@ let add_tasks (id, parent_id, domain, ts, kind) =
   Task_tree.add tasks task
 
 let update_loc i loc =
-  Task_tree.update tasks (Task.Id.eio_of_int i) (fun t ->
+  Task_tree.update tasks (Task.Id.extern_of_int i) (fun t ->
       { t with Task.loc = loc :: t.loc })
 
 let update_logs i logs =
-  Task_tree.update tasks (Task.Id.eio_of_int i) (fun t ->
+  Task_tree.update tasks (Task.Id.extern_of_int i) (fun t ->
+      (* ignore (failwith(Fmt.str "Log updating for %a" Task.Id.pp t.id)); *)
       { t with Task.logs = logs :: t.logs })
 
 let update_name i name =
-  Task_tree.update tasks (Task.Id.eio_of_int i) (fun t ->
+  Task_tree.update tasks (Task.Id.extern_of_int i) (fun t ->
       { t with Task.name = name :: t.name })
 
 let switch_to ~id ~domain:_ ts =
-  Task_tree.update_active tasks ~id:(Task.Id.eio_of_int id)
+  Task_tree.update_active tasks ~id:(Task.Id.extern_of_int id)
     (Runtime_events.Timestamp.to_int64 ts)
 
 let resolved v ts =
-  Task_tree.update tasks (Task.Id.eio_of_int v) (fun t ->
+  Task_tree.update tasks (Task.Id.extern_of_int v) (fun t ->
       { t with status = Resolved (Runtime_events.Timestamp.to_int64 ts) })
 
 let terminated status ts =

src/lib/state.mli

@@ -1,7 +1,8 @@
 val tasks : Task_tree.t
 
 val add_tasks :
-  int * int * int * Runtime_events.Timestamp.t * Eio.Private.Ctf.event -> unit
+  int * int * int * Runtime_events.Timestamp.t * Meio_runtime_events.event ->
+  unit
 
 val update_loc : int -> string -> unit
 val update_logs : int -> string -> unit

src/lib/task.ml

@@ -54,29 +54,29 @@ end
 type display = Auto | Yes | No | Toggle_requested
 
 module Id = struct
-  type t = { eio : int; counter : int; global_counter : int ref }
+  type t = { extern : int; counter : int; global_counter : int ref }
 
-  let make eio = { eio; counter = 0; global_counter = ref 0 }
+  let make extern = { extern; counter = 0; global_counter = ref 0 }
 
   let fork t =
     incr t.global_counter;
     { t with counter = !(t.global_counter) }
 
   let pp fmt t =
-    if t.counter == 0 then Fmt.int fmt t.eio
-    else Fmt.pf fmt "%d.%d" t.eio t.counter
+    if t.counter == 0 then Fmt.int fmt t.extern
+    else Fmt.pf fmt "%d.%d" t.extern t.counter
 
   let compare a b =
-    match Int.compare a.eio b.eio with
+    match Int.compare a.extern b.extern with
     | 0 -> Int.compare a.counter b.counter
     | v -> v
 
-  type eio = int
+  type extern = int
 
-  let pp_eio = Fmt.int
-  let eio t = t.eio
+  let pp_extern = Fmt.int
+  let to_extern t = t.extern
 
-  let eio_of_int t =
+  let extern_of_int t =
     assert (t >= -1);
     t
 end
@@ -91,7 +91,7 @@ type t = {
   loc : string list;
   logs : string list;
   status : status;
-  kind : Eio.Private.Ctf.event;
+  kind : Meio_runtime_events.event;
   selected : bool ref;
   display : display ref;
 }