Alive poller

lib/olly_common/alive_poller.ml

@@ -0,0 +1,32 @@
+type t = {
+  stop_flag : bool Atomic.t;
+  domain : unit Domain.t;
+  alive : bool Atomic.t;
+}
+
+let start ~check ~interval =
+  let stop_flag = Atomic.make false in
+  let alive = Atomic.make true in
+  let domain =
+    Domain.spawn (fun () ->
+        let rec loop () =
+          if Atomic.get stop_flag then ()
+          else begin
+            let still_alive = check () in
+            Atomic.set alive still_alive;
+            if still_alive then begin
+              (try Unix.sleepf interval
+               with Unix.Unix_error (Unix.EINTR, _, _) -> ());
+              loop ()
+            end
+          end
+        in
+        loop ())
+  in
+  { stop_flag; domain; alive }
+
+let is_alive t = Atomic.get t.alive
+
+let stop t =
+  Atomic.set t.stop_flag true;
+  Domain.join t.domain

lib/olly_common/alive_poller.mli

@@ -0,0 +1,13 @@
+(** Sample whether a process is alive from a dedicated domain. *)
+
+type t
+
+val start : check:(unit -> bool) -> interval:float -> t
+(** [start ~check ~interval] spawns a domain that samples the function [check]
+    every [interval] seconds. *)
+
+val is_alive : t -> bool
+(** retrieve last process status sampled *)
+
+val stop : t -> unit
+(** [stop t] signals the poller to stop, waits for it to finish. *)

lib/olly_common/cli.ml

@@ -26,6 +26,17 @@ let freq_option =
     & opt float 0.1 (* Poll at 10Hz by default. *)
     & info [ "freq" ] ~docv:"freq" ~doc)
 
+let rss_freq_option =
+  let doc =
+    "Set the interval, in seconds, at which the peak RSS of the monitored \
+     process is sampled. Sampling runs on a dedicated domain, independently of \
+     $(b,--freq). A value of 0.0 samples as fast as possible."
+  in
+  Arg.(
+    value
+    & opt float 0.1 (* Sample at 10Hz by default. *)
+    & info [ "rss-freq" ] ~docv:"rss-freq" ~doc)
+
 let runtime_events_dir =
   let doc =
     "Sets the directory where the .events files containing the runtime event \

lib/olly_common/launch.ml

@@ -1,7 +1,11 @@
 external is_process_alive : int -> bool = "olly_is_process_alive"
 
