X-Git-Url: https://git.xandkar.net/?p=dups.git;a=blobdiff_plain;f=dups.ml;h=e214d2562d9480af829693df19b98fddcdaf47f4;hp=44acdf3090a22af26dc943636f8e4e54fe23718a;hb=dbb52e5c345aeafd3b7a2f142ca6bf2039616574;hpb=7c17443d85a81ab5c7b9b77a076ac542af2234c6 diff --git a/dups.ml b/dups.ml index 44acdf3..e214d25 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)) @@ -271,51 +283,48 @@ end = struct end let lord t ~njobs ~vassals ~ic ~ocs = - eprintf "[debug] [lord] started\n%!"; let active_vassals = ref njobs in let results = Queue.create () in - let rec dispatch () = - match Ipc.recv ic with - | ((Exiting i) : ('input, 'output) msg_from_vassal) -> + let rec loop () = + match ((Ipc.recv ic) : ('input, 'output) msg_from_vassal) with + | Exiting i -> close_out ocs.(i); decr active_vassals; - if !active_vassals = 0 then - () - else - dispatch () - | ((Ready i) : ('input, 'output) msg_from_vassal) -> + if !active_vassals = 0 then () else loop () + | Ready i -> Ipc.send ocs.(i) (Job (next t)); - dispatch () - | ((Result (i, result)) : ('input, 'output) msg_from_vassal) -> + loop () + | Result (i, result) -> Queue.add result results; Ipc.send ocs.(i) (Job (next t)); - dispatch () + loop () in let rec wait = function - | [] -> () + | [] -> + () | vassals -> let pid, _process_status = Unix.wait () in (* TODO: handle process_status *) wait (List.filter vassals ~f:(fun p -> p <> pid)) in - dispatch (); + loop (); close_in ic; wait vassals; of_queue results let vassal i ~f ~vassal_pipe_r ~lord_pipe_w = - eprintf "[debug] [vassal %d] started\n%!" i; let ic = Unix.in_channel_of_descr vassal_pipe_r in let oc = Unix.out_channel_of_descr lord_pipe_w in - let rec work msg = - Ipc.send oc msg; - match Ipc.recv ic with - | (Job (Some x) : 'input msg_from_lord) -> - work (Result (i, (x, f x))) - | (Job None : 'input msg_from_lord) -> + let rec loop () = + match (Ipc.recv ic : 'input msg_from_lord) with + | Job (Some x) -> + Ipc.send oc (Result (i, (x, f x))); + loop () + | Job None -> Ipc.send oc (Exiting i) in - work (Ready i); + Ipc.send oc (Ready i); + loop (); close_in ic; close_out oc; exit 0 @@ -369,6 +378,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 +443,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 +467,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 +526,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 +550,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 +601,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 +612,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 +642,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 ->