Source Code for Module roshlaunch.nodeprocess

  1  # Software License Agreement (BSD License) 
  2  # 
  3  # Copyright (c) 2008, Willow Garage, Inc. 
  4  # All rights reserved. 
  5  # 
  6  # Redistribution and use in source and binary forms, with or without 
  7  # modification, are permitted provided that the following conditions 
  8  # are met: 
  9  # 
 10  #  * Redistributions of source code must retain the above copyright 
 11  #    notice, this list of conditions and the following disclaimer. 
 12  #  * Redistributions in binary form must reproduce the above 
 13  #    copyright notice, this list of conditions and the following 
 14  #    disclaimer in the documentation and/or other materials provided 
 15  #    with the distribution. 
 16  #  * Neither the name of Willow Garage, Inc. nor the names of its 
 17  #    contributors may be used to endorse or promote products derived 
 18  #    from this software without specific prior written permission. 
 19  # 
 20  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 21  # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
 22  # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS 
 23  # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE 
 24  # COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
 25  # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 
 26  # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; 
 27  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 
 28  # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 
 29  # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN 
 30  # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 31  # POSSIBILITY OF SUCH DAMAGE. 
 32  # 
 33  # Revision $Id: nodeprocess.py 10265 2010-07-06 18:52:04Z kwc $ 
 34   
 35  """ 
 36  Local process implementation for running and monitoring nodes. 
 37  """ 
 38   
 39  import os 
 40  import sys 
 41  import signal 
 42  import socket 
 43  import subprocess  
 44  import time 
 45  import traceback 
 46   
 47  import roslib.rosenv  
 48  import roslib.network  
 49   
 50  from roslaunch.core import * 
 51  from roslaunch.node_args import create_local_process_env, create_local_process_args 
 52  from roslaunch.pmon import Process, FatalProcessLaunch 
 53   
 54  import logging 
 55  _logger = logging.getLogger("roslaunch") 
 56   
 57  _TIMEOUT_SIGINT  = 15.0 #seconds 
 58  _TIMEOUT_SIGTERM = 2.0 #seconds 
 59   
 60  _counter = 0 
 61 -def _next_counter(): 
 62      global _counter 
 63      _counter += 1 
 64      return _counter 
 65   
 66 -def create_master_process(run_id, type_, ros_root, port): 
 67      """ 
 68      Launch a master 
 69      @param type_: name of master executable 
 70      @type  type_: str 
 71      @param ros_root: ROS_ROOT environment setting 
 72      @type  ros_root: str 
 73      @param port: port to launch master on 
 74      @type  port: int 
 75      @raise RLException: if type_ or port is invalid 
 76      """     
 77      if port < 1 or port > 65535: 
 78          raise RLException("invalid port assignment: %s"%port) 
 79   
 80      _logger.info("create_master_process: %s, %s, %s", type_, ros_root, port) 
 81      master = os.path.join(ros_root, 'bin', type_) 
 82      # zenmaster is deprecated and aliased to rosmaster 
 83      if type_ == Master.ROSMASTER:         
 84          package = 'rosmaster'         
 85          args = [master, '--core', '-p', str(port)] 
 86      else: 
 87          raise RLException("unknown master typ_: %s"%type_) 
 88   
 89      _logger.info("process[master]: launching with args [%s]"%args) 
 90      log_output = False 
 91      return LocalProcess(run_id, package, 'master', args, os.environ, log_output, None) 
 92   
 93 -def create_node_process(run_id, node, master_uri): 
 94      """ 
 95      Factory for generating processes for launching local ROS 
 96      nodes. Also registers the process with the L{ProcessMonitor} so that 
 97      events can be generated when the process dies. 
 98       
 99      @param run_id: run_id of launch 
100      @type  run_id: str 
101      @param node: node to launch. Node name must be assigned. 
102      @type  node: L{Node} 
103      @param master_uri: API URI for master node 
104      @type  master_uri: str 
105      @return: local process instance 
106      @rtype: L{LocalProcess} 
107      @raise NodeParamsException: If the node's parameters are improperly specific 
108      """     
109      _logger.info("create_node_process: package[%s] type[%s] machine[%s] master_uri[%s]", node.package, node.type, node.machine, master_uri) 
110      # check input args 
111      machine = node.machine 
112      if machine is None: 
113          raise RLException("Internal error: no machine selected for node of type [%s/%s]"%(node.package, node.type)) 
114      if not node.name: 
115          raise ValueError("node name must be assigned") 
116   
117      # - setup env for process (vars must be strings for os.environ) 
118      env = create_local_process_env(node, machine, master_uri) 
119   
120      if not node.name: 
121          raise ValueError("node name must be assigned") 
122       
123      # we have to include the counter to prevent potential name 
124      # collisions between the two branches 
125      name = "%s-%s"%(node.name, _next_counter()) 
126   
127      _logger.info('process[%s]: env[%s]', name, env) 
128   
129      args = create_local_process_args(node, machine) 
130      _logger.info('process[%s]: args[%s]', name, args)         
131   
132      # default for node.output not set is 'log' 
133      log_output = node.output != 'screen' 
134      _logger.debug('process[%s]: returning LocalProcess wrapper') 
135      return LocalProcess(run_id, node.package, name, args, env, log_output, respawn=node.respawn, required=node.required, cwd=node.cwd) 
136   
137   
138  # ROSHTODO: create LocalProcess and NodeProcess 
139 -class LocalProcess(Process): 
140      """ 
141      Process launched on local machine 
142      """ 
143       
144 -    def __init__(self, run_id, package, name, args, env, log_output, respawn=False, required=False, cwd=None, is_node=True): 
145          """ 
146          @param run_id: unique run ID for this roslaunch. Used to 
147            generate log directory location. run_id may be None if this 
148            feature is not being used. 
149          @type  run_id: str 
150          @param package: name of package process is part of 
151          @type  package: str 
152          @param name: name of process 
153          @type  name: str 
154          @param args: list of arguments to process 
155          @type  args: [str] 
156          @param env: environment dictionary for process 
157          @type  env: {str : str} 
158          @param log_output: if True, log output streams of process 
159          @type  log_output: bool 
160          @param respawn: respawn process if it dies (default is False) 
161          @type  respawn: bool 
162          @param cwd: working directory of process, or None 
163          @type  cwd: str 
164          @param is_node: (optional) if True, process is ROS node and accepts ROS node command-line arguments. Default: True 
165          @type  is_node: False 
166          """     
167          super(LocalProcess, self).__init__(package, name, args, env, respawn, required) 
168          self.run_id = run_id 
169          self.popen = None 
170          self.log_output = log_output 
171          self.started = False 
172          self.stopped = False 
173          self.cwd = cwd 
174          self.log_dir = None 
175          self.pid = -1 
176          self.is_node = is_node 
177   
178      # NOTE: in the future, info() is going to have to be sufficient for relaunching a process 
179 -    def get_info(self): 
180          """ 
181          Get all data about this process in dictionary form 
182          """     
183          info = super(LocalProcess, self).get_info() 
184          info['pid'] = self.pid 
185          if self.run_id: 
186              info['run_id'] = self.run_id 
187          info['log_output'] = self.log_output 
188          if self.cwd is not None: 
189              info['cwd'] = self.cwd 
190          return info 
191   
192 -    def _configure_logging(self): 
193          """ 
194          Configure logging of node's log file and stdout/stderr 
195          @return: stdout log file name, stderr log file 
196          name. Values are None if stdout/stderr are not logged. 
197          @rtype: str, str 
198          """     
199          log_dir = roslib.rosenv.get_log_dir(env=os.environ) 
200          if self.run_id: 
201              log_dir = os.path.join(log_dir, self.run_id) 
202          if not os.path.exists(log_dir): 
203              try: 
204                  os.makedirs(log_dir) 
205              except OSError, (errno, msg): 
206                  if errno == 13: 
207                      raise RLException("unable to create directory for log file [%s].\nPlease check permissions."%log_dir) 
208                  else: 
209                      raise RLException("unable to create directory for log file [%s]: %s"%(log_dir, msg)) 
210          # #973: save log dir for error messages 
211          self.log_dir = log_dir 
212   
213          # send stdout/stderr to file. in the case of respawning, we have to 
214          # open in append mode 
215          # note: logfileerr: disabling in favor of stderr appearing in the console. 
216          # will likely reinstate once roserr/rosout is more properly used. 
217          logfileout = logfileerr = None 
218   
219          # ROSHTODO: get rid of is_child_mode() and replace with 'logerr_output' option 
220          if self.log_output: 
221              outf, errf = [os.path.join(log_dir, '%s-%s.log'%(self.name, n)) for n in ['stdout', 'stderr']] 
222              if self.respawn: 
223                  mode = 'a' 
224              else: 
225                  mode = 'w' 
226              logfileout = open(outf, mode) 
227              if is_child_mode(): 
228                  logfileerr = open(errf, mode) 
229   
230          # #986: pass in logfile name to node 
231          node_log_file = log_dir 
232          if self.is_node: 
233              # #1595: on respawn, these keep appending 
234              self.args = _cleanup_remappings(self.args, '__log:=') 
235              self.args.append("__log:=%s"%os.path.join(log_dir, "%s.log"%self.name)) 
236   
237          return logfileout, logfileerr 
238   
239 -    def start(self): 
240          """ 
241          Start the process. 
242           
243          @raise FatalProcessLaunch: if process cannot be started and it 
244          is not likely to ever succeed 
245          """ 
246          super(LocalProcess, self).start() 
247          try: 
248              self.lock.acquire() 
249              if self.started: 
250                  _logger.info("process[%s]: restarting os process", self.name) 
251              else: 
252                  _logger.info("process[%s]: starting os process", self.name) 
253              self.started = self.stopped = False 
254   
255              full_env = self.env 
256   
257              # _configure_logging() can mutate self.args 
258              try: 
259                  logfileout, logfileerr = self._configure_logging() 
260              except Exception, e: 
261                  _logger.error(traceback.format_exc()) 
262                  printerrlog("[%s] ERROR: unable to configure logging [%s]"%(self.name, str(e))) 
263                  # it's not safe to inherit from this process as 
264                  # rostest changes stdout to a StringIO, which is not a 
265                  # proper file. 
266                  logfileout, logfileerr = subprocess.PIPE, subprocess.PIPE 
267   
268              if self.cwd == 'node': 
269                  cwd = os.path.dirname(self.args[0]) 
270              elif self.cwd == 'cwd': 
271                  cwd = os.getcwd() 
272              elif self.cwd == 'ros-root': 
273                  cwd = get_ros_root() 
274              else: 
275                  cwd = roslib.rosenv.get_ros_home() 
276   
277              _logger.info("process[%s]: start w/ args [%s]", self.name, self.args) 
278              _logger.info("process[%s]: cwd will be [%s]", self.name, cwd) 
279   
280              try: 
281                  self.popen = subprocess.Popen(self.args, cwd=cwd, stdout=logfileout, stderr=logfileerr, env=full_env, close_fds=True, preexec_fn=os.setsid) 
282              except OSError, (errno, msg): 
283                  self.started = True # must set so is_alive state is correct 
284                  _logger.error("OSError(%d, %s)", errno, msg) 
285                  if errno == 8: #Exec format error 
286                      raise FatalProcessLaunch("Unable to launch [%s]. \nIf it is a script, you may be missing a '#!' declaration at the top."%self.name) 
287                  elif errno == 2: #no such file or directory 
288                      raise FatalProcessLaunch("""Roslaunch got a '%s' error while attempting to run: 
289   
290  %s 
291   
292  Please make sure that all the executables in this command exist and have 
293  executable permission. This is often caused by a bad launch-prefix."""%(msg, ' '.join(self.args))) 
294                  else: 
295                      raise FatalProcessLaunch("unable to launch [%s]: %s"%(' '.join(self.args), msg)) 
296                   
297              self.started = True 
298              # Check that the process is either still running (poll returns 
299              # None) or that it completed successfully since when we 
300              # launched it above (poll returns the return code, 0). 
301              poll_result = self.popen.poll() 
302              if poll_result is None or poll_result == 0: 
303                  self.pid = self.popen.pid 
304                  printlog_bold("process[%s]: started with pid [%s]"%(self.name, self.pid)) 
305                  return True 
306              else: 
307                  printerrlog("failed to start local process: %s"%(' '.join(self.args))) 
308                  return False 
309          finally: 
310              self.lock.release() 
311   
312 -    def is_alive(self): 
313          """ 
314          @return: True if process is still running 
315          @rtype: bool 
316          """ 
317          if not self.started: #not started yet 
318              return True 
319          if self.stopped or self.popen is None: 
320              return False 
321          self.exit_code = self.popen.poll() 
322          if self.exit_code is not None: 
323              return False 
324          return True 
325   
326 -    def get_exit_description(self): 
327          """ 
328          @return: human-readable description of exit state  
329          @rtype: str 
330          """ 
331          # #973: include location of output location in message 
332          if self.exit_code is not None: 
333              if self.exit_code: 
334                  if self.log_dir: 
335                      return 'process has died [pid %s, exit code %s].\nlog files: %s*.log'%(self.pid, self.exit_code, os.path.join(self.log_dir, self.name)) 
336                  else: 
337                      return 'process has died [pid %s, exit code %s]'%(self.pid, self.exit_code) 
338              else: 
339                  if self.log_dir: 
340                      return 'process has finished cleanly.\nlog file: %s*.log'%(os.path.join(self.log_dir, self.name)) 
341                  else: 
342                      return 'process has finished cleanly' 
343          else: 
344              return 'process has died' 
345   
346 -    def _stop_unix(self, errors): 
347          """ 
348          UNIX implementation of process killing 
349   
350          @param errors: error messages. stop() will record messages into this list. 
351          @type  errors: [str] 
352          """ 
353          self.exit_code = self.popen.poll()  
354          if self.exit_code is not None: 
355              _logger.debug("process[%s].stop(): process has already returned %s", self.name, self.exit_code) 
356              #print "process[%s].stop(): process has already returned %s"%(self.name, self.exit_code)                 
357              self.popen = None 
358              self.stopped = True 
359              return 
360   
361          pid = self.popen.pid 
362          pgid = os.getpgid(pid) 
363          _logger.info("process[%s]: killing os process with pid[%s] pgid[%s]", self.name, pid, pgid) 
364   
365          try: 
366              # Start with SIGINT and escalate from there. 
367              _logger.info("[%s] sending SIGINT to pgid [%s]", self.name, pgid)                                     
368              os.killpg(pgid, signal.SIGINT) 
369              _logger.info("[%s] sent SIGINT to pgid [%s]", self.name, pgid) 
370              timeout_t = time.time() + _TIMEOUT_SIGINT 
371              retcode = self.popen.poll()                 
372              while time.time() < timeout_t and retcode is None: 
373                  time.sleep(0.1) 
374                  retcode = self.popen.poll() 
375              # Escalate non-responsive process 
376              if retcode is None: 
377                  printerrlog("[%s] escalating to SIGTERM"%self.name) 
378                  timeout_t = time.time() + _TIMEOUT_SIGTERM 
379                  os.killpg(pgid, signal.SIGTERM)                 
380                  _logger.info("[%s] sent SIGTERM to pgid [%s]"%(self.name, pgid)) 
381                  retcode = self.popen.poll() 
382                  while time.time() < timeout_t and retcode is None: 
383                      time.sleep(0.2) 
384                      _logger.debug('poll for retcode') 
385                      retcode = self.popen.poll() 
386                  if retcode is None: 
387                      printerrlog("[%s] escalating to SIGKILL"%self.name) 
388                      errors.append("process[%s, pid %s]: required SIGKILL. May still be running."%(self.name, pid)) 
389                      try: 
390                          os.killpg(pgid, signal.SIGKILL) 
391                          _logger.info("[%s] sent SIGKILL to pgid [%s]"%(self.name, pgid)) 
392                          # #2096: don't block on SIGKILL, because this results in more orphaned processes overall 
393                          #self.popen.wait() 
394                          #os.wait() 
395                          _logger.info("process[%s]: sent SIGKILL", self.name) 
396                      except OSError, e: 
397                          if e.args[0] == 3: 
398                              printerrlog("no [%s] process with pid [%s]"%(self.name, pid)) 
399                          else: 
400                              printerrlog("errors shutting down [%s], see log for details"%self.name) 
401                              _logger.error(traceback.format_exc()) 
402                  else: 
403                      _logger.info("process[%s]: SIGTERM killed with return value %s", self.name, retcode) 
404              else: 
405                  _logger.info("process[%s]: SIGINT killed with return value %s", self.name, retcode) 
406                   
407          finally: 
408              self.popen = None 
409   
410 -    def stop(self, errors=[]): 
411          """ 
412          Stop the process. Record any significant error messages in the errors parameter 
413           
414          @param errors: error messages. stop() will record messages into this list. 
415          @type  errors: [str] 
416          """ 
417          super(LocalProcess, self).stop(errors) 
418          self.lock.acquire()         
419          try: 
420              try: 
421                  _logger.debug("process[%s].stop() starting", self.name) 
422                  if self.popen is None: 
423                      _logger.debug("process[%s].stop(): popen is None, nothing to kill")  
424                      return 
425                  #NOTE: currently POSIX-only. Need to add in Windows code once I have a test environment: 
426                  # http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/347462 
427                  self._stop_unix(errors) 
428              except: 
429                  #traceback.print_exc()  
430                  _logger.error("[%s] EXCEPTION %s", self.name, traceback.format_exc())                                 
431          finally: 
432              self.stopped = True 
433              self.lock.release() 
434   
435   
436  # #1595 
437 -def _cleanup_remappings(args, prefix): 
438      """ 
439      Remove all instances of args that start with prefix. This is used 
440      to remove args that were previously added (and are now being 
441      regenerated due to respawning) 
442      """     
443      existing_args = [a for a in args if a.startswith(prefix)] 
444      for a in existing_args: 
445          args.remove(a) 
446      return args 
447