diff --git a/cache.ml b/cache.ml index 830527e..222458e 100644 --- a/cache.ml +++ b/cache.ml @@ -60,37 +60,39 @@ module TimeLimited2(E: Set.OrderedType) end module Count = struct - open Hashtbl - type 'a t = ('a, int ref) Hashtbl.t - let create () : 'a t = create 16 - let clear = Hashtbl.clear - let entry t x = match find t x with r -> r | exception Not_found -> let r = ref 0 in Hashtbl.add t x r; r - let plus t x n = entry t x += n - let minus t x n = entry t x -= n + module H = ExtThread.ShardedHashTrie + (* counters are atomic, the table only grows (or is cleared) *) + type 'a t = ('a, int Atomic.t) H.t + let create () : 'a t = H.create ~shard_bits:6 () + let clear = H.clear + let entry t x = H.get_or_create t x ~f:(fun _ -> Atomic.make 0) + let plus t x n = ignore (Atomic.fetch_and_add (entry t x) n : int) + let minus t x n = plus t x (-n) let of_enum e = let h = create () in Enum.iter (fun (k,n) -> plus h k n) e; h let of_list l = of_enum @@ List.enum l let add t x = plus t x 1 let del t x = minus t x 1 - let enum t = enum t |> Enum.map (fun (k,n) -> k, !n) - let iter t f = iter (fun k n -> f k !n) t - let fold t f acc = Hashtbl.fold (fun k n acc -> f k !n acc) t acc - let count t k = match Hashtbl.find t k with n -> !n | exception Not_found -> 0 - let count_all t = Hashtbl.fold (fun _ n acc -> acc + !n) t 0 - let size = Hashtbl.length - let show t ?(sep=" ") f = enum t |> - List.of_enum |> List.sort ~cmp:(Action.compare_by fst) |> + let iter t f = H.iter t (fun k n -> f k (Atomic.get n)) + let fold t f acc = H.fold t (fun k n acc -> f k (Atomic.get n) acc) acc + let to_list t = fold t (fun k n acc -> (k,n) :: acc) [] + let enum t = List.enum (to_list t) + let count t k = match H.find_opt t k with Some n -> Atomic.get n | None -> 0 + let count_all t = fold t (fun _ n acc -> acc + n) 0 + let size = H.length + let show t ?(sep=" ") f = to_list t |> + List.sort ~cmp:(Action.compare_by fst) |> List.map (fun (x,n) -> sprintf "%S: %u" (f x) n) |> String.concat sep - let show_sorted t ?limit ?(sep="\n") f = enum t |> - List.of_enum |> List.sort ~cmp:(flip @@ Action.compare_by snd) |> + let show_sorted t ?limit ?(sep="\n") f = to_list t |> + List.sort ~cmp:(flip @@ Action.compare_by snd) |> (match limit with None -> id | Some n -> List.take n) |> List.map (fun (x,n) -> sprintf "%6d : %S" n (f x)) |> String.concat sep let stats t ?(cmp=compare) f = - if Hashtbl.length t = 0 then + let a = to_list t |> Array.of_list in + if Array.length a = 0 then "" else - let a = Array.of_enum (enum t) in let total = Array.fold_left (fun t (_,n) -> t + n) 0 a in let half = total / 2 in let cmp (x,_) (y,_) = cmp x y in @@ -108,10 +110,10 @@ module Count = struct sprintf "total %d median %s min %s max %s" total (match !med with None -> "?" | Some x -> show x) (show mi) (show ma) let distrib t = - if Hashtbl.length t = 0 then + let a = to_list t |> Array.of_list in + if Array.length a = 0 then [||] else - let a = Array.of_enum (enum t) in let total = Array.fold_left (fun t (_,n) -> t + n) 0 a in let limits = Array.init 10 (fun i -> total * (i + 1) / 10) in let cmp (x,_) (y,_) = compare (x:float) y in @@ -133,7 +135,7 @@ module Count = struct let data = show_sorted t ?limit ~sep f in let stats = stats t ?cmp f in stats^sep^data - let names (t : 'a t) = List.of_enum @@ Hashtbl.keys t + let names (t : 'a t) = H.fold t (fun k _ acc -> k :: acc) [] end diff --git a/cache.mli b/cache.mli index e6c5e40..f6b71c9 100644 --- a/cache.mli +++ b/cache.mli @@ -38,7 +38,12 @@ module LRU(K : Hashtbl.HashedType) : sig val lfu_free : 'v t -> int end -(** Count elements *) +(** Count elements. Domain safe. + + Collection-wide operations such as [iter], [fold], [size], + [clear], etc. can race against individual modifications (ie it's + possible that a call to [clear t] races against [plus t "x" 42] + and an empty structure is never observable) *) module Count : sig type 'a t val create : unit -> 'a t diff --git a/log.ml b/log.ml index e0da2b9..2872937 100644 --- a/log.ml +++ b/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 diff --git a/logger.ml b/logger.ml index 3a27585..fd35962 100644 --- a/logger.ml +++ b/logger.ml @@ -1,6 +1,13 @@ +(** 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 @@ -8,15 +15,15 @@ let int_level = function | `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) } diff --git a/test.ml b/test.ml index 05a6ece..ca7f43d 100644 --- a/test.ml +++ b/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 diff --git a/test_log_rate_limit.ml b/test_log_rate_limit.ml index b2dc860..3c4d275 100644 --- a/test_log_rate_limit.ml +++ b/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 diff --git a/tests/dune b/tests/dune index df491fd..b8584a6 100644 --- a/tests/dune +++ b/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)) diff --git a/tests/logger_domains_test.ml b/tests/logger_domains_test.ml new file mode 100644 index 0000000..ff9a7de --- /dev/null +++ b/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 ])