domain safe logger

log.ml

@@ -66,7 +66,7 @@ module State = struct
       Hashtbl.find all name
     with
       Not_found ->
-        let x = { Logger.name = name; show = Logger.int_level !default_level } in
+        let x = { Logger.name = name; show = Atomic.make (Logger.int_level !default_level) } in
         Hashtbl.add all name x;
         x
 
@@ -150,7 +150,7 @@ module State = struct
   (** Main logger *)
   let logger_target = {Logger.
     format;
-    output = output_simple;
+    output = Atomic.make output_simple;
   }
   let logger = Logger.put_simple logger_target
 

logger.ml

@@ -1,22 +1,29 @@
+(** Logger primitives.
+
+  Domain safety: [t.put], {!allowed}, {!get_level} and {!set_filter} are safe to call
+  from any domain, and [target.output] may be swapped (with [Atomic.set]) from any domain.
+  Reconfiguration is not atomic with respect to messages being logged concurrently:
+  a message that passed the old filter may still be emitted, possibly via the new output.
+  [target.output] itself is called from whichever domain logs, so it must be domain-safe. *)
 
 type level = [`Debug | `Info | `Warn | `Error | `Critical | `Nothing]
-type facil = { name : string; mutable show : int; }
+type facil = { name : string; show : int Atomic.t; }
 let int_level = function
   | `Debug -> 0
   | `Info -> 1
   | `Warn -> 2
   | `Error -> 3
   | `Critical -> 4
   | `Nothing -> 100
-let set_filter facil level = facil.show <- int_level level
-let get_level facil = match facil.show with
+let set_filter facil level = Atomic.set facil.show (int_level level)
+let get_level facil = match Atomic.get facil.show with
   | 0 -> `Debug
   | 1 -> `Info
   | 2 -> `Warn
   | 3 -> `Error
   | x when x = 100 -> `Nothing
   | _ -> `Critical (* ! *)
-let allowed facil level = level <> `Nothing && int_level level >= facil.show
+let allowed facil level = level <> `Nothing && int_level level >= Atomic.get facil.show
 
 let string_level = function
   | `Debug -> "debug"
@@ -42,7 +49,7 @@ end
 
 type target = {
   format : level -> facil -> Time.t -> Pairs.t -> string -> string;
-  mutable output : level -> facil -> string -> unit;
+  output : (level -> facil -> string -> unit) Atomic.t;
 }
 
 (** A logger *)
@@ -55,5 +62,5 @@ let put_simple (t:target) : t = {
   allowed;
   put = fun level facil ts pairs str ->
     if allowed facil level then
-      t.output level facil (t.format level facil ts pairs str)
+      (Atomic.get t.output) level facil (t.format level facil ts pairs str)
 }

test.ml

@@ -741,9 +741,9 @@ let () = test "Logfmt.Parser" begin fun () ->
 end
 
 let without_logging f =
-  let log_output = Log.State.logger_target.output in
-  Log.State.logger_target.output <- (fun level facil s -> !Log.State.hook level facil s);
-  Std.finally (fun () -> Log.State.logger_target.output <- log_output) f ()
+  let log_output = Atomic.get Log.State.logger_target.output in
+  Atomic.set Log.State.logger_target.output (fun level facil s -> !Log.State.hook level facil s);
+  Std.finally (fun () -> Atomic.set Log.State.logger_target.output log_output) f ()
 
 let with_log_hook f =
   let buf = Buffer.create 128 in

test_log_rate_limit.ml

@@ -38,7 +38,7 @@ let () =
     format = (fun _level _facility _timestamp _pairs message -> message);
     output = (fun _level _facility message ->
       Buffer.add_string output message;
-      Buffer.add_char output '\n');
+      Buffer.add_char output '\n') |> Atomic.make;
   } in
   let logger = Logger.put_simple target in
   let log = new Log.logger ~logger (Log.facility "rate-limit-test") in

tests/dune

@@ -1,3 +1,3 @@
 (tests
- (names sharded_hash_trie_test cache_count_test)
+ (names sharded_hash_trie_test cache_count_test logger_domains_test)
  (libraries devkit qcheck-core qcheck-core.runner unix))

tests/logger_domains_test.ml

@@ -0,0 +1,51 @@
+open Devkit
+open QCheck2
+
+let sink counter = fun _level _facil _s -> Atomic.incr counter
+
+let mk_target counter =
+  { Logger.format = (fun _level _facil _ts _pairs msg -> msg); output = Atomic.make (sink counter) }
+
+(* [n_domains] domains each log [n_msgs] times while the main domain
+   runs [reconfigure] until they are done *)
+let run ~reconfigure ~facil ~target n_domains n_msgs =
+  let logger = Logger.put_simple target in
+  let done_ = Atomic.make 0 in
+  let ds = List.init n_domains (fun _ -> Domain.spawn (fun () ->
+    for i = 1 to n_msgs do
+      logger.Logger.put `Info facil 0. [] (string_of_int i)
+    done;
+    Atomic.incr done_))
+  in
+  let i = ref 0 in
+  while Atomic.get done_ < n_domains do
+    reconfigure !i; incr i; Domain.cpu_relax ()
+  done;
+  List.iter Domain.join ds
+
+let gen = Gen.(pair (int_range 1 8) (int_range 0 2000))
+
+(* swapping outputs never loses nor duplicates a message *)
+let swap_output =
+  Test.make ~name:"swap output" ~count:30 gen (fun (n_domains, n_msgs) ->
+    let a = Atomic.make 0 and b = Atomic.make 0 in
+    let target = mk_target a in
+    let facil = { Logger.name = "test"; show = Atomic.make (Logger.int_level `Debug) } in
+    run ~facil ~target n_domains n_msgs
+      ~reconfigure:(fun i -> Atomic.set target.output (sink (if i land 1 = 0 then b else a)));
+    Atomic.get a + Atomic.get b = n_domains * n_msgs)
+
+(* toggling the filter concurrently with logging *)
+let toggle_filter =
+  Test.make ~name:"toggle filter" ~count:30 gen (fun (n_domains, n_msgs) ->
+    let a = Atomic.make 0 in
+    let target = mk_target a in
+    let facil = { Logger.name = "test"; show = Atomic.make (Logger.int_level `Debug) } in
+    run ~facil ~target n_domains n_msgs
+      ~reconfigure:(fun i -> Logger.set_filter facil (if i land 1 = 0 then `Nothing else `Debug));
+    Logger.set_filter facil `Error;
+    Atomic.get a <= n_domains * n_msgs && Logger.get_level facil = `Error)
+
+let () =
+  ignore (Unix.alarm 300 : int);
+  exit (QCheck_base_runner.run_tests ~verbose:true [ swap_output; toggle_filter ])