Class: ProcessExecuter::MonitoredPipe

Inherits:
Object
  • Object
show all
Includes:
TrackOpenInstances
Defined in:
lib/process_executer/monitored_pipe.rb

Overview

Acts as a pipe that writes the data written to it to one or more destinations

MonitoredPipe was created to expand the output redirection options for Process.spawn and methods derived from it within the ProcessExecuter module.

This class's initializer accepts any redirection destination supported by Process.spawn (this is the value part of the file redirection option described in the File Redirection section of Process.spawn.

In addition to the standard redirection destinations, MonitoredPipe also supports these additional types of destinations:

  • Arbitrary Writers

You can redirect subprocess output to any Ruby object that implements the #write method. This is particularly useful for:

- capturing command output in in-memory buffers like `StringIO`
- sending command output to custom logging objects that do not have a file
  descriptor
- processing with a streaming parser to parse and process command output as
  the command is running
  • Multiple Destinations

MonitoredPipe supports duplicating (or "teeing") output to multiple destinations simultaneously. This is achieved by providing a redirection destination in the form [:tee, destination1, destination2, ...], where each destination can be any value that MonitoredPipe itself supports (including another tee or MonitoredPipe).

When a new MonitoredPipe is created, a pipe is created (via IO.pipe) and a thread is created to read data written to the pipe. As data is read from the pipe, it is written to the destination provided in the MonitoredPipe initializer.

If the destination raises an exception (of any class, not just StandardError), the monitoring thread will exit, the pipe will be closed, and the exception will be saved in #exception.

⚠️ WARNING

#close must be called to ensure that (1) the pipe is closed, (2) all data is read from the pipe and written to the destination, and (3) the monitoring thread is killed.

File descriptor usage

Each MonitoredPipe holds four file descriptors: two for the data pipe the subprocess writes to and two for an internal wake pipe that interrupts the monitoring thread when the pipe is closed. All four are held concurrently from construction until the pipe is closed, so a ProcessExecuter.run that captures stdout and stderr holds 8 pipe file descriptors for the duration of the subprocess.

File descriptor limits are per-process. An application spawning many commands concurrently can hit the soft RLIMIT_NOFILE limit -- macOS defaults to 256 and Linux commonly to 1024. The workaround is to raise the limit in the spawning process, for example with Process.setrlimit(:NOFILE, 10_240) or ulimit -n.

Examples:

Collect pipe data into a StringIO object

pipe_data = StringIO.new
begin
  pipe = ProcessExecuter::MonitoredPipe.new(pipe_data)
  pipe.write("Hello World")
ensure
  pipe.close
end
pipe_data.string #=> "Hello World"

Collect pipe data into a string AND a file

pipe_data_string = StringIO.new
pipe_data_file = File.open("pipe_data.txt", "w")
begin
  pipe = ProcessExecuter::MonitoredPipe.new([:tee, pipe_data_string, pipe_data_file])
  pipe.write("Hello World")
ensure
  pipe.close
end
pipe_data_string.string #=> "Hello World"
# It is your responsibility to close the file you opened
pipe_data_file.close
File.read("pipe_data.txt") #=> "Hello World"

Using a MonitoredPipe with Process.spawn

stdout_buffer = StringIO.new
begin
  stdout_pipe = ProcessExecuter::MonitoredPipe.new(stdout_buffer)
  pid = Process.spawn('echo Hello World', out: stdout_pipe)
  _waited_pid, status = Process.wait2(pid)
ensure
  stdout_pipe.close
end
stdout_buffer.string #=> "Hello World\n"

Constant Summary collapse

DEFAULT_CLOSE_TIMEOUT =

The default number of seconds #close waits to drain remaining pipe data

Draining normally finishes in well under a second: it only has to read whatever is still buffered in the pipe once every copy of the pipe's write fd is closed. The timeout exists so that a write fd inherited by a process outside this object's control (such as an orphaned descendant of a killed subprocess) cannot make #close block indefinitely.

Returns:

  • (Numeric) 
10

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redirection_destination, chunk_size: 100_000) ⇒ MonitoredPipe

Create a new monitored pipe

Creates an IO.pipe and starts a monitoring thread to read data written to the pipe.

Examples:

redirection_destination = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(redirection_destination)

Parameters:

  • redirection_destination (Object) —

    as data is read from the pipe, it is written to this destination

    Accepts any redirection destination supported by Process.spawn. This is the value part of the file redirection option described in the File Redirection section of Process.spawn.

    In addition to the standard redirection destinations, MonitoredPipe also accepts (1) another monitored pipe, (2) any object that implements a #write method and (3) an array in the form [:tee, destination1, destination2, ...] where each destination can be any value that MonitoredPipe itself supports (including another tee or MonitoredPipe).

  • chunk_size (Integer) (defaults to: 100_000) —

    the size of the chunks to read from the pipe



