Report lost events at end of run

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;
@@ -140,7 +144,7 @@ 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))
@@ -150,7 +154,6 @@ let collect_events poll_sleep ~on_poll child callbacks =
     (fun () ->
       (* Read from the child process *)
       while child.alive () && not (Atomic.get interrupted) do
-        on_poll child.pid;
         Runtime_events.read_poll child.cursor callbacks None |> ignore;
         if poll_sleep > 0.0 then
           try Unix.sleepf poll_sleep
@@ -169,7 +172,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 +187,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 +197,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 +232,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. *)