Class: RightScale::RightPopen::ProcessBase

Inherits:
Object
  • Object
show all
Defined in:
lib/right_popen/process_base.rb

Direct Known Subclasses

Process

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ ProcessBase

Parameters

Parameters:

  • options (Hash) (defaults to: {}) —

    see RightScale.popen3_async for details



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/right_popen/process_base.rb', line 38

def initialize(options={})
  @options = options
  @stdin = nil
  @stdout = nil
  @stderr = nil
  @status_fd = nil
  @last_interrupt = nil
  @pid = nil
  @start_time = nil
  @stop_time = nil
  @watch_directory = nil
  @size_limit_bytes = nil
  @cmd = nil
  @target = nil
  @status = nil
  @channels_to_finish = nil
  @needs_watching = !!(
    @options[:timeout_seconds] ||
    @options[:size_limit_bytes] ||
    @options[:watch_handler])
end

Instance Attribute Details

#channels_to_finish ⇒ Object (readonly)

Returns the value of attribute channels_to_finish.



34
35
36
 
# File 'lib/right_popen/process_base.rb', line 34

def channels_to_finish
  @channels_to_finish
end

#pid ⇒ Object (readonly)

Returns the value of attribute pid.



33
34
35
 
# File 'lib/right_popen/process_base.rb', line 33

def pid
  @pid
end

#start_time ⇒ Object (readonly)

Returns the value of attribute start_time.



34
35
36
 
# File 'lib/right_popen/process_base.rb', line 34

def start_time
  @start_time
end

#status ⇒ Object (readonly)

Returns the value of attribute status.



33
34
35
 
# File 'lib/right_popen/process_base.rb', line 33

def status
  @status
end

#status_fd ⇒ Object (readonly)

Returns the value of attribute status_fd.



33
34
35
 
# File 'lib/right_popen/process_base.rb', line 33

def status_fd
  @status_fd
end

#stderr ⇒ Object (readonly)

Returns the value of attribute stderr.



33
34
35
 
# File 'lib/right_popen/process_base.rb', line 33

def stderr
  @stderr
end

#stdin ⇒ Object (readonly)

Returns the value of attribute stdin.



33
34
35
 
# File 'lib/right_popen/process_base.rb', line 33

def stdin
  @stdin
end

#stdout ⇒ Object (readonly)

Returns the value of attribute stdout.



33
34
35
 
# File 'lib/right_popen/process_base.rb', line 33

def stdout
  @stdout
end

#stop_time ⇒ Object (readonly)

Returns the value of attribute stop_time.



34
35
36
 
# File 'lib/right_popen/process_base.rb', line 34

def stop_time
  @stop_time
end

Instance Method Details

#alive? ⇒ TrueClass|FalseClass

Determines if the process is still running.

Return

Returns:

  • (TrueClass|FalseClass) —

    true if running

Raises:

  • (NotImplementedError) 


64
65
66
 
# File 'lib/right_popen/process_base.rb', line 64

def alive?
  raise NotImplementedError, 'Must be overridden'
end

#drain_all_upon_death? ⇒ TrueClass|FalseClass

Determines whether or not to drain all open streams upon death of child or else only those where IO.select indicates data available. This decision is platform-specific.

Return

Returns:

  • (TrueClass|FalseClass) —

    true if draining all

Raises:

  • (NotImplementedError) 


74
75
76
 
# File 'lib/right_popen/process_base.rb', line 74

def drain_all_upon_death?
  raise NotImplementedError, 'Must be overridden'
end

#interrupt ⇒ TrueClass|FalseClass

Interrupts the running process (without abandoning watch) in increasing degrees of signalled severity.

Return

Returns:

  • (TrueClass|FalseClass) —

    true if process was alive and interrupted, false if dead before (first) interrupt



287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
 
# File 'lib/right_popen/process_base.rb', line 287

def interrupt
  while alive?
    if !@kill_time || Time.now >= @kill_time
      # soft then hard interrupt (assumed to be called periodically until
      # process is gone).
      sigs = signals_for_interrupt
      if @last_interrupt
        last_index = sigs.index(@last_interrupt)
        next_interrupt = sigs[last_index + 1]
      else
        next_interrupt = sigs.first
      end
      unless next_interrupt
        raise ::RightScale::RightPopen::ProcessError,
              'Unable to kill child process'
      end
      @last_interrupt = next_interrupt

      # kill
      result = ::Process.kill(next_interrupt, @pid) rescue nil
      if result
        @kill_time = Time.now + 3 # more seconds until next attempt
        break
      end
    end
  end
  interrupted?
