Module: Polyrun::WorkerOutput

Defined in:
lib/polyrun/worker_output.rb,
lib/polyrun/worker_output_forwarders.rb,
sig/polyrun/worker_output.rbs

Overview

Routes parallel worker stdout/stderr through per-shard log files with optional prefixed TTY echo.

Opt-in: POLYRUN_WORKER_OUTPUT_ROUTING=1 (or set POLYRUN_WORKER_LOG_DIR). POLYRUN_WORKER_OUTPUT_ROUTING=0 keeps inherited stdio (default polyrun behavior). POLYRUN_WORKER_OUTPUT_PREFIX=0 writes logs only (no live prefixed echo).

Defined Under Namespace

Classes: LineForwarder, WorkerForwarder

Class Method Summary collapse

Class Method Details

.default_log_directory ⇒ Object 



111
112
113
 
# File 'lib/polyrun/worker_output.rb', line 111

def default_log_directory
  "tmp/polyrun/workers"
end

.disabled_by_env? ⇒ Boolean

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/polyrun/worker_output.rb', line 115

def disabled_by_env?
  %w[0 false no off].include?(ENV["POLYRUN_WORKER_OUTPUT_ROUTING"].to_s.strip.downcase)
end

.finish_worker(pid) ⇒ void

This method returns an undefined value.

Parameters:

  • pid (Integer) 


99
100
101
102
103
104
 
# File 'lib/polyrun/worker_output.rb', line 99

def finish_worker(pid)
  entry = registry.delete(pid)
  return unless entry

  drain_forwarders(entry)
end

.log_directory ⇒ Object 



19
20
21
22
23
24
25
 
# File 'lib/polyrun/worker_output.rb', line 19

def log_directory
  value = ENV["POLYRUN_WORKER_LOG_DIR"]
  return nil if value.nil?

  stripped = value.to_s.strip
  stripped.empty? ? nil : stripped
end

.log_path_for(shard) ⇒ String

Parameters:

  • shard (Integer, String)

Returns:

  • (String) 


33
34
35
 
# File 'lib/polyrun/worker_output.rb', line 33

def log_path_for(shard)
  File.expand_path("shard-#{shard}.log", log_directory || default_log_directory)
end

.prefix_live? ⇒ Boolean

Returns:

  • (Boolean) 


27
28
29
30
31
 
# File 'lib/polyrun/worker_output.rb', line 27

def prefix_live?
  return true if ENV["POLYRUN_WORKER_OUTPUT_PREFIX"].nil?

  truthy?(ENV["POLYRUN_WORKER_OUTPUT_PREFIX"])
end

.prepare_log_dir! ⇒ void

This method returns an undefined value.



41
42
43
44
45
 
# File 'lib/polyrun/worker_output.rb', line 41

def prepare_log_dir!
  directory = log_directory || default_log_directory
  FileUtils.mkdir_p(directory)
  Dir.glob(File.join(directory, "shard-*.log")).each { |path| File.delete(path) }
end

.routing_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


13
14
15
16
17
 
# File 'lib/polyrun/worker_output.rb', line 13

def routing_enabled?
  return false if disabled_by_env?

  truthy?(ENV["POLYRUN_WORKER_OUTPUT_ROUTING"]) || !log_directory.nil?
end

.shutdown_all! ⇒ void

This method returns an undefined value.



106
107
108
109
 
# File 'lib/polyrun/worker_output.rb', line 106

def shutdown_all!
  registry.each_value { |entry| drain_forwarders(entry) }
  registry.clear
end

.spawn_worker(child_env, cmd, paths, hook_cfg) ⇒ Integer

Parameters:

  • child_env (Hash[Symbol, untyped])
  • cmd (Array[String])
  • paths (Array[String])
  • hook_cfg (Object)

Returns:

  • (Integer) 


53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/polyrun/worker_output.rb', line 53

def spawn_worker(child_env, cmd, paths, hook_cfg)
  shard = child_env.fetch("POLYRUN_SHARD_INDEX", "?")
  out_read, out_write = IO.pipe
  err_read, err_write = IO.pipe

  pid =
    if hook_cfg.worker_hooks? && !Polyrun::Hooks.disabled?
      Process.spawn(
        child_env,
        "sh",
        "-c",
        hook_cfg.build_worker_shell_script(cmd, paths),
        out: out_write,
        err: err_write
      )
    else
      Process.spawn(child_env, *cmd, *paths, out: out_write, err: err_write)
    end

  out_write.close
  err_write.close

  start_forwarders(
    pid: pid,
    shard: shard,
    stdout_io: out_read,
    stderr_io: err_read,
    log_path: log_path_for(shard)
  )
  pid
end

.start_forwarders(pid:, shard:, stdout_io:, stderr_io:, log_path:) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/polyrun/worker_output.rb', line 85

def start_forwarders(pid:, shard:, stdout_io:, stderr_io:, log_path:)
  forwarder = WorkerForwarder.new(
    shard: shard,
    pid: pid,
    log_path: log_path,
    prefix_live: prefix_live?
  )
  threads = [
    forwarder_thread(stdout_io, forwarder, :stdout),
    forwarder_thread(stderr_io, forwarder, :stderr)
  ]
  registry[pid] = {threads: threads, ios: [stdout_io, stderr_io], forwarder: forwarder}
end

.warn_shard_log(shard) ⇒ void

This method returns an undefined value.

Parameters:

  • shard (Integer, String) 


47
48
49
50
51
 
# File 'lib/polyrun/worker_output.rb', line 47

def warn_shard_log(shard)
  return unless routing_enabled?

  Polyrun::Log.warn "polyrun run-shards: shard #{shard} worker log → #{log_path_for(shard)}"
end

.worker_log_directory_label ⇒ Object 



37
38
39
 
# File 'lib/polyrun/worker_output.rb', line 37

def worker_log_directory_label
  log_directory || default_log_directory
end