X-Git-Url: https://git.xandkar.net/?p=dups.git;a=blobdiff_plain;f=dups.ml;h=6628d99a4260ccecd0f2be5077ad557c3d99cc1d;hp=44acdf3090a22af26dc943636f8e4e54fe23718a;hb=eb6d0f387ce3495e4263d8dd0866bd0d41e476bc;hpb=7c17443d85a81ab5c7b9b77a076ac542af2234c6 diff --git a/dups.ml b/dups.ml index 44acdf3..6628d99 100644 --- a/dups.ml +++ b/dups.ml @@ -12,10 +12,14 @@ module Metrics : sig : unit -> t val report : t - -> time_all:float - -> time_group_by_size:float - -> time_group_by_head:float - -> time_group_by_digest:float + -> wall_time_all:float + -> wall_time_group_by_size:float + -> wall_time_group_by_head:float + -> wall_time_group_by_digest:float + -> proc_time_all:float + -> proc_time_group_by_size:float + -> proc_time_group_by_head:float + -> proc_time_group_by_digest:float -> unit val file_considered @@ -115,39 +119,47 @@ end = struct let report t - ~time_all - ~time_group_by_size - ~time_group_by_head - ~time_group_by_digest + ~wall_time_all + ~wall_time_group_by_size + ~wall_time_group_by_head + ~wall_time_group_by_digest + ~proc_time_all + ~proc_time_group_by_size + ~proc_time_group_by_head + ~proc_time_group_by_digest = let b_to_mb b = (float_of_int b) /. 1024. /. 1024. in let b_to_gb b = (b_to_mb b) /. 1024. in - eprintf "Time : %8.2f seconds\n%!" - time_all; + eprintf "Total time : %.2f wall sec %.2f proc sec\n%!" + wall_time_all + proc_time_all; eprintf "Considered : %8d files %6.2f Gb\n%!" !(t.considered_files) (b_to_gb !(t.considered_bytes)); eprintf "Sampled : %8d files %6.2f Gb\n%!" !(t.sampled_files) (b_to_gb !(t.sampled_bytes)); - eprintf "Hashed : %8d files %6.2f Gb %6.2f seconds\n%!" + eprintf "Hashed : %8d files %6.2f Gb %6.2f wall sec %6.2f proc sec\n%!" !(t.hashed_files) (b_to_gb !(t.hashed_bytes)) - time_group_by_digest; + wall_time_group_by_digest + proc_time_group_by_digest; eprintf "Digests : %8d\n%!" !(t.digests); eprintf "Duplicates (Hashed - Digests): %8d files %6.2f Gb\n%!" (!(t.hashed_files) - !(t.digests)) (b_to_gb !(t.redundant_data)); eprintf "Skipped due to 0 size : %8d files\n%!" !(t.empty); - eprintf "Skipped due to unique size : %8d files %6.2f Gb %6.2f seconds\n%!" + eprintf "Skipped due to unique size : %8d files %6.2f Gb %6.2f wall sec %6.2f proc sec\n%!" !(t.unique_size_files) (b_to_gb !(t.unique_size_bytes)) - time_group_by_size; - eprintf "Skipped due to unique sample : %8d files %6.2f Gb %6.2f seconds\n%!" + wall_time_group_by_size + proc_time_group_by_size; + eprintf "Skipped due to unique sample : %8d files %6.2f Gb %6.2f wall sec %6.2f proc sec\n%!" !(t.unique_sample_files) (b_to_gb !(t.unique_sample_bytes)) - time_group_by_head; + wall_time_group_by_head + proc_time_group_by_head; eprintf "Ignored due to regex match : %8d files %6.2f Gb\n%!" !(t.ignored_files) (b_to_gb !(t.ignored_bytes)) @@ -369,6 +381,8 @@ module File : sig val lookup : string Stream.t -> t Stream.t (** Lookup file info for given paths *) + val head : t -> len:int -> metrics:M.t -> string + val filter_out_unique_sizes : t Stream.t -> metrics:M.t -> t Stream.t val filter_out_unique_heads : t Stream.t -> len:int -> metrics:M.t -> t Stream.t end = struct @@ -432,7 +446,8 @@ end = struct ~group:(fun {size; _} -> size) ~handle_singleton:(fun (size, _, _) -> M.file_unique_size metrics ~size) - let head path ~len ~metrics = + let head {path; _} ~len ~metrics = + M.file_sampled metrics; let buf = Bytes.make len ' ' in let ic = open_in_bin path in let rec read pos len = @@ -455,10 +470,7 @@ end = struct let filter_out_unique_heads files ~len ~metrics = filter_out_singletons files - ~group:(fun {path; _} -> - M.file_sampled metrics; - head path ~len ~metrics - ) + ~group:(head ~len ~metrics) ~handle_singleton:(fun (_, _, files) -> let {size; _} = List.hd files in (* Guaranteed non-empty *) M.file_unique_sample metrics ~size @@ -517,11 +529,15 @@ let make_output_fun = function ); close_out oc -let time () = +let time_wall () = Unix.gettimeofday () +let time_proc () = + Sys.time () + let main {input; output; ignore; sample = sample_len; njobs} = - let t0_all = time () in + let wt0_all = time_wall () in + let pt0_all = time_proc () in let metrics = M.init () in let output = make_output_fun output in let input = make_input_stream input ignore ~metrics in @@ -537,24 +553,42 @@ let main {input; output; ignore; sample = sample_len; njobs} = let files = input in - let t0_group_by_size = time () in + let wt0_group_by_size = time_wall () in + let pt0_group_by_size = time_proc () in eprintf "[debug] filtering-out files with unique size\n%!"; let files = File.filter_out_unique_sizes files ~metrics in - let t1_group_by_size = time () in + let pt1_group_by_size = time_proc () in + let wt1_group_by_size = time_wall () in - let t0_group_by_sample = t1_group_by_size in + let wt0_group_by_sample = wt1_group_by_size in + let pt0_group_by_sample = pt1_group_by_size in eprintf "[debug] filtering-out files with unique heads\n%!"; - let files = File.filter_out_unique_heads files ~len:sample_len ~metrics in - let t1_group_by_sample = time () in + let files = + if njobs > 1 then begin + let q = Queue.create () in + files + |> Stream.bag_map ~njobs ~f:(File.head ~len:sample_len ~metrics) + |> Stream.group_by ~f:snd + |> Stream.map ~f:(fun (d, n, pairs) -> (d, n, List.map pairs ~f:fst)) + |> Stream.filter ~f:(fun (_, n, _) -> n > 1) + |> Stream.iter ~f:(fun (_, _, fs) -> List.iter fs ~f:(fun f -> Queue.add f q)) + ; + Stream.of_queue q + end else + File.filter_out_unique_heads files ~len:sample_len ~metrics + in + let pt1_group_by_sample = time_proc () in + let wt1_group_by_sample = time_wall () in - let t0_group_by_digest = t1_group_by_sample in + let wt0_group_by_digest = wt1_group_by_sample in + let pt0_group_by_digest = pt1_group_by_sample in eprintf "[debug] hashing\n%!"; let groups = if njobs > 1 then - let digests = + let with_digests = Stream.bag_map files ~njobs ~f:(fun {File.path; _} -> Digest.file path) in - Stream.map (Stream.group_by digests ~f:(fun (_, d) -> d)) ~f:( + Stream.map (Stream.group_by with_digests ~f:snd) ~f:( fun (digest, n, file_digest_pairs) -> let files = List.map file_digest_pairs ~f:(fun (file, _) -> @@ -570,7 +604,8 @@ let main {input; output; ignore; sample = sample_len; njobs} = Digest.file path ) in - let t1_group_by_digest = time () in + let pt1_group_by_digest = time_proc () in + let wt1_group_by_digest = time_wall () in eprintf "[debug] reporting\n%!"; Stream.iter groups ~f:(fun (d, n, files) -> @@ -580,13 +615,18 @@ let main {input; output; ignore; sample = sample_len; njobs} = output d n files ); - let t1_all = time () in + let pt1_all = time_proc () in + let wt1_all = time_wall () in M.report metrics - ~time_all: (t1_all -. t0_all) - ~time_group_by_size: (t1_group_by_size -. t0_group_by_size) - ~time_group_by_head: (t1_group_by_sample -. t0_group_by_sample) - ~time_group_by_digest:(t1_group_by_digest -. t0_group_by_digest) + ~wall_time_all: (wt1_all -. wt0_all) + ~wall_time_group_by_size: (wt1_group_by_size -. wt0_group_by_size) + ~wall_time_group_by_head: (wt1_group_by_sample -. wt0_group_by_sample) + ~wall_time_group_by_digest:(wt1_group_by_digest -. wt0_group_by_digest) + ~proc_time_all: (pt1_all -. pt0_all) + ~proc_time_group_by_size: (pt1_group_by_size -. pt0_group_by_size) + ~proc_time_group_by_head: (pt1_group_by_sample -. pt0_group_by_sample) + ~proc_time_group_by_digest:(pt1_group_by_digest -. pt0_group_by_digest) let get_opt () : opt = let assert_ test x msg = @@ -605,7 +645,7 @@ let get_opt () : opt = let output = ref Stdout in let ignore = ref (fun _ -> false) in let sample = ref 512 in - let njobs = ref 8 in + let njobs = ref 6 in let spec = [ ( "-out" , Arg.String (fun path ->