Merge pull request #5396 from kkeundotnet/mem2

Run garbage collection once when the main process is waiting for

master_changes.md

@@ -120,6 +120,7 @@ users)
   * Prefer curl over any other download tools on every systems, if available [#6305 @kit-ty-kate]
   * Avoid issues when using wget2 where the requested url might return an html page instead of the expected content [#6303 @kit-ty-kate]
   * Ensure each repositories stored in repos-config is associated with an URL [#6249 @kit-ty-kate]
+  * Run `Gc.compact` in OpamParallel, when the main process is waiting for the children processes for the first time [#5396 @kkeundotnet]
 
 ## Internal: Windows
 
@@ -174,3 +175,4 @@ users)
   * `OpamStubs.get_stdout_ws_col`: new Unix-only function returning the number of columns of the current terminal window [#6244 @kit-ty-kate]
   * `OpamSystem`: add `is_archive_from_string` that does the same than `is_archive` but without looking at the file, only analysing the string (extension) [#6219 @rjbou]
   * `OpamSystem.remove_dir`: do not fail with an exception when directory is a symbolic link [#6276 @btjorge @rjbou - fix #6275]
+  * `OpamParallel.*.{map,reduce,iter}`: Run `Gc.compact` when the main process is waiting for the children processes for the first time [#5396 @kkeundotnet]

src/core/opamParallel.ml

@@ -58,6 +58,16 @@ module type SIG = sig
   exception Cyclic of G.V.t list list
 end
 
+let gc_compact () =
+  let get_heap () =
+    let {Gc.heap_words; _} = Gc.quick_stat () in
+    heap_words * Sys.word_size / 8 / 1024 / 1024
+  in
+  let before = get_heap () in
+  Gc.compact ();
+  let after = get_heap () in
+  log "GC compact (heap %d MB -> %d MB)" before after
+
 module Make (G : G) = struct
 
   module G = G
@@ -92,6 +102,8 @@ module Make (G : G) = struct
       default :: defined
     in
 
+    let gc_compacted = ref false in
+
     if G.has_cycle g then (
       let sccs = G.scc_list g in
       let sccs = List.filter (function _::_::_ -> true | _ -> false) sccs in
@@ -264,28 +276,31 @@ module Make (G : G) = struct
             (get_slots nslots n)
         in
         run_seq_command nslots ready n cmd
-      else
-      (* Wait for a process to end *)
-      let processes =
-        M.fold (fun n (p,x,_) acc -> (p,(n,x)) :: acc) running []
-      in
-      let process, result =
-        if dry_run then
-          OpamProcess.dry_wait_one (List.map fst processes)
-        else try match processes with
-          | [p,_] -> p, OpamProcess.wait p
-          | _ -> OpamProcess.wait_one (List.map fst processes)
-        with e -> fail (fst (snd (List.hd processes))) e
-      in
-      let n,cont = OpamStd.(List.assoc Compare.equal process processes) in
-      log "Collected task for job %a (ret:%d)"
-        (slog (string_of_int @* V.hash)) n result.OpamProcess.r_code;
-      let next =
-        try cont result with e ->
-          OpamProcess.cleanup result;
-          fail n e in
-      OpamProcess.cleanup result;
-      run_seq_command nslots ready n next
+      else (
+        (* Wait for a process to end *)
+        if not !gc_compacted then
+          (gc_compact ();
+           gc_compacted := true);
+        let processes =
+          M.fold (fun n (p,x,_) acc -> (p,(n,x)) :: acc) running []
+        in
+        let process, result =
+          if dry_run then
+            OpamProcess.dry_wait_one (List.map fst processes)
+          else try match processes with
+            | [p,_] -> p, OpamProcess.wait p
+            | _ -> OpamProcess.wait_one (List.map fst processes)
+            with e -> fail (fst (snd (List.hd processes))) e
+        in
+        let n,cont = OpamStd.(List.assoc Compare.equal process processes) in
+        log "Collected task for job %a (ret:%d)"
+          (slog (string_of_int @* V.hash)) n result.OpamProcess.r_code;
+        let next =
+          try cont result with e ->
+            OpamProcess.cleanup result;
+            fail n e in
+        OpamProcess.cleanup result;
+        run_seq_command nslots ready n next)
     in
     let roots =
       G.fold_vertex