-let lost_events ring_id num =
-  Printf.eprintf "[ring_id=%d] Lost %d events\n%!" ring_id num
+let lost_events_count = ref 0
+
+let lost_events _ring_id num =
+  let sum = !lost_events_count + num in
+  (* detect overflow and stay at [max_int] *)
+  lost_events_count := if sum < 0 then max_int else sum
 
 type subprocess = {
   alive : unit -> bool;
@@ -92,6 +96,9 @@ let exec_process (config : runtime_events_config) (args : string list) :
     | _, _ -> assert false
     | exception Unix.Unix_error (Unix.EINTR, _, _) -> true
   and close () =
+    (* reap child in case we were interrupted *)
+    (try Unix.waitpid [] child_pid |> ignore
+     with Unix.Unix_error (Unix.ECHILD, _, _) -> ());
     Runtime_events.free_cursor cursor;
     (* We need to remove the ring buffers ourselves because we told
        the child process not to remove them. However, if the user
@@ -140,17 +147,21 @@ let launch_process config (exec_args : exec_config) : subprocess =
 
 let interrupted = Atomic.make false
 
-let collect_events poll_sleep ~on_poll child callbacks =
+let collect_events poll_sleep child callbacks =
   let old_handler =
     Sys.signal Sys.sigint
       (Sys.Signal_handle (fun _ -> Atomic.set interrupted true))
   in
+  let alive_poller = Alive_poller.start ~check:child.alive ~interval:0.1 in
   Fun.protect
-    ~finally:(fun () -> Sys.set_signal Sys.sigint old_handler)
+    ~finally:(fun () ->
+      Alive_poller.stop alive_poller;
+      Sys.set_signal Sys.sigint old_handler)
     (fun () ->
       (* Read from the child process *)
-      while child.alive () && not (Atomic.get interrupted) do
-        on_poll child.pid;
+      while
+        Alive_poller.is_alive alive_poller && not (Atomic.get interrupted)
+      do
         Runtime_events.read_poll child.cursor callbacks None |> ignore;
         if poll_sleep > 0.0 then
           try Unix.sleepf poll_sleep
@@ -169,7 +180,7 @@ type consumer_config = {
   extra : Runtime_events.Callbacks.t -> Runtime_events.Callbacks.t;
   init : unit -> unit;
   cleanup : unit -> unit;
-  on_poll : int -> unit;
+  on_launch : subprocess -> unit;
   poll_sleep : float;
   runtime_events_dir : string option;
   runtime_events_log_wsize : int option;
@@ -184,7 +195,7 @@ let empty_config =
     extra = Fun.id;
     init = (fun () -> ());
     cleanup = (fun () -> ());
-    on_poll = (fun _ -> ());
+    on_launch = (fun _ -> ());
     poll_sleep = 0.1 (* Poll at 10Hz *);
     runtime_events_dir = None;
     (* Use default tmp directory *)
@@ -194,14 +205,25 @@ let empty_config =
 
 let olly config exec_args =
   config.init ();
-  Fun.protect ~finally:config.cleanup (fun () ->
+  let finally () =
+    config.cleanup ();
+    if !lost_events_count > 0 then begin
+      Printf.eprintf "Lost %d events, stats not reliable%s\n%!"
+        !lost_events_count
+        (if !lost_events_count = max_int then
+           " (possible counter overflow detected)\n%!"
+         else "")
+    end
+  in
+  Fun.protect ~finally (fun () ->
       let runtime_config =
         {
           dir = config.runtime_events_dir;
           log_wsize = config.runtime_events_log_wsize;
         }
       in
       let child = launch_process runtime_config exec_args in
+      config.on_launch child;
       Fun.protect ~finally:child.close (fun () ->
           let callbacks =
             let {
@@ -218,5 +240,4 @@ let olly config exec_args =
               ~runtime_counter ~lifecycle ~lost_events ()
             |> extra
           in
-          collect_events config.poll_sleep ~on_poll:config.on_poll child
-            callbacks))
+          collect_events config.poll_sleep child callbacks))

lib/olly_common/max_rss.ml

@@ -1,11 +1,5 @@
-external get_rss_kb : int -> int = "olly_get_rss_kb"
-
 type t = { mutable max_rss_kb : int }
 
 let create () = { max_rss_kb = 0 }
-
-let sample t pid =
-  let rss = get_rss_kb pid in
-  if rss > t.max_rss_kb then t.max_rss_kb <- rss
-
+let set t kb = if kb > t.max_rss_kb then t.max_rss_kb <- kb
 let max_rss_kb t = t.max_rss_kb

lib/olly_common/rss_poller.ml

@@ -0,0 +1,25 @@
+external get_rss_kb : int -> int = "olly_get_rss_kb"
+
+type t = { stop_flag : bool Atomic.t; domain : int Domain.t }
+
+let start ~pid ~interval =
+  let stop_flag = Atomic.make false in
+  let domain =
+    Domain.spawn (fun () ->
+        let rec loop peak =
+          if Atomic.get stop_flag then peak
+          else begin
+            (* sample before waiting so that we get a reading at launch *)
+            let rss = get_rss_kb pid in
+            (try Unix.sleepf interval
+             with Unix.Unix_error (Unix.EINTR, _, _) -> ());
+            loop (max rss peak)
+          end
+        in
+        loop 0)
+  in
+  { stop_flag; domain }
+
+let stop t =
+  Atomic.set t.stop_flag true;
+  Domain.join t.domain

lib/olly_common/rss_poller.mli

@@ -0,0 +1,11 @@
+(** Sample a process's peak resident set size from a dedicated domain. *)
+
+type t
+
+val start : pid:int -> interval:float -> t
+(** [start ~pid ~interval] spawns a domain that samples the RSS of [pid] every
+    [interval] seconds, tracking the peak. *)
+
+val stop : t -> int
+(** [stop t] signals the poller to stop, waits for it to finish, and returns the
+    peak RSS observed, in kB. *)