Class: Roby::Tasks::ExternalProcess

Inherits:
Roby::Task show all
Defined in:
lib/roby/tasks/external_process.rb

Overview

This task class can be used to monitor the execution of an external process.

Importantly, the task by default is not interruptible, because there is no good common way to gracefully terminate an external program. To use the common way to stop the task with a signal, use ExternalProcess.interruptible_with_signal instead of new to create a task instance, which will set it up by default for you

Among the useful features, it can redirect standard output and error output to files.

The events will act as follows:

  • the start command starts the process per se. The event is emitted once the process has been spawned with success

  • the signaled event is emitted when the process dies because of a signal. The event’s context is the Process::Status object.

  • the failed event is emitted whenever the process exits with a nonzero status. The event’s context is the Process::Status object.

  • the success event is emitted when the process exits with a zero status

  • the stop event is emitted when the process exits, regardless of how

Direct Known Subclasses

InterruptibleWithSignal

Defined Under Namespace

Classes: InterruptibleWithSignal

Constant Summary

Constants included from Models::Arguments

Models::Arguments::NO_DEFAULT_ARGUMENT

Class Attribute Summary collapse

Instance Attribute Summary collapse

Attributes inherited from Roby::Task

#arguments, #bound_events, #data, #execute_handlers, #failed_to_start_time, #failure_event, #failure_reason, #history, #poll_handlers, #quarantine_reason, #state_machine, #terminal_event

Attributes included from GUI::RelationsCanvasTask

#displayed_state, #last_event

Attributes inherited from PlanObject

#addition_time, #executable, #execution_engine, #finalization_handlers, #finalization_time, #model, #plan, #promise_executor, #removed_at

Attributes included from Roby::Transaction::Proxying::Cache

#transaction_forwarder_module, #transaction_proxy_module

Attributes included from Relations::DirectedRelationSupport

#relation_graphs

Attributes inherited from DistributedObject

#local_owner_id, #owners

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Roby::Task

#+, #abstract?, #action_state_machine, #add_child_object, #add_coordination_object, #apply_terminal_flags, #as_plan, #as_service, #assign_argument, #assign_arguments, #can_merge?, #can_replace?, #check_emission_validity, #clear_events_external_relations, #clear_relations, #commit_transaction, #compatible_state?, #compute_replacement_candidates, #compute_subplan_replacement_operation, #compute_task_replacement_operation, #create_fresh_copy, create_script, #create_transaction_proxy, #current_state, #current_state?, #do_not_reuse, #do_poll, #each_coordination_object, #each_event, #each_exception_handler, #emit, #end_time, #ensure_poll_handler_called, #event, #event_model, #executable=, #executable?, #execute, #failed_to_start!, #failed_to_start?, #filter_events_from_strongly_related_tasks, #find_event, #fired_event, #forcefully_terminate, #forward_to, #freeze_delayed_arguments, #fullfills?, #fully_instanciated?, #garbage!, #goal, goal, #handle_exception, #has_argument?, #has_event?, #initialize_copy, #initialize_replacement, #inspect, #interruptible?, #invalidate_terminal_flag, #invalidated_terminal_flag?, #last_event, #lifetime, #list_unset_arguments, #mark_failed_to_start, #match, #meaningful_arguments, #name, #null?, #on, #partially_instanciated?, #plan=, #poll, #poll_handler, #pretty_print, #promise, #quarantined!, #quarantined?, #related_events, #related_tasks, #remove_coordination_object, #remove_poll_handler, #replace_by, #replace_subplan_by, #resolve_goals, #resolve_state_sources, #respawn, #reusable?, #running?, script, #script, #signals, #simulate, #start_time, state, #state, #terminal_events, #to_execution_exception, #to_s, #to_task, #transform_candidates_into_operations, #transition!, #update_task_status, #update_terminal_flag, #updated_data, #use_fault_response_table, #when_finalized, #|

Methods included from Models::Task

#abstract, #all_models, #as_plan, #can_merge?, #causal_link, #clear_model, #compute_terminal_events, #define_command_method, #define_event_methods, define_method_unless_present, #discover_terminal_events, #enum_events, #event, #event_model, #find_event_model, #forward, #from, #from_state, #fullfills?, #instantiate_event_relations, #interruptible, #invalidate_template, #match, model_attribute_list, model_relation, #on, #on_exception, #poll, #precondition, #provided_services, #query, #signal, #template, #terminal_events, #terminates, #to_coordination_task, #to_execution_exception_matcher, #update_terminal_flag, #with_arguments

Methods included from Models::Arguments

#argument, #arguments, #default_argument, #fullfills?, #meaningful_arguments

Methods included from Roby::TaskStateHelper

