+27 -26

bin/obs/cmd_report.ml

View file Reviewed

··· 147 147 still measured." 148 148 ctf) 149 149 150 150 + let read_traces model traces = 151 151 + List.iter 152 152 + (fun path -> 153 153 + match trace_kind path with 154 154 + | `Fxt -> 155 155 + Log.info (fun m -> m "reading scheduler/GC trace %s" path); 156 156 + Fxt_adapter.feed_file model path 157 157 + | `Ctf -> 158 158 + Log.info (fun m -> m "reading allocation trace %s" path); 159 159 + read_memtrace model path) 160 160 + traces 161 161 + 162 162 + let existing_cpu input = 163 163 + Option.bind input.cpu (fun path -> 164 164 + if Sys.file_exists path then Some path else None) 165 165 + 166 166 + let render_native_profile inputs native = 167 167 + match (List.find_map existing_cpu inputs, native) with 168 168 + | Some cpu, Some (backend, target_pid) -> 169 169 + let frames = Cpu_sampler.parse_saved cpu in 170 170 + if frames <> [] then 171 171 + Fmt.pr "%a" (Cpu_sampler.pp_profile ~backend ~target_pid) frames 172 172 + | _ -> () 173 173 + 150 174 (* Offline analysis of saved traces. Each TRACE is fed into one shared model: a 151 175 [.fxt] supplies scheduler and GC events, a [.ctf] supplies allocation 152 176 samples. Feeding both produces the unified report; the alloc->span join stays ··· 173 197 let model = Observer_model.Model.v () in 174 198 (* Each fxt record is routed to the (process, ring) carried by its 175 199 thread, so multiple producers in one trace stay isolated. *) 176 176 - List.iter 177 177 - (fun path -> 178 178 - match trace_kind path with 179 179 - | `Fxt -> 180 180 - Log.info (fun m -> m "reading scheduler/GC trace %s" path); 181 181 - Fxt_adapter.feed_file model path 182 182 - | `Ctf -> 183 183 - Log.info (fun m -> m "reading allocation trace %s" path); 184 184 - read_memtrace model path) 185 185 - traces; 200 200 + read_traces model traces; 186 201 Option.iter 187 202 (fun (start, stop) -> 188 203 Observer_model.Model.set_wall_clock model ~start ~stop) ··· 192 207 Observer_model.Model.set_process_context model ~process ~parent_span) 193 208 meta.ctx; 194 209 Cmd_common.render_report opts (Observer_model.Model.report model); 195 195 - (if not (Cmd_common.json_enabled ()) then 196 196 - match 197 197 - ( List.find_map 198 198 - (fun input -> 199 199 - Option.bind input.cpu (fun path -> 200 200 - if Sys.file_exists path then Some path else None)) 201 201 - inputs, 202 202 - meta.native ) 203 203 - with 204 204 - | Some cpu, Some (backend, target_pid) -> 205 205 - let frames = Cpu_sampler.parse_saved cpu in 206 206 - if frames <> [] then 207 207 - Fmt.pr "%a" 208 208 - (Cpu_sampler.pp_profile ~backend ~target_pid) 209 209 - frames 210 210 - | _ -> ()); 210 210 + if not (Cmd_common.json_enabled ()) then 211 211 + render_native_profile inputs meta.native; 211 212 Ok () 212 213 end 213 214

+33 -17

bin/obs/cmd_run.ml

View file Reviewed