end

#interrupted? ⇒ TrueClass|FalseClass

Returns interrupted as true if child process was interrupted by watcher.

Returns:

  • (TrueClass|FalseClass) —

    interrupted as true if child process was interrupted by watcher



113
 
# File 'lib/right_popen/process_base.rb', line 113

def interrupted?; !!@last_interrupt; end

#needs_watching? ⇒ TrueClass|FalseClass

Determines if this process needs to be watched (beyond waiting for the process to exit).

Return

Returns:

  • (TrueClass|FalseClass) —

    true if needs watching



83
 
# File 'lib/right_popen/process_base.rb', line 83

def needs_watching?; @needs_watching; end

#safe_close_io ⇒ TrueClass

Safely closes any open I/O objects associated with this process.

Return

Returns:

  • (TrueClass) —

    alway true



320
321
322
323
324
325
326
 
# File 'lib/right_popen/process_base.rb', line 320

def safe_close_io
  @stdin.close rescue nil if @stdin && !@stdin.closed?
  @stdout.close rescue nil if @stdout && !@stdout.closed?
  @stderr.close rescue nil if @stderr && !@stderr.closed?
  @status_fd.close rescue nil if @status_fd && !@status_fd.closed?
  true
end

#signals_for_interrupt ⇒ Array

Returns escalating termination signals for this platform.

Returns:

  • (Array) —

    escalating termination signals for this platform

Raises:

  • (NotImplementedError) 


278
279
280
 
# File 'lib/right_popen/process_base.rb', line 278

def signals_for_interrupt
  raise NotImplementedError, 'Must be overridden'
end

#size_limit_exceeded? ⇒ TrueClass|FalseClass

Determines if total size of files created by child process has exceeded the limit specified, if any.

Return

Returns:

  • (TrueClass|FalseClass) —

    true if size limit exceeded



98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/right_popen/process_base.rb', line 98

def size_limit_exceeded?
  if @watch_directory
    globbie = ::File.join(@watch_directory, '**/*')
    size = 0
    ::Dir.glob(globbie) do |f|
      size += ::File.stat(f).size rescue 0 if ::File.file?(f)
      break if size > @size_limit_bytes
    end
    size > @size_limit_bytes
  else
    false
  end
end

#spawn(cmd, target) ⇒ TrueClass

Spawns a child process using given command and handler target in a platform-independant manner.

must be overridden and override must call super.

Parameters

Return

Parameters:

  • cmd (String|Array) —

    as shell command or binary to execute

  • target (Object) —

    that implements all handlers (see TargetProxy)

Returns:

  • (TrueClass) —

    always true



142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/right_popen/process_base.rb', line 142

def spawn(cmd, target)
  @cmd = cmd
  @target = target
  @kill_time = nil
  @pid = nil
  @status = nil
  @last_interrupt = nil
  @channels_to_finish = nil
  @wait_thread = nil

  if @size_limit_bytes = @options[:size_limit_bytes]
    @watch_directory = @options[:watch_directory] || @options[:directory] || ::Dir.pwd
  end
end

#sync_all(cmd, target) ⇒ TrueClass

Performs all process operations in synchronous fashion. It is possible for errors or callback behavior to conditionally short-circuit the synchronous operations.

Parameters

Return

Parameters:

  • cmd (String|Array) —

    as shell command or binary to execute

  • target (Object) —

    that implements all handlers (see TargetProxy)

Returns:

  • (TrueClass) —

    always true



125
126
127
128
129
 
# File 'lib/right_popen/process_base.rb', line 125

def sync_all(cmd, target)
  spawn(cmd, target)
  sync_exit_with_target if sync_pid_with_target
  true
end

#sync_exit_with_target ⇒ TrueClass

Monitors I/O from child process and directly notifies target of any events. Blocks until child exits.

Return

Returns:

  • (TrueClass) —

    always true



208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
 
# File 'lib/right_popen/process_base.rb', line 208