#import_events_to_roby, #namespace, #namespace=, #refine_running_state, #state_machine

Methods included from DRoby::Identifiable

#droby_id, #initialize_copy

Methods included from DRoby::V5::Models::TaskDumper

#droby_dump

Methods included from DRoby::V5::ModelDumper

#droby_dump, #droby_marshallable?

Methods included from DRoby::V5::TaskDumper

#droby_dump

Methods included from GUI::GraphvizTask

#apply_layout, #dot_label, #to_dot_events

Methods included from GUI::GraphvizPlanObject

#apply_layout, #dot_label, #to_dot

Methods included from GUI::RelationsCanvasTask

#display, #display_create, #display_name, #display_time_end, #display_time_start, #layout_events, to_svg, #update_graphics

Methods included from GUI::RelationsCanvasPlanObject

#display, #display_create, #display_events, #display_name, #display_parent

Methods included from ExceptionHandlingObject

#add_error, #handle_exception, #pass_exception

Methods inherited from PlanObject

#add_child_object, #apply_relation_changes, #as_plan, #can_finalize?, #commit_transaction, #concrete_model, #connection_space, #each_finalization_handler, #each_in_neighbour_merged, #each_out_neighbour_merged, #each_plan_child, #engine, #executable?, #finalized!, #finalized?, #forget_peer, #fullfills?, #garbage!, #garbage?, #initialize_copy, #initialize_replacement, #merged_relations, #promise, #read_write?, #real_object, #remotely_useful?, #replace_by, #replace_subplan_by, #root_object, #root_object?, #subscribed?, #transaction_proxy?, #transaction_stack, #update_on?, #updated_by?, #when_finalized

Methods included from Models::PlanObject

#child_plan_object, #finalization_handler, #match, #when_finalized

Methods included from Relations::DirectedRelationSupport

#[], #[]=, #add_child_object, #add_parent_object, #child_object?, #child_objects, #clear_vertex, #each_child_object, #each_in_neighbour, #each_out_neighbour, #each_parent_object, #each_relation, #each_relation_graph, #each_relation_sorted, #each_root_relation_graph, #enum_child_objects, #enum_parent_objects, #enum_relations, #leaf?, #parent_object?, #parent_objects, #related_object?, #related_objects, #relation_graph_for, #relations, #remove_child_object, #remove_children, #remove_parent_object, #remove_parents, #remove_relations, #root?, #sorted_relations

Methods inherited from DistributedObject

#add_owner, #clear_owners, #initialize_copy, #owned_by?, #remove_owner

Constructor Details

#initialize(command_line: nil, **arguments) ⇒ ExternalProcess

Returns a new instance of ExternalProcess.



86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/roby/tasks/external_process.rb', line 86

def initialize(command_line: nil, **arguments)
    command_line = Array(command_line) if command_line

    @pid = nil
    @buffer = nil
    @redirection = {}

    if arguments[:stub_subprocess].nil?
        arguments[:stub_subprocess] =
            Roby.app.simulation? &&
            ExternalProcess.stub_in_roby_simulation_mode?
    end
    super(command_line: command_line, **arguments)
end

Class Attribute Details

.stub_in_roby_simulation_mode=(value) ⇒ Object (writeonly)

Sets the default behavior of all ExternalProcess tasks regarding roby simulation mode (i.e. roby test)

If true, tasks will not actually start the subprocess, but will check that it is in PATH and executable. If false, the subprocess is started. The flag can be overriden on a per-task basis by setting the stub_in_roby_simulation_mode argument to either ‘true` or `false`. Use `nil` to use the default

The default is false for backward compatibility reasons

See Also:

  • {ExternalProcess#stub_in_roby_simulation_mode}


65
66
67
 
# File 'lib/roby/tasks/external_process.rb', line 65

def stub_in_roby_simulation_mode=(value)
  @stub_in_roby_simulation_mode = value
end

Instance Attribute Details

#chdir ⇒ Object

In which directory the program should be executed



43
 
# File 'lib/roby/tasks/external_process.rb', line 43

argument :chdir, default: nil

#command_line ⇒ Array, String

executable to start and the rest the arguments that need to be passed to it. If a string, it is interpreted as the executable name with no arguments.

Returns:

  • (Array, String) —

    If an array, its first element is the



35
 
# File 'lib/roby/tasks/external_process.rb', line 35

argument :command_line

#pid ⇒ Object (readonly)

The PID of the child process, or nil if the child process is not running



84
85
86
 
# File 'lib/roby/tasks/external_process.rb', line 84

def pid
  @pid
end

#stub_in_roby_simulation_mode ⇒ Object

Controls whether the task should actually start the subprocess or not. If ‘nil`, the behavior is controlled by #stub_in_roby_simulation_mode



