Merge pull request #24 from c-cube/wip-fuchsia

fuchsia trace format

.gitignore

@@ -2,3 +2,5 @@ _opam
 _build
 *.json
 *.exe
+perf.*
+*.fxt

README.md

@@ -117,6 +117,21 @@ let f x y z =
     raise e
 ```
 
+Alternatively, a name can be provided for the span, which is useful if you want
+to access it and use functions like `Trace.add_data_to_span`:
+
+
+```ocaml
+let%trace f x y z =
+  do_sth x;
+  do_sth y;
+  begin
+    let%trace _sp = "sub-span" in
+    do_sth z;
+    Trace.add_data_to_span _sp ["x", `Int 42]
+  end
+```
+
 ### Dune configuration
 
 In your `library` or `executable` stanza, add: `(preprocess (pps ppx_trace))`.

bench/bench_fuchsia_write.ml

@@ -0,0 +1,58 @@
+open Trace_fuchsia_write
+module B = Benchmark
+
+let pf = Printf.printf
+
+let encode_1_span (out : Output.t) () =
+  Event.Duration_complete.encode out ~name:"span" ~t_ref:(Thread_ref.Ref 5)
+    ~time_ns:100_000L ~end_time_ns:5_000_000L ~args:[] ()
+
+let encode_3_span (out : Output.t) () =
+  Event.Duration_complete.encode out ~name:"outer" ~t_ref:(Thread_ref.Ref 5)
+    ~time_ns:100_000L ~end_time_ns:5_000_000L ~args:[] ();
+  Event.Duration_complete.encode out ~name:"inner" ~t_ref:(Thread_ref.Ref 5)
+    ~time_ns:180_000L ~end_time_ns:4_500_000L ~args:[] ();
+  Event.Instant.encode out ~name:"hello" ~time_ns:1_234_567L
+    ~t_ref:(Thread_ref.Ref 5)
+    ~args:[ "x", `Int 42 ]
+    ()
+
+let time_per_iter_ns (samples : B.t list) : float =
+  let n_iters = ref 0L in
+  let time = ref 0. in
+  List.iter
+    (fun (s : B.t) ->
+      n_iters := Int64.add !n_iters s.iters;
+      time := !time +. s.stime +. s.utime)
+    samples;
+  !time *. 1e9 /. Int64.to_float !n_iters
+
+let () =
+  let buf_pool = Buf_pool.create () in
+  let out =
+    Output.create ~buf_pool
+      ~send_buf:(fun buf -> Buf_pool.recycle buf_pool buf)
+      ()
+  in
+
+  let samples = B.throughput1 4 ~name:"encode_1_span" (encode_1_span out) () in
+  B.print_gc samples;
+
+  let [ (_, samples) ] = samples [@@warning "-8"] in
+
+  let iter_per_ns = time_per_iter_ns samples in
+  pf "%.3f ns/iter\n" iter_per_ns;
+
+  ()
+
+let () =
+  let buf_pool = Buf_pool.create () in
+  let out =
+    Output.create ~buf_pool
+      ~send_buf:(fun buf -> Buf_pool.recycle buf_pool buf)
+      ()
+  in
+
+  let samples = B.throughput1 4 ~name:"encode_3_span" (encode_3_span out) () in
+  B.print_gc samples;
+  ()

bench/dune

@@ -1,4 +1,16 @@
 
 (executable
   (name trace1)
+  (modules trace1)
   (libraries trace.core trace-tef))
+
+(executable
+  (name trace_fx)
+  (modules trace_fx)
+  (preprocess (pps ppx_trace))
+  (libraries trace.core trace-fuchsia))
+
+(executable
+  (name bench_fuchsia_write)
+  (modules bench_fuchsia_write)
+  (libraries benchmark trace-fuchsia.write))

bench/trace_fx.ml

@@ -0,0 +1,50 @@
+module Trace = Trace_core
+
+let ( let@ ) = ( @@ )
+
+let work ~dom_idx ~n () : unit =
+  Trace_core.set_thread_name (Printf.sprintf "worker%d" dom_idx);
+  for _i = 1 to n do
+    let%trace _sp = "outer" in
+    Trace_core.add_data_to_span _sp [ "i", `Int _i ];
+    for _k = 1 to 10 do
+      let%trace _sp = "inner" in
+      ()
+    done;
+
+    (* Thread.delay 1e-6 *)
+    if dom_idx = 0 && _i mod 4096 = 0 then (
+      Trace_core.message "gc stats";
+      let stat = Gc.quick_stat () in
+      Trace_core.counter_float "gc.minor" (8. *. stat.minor_words);
+      Trace_core.counter_float "gc.major" (8. *. stat.major_words)
+    )
+  done
+
+let main ~n ~j () : unit =
+  let domains =
+    Array.init j (fun dom_idx -> Domain.spawn (fun () -> work ~dom_idx ~n ()))
+  in
+
+  let%trace () = "join" in
+  Array.iter Domain.join domains
+
+let () =
+  let@ () = Trace_fuchsia.with_setup () in
+  Trace_core.set_process_name "trace_fxt1";
+  Trace_core.set_thread_name "main";
+
+  let%trace () = "main" in
+
+  let n = ref 10_000 in
+  let j = ref 4 in
+
+  let args =
+    [
+      "-n", Arg.Set_int n, " number of iterations";
+      "-j", Arg.Set_int j, " set number of workers";
+    ]
+    |> Arg.align
+  in
+  Arg.parse args ignore "bench1";
+  main ~n:!n ~j:!j ()

bench_fx.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+DUNE_OPTS="--profile=release --display=quiet"
+exec dune exec $DUNE_OPTS bench/trace_fx.exe -- $@

dune

@@ -1,4 +1,4 @@
 
 (env
-  (_ (flags :standard -strict-sequence -warn-error -a+8+26+27 -w +a-4-40-70)))
+  (_ (flags :standard -strict-sequence -warn-error -a+8+26+27 -w +a-4-40-42-44-70)))
 

