From b6efe8b498b0c032c23767e6b6d60005086c2fd9 Mon Sep 17 00:00:00 2001 From: Julien Portalier Date: Tue, 30 Jan 2024 09:06:37 +0100 Subject: [PATCH 1/4] Codegen: on demand distribution to forked processes Instead of pre-slicing the list of compilation units then having each forked process handle its personal list, with some finishing before the others, this patch adds bidirectional pipes to push the compilation units as the forked processes make progress, so that each forked process will continue to compile for as long as there is something to compile. The forker processes are overqueued with one compilation unit to avoid latencies (they compile the next unit while the main process pushes the next). This only improves performance when there are multiple compilation units. It won't have any effect when `--single-module` (implied by `--release`) is set, and will have little impact when a module is extra large (e.g. crystal specs) unless maybe when there are multiple of them and they were set to the same forked process. --- src/compiler/crystal/compiler.cr | 105 +++++++++++++++++++++---------- 1 file changed, 73 insertions(+), 32 deletions(-) diff --git a/src/compiler/crystal/compiler.cr b/src/compiler/crystal/compiler.cr index 899ef242f318..f6c8273a5cee 100644 --- a/src/compiler/crystal/compiler.cr +++ b/src/compiler/crystal/compiler.cr @@ -551,57 +551,98 @@ module Crystal {% elsif flag?(:preview_mt) %} raise "Cannot fork compiler in multithread mode" {% else %} - jobs_count = 0 - wait_channel = Channel(Array(String)).new(@n_threads) + workers = fork_workers do |input, output| + while i = input.gets(chomp: true).presence + unit = units[i.to_i] + unit.compile + result = {name: unit.name, reused: unit.reused_previous_compilation?} + output.puts result.to_json + end + end + + overqueue = 1 + indexes = Atomic(Int32).new(0) + channel = Channel(String).new(@n_threads) + completed = Channel(Nil).new(@n_threads) - units.each_slice(Math.max(units.size // @n_threads, 1)) do |slice| - jobs_count += 1 + workers.each do |pid, input, output| spawn do - # For stats output we want to count how many previous - # .o files were reused, mainly to detect performance regressions. - # Because we fork, we must communicate using a pipe. - reused = [] of String - if wants_stats_or_progress - pr, pw = IO.pipe - spawn do - pr.each_line do |line| - unit = JSON.parse(line) - reused << unit["name"].as_s if unit["reused"].as_bool - @progress_tracker.stage_progress += 1 - end + overqueue.times do + if (index = indexes.add(1)) < units.size + input.puts index end end - codegen_process = Crystal::System::Process.fork do - pipe_w = pw - slice.each do |unit| - unit.compile - if pipe_w - unit_json = {name: unit.name, reused: unit.reused_previous_compilation?}.to_json - pipe_w.puts unit_json - end + while (index = indexes.add(1)) < units.size + input.puts index + + if response = output.gets(chomp: true) + channel.send response end end - Process.new(codegen_process).wait - if pipe_w = pw - pipe_w.close - Fiber.yield + overqueue.times do + if response = output.gets(chomp: true) + channel.send response + end end - wait_channel.send reused + input << '\n' + input.close + output.close + + Process.new(pid).wait + completed.send(nil) end end - jobs_count.times do - reused = wait_channel.receive - all_reused.concat(reused) + spawn do + @n_threads.times { completed.receive } + channel.close + end + + while response = channel.receive? + next unless wants_stats_or_progress + + result = JSON.parse(response) + all_reused << result["name"].as_s if result["reused"].as_bool + @progress_tracker.stage_progress += 1 end all_reused {% end %} end + private def fork_workers + workers = [] of {Int32, IO::FileDescriptor, IO::FileDescriptor} + + @n_threads.times do + iread, iwrite = IO.pipe + oread, owrite = IO.pipe + + iwrite.flush_on_newline = true + owrite.flush_on_newline = true + + pid = Crystal::System::Process.fork do + iwrite.close + oread.close + + yield iread, owrite + + iread.close + owrite.close + exit 0 + end + + iread.close + owrite.close + + workers << {pid, iwrite, oread} + end + + workers + end + private def print_macro_run_stats(program) return unless @progress_tracker.stats? return if program.compiled_macros_cache.empty? From 2f1af8841e53ffe4d7d80954b3ea6a4e95f5f435 Mon Sep 17 00:00:00 2001 From: Julien Portalier Date: Tue, 30 Jan 2024 10:33:24 +0100 Subject: [PATCH 2/4] Fix: Crystal::System::Process.fork exists on windows The method exists but raises NotImplementedError which causes Process.new(pid) to fail because it doesn't exist with a `PidT`). This patch fixes the issue by checking LibC.fork directly. Weirdly, both were used before these changes and compilation did succeed. --- src/compiler/crystal/compiler.cr | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/compiler/crystal/compiler.cr b/src/compiler/crystal/compiler.cr index f6c8273a5cee..5aca867e3a2f 100644 --- a/src/compiler/crystal/compiler.cr +++ b/src/compiler/crystal/compiler.cr @@ -546,7 +546,7 @@ module Crystal return all_reused end - {% if !Crystal::System::Process.class.has_method?("fork") %} + {% if !LibC.has_method?("fork") %} raise "Cannot fork compiler. `Crystal::System::Process.fork` is not implemented on this system." {% elsif flag?(:preview_mt) %} raise "Cannot fork compiler in multithread mode" From 434edae499964f3b430645b0672190b5c94dfa3a Mon Sep 17 00:00:00 2001 From: Julien Portalier Date: Tue, 13 Feb 2024 13:44:06 +0100 Subject: [PATCH 3/4] Fix: expect N responses from forked processes --- src/compiler/crystal/compiler.cr | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/src/compiler/crystal/compiler.cr b/src/compiler/crystal/compiler.cr index 5aca867e3a2f..3921bf5ea937 100644 --- a/src/compiler/crystal/compiler.cr +++ b/src/compiler/crystal/compiler.cr @@ -567,24 +567,25 @@ module Crystal workers.each do |pid, input, output| spawn do + overqueued = 0 + overqueue.times do if (index = indexes.add(1)) < units.size input.puts index + overqueued += 1 end end while (index = indexes.add(1)) < units.size input.puts index - if response = output.gets(chomp: true) - channel.send response - end + response = output.gets(chomp: true).not_nil! + channel.send response end - overqueue.times do - if response = output.gets(chomp: true) - channel.send response - end + overqueued.times do + response = output.gets(chomp: true).not_nil! + channel.send response end input << '\n' From 73fb8fd1e907403aa0a6fd27e4ecda4c11ddadaa Mon Sep 17 00:00:00 2001 From: Julien Portalier Date: Tue, 5 Mar 2024 11:46:38 +0100 Subject: [PATCH 4/4] Fix: don't fork more processes than compilation units --- src/compiler/crystal/compiler.cr | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/compiler/crystal/compiler.cr b/src/compiler/crystal/compiler.cr index 3921bf5ea937..ddcd9e205c40 100644 --- a/src/compiler/crystal/compiler.cr +++ b/src/compiler/crystal/compiler.cr @@ -535,10 +535,13 @@ module Crystal wants_stats_or_progress = @progress_tracker.stats? || @progress_tracker.progress? - # If threads is 1 and no stats/progress is needed we can avoid - # fork/spawn/channels altogether. This is particularly useful for - # CI because there forking eventually leads to "out of memory" errors. - if @n_threads == 1 + # Don't start more processes than compilation units + n_threads = @n_threads.clamp(1..units.size) + + # If threads is 1 we can avoid fork/spawn/channels altogether. This is + # particularly useful for CI because there forking eventually leads to + # "out of memory" errors. + if n_threads == 1 units.each do |unit| unit.compile all_reused << unit.name if wants_stats_or_progress && unit.reused_previous_compilation? @@ -551,7 +554,7 @@ module Crystal {% elsif flag?(:preview_mt) %} raise "Cannot fork compiler in multithread mode" {% else %} - workers = fork_workers do |input, output| + workers = fork_workers(n_threads) do |input, output| while i = input.gets(chomp: true).presence unit = units[i.to_i] unit.compile @@ -562,8 +565,8 @@ module Crystal overqueue = 1 indexes = Atomic(Int32).new(0) - channel = Channel(String).new(@n_threads) - completed = Channel(Nil).new(@n_threads) + channel = Channel(String).new(n_threads) + completed = Channel(Nil).new(n_threads) workers.each do |pid, input, output| spawn do @@ -598,7 +601,7 @@ module Crystal end spawn do - @n_threads.times { completed.receive } + n_threads.times { completed.receive } channel.close end @@ -614,10 +617,10 @@ module Crystal {% end %} end - private def fork_workers + private def fork_workers(n_threads) workers = [] of {Int32, IO::FileDescriptor, IO::FileDescriptor} - @n_threads.times do + n_threads.times do iread, iwrite = IO.pipe oread, owrite = IO.pipe