Class: Pikuri::Subprocess

Inherits:
Object
  • Object
show all
Defined in:
lib/pikuri/subprocess.rb

Overview

Chokepoint for all subprocess spawning in pikuri. Forces a new process group per invocation and tracks pgids so descendants (commands backgrounded with &) can be cleaned up at exit. Two front doors: Subprocess.spawn (combined stdout+stderr through one pipe — the shell-command shape) and Subprocess.run (stdin from a String/IO, stdout to a file, stderr captured — the filter shape).

Seam discipline

All subprocess spawning in lib/ goes through Subprocess.spawn/Subprocess.run; direct Process.spawn / Open3.* / system / backticks are bugs. Grep-enforceable: grep -rnE 'Process\.spawn|Open3\.|\bsystem\(' lib/ should hit only this file (plus the MCP-exception comment in pikuri-mcp/lib/pikuri/mcp/servers.rb).

Timeouts are the caller's job

Subprocess.spawn implements no timeout — Ruby's Timeout.timeout can't kill subprocesses cleanly. Callers wrap their argv with coreutils' timeout:

Pikuri::Subprocess.spawn(
'timeout', '--signal=TERM', '--kill-after=5s', '120s',
'bash', '-c', command, chdir: workspace.cwd.to_s)

When timeout and its FD-inheriting children die, the output pipe closes and #wait's io.read returns — timeout handles the SIGTERM-then-SIGKILL race-free, no Ruby-side machinery.

Backgrounded subprocesses

A command that backgrounds work with & leaves a process in our pgroup. #wait returns when the direct child exits, but Subprocess.active keeps the pgid tracked while any group member is alive (+kill(0, -pgid)+); Subprocess.cleanup! SIGTERMs every tracked group at exit. The model opts out with nohup / setsid (both detach from our group).

State is process-global

One @active Set for the process, swept once at exit via Finalizers. A Mutex guards register/prune/cleanup.

Defined Under Namespace

Classes: Result

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(io:, wait_thr:) ⇒ Subprocess

Returns a new instance of Subprocess.



141
142
143
144
145
146
 
# File 'lib/pikuri/subprocess.rb', line 141

def initialize(io:, wait_thr:)
  @io       = io
  @wait_thr = wait_thr
  @pid      = wait_thr.pid
  @pgid     = wait_thr.pid # pgroup:true → pgid == pid
end

Instance Attribute Details

#io ⇒ IO (readonly)

Returns read end of the combined stdout+stderr pipe. Callers normally go straight to #wait, which drains it.

Returns:

  • (IO) —

    read end of the combined stdout+stderr pipe. Callers normally go straight to #wait, which drains it.



138
139
140
 
# File 'lib/pikuri/subprocess.rb', line 138

def io
  @io
end

#pgid ⇒ Integer (readonly)

Returns process group id. Equal to #pid since the child was spawned with pgroup: true (it's the group leader).

Returns:

  • (Integer) —

    process group id. Equal to #pid since the child was spawned with pgroup: true (it's the group leader).



134
135
136
 
# File 'lib/pikuri/subprocess.rb', line 134

def pgid
  @pgid
end

#pid ⇒ Integer (readonly)

Returns direct child's pid.

Returns:

  • (Integer) —

    direct child's pid



130
131
132
 
# File 'lib/pikuri/subprocess.rb', line 130

def pid
  @pid
end

Class Method Details

.active ⇒ Array<Integer>

Currently-tracked process groups, with dead ones pruned as a side effect.

Returns:

  • (Array<Integer>) 


181
182
183
184
185
186
 
# File 'lib/pikuri/subprocess.rb', line 181

def active
  @mutex.synchronize do
    @active.delete_if { |g| !alive?(g) }
    @active.to_a
  end
end

.cleanup! ⇒ void

This method returns an undefined value.

SIGTERM every tracked process group. Run at process exit via Finalizers, and from specs' after blocks. Best-effort.



192
193
194
195
196
197
 
# File 'lib/pikuri/subprocess.rb', line 192

def cleanup!
  @mutex.synchronize do
    @active.each { |g| Process.kill('-TERM', g) rescue nil }
    @active.clear
  end
end

.run(*argv, stdin_data:, stdout:, chdir:, env: {}) ⇒ Result

Run argv as a one-shot filter: feed it stdin_data, redirect stdout to stdout (an open File), capture stderr through a pipe, block until it exits. Built for the stdin→markdown converters (pikuri-extractors), where spawn's shape is wrong twice: it closes stdin immediately and merges stderr onto stdout — fatal when stdout is the payload.