def sync_exit_with_target
  abandon = false
  status_fd_data = []
  begin
    while true
      channels_to_watch = @channels_to_finish.map { |ctf| ctf.last }
      ready = ::IO.select(channels_to_watch, nil, nil, 0.1) rescue nil
      dead = !alive?
      channels_to_read = ready && ready.first
      if dead && drain_all_upon_death?
        # finish reading all dead channels.
        channels_to_read = @channels_to_finish.map { |ctf| ctf.last }
      end
      if channels_to_read
        channels_to_read.each do |channel|
          index = @channels_to_finish.index { |ctf| ctf.last == channel }
          key = @channels_to_finish[index].first
          data = dead ? channel.gets(nil) : channel.gets
          if data
            if key == :status_fd
              status_fd_data << data
            else
              @target.method(key).call(data)
            end
          else
            # nothing on channel indicates EOF
            @channels_to_finish.delete_at(index)
          end
        end
      end
      if dead
        break
      elsif (interrupted? || timer_expired? || size_limit_exceeded?)
        interrupt
      elsif abandon = !@target.watch_handler(self)
        return true  # bypass any remaining callbacks
      end
    end
    wait_for_exit_status
    unless status_fd_data.empty?
      data = status_fd_data.join
      error_data = ::YAML.load(data)
      status_fd_error = ::RightScale::RightPopen::ProcessError.new(
        "#{error_data['class']}: #{error_data['message']}")
      if error_data['backtrace']
        status_fd_error.set_backtrace(error_data['backtrace'])
      end
      raise status_fd_error
    end
    @target.timeout_handler if timer_expired?
    @target.size_limit_handler if size_limit_exceeded?
    @target.exit_handler(@status)
  ensure
    # abandon will not close I/O objects; caller takes responsibility via
    # process object passed to watch_handler. if anyone calls interrupt
    # then close I/O regardless of abandon to try to force child to die.
    safe_close_io if !abandon || interrupted?
  end
  true
end

#sync_pid_with_target ⇒ TrueClass|FalseClass

Performs initial handler callbacks before consuming I/O. Represents any code that must not be invoked twice (unlike sync_exit_with_target).

Return

Returns:

  • (TrueClass|FalseClass) —

    true to begin watch, false to abandon



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/right_popen/process_base.rb', line 162

def sync_pid_with_target
  # early handling in case caller wants to stream to/from the pipes
  # directly (as in a classic popen3/4 scenario).
  @target.pid_handler(@pid)
  if input_text = @options[:input]
    @stdin.write(input_text)
  end

  # one-time initialization of the stateful channels_to_finish hash to
  # allow for multiple invocations of the sync_exit_with_target with a
  # possible abandon in between.
  #
  # note that calling IO.select on pipes which have already had all
  # of their output consumed can cause segfault (in Ubuntu?) so it is
  # important to keep track of when all I/O has been consumed.
  @channels_to_finish = [
    [:stdout_handler, @stdout],
    [:stderr_handler, @stderr],
  ]
  @channels_to_finish << [:status_fd, @status_fd] if @status_fd

  # sync watch_handler has the option to abandon watch as soon as child
  # process comes alive and before streaming any output.
  if @target.watch_handler(self)
    # can close stdin if not returning control to caller.
    @stdin.close rescue nil
    return true
  else
    # caller is reponsible for draining and/or closing all pipes. this can
    # be accomplished by explicity calling either sync_exit_with_target or
    # safe_close_io plus wait_for_exit_status (if data has been read from
    # the I/O streams). it is unsafe to read some data and then call
    # sync_exit_with_target because IO.select may segfault if all of the
    # data in a stream has been already consumed.
    return false
  end
rescue
  safe_close_io
  raise
end

#timer_expired? ⇒ TrueClass|FalseClass

Determines if timeout on child process has expired, if any.

Return

Returns:

  • (TrueClass|FalseClass) —

    true if timer expired



89
90
91
 
# File 'lib/right_popen/process_base.rb', line 89

def timer_expired?
  !!(@stop_time && Time.now >= @stop_time)
end

#wait_for_exit_status ⇒ ProcessStatus

blocks waiting for process exit status.

Return

Returns:

Raises:

  • (NotImplementedError) 


273
274
275
 
# File 'lib/right_popen/process_base.rb', line 273

def wait_for_exit_status
  raise NotImplementedError, 'Must be overridden'
end