··· 268 268 (Collector_server.clock_domain_count server) 269 269 else prerr_endline "obs: clock: exact (single clock domain)" 270 270 271 271 - (* Collection ended: write the bundle from the collector's sinks, then build the 272 272 - model and render the report -- the same path obs report takes offline. *) 273 273 - let finalize ~server ~sink ~sqlite ~sqlite_file ~catapult ~memtrace_dir ~fxt 274 274 - ~base ~run_id ~start ~stop ~cpu opts = 271 271 + (* Persist the streaming capture before native-sampler post-processing. xctrace 272 272 + export is an independent, occasionally slow sidecar operation; a supervisor 273 273 + must never be able to interrupt it and lose the already-complete runtime, 274 274 + allocation, Catapult, and sqlite observations. *) 275 275 + let flush_capture ~server ~sink ~sqlite ~sqlite_file ~catapult ~memtrace_dir 276 276 + ~fxt = 275 277 Fxt_adapter.Sink.to_file sink fxt; 276 278 prerr_endline ("obs: wrote scheduler/GC trace " ^ fxt); 277 279 close_output_sinks ~sqlite ~sqlite_file ~catapult; 278 280 report_clock server; 279 279 - write_memtrace_ctfs ~memtrace_dir server; 281 281 + write_memtrace_ctfs ~memtrace_dir server 282 282 + 283 283 + (* The durable capture is on disk. Add metadata and render the same report that 284 284 + [obs report] reconstructs offline. *) 285 285 + let finalize_report ~server ~memtrace_dir ~fxt ~base ~run_id ~start ~stop ~cpu 286 286 + opts = 280 287 let producers = Collector_server.producers server in 281 288 let model = Model.v () in 282 289 Fxt_adapter.feed_file model fxt; ··· 348 355 do 349 356 Eio.Time.sleep clock 0.005 350 357 done; 351 351 - (status, finish_cpu (), Some (child_start, child_stop)) 358 358 + (status, finish_cpu, Some (child_start, child_stop)) 352 359 | None -> 353 360 prerr_endline ("obs: collecting on unix:" ^ sock); 354 361 prerr_endline ("obs: run producers with OBS_COLLECTOR=unix:" ^ sock); 355 362 Collector_server.wait_stop server ~clock ~grace ~max_secs ~interrupted; 356 356 - (Unix.WEXITED 0, None, None) 363 363 + (Unix.WEXITED 0, (fun () -> None), None) 357 364 358 365 let sqlite_sink ~fs ~sw = 359 366 Option.map (fun file -> Sqlite_adapter.Sink.v ~fs ~sw file) ··· 377 384 Sys.set_signal Sys.sigterm (Sys.Signal_handle request_stop); 378 385 (interrupted, termination_signal) 379 386 387 387 + let finish_capture ~server ~sink ~sqlite ~sqlite_file ~catapult ~memtrace_dir 388 388 + ~fxt ~base ~run_id ~collector_start ~collector_stop ~child_wall ~finish_cpu 389 389 + opts = 390 390 + (* For a spawned command, exclude collector setup and drain/serialization 391 391 + from the application's headline wall budget. They are especially visible 392 392 + on sub-second benchmarks and previously appeared as phantom application 393 393 + CPU. Listen-only captures retain the collector window. *) 394 394 + let start, stop = 395 395 + Option.value child_wall ~default:(collector_start, collector_stop) 396 396 + in 397 397 + flush_capture ~server ~sink ~sqlite ~sqlite_file ~catapult ~memtrace_dir ~fxt; 398 398 + let cpu = finish_cpu () in 399 399 + finalize_report ~server ~memtrace_dir ~fxt ~base ~run_id ~start ~stop ~cpu 400 400 + opts 401 401 + 380 402 let run cmd args out catapult sqlite_file sample sample_trigger vsock tcp grace 381 403 max_secs opts = 382 404 let run_id, base, memtrace_dir, fxt, catapult = resolve_paths out catapult in ··· 406 428 Collector_server.serve ~sw ~net ~mono server ~path:sock; 407 429 serve_vsock_if_requested ~sw ~mono server vsock; 408 430 serve_tcp_if_requested ~sw ~net ~mono server tcp; 409 409 - let st, cpu, child_wall = 431 431 + let st, finish_cpu, child_wall = 410 432 collect_cpu ~sw ~clock ~sample ~sample_trigger ~base ~sock ~grace 411 433 ~max_secs ~interrupted ~termination_signal ~server cmd args 412 434 in 413 435 status := st; 414 436 let collector_stop = Unix.gettimeofday () in 415 415 - (* For a spawned command, exclude collector setup and drain/serialization 416 416 - from the application's headline wall budget. They are especially 417 417 - visible on sub-second benchmarks and previously appeared as phantom 418 418 - application CPU. Listen-only captures retain the collector window. *) 419 419 - let start, stop = 420 420 - Option.value child_wall ~default:(collector_start, collector_stop) 421 421 - in 422 422 - finalize ~server ~sink ~sqlite ~sqlite_file ~catapult ~memtrace_dir ~fxt 423 423 - ~base ~run_id ~start ~stop ~cpu opts); 437 437 + finish_capture ~server ~sink ~sqlite ~sqlite_file ~catapult ~memtrace_dir 438 438 + ~fxt ~base ~run_id ~collector_start ~collector_stop ~child_wall 439 439 + ~finish_cpu opts); 424 440 (* The Eio listener removes the socket on switch close; tolerate that it may 425 441 already be gone. *) 426 442 (try Unix.unlink sock with Unix.Unix_error _ -> ());

+35 -18

lib/obs/offline/cpu_sampler.ml

View file Reviewed

··· 35 35 mutable child : int option; 36 36 } 37 37 38 38 - (* Spawn the sampler on [pid] without blocking; it stops itself when the child 39 39 - dies ([-mayDie] / the [sleep] bound) or after [max_secs]. *) 38 38 + (* Spawn the sampler on [pid] without blocking. Both native profilers attach to 39 39 + that process: a default observation must profile the command it was asked to 40 40 + run, not every unrelated process on the machine. [stop] interrupts xctrace 41 41 + at the observed child's end; [max_secs] remains a safety ceiling. *) 40 42 let start ~backend ~pid ~max_secs ~raw = 41 43 (try Unix.unlink raw with Unix.Unix_error (ENOENT, _, _) -> ()); 42 44 let secs = Fmt.str "%g" max_secs in ··· 69 71 "--no-prompt"; 70 72 "--template"; 71 73 "Time Profiler"; 72 72 - "--all-processes"; 74 74 + "--attach"; 75 75 + string_of_int pid; 73 76 "--time-limit"; 74 77 secs ^ "s"; 75 78 "--notify-tracing-started"; ··· 103 106 in 104 107 { backend; raw; trace; ready; child } 105 108 106 106 - let wait = function 107 107 - | None -> () 108 108 - | Some pid -> ( 109 109 - try ignore (Unix.waitpid [] pid) with Unix.Unix_error _ -> ()) 109 109 + let rec waitpid pid = 110 110 + try Some (snd (Unix.waitpid [] pid)) with 111 111 + | Unix.Unix_error (EINTR, _, _) -> waitpid pid 112 112 + | Unix.Unix_error (ECHILD, _, _) -> None 113 113 + 114 114 + let wait = function None -> () | Some pid -> ignore (waitpid pid) 110 115 111 116 let await_ready t = 112 117 wait t.ready; ··· 138 143 |] 139 144 Unix.stdin out Unix.stderr 140 145 in 141 141 - ignore (Unix.waitpid [] pid) 146 146 + ignore (waitpid pid) 142 147 with Unix.Unix_error _ -> ()) 143 148 144 149 let xctrace_export t = ··· 167 172 let pid = 168 173 Unix.create_process argv.(0) argv Unix.stdin Unix.stderr Unix.stderr 169 174 in 170 170 - let status = snd (Unix.waitpid [] pid) in 171 171 - status = WEXITED 0 && Sys.file_exists tmp 175 175 + let status = waitpid pid in 176 176 + status = Some (WEXITED 0) && Sys.file_exists tmp 172 177 && (Unix.stat tmp).st_size > 0 173 178 with Unix.Unix_error _ -> false 174 179 in 175 175 - if 176 176 - run () 177 177 - || 178 178 - (Unix.sleepf 0.25; 179 179 - run ()) 180 180 - then Unix.rename tmp t.raw 180 180 + (* [xctrace record] can exit before its trace service has made the final 181 181 + table exportable. A fixed 250 ms retry lost valid captures in real VZ 182 182 + runs. Retry only this sidecar operation, with a short bounded backoff; 183 183 + the runtime and allocation traces are already durable at this point. *) 184 184 + let deadline = Unix.gettimeofday () +. 8. in 185 185 + let rec export delay = 186 186 + if run () then true 187 187 + else if Unix.gettimeofday () >= deadline then false 188 188 + else begin 189 189 + Unix.sleepf delay; 190 190 + export (Float.min 2. (delay *. 2.)) 191 191 + end 192 192 + in 193 193 + if export 0.25 then Unix.rename tmp t.raw 181 194 else begin 182 195 (try Unix.unlink tmp with Unix.Unix_error (ENOENT, _, _) -> ()); 183 196 Fmt.epr 184 184 - "obs: xctrace export failed twice; retained native capture %s for \ 185 185 - recovery@." 197 197 + "obs: xctrace export did not become ready within 8s; retained native \ 198 198 + capture %s for recovery@." 186 199 trace 187 200 end 188 201 189 202 let stop t = 190 203 await_ready t; 204 204 + (match (t.backend, t.child) with 205 205 + | Xctrace, Some pid -> ( 206 206 + try Unix.kill pid Sys.sigint with Unix.Unix_error (ESRCH, _, _) -> ()) 207 207 + | Perf, _ | Xctrace, None -> ()); 191 208 wait t.child; 192 209 t.child <- None; 193 210 (match t.backend with Perf -> perf_report t | Xctrace -> xctrace_export t);

+3 -2

lib/obs/offline/cpu_sampler.mli

View file Reviewed

··· 23 23 24 24 val start : backend:backend -> pid:int -> max_secs:float -> raw:string -> t 25 25 (** [start ~backend ~pid ~max_secs ~raw] spawns the sampler on [pid] without 26 26 - blocking, writing its own output under [raw]. It stops itself when the child 27 27 - exits or after [max_secs]. *) 26 26 + blocking, writing its own output under [raw]. {!stop} ends an xctrace 27 27 + recording when the observed child exits; [max_secs] is the safety ceiling. 28 28 + Perf attaches to [pid] and keeps the same ceiling. *) 28 29 29 30 val await_ready : t -> unit 30 31 (** Wait until capture is active. This is meaningful for [xctrace], whose