49
 
# File 'lib/roby/tasks/external_process.rb', line 49

argument :stub_subprocess, default: nil

#working_directory ⇒ Object

DO NOT USE THIS. Use #chdir instead.



39
 
# File 'lib/roby/tasks/external_process.rb', line 39

argument :working_directory, default: nil

Class Method Details

.interruptible_with_signal(signal: "INT", **arguments) ⇒ Object

Create an ExternalProcess task that can be interrupted with the given signal



351
352
353
 
# File 'lib/roby/tasks/external_process.rb', line 351

def self.interruptible_with_signal(signal: "INT", **arguments)
    InterruptibleWithSignal.new(signal: signal, **arguments)
end

.stub_in_roby_simulation_mode? ⇒ Boolean

Returns:

  • (Boolean)

See Also:

  • {{.stub_in_roby_simulation_mode=}


68
69
70
 
# File 'lib/roby/tasks/external_process.rb', line 68

def stub_in_roby_simulation_mode?
    @stub_in_roby_simulation_mode
end

Instance Method Details

#actual_working_directory ⇒ Object 



219
220
221
 
# File 'lib/roby/tasks/external_process.rb', line 219

def actual_working_directory
    chdir || Dir.pwd
end

#create_redirection(redir_target) ⇒ Object

Handle redirection for a single stream (out or err)



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/roby/tasks/external_process.rb', line 158

def create_redirection(redir_target)
    if !redir_target
        [[], nil]
    elsif redir_target == :close
        [[], :close]
    elsif redir_target == :pipe
        pipe, io = IO.pipe
        [[[:close, io]], io, pipe, "".dup]
    elsif redir_target !~ /%p/
        # Assume no replacement in redirection, just open the file
        filename, mode =
            if redir_target[0, 1] == "+"
                [redir_target[1..-1], "a"]
            else
                [redir_target, "w"]
            end

        full_path = File.expand_path(filename, redirection_base_path)
        io = File.open(full_path, mode)

        [[[:close, io]], io]
    else
        full_path = File.expand_path(redir_target, redirection_base_path)
        dir = File.dirname(full_path)
        io = open_redirection(dir)
        [[[full_path, io]], io]
    end
end

#dead!(result) ⇒ Object

Called to announce that this task has been killed. result is the corresponding Process::Status object.



103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/roby/tasks/external_process.rb', line 103

def dead!(result)
    if !result
        failed_event.emit
    elsif result.success?
        success_event.emit
    elsif result.signaled?
        signaled_event.emit(result)
    else
        failed_event.emit(result)
    end
end

#handle_redirection ⇒ Object

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.

Setup redirections pre-spawn



190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/roby/tasks/external_process.rb', line 190

def handle_redirection
    return [], {} if !@redirection[:stdout] && !@redirection[:stderr]

    if (@redirection[:stdout] == @redirection[:stderr]) &&
       !%i[pipe close].include?(@redirection[:stdout])
        redir_target = @redirection[:stdout]
        path = File.expand_path(redir_target, redirection_base_path)
        dir = File.dirname(path)
        io = open_redirection(dir)
        return [[@redirection[:stdout], io]], Hash[out: io, err: io]
    end

    out_open, out_io, @out_pipe, @out_buffer =
        create_redirection(@redirection[:stdout])
    err_open, err_io, @err_pipe, @err_buffer =
        create_redirection(@redirection[:stderr])

    @read_buffer = "".dup if @out_buffer || @err_buffer

    spawn_options = {}
    spawn_options[:out] = out_io if out_io
    spawn_options[:err] = err_io if err_io
    [(out_open + err_open), spawn_options]
end

#kill(signo) ⇒ Object

Kills the child process

Parameters:

  • signo (String, Integer) —

    the signal name or number, as accepted by Process#kill



284
285
286
 
# File 'lib/roby/tasks/external_process.rb', line 284

def kill(signo)
    Process.kill(signo, pid)
end

#normalize_redirection_mode(mode) ⇒ Object

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.

Normalize the redirection target argument of #redirect_output



145
146
147
148
149
150
151
152
153
 
# File 'lib/roby/tasks/external_process.rb', line 145

def normalize_redirection_mode(mode)
    return unless mode

    if %i[pipe close].include?(mode)
        mode
    else
        mode.to_str
    end
end

#open_redirection(dir) ⇒ Object

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.

Open the output file for redirection, before spawning



274
275
276
277
278
 
# File 'lib/roby/tasks/external_process.rb', line 274

def open_redirection(dir)
    Dir::Tmpname.create "roby-external-process", dir do |path, _|
        return File.open(path, "w+")
    end
end

#poll_live_process ⇒ Object 



337
338
339
340
341
 
# File 'lib/roby/tasks/external_process.rb', line 337

def poll_live_process
    read_pipes
    pid, exit_status = ::Process.waitpid2(self.pid, ::Process::WNOHANG)
    dead!(exit_status) if pid
end

#read_pipe(pipe, buffer) ⇒ Object

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.

Read a given pipe, when an output is redirected to pipe



291
292
293
294
295
296
297
298
299
300
301
302
303
 
# File 'lib/roby/tasks/external_process.rb', line 291

def read_pipe(pipe, buffer)
    received = false
    loop do
        pipe.read_nonblock 1024, @read_buffer
        received = true
        buffer.concat(@read_buffer)
    end
rescue EOFError
    pipe.close
    [true, buffer.dup]
rescue IO::WaitReadable
    [false, buffer.dup] if received
end

#read_pipes ⇒ Object 



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
 
# File 'lib/roby/tasks/external_process.rb', line 315

def read_pipes
    if @out_pipe
        eos, data = read_pipe(@out_pipe, @out_buffer)
        @out_pipe = nil if eos
        stdout_received(data) if data
        @out_buffer.clear
    end

    if @err_pipe
        eos, data = read_pipe(@err_pipe, @err_buffer)
        @err_pipe = nil if eos
        stderr_received(data) if data
        @err_buffer.clear
    end

    nil
end

#redirect_output(common) ⇒ Object #redirect_output(stdout: nil, stderr: nil) ⇒ Object

Overloads:

  • #redirect_output(stdout: nil, stderr: nil) ⇒ Object

    Redirect either stdout and stderr. The redirection target can either be a string, which is interpreted as a path, or one of :pipe and :close.

    If redirecting to a string, %p is replaced by the process actual PID. Both can be redirected to the same output file

    The special value :pipe shall be used to make the task read the process output and call #stdout_received (resp. #stderr_received) with it. :close will make the task close this output

    Pass ‘nil` to not redirect this particular output. Calling the method with a single argument applies this redirection to both outputs.



131
132
133
134
135
136
137
138
139
140
 
# File 'lib/roby/tasks/external_process.rb', line 131

def redirect_output(common = nil, stdout: nil, stderr: nil)
    raise "cannot change redirection after task start" if @pid

    stdout = stderr = common if common

    @redirection = {
        stdout: normalize_redirection_mode(common || stdout),
        stderr: normalize_redirection_mode(common || stderr)
    }
end

#redirection_base_path ⇒ Object 



215
216
217
 
# File 'lib/roby/tasks/external_process.rb', line 215

def redirection_base_path
    chdir || working_directory || Dir.pwd
end

#redirection_path(pattern, pid) ⇒ Object

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.

Returns the file name based on the redirection pattern and the current PID value.



267
268
269
 
# File 'lib/roby/tasks/external_process.rb', line 267

def redirection_path(pattern, pid) # :nodoc:
    pattern.gsub "%p", pid.to_s
end

#start ⇒ Object

:method: start!

Starts the child process. Emits start when the process is actually started.



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/roby/tasks/external_process.rb', line 228

event :start do |_|
    opened_ios, spawn_options = handle_redirection

    if stub_subprocess
        @pid = rand(65_535)
        validate_program(command_line[0])
    else
        @pid = Process.spawn(
            *command_line, chdir: chdir || Dir.pwd, **spawn_options
        )
    end

    opened_ios.each do |pattern, io|
        if pattern != :close
            target_path = File.expand_path(
                redirection_path(pattern, @pid), redirection_base_path
            )
            FileUtils.mv io.path, target_path
        end
        io.close
    end

    start_event.emit
end

#stderr_received(data) ⇒ Object

Method called when data is received on an intercepted stderr

Intercept stdout by calling redirect_output(stderr: :pipe)



313
 
# File 'lib/roby/tasks/external_process.rb', line 313

def stderr_received(data); end

#stdout_received(data) ⇒ Object

Method called when data is received on an intercepted stdout

Intercept stdout by calling redirect_output(stdout: :pipe)



308
 
# File 'lib/roby/tasks/external_process.rb', line 308

def stdout_received(data); end

#validate_program(cmd) ⇒ Object

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.

Emulates error handling of Process.spawn when stub_subprocess is set

Raises:

  • (Errno::ENOENT) 


256
257
258
259
260
261
 
# File 'lib/roby/tasks/external_process.rb', line 256

def validate_program(cmd)
    raise Errno::ENOENT, cmd unless (absolute = Roby.find_in_path(cmd))
    raise Errno::EACCES, cmd unless File.executable?(absolute)

    nil
end