dune-project

@@ -41,9 +41,21 @@
    (trace (= :version))
    (mtime (>= 2.0))
    base-unix
-   atomic
    dune)
  (tags
   (trace tracing catapult)))
 
+(package
+ (name trace-fuchsia)
+ (synopsis "A high-performance backend for trace, emitting a Fuchsia trace into a file")
+ (depends
+   (ocaml (>= 4.08))
+   (trace (= :version))
+   (mtime (>= 2.0))
+   base-bigarray
+   base-unix
+   dune)
+ (tags
+  (trace tracing fuchsia)))
+
 ; See the complete stanza docs at https://dune.readthedocs.io/en/stable/dune-files.html#dune-project

src/fuchsia/bg_thread.ml

@@ -0,0 +1,63 @@
+open Common_
+
+type out =
+  [ `Stdout
+  | `Stderr
+  | `File of string
+  ]
+
+type event =
+  | E_write_buf of Buf.t
+  | E_tick
+
+type state = {
+  buf_pool: Buf_pool.t;
+  oc: out_channel;
+  events: event B_queue.t;
+}
+
+let with_out_ (out : out) f =
+  let oc, must_close =
+    match out with
+    | `Stdout -> stdout, false
+    | `Stderr -> stderr, false
+    | `File path -> open_out path, true
+  in
+
+  if must_close then (
+    let finally () = close_out_noerr oc in
+    Fun.protect ~finally (fun () -> f oc)
+  ) else
+    f oc
+
+let handle_ev (self : state) (ev : event) : unit =
+  match ev with
+  | E_tick -> flush self.oc
+  | E_write_buf buf ->
+    output self.oc buf.buf 0 buf.offset;
+    Buf_pool.recycle self.buf_pool buf
+
+let bg_loop (self : state) : unit =
+  let continue = ref true in
+
+  while !continue do
+    match B_queue.pop_all self.events with
+    | exception B_queue.Closed -> continue := false
+    | evs -> List.iter (handle_ev self) evs
+  done
+
+let bg_thread ~buf_pool ~out ~(events : event B_queue.t) () : unit =
+  let@ oc = with_out_ out in
+  let st = { oc; buf_pool; events } in
+  bg_loop st
+
+(** Thread that simply regularly "ticks", sending events to
+     the background thread so it has a chance to write to the file,
+     and call [f()] *)
+let tick_thread events : unit =
+  try
+    while true do
+      Thread.delay 0.5;
+      B_queue.push events E_tick
+    done
+  with B_queue.Closed -> ()

src/fuchsia/common_.ml

@@ -0,0 +1,12 @@
+module A = Trace_core.Internal_.Atomic_
+module FWrite = Trace_fuchsia_write
+module B_queue = Trace_private_util.B_queue
+module Buf = FWrite.Buf
+module Buf_pool = FWrite.Buf_pool
+module Output = FWrite.Output
+
+let on_tracing_error =
+  ref (fun s -> Printf.eprintf "trace-fuchsia error: %s\n%!" s)
+
+let ( let@ ) = ( @@ )
+let spf = Printf.sprintf

src/fuchsia/dune

@@ -0,0 +1,9 @@
+
+
+(library
+  (name trace_fuchsia)
+  (public_name trace-fuchsia)
+  (synopsis "A high-performance backend for trace, emitting a Fuchsia trace into a file")
+  (libraries trace.core trace.private.util thread-local-storage
+             (re_export trace-fuchsia.write) bigarray
+             mtime mtime.clock.os atomic unix threads))