150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/process_executer/monitored_pipe.rb', line 150

def initialize(redirection_destination, chunk_size: 100_000)
  @destination = Destinations.factory(redirection_destination)
  complete_initialization(chunk_size)
rescue Exception # rubocop:disable Lint/RescueException
  # The destination may hold resources (e.g. the File opened by
  # Destinations::FilePath), so a failure partway through construction must
  # close whatever was created so far -- the destination and, for each
  # IO.pipe that succeeded, both pipe IOs -- or they leak. A failed
  # initialize never returns a MonitoredPipe instance for the caller (or
  # #close) to clean up.
  [destination, pipe_reader, pipe_writer, wake_reader, wake_writer].each { |resource| resource&.close }
  raise
end

Instance Attribute Details

#chunk_size ⇒ Integer (readonly)

The size of the chunks to read from the pipe

Examples:

require 'stringio'
data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.chunk_size #=> 100_000

Returns:

  • (Integer) —

    the size of the chunks to read from the pipe



312
313
314
 
# File 'lib/process_executer/monitored_pipe.rb', line 312

def chunk_size
  @chunk_size
end

#destination ⇒ ProcessExecuter::Destinations::DestinationBase (readonly)

The redirection destination to write data that is read from the pipe

Examples:

require 'stringio'
data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.destination #=> #<ProcessExecuter::Destinations::Writer>

Returns:



326
327
328
 
# File 'lib/process_executer/monitored_pipe.rb', line 326

def destination
  @destination
end

#exception ⇒ Exception? (readonly)

The exception raised by a destination

If an exception is raised by a destination, it is stored here. Otherwise, it is nil.

Examples:

pipe.exception #=> nil

Returns:

  • (Exception, nil) —

    the exception raised by a destination or nil if no exception was raised



359
360
361
 
# File 'lib/process_executer/monitored_pipe.rb', line 359

def exception
  @exception
end

#fileno ⇒ Integer (readonly)

The file descriptor for the write end of the pipe

Examples:

require 'stringio'
data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.fileno == pipe.to_io.fileno #=> true

Returns:

  • (Integer) —

    the file descriptor for the write end of the pipe



247
248
249
 
# File 'lib/process_executer/monitored_pipe.rb', line 247

def fileno
  pipe_writer.fileno
end

#pipe_reader ⇒ IO (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The read end of the pipe

Examples:

pipe = ProcessExecuter::MonitoredPipe.new($stdout)
pipe.pipe_reader #=> #<IO:fd 11>

Returns:

  • (IO) 


408
409
410
 
# File 'lib/process_executer/monitored_pipe.rb', line 408

def pipe_reader
  @pipe_reader
end

#pipe_writer ⇒ IO (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The write end of the pipe

Examples:

pipe = ProcessExecuter::MonitoredPipe.new($stdout)
pipe.pipe_writer #=> #<IO:fd 12>

Returns:

  • (IO) —

    the write end of the pipe



422
423
424
 
# File 'lib/process_executer/monitored_pipe.rb', line 422

def pipe_writer
  @pipe_writer
end

#state ⇒ Symbol (readonly)

The state of the pipe

Must be either :open, :closing, or :closed

  • :open - the pipe is open and data can be written to it
  • :closing - the pipe is being closed and data can no longer be written to it
  • :closed - the pipe is closed and data can no longer be written to it

Examples:

pipe = ProcessExecuter::MonitoredPipe.new($stdout)
pipe.state #=> :open
pipe.close
pipe.state #=> :closed

Returns:

  • (Symbol) —

    the state of the pipe



346
347
348
 
# File 'lib/process_executer/monitored_pipe.rb', line 346

def state
  @state
end

#thread ⇒ Thread (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The thread that monitors the pipe

Examples:

require 'stringio'
data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.thread #=> #<Thread:0x00007f8b1a0b0e00>

Returns:

  • (Thread) 


394
395
396
 
# File 'lib/process_executer/monitored_pipe.rb', line 394

def thread
  @thread
end

#wake_reader ⇒ IO (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The read end of the internal wake pipe