Redirecting stdout to a file (not a pipe) is what makes the I/O deadlock-free with one writer thread: the child never blocks on output, so it keeps draining stdin while the parent drains the low-volume stderr pipe. Pikuri::Subprocess::Result#output is the captured stderr, not the payload — the payload is in stdout, whose file offset is shared with the child, so rewind before reading it back. Same discipline as spawn (new group, exit-sweep tracked, no built-in timeout).

Parameters:

  • argv (Array<String>) —

    command + arguments; no implicit shell.

  • stdin_data (String, IO, StringIO) —

    the child's stdin: a String is written as-is, an IO is streamed via IO.copy_stream from its current position (so a large file never hits the Ruby heap). Closed (EOF) after. May be empty.

  • stdout (File) —

    open writable file stdout is redirected to.

  • chdir (String, Pathname) —

    working directory.

  • env (Hash{String=>String}) (defaults to: {}) —

    extra env vars, as for spawn.

Returns:

  • (Result) —

    output is the captured stderr; status the exit status.

Raises:

  • (SystemCallError) —

    whatever an IO stdin_data raises mid-stream, re-raised after the child is reaped.



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/pikuri/subprocess.rb', line 97

def self.run(*argv, stdin_data:, stdout:, chdir:, env: {})
  in_r, in_w = IO.pipe
  err_r, err_w = IO.pipe
  pid = Process.spawn(env, *argv, chdir: chdir.to_s, pgroup: true,
                      in: in_r, out: stdout, err: err_w)
  in_r.close
  err_w.close
  register(pid)
  writer = Thread.new do
    in_w.binmode
    if stdin_data.respond_to?(:read)
      IO.copy_stream(stdin_data, in_w)
    else
      in_w.write(stdin_data)
    end
  rescue Errno::EPIPE
    nil # child exited without draining stdin; its status tells the story
  ensure
    in_w.close
  end
  stderr = err_r.read
  err_r.close
  # Reap before joining the writer: if an IO source raised
  # mid-stream, #join re-raises it, and the child (already exited —
  # err_r hit EOF) must not be left a zombie.
  _, status = Process.waitpid2(pid)
  writer.join
  Result.new(output: stderr, status: status)
ensure
  prune(pid) if pid
end

.spawn(*argv, chdir:, env: {}) ⇒ Subprocess

Spawn argv in a new process group with stderr on stdout. Tracked for cleanup.

Parameters:

  • argv (Array<String>) —

    command + arguments, passed to exec directly — no implicit shell (caller wraps in 'bash', '-c', cmd when shell interpretation is wanted).

  • chdir (String, Pathname) —

    working directory

  • env (Hash{String=>(String,nil)}) (defaults to: {}) —

    extra env vars for the child; it otherwise inherits the parent's full environment (default {}, pure inheritance). A nil value unsets that variable, which is how BundlerEnv hands back the environment pikuri's own boot rewrote.

Returns:



64
65
66
67
68
69
 
# File 'lib/pikuri/subprocess.rb', line 64

def self.spawn(*argv, chdir:, env: {})
  stdin, io, wait_thr = Open3.popen2e(env, *argv, chdir: chdir.to_s, pgroup: true)
  stdin.close
  register(wait_thr.pid)
  new(io: io, wait_thr: wait_thr)
end

Instance Method Details

#terminate ⇒ void

This method returns an undefined value.

SIGTERM the whole process group without blocking for output — the stop button for a daemon child (one the caller never #waits on, e.g. Memory::Mem0Server's socat relay). Best-effort and idempotent; the group stays in the exit-sweep set until it dies, so a child ignoring SIGTERM is re-signalled by cleanup! at exit.



168
169
170
171
172
173
174
 
# File 'lib/pikuri/subprocess.rb', line 168

def terminate
  Process.kill('-TERM', @pgid)
rescue Errno::ESRCH
  # already gone
ensure
  self.class.send(:prune, @pgid)
end

#wait ⇒ Result

Block until the direct child exits, drain the combined-output pipe, return a Result. The pgid stays tracked if the group still has live members (backgrounded children); pruned if everything's gone.

Returns:



153
154
155
156
157
158
159
 
# File 'lib/pikuri/subprocess.rb', line 153

def wait
  output = @io.read
  @io.close
  Result.new(output: output, status: @wait_thr.value)
ensure
  self.class.send(:prune, @pgid)
end