The monitoring thread waits on this IO (along with #pipe_reader) so that #close can interrupt its wait for pipe data.

Returns:

  • (IO) 


435
436
437
 
# File 'lib/process_executer/monitored_pipe.rb', line 435

def wake_reader
  @wake_reader
end

#wake_writer ⇒ IO (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The write end of the internal wake pipe

#close writes a single byte to this IO -- at most once in the pipe's life -- to interrupt the monitoring thread's wait for pipe data.

Returns:

  • (IO) 


448
449
450
 
# File 'lib/process_executer/monitored_pipe.rb', line 448

def wake_writer
  @wake_writer
end

Instance Method Details

#close(timeout: DEFAULT_CLOSE_TIMEOUT)

This method returns an undefined value.

Set the state to :closing and wait for the state to be set to :closed

A byte written to the internal wake pipe interrupts the monitoring thread's wait for pipe data; the thread then sees that the state has changed and closes the pipe.

Remaining pipe data is drained to the destination for at most timeout seconds. The pipe only reaches EOF once every copy of its write fd is closed -- including copies inherited by processes outside this object's control -- so without a timeout this method could block indefinitely. When the timeout expires before EOF, the pipe is closed anyway, #truncated? returns true, and data still in the pipe is discarded.

The timeout is one absolute deadline for the whole drain: time spent writing to the destination counts against it too. Only the waits for pipe data or EOF are cut short when the deadline passes, though -- a destination #write already in progress is never interrupted, so a destination that blocks can still delay this method past the timeout.

An exception that escapes the monitoring thread's work is recorded in #exception by the monitoring thread itself before it terminates, so the Thread#join in this method never re-raises one. An exception raised at the join -- such as an Interrupt delivered to the calling thread -- is directed at the caller and propagates.

Examples:

data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.state #=> :open
pipe.write('Hello World')
pipe.close
pipe.state #=> :closed
data_collector.string #=> "Hello World"

Parameters:

  • timeout (Numeric, nil) (defaults to: DEFAULT_CLOSE_TIMEOUT) —

    the number of seconds to spend draining remaining pipe data to the destination before giving up, or nil to wait without a time limit. The deadline is absolute -- time in the destination's #write counts against it -- but a write in progress is never interrupted, so a blocking destination can overrun it.



206
207
208
209
210
211
212
 
# File 'lib/process_executer/monitored_pipe.rb', line 206

def close(timeout: DEFAULT_CLOSE_TIMEOUT)
  initiate_close_and_wait_until_closed(timeout)

  thread.join
  destination.close
  self.class.remove_open_instance(self)
end

#to_io ⇒ IO

Return the write end of the pipe so that data can be written to it

Data written to this end of the pipe will be read by the monitor thread and written to the destination.

This is so we can provide a MonitoredPipe to Process.spawn as a FD

Examples:

require 'stringio'
data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.to_io.write('Hello World')
pipe.close
data_collector.string #=> "Hello World"

Returns:

  • (IO) —

    the write end of the pipe



231
232
233
 
# File 'lib/process_executer/monitored_pipe.rb', line 231

def to_io
  pipe_writer
end

#truncated? ⇒ Boolean

Whether #close gave up draining the pipe before reaching EOF

true when the close timeout expired before the pipe reached EOF: some copy of the pipe's write fd was still open (for instance, held by an orphaned descendant of a killed subprocess) or unread data remained, and what was left was discarded instead of being written to the destination. An expired timeout on a pipe with nothing left to drain closes normally and stays false.

Examples:

data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.close
pipe.truncated? #=> false

Returns:

  • (Boolean) 


378
 
# File 'lib/process_executer/monitored_pipe.rb', line 378

def truncated? = @truncated

#write(data) ⇒ Integer

Writes data to the pipe so that it can be read by the monitor thread

Primarily used for testing.

Examples:

require 'stringio'
data_collector = StringIO.new
pipe = ProcessExecuter::MonitoredPipe.new(data_collector)
pipe.write('Hello World')
pipe.close
data_collector.string #=> "Hello World"

Parameters:

  • data (String) —

    the data to write to the pipe

Returns:

  • (Integer) —

    the number of bytes written to the pipe

Raises:

  • (IOError) —

    if the pipe is not open

    The pipe is only checked before the write begins. If the destination raises (or another thread calls #close) while a large write is still in progress, the monitoring thread closes the pipe and the in-progress write fails with an IOError too.



274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/process_executer/monitored_pipe.rb', line 274

def write(data)
  # The mutex is released before writing to the pipe. `pipe_writer.write`
  # blocks once the OS pipe buffer is full, and it can only be unblocked by
  # the monitoring thread draining the pipe. Holding the mutex across the
  # write would stop the monitoring thread from taking the mutex in its own
  # error path, deadlocking both threads.
  mutex.synchronize { raise IOError, 'closed stream' unless state == :open }

  pipe_writer.write(data)
rescue SystemCallError
  # Engines disagree about how a write that is already blocked reacts to the
  # monitoring thread closing the pipe. MRI and JRuby raise an IOError.
  # TruffleRuby lets the write continue and fail at the system call, and which
  # errno that is depends on the platform: EPIPE on macOS, EBADF on Linux.
  #
  # The monitoring thread is the only reader of this pipe, so any error from
  # the operating system means the same thing the IOError does: the pipe went
  # away mid-write. Report it as an IOError so #write has one documented
  # contract on every supported engine. The original error is still available
  # through Exception#cause.
  #
  # :nocov: only reached on engines that do not interrupt the blocked write
  raise IOError, 'closed stream'
  # :nocov:
end