1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35 from __future__ import print_function
36
37 """
38 Top-level implementation of launching processes. Coordinates
39 lower-level libraries.
40 """
41
42 import os
43 import logging
44 import subprocess
45 import sys
46 import time
47 import traceback
48
49 import rosgraph
50 import rosgraph.names
51 import rosgraph.network
52
53 from roslaunch.core import *
54
55 from roslaunch.nodeprocess import create_master_process, create_node_process
56 from roslaunch.pmon import start_process_monitor, ProcessListener
57
58 from roslaunch.rlutil import update_terminal_name
59
60 from rosmaster.master_api import NUM_WORKERS
61
62 _TIMEOUT_MASTER_START = 10.0
63 _TIMEOUT_MASTER_STOP = 10.0
64
65 _ID = '/roslaunch'
66
68
70 """
71 Validate the configuration of a master we are about to launch. Ths
72 validation already assumes that no existing master is running at
73 this configuration and merely checks configuration for a new
74 launch.
75 """
76
77
78
79
80
81
82
83 if not rosgraph.network.is_local_address(m.get_host()):
84
85
86 if is_core:
87
88
89
90 try:
91 reverse_ips = [host[4][0] for host in socket.getaddrinfo(m.get_host(), 0, 0, 0, socket.SOL_TCP)]
92 local_addrs = rosgraph.network.get_local_addresses()
93 printerrlog("""WARNING: IP addresses %s for local hostname '%s' do not appear to match
94 any local IP address (%s). Your ROS nodes may fail to communicate.
95
96 Please use ROS_IP to set the correct IP address to use."""%(','.join(reverse_ips), m.get_host(), ','.join(local_addrs)))
97 except:
98 printerrlog("""WARNING: local hostname '%s' does not map to an IP address.
99 Your ROS nodes may fail to communicate.
100
101 Please use ROS_IP to set the correct IP address to use."""%(m.get_host()))
102
103 else:
104
105 raise RLException("ERROR: unable to contact ROS master at [%s]"%(m.uri))
106
107 if is_core and not is_rostest:
108
109
110 env_uri = rosgraph.get_master_uri()
111 env_host, env_port = rosgraph.network.parse_http_host_and_port(env_uri)
112
113 if not rosgraph.network.is_local_address(env_host):
114
115 printerrlog("WARNING: ROS_MASTER_URI [%s] host is not set to this machine"%(env_uri))
116 elif env_port != m.get_port():
117
118 printerrlog("WARNING: ROS_MASTER_URI port [%s] does not match this roscore [%s]"%(env_port, m.get_port()))
119
121 splits = [s for s in p.split(rosgraph.names.SEP) if s]
122 curr =rosgraph.names.SEP
123 parents = [curr]
124 for s in splits[:-1]:
125 next_ = curr + s + rosgraph.names.SEP
126 parents.append(next_)
127 curr = next_
128 return parents
129
131 """
132 Reduce clear_params such that only the highest-level namespaces
133 are represented for overlapping namespaces. e.g. if both /foo/ and
134 /foo/bar are present, return just /foo/.
135
136 @param params: paremter namees
137 @type params: [str]
138 @return: unified parameters
139 @rtype: [str]
140 """
141
142
143
144 canon_params = []
145 for p in params:
146 if not p[-1] == rosgraph.names.SEP:
147 p += rosgraph.names.SEP
148 if not p in canon_params:
149 canon_params.append(p)
150
151 clear_params = canon_params[:]
152 for p in canon_params:
153 for parent in _all_namespace_parents(p):
154 if parent in canon_params and p in clear_params and p != '/':
155 clear_params.remove(p)
156 return clear_params
157
159 """
160 Utility function to strip illegal characters from hostname. This
161 is implemented to be correct, not efficient."""
162
163
164 fixed = 'host_'
165
166 hostname = hostname.lower()
167 for c in hostname:
168
169 if (c >= 'a' and c <= 'z') or \
170 (c >= '0' and c <= '9'):
171 fixed+=c
172 else:
173 fixed+='_'
174 return fixed
175
177 """
178 Helper class to manage distributing process events. It functions as
179 a higher level aggregator of events. In particular, events about
180 remote and local processes require different mechanisms for being
181 caught and reported.
182 """
184 self.process_listeners = []
185
187 """
188 Add listener to list of listeners. Not threadsafe.
189 @param l: listener
190 @type l: L{ProcessListener}
191 """
192 self.process_listeners.append(l)
193
195 """
196 ProcessListener callback
197 """
198 for l in self.process_listeners:
199 try:
200 l.process_died(process_name, exit_code)
201 except Exception as e:
202 logging.getLogger('roslaunch').error(traceback.format_exc())
203
205 """
206 Listener interface for events related to ROSLaunch.
207 ROSLaunchListener is currently identical to the lower-level
208 L{roslaunch.pmon.ProcessListener} interface, but is separate for
209 architectural reasons. The lower-level
210 L{roslaunch.pmon.ProcessMonitor} does not provide events about
211 remotely running processes.
212 """
213
215 """
216 Notifies listener that process has died. This callback only
217 occurs for processes that die during normal process monitor
218 execution -- processes that are forcibly killed during
219 L{roslaunch.pmon.ProcessMonitor} shutdown are not reported.
220 @param process_name: name of process
221 @type process_name: str
222 @param exit_code int: exit code of process. If None, it means
223 that L{roslaunch.pmon.ProcessMonitor} was unable to determine an exit code.
224 @type exit_code: int
225 """
226 pass
227
229 """
230 Runs a roslaunch. The normal sequence of API calls is L{launch()}
231 followed by L{spin()}. An external thread can call L{stop()}; otherwise
232 the runner will block until an exit signal. Another usage is to
233 call L{launch()} followed by repeated calls to L{spin_once()}. This usage
234 allows the main thread to continue to do work while processes are
235 monitored.
236 """
237
238 - def __init__(self, run_id, config, server_uri=None, pmon=None, is_core=False, remote_runner=None, is_child=False, is_rostest=False, num_workers=NUM_WORKERS, timeout=None, master_logger_level=False):
239 """
240 @param run_id: /run_id for this launch. If the core is not
241 running, this value will be used to initialize /run_id. If
242 the core is already running, this value will be checked
243 against the value stored on the core. L{ROSLaunchRunner} will
244 fail during L{launch()} if they do not match.
245 @type run_id: str
246 @param config: roslauch instance to run
247 @type config: L{ROSLaunchConfig}
248 @param server_uri: XML-RPC URI of roslaunch server.
249 @type server_uri: str
250 @param pmon: optionally override the process
251 monitor the runner uses for starting and tracking processes
252 @type pmon: L{ProcessMonitor}
253
254 @param is_core: if True, this runner is a roscore
255 instance. This affects the error behavior if a master is
256 already running -- aborts if is_core is True and a core is
257 detected.
258 @type is_core: bool
259 @param remote_runner: remote roslaunch process runner
260 @param is_rostest: if True, this runner is a rostest
261 instance. This affects certain validation checks.
262 @type is_rostest: bool
263 @param num_workers: If this is the core, the number of worker-threads to use.
264 @type num_workers: int
265 @param timeout: If this is the core, the socket-timeout to use.
266 @type timeout: Float or None
267 @param master_logger_level: Specify roscore's rosmaster.master logger level, use default if it is False.
268 @type master_logger_level: str or False
269 """
270 if run_id is None:
271 raise RLException("run_id is None")
272 self.run_id = run_id
273
274
275
276
277
278
279 self.config = config
280 self.server_uri = server_uri
281 self.is_child = is_child
282 self.is_core = is_core
283 self.is_rostest = is_rostest
284 self.num_workers = num_workers
285 self.timeout = timeout
286 self.master_logger_level = master_logger_level
287 self.logger = logging.getLogger('roslaunch')
288 self.pm = pmon or start_process_monitor()
289
290
291
292
293 self.listeners = _ROSLaunchListeners()
294 if self.pm is None:
295 raise RLException("unable to initialize roslaunch process monitor")
296 if self.pm.is_shutdown:
297 raise RLException("bad roslaunch process monitor initialization: process monitor is already dead")
298
299 self.pm.add_process_listener(self.listeners)
300
301 self.remote_runner = remote_runner
302
304 """
305 Add listener to list of listeners. Not threadsafe. Must be
306 called before processes started.
307 @param l: listener
308 @type l: L{ProcessListener}
309 """
310 self.listeners.add_process_listener(l)
311
313 """
314 Load parameters onto the parameter server
315 """
316 self.logger.info("load_parameters starting ...")
317 config = self.config
318 param_server = config.master.get()
319 p = None
320 try:
321
322 param_server_multi = config.master.get_multi()
323
324
325
326 for p in _unify_clear_params(config.clear_params):
327 if param_server.hasParam(_ID, p)[2]:
328
329 param_server_multi.deleteParam(_ID, p)
330 r = param_server_multi()
331 for code, msg, _ in r:
332 if code != 1:
333 raise RLException("Failed to clear parameter: %s"%(msg))
334
335
336 param_server_multi = config.master.get_multi()
337 for p in config.params.values():
338
339
340 param_server_multi.setParam(_ID, p.key, p.value)
341 r = param_server_multi()
342 for code, msg, _ in r:
343 if code != 1:
344 raise RLException("Failed to set parameter: %s"%(msg))
345 except RLException:
346 raise
347 except Exception as e:
348 printerrlog("load_parameters: unable to set parameters (last param was [%s]): %s"%(p,e))
349 raise
350 self.logger.info("... load_parameters complete")
351
353 """
354 Launch all the declared nodes/master
355 @return: two lists of node names where the first
356 is the nodes that successfully launched and the second is the
357 nodes that failed to launch.
358 @rtype: [[str], [str]]
359 """
360 config = self.config
361 succeeded = []
362 failed = []
363 self.logger.info("launch_nodes: launching local nodes ...")
364 local_nodes = config.nodes
365
366
367 local_nodes = [n for n in config.nodes if is_machine_local(n.machine)]
368
369 for node in local_nodes:
370 proc, success = self.launch_node(node)
371 if success:
372 succeeded.append(str(proc))
373 else:
374 failed.append(str(proc))
375
376 if self.remote_runner:
377 self.logger.info("launch_nodes: launching remote nodes ...")
378 r_succ, r_fail = self.remote_runner.launch_remote_nodes()
379 succeeded.extend(r_succ)
380 failed.extend(r_fail)
381
382 self.logger.info("... launch_nodes complete")
383 return succeeded, failed
384
386 """
387 Launches master if requested.
388 @return: True if a master was launched, False if a master was
389 already running.
390 @rtype: bool
391 @raise RLException: if master launch fails
392 """
393 m = self.config.master
394 is_running = m.is_running()
395
396 if self.is_core and is_running:
397 raise RLException("roscore cannot run as another roscore/master is already running. \nPlease kill other roscore/master processes before relaunching.\nThe ROS_MASTER_URI is %s"%(m.uri))
398
399 if not is_running:
400 validate_master_launch(m, self.is_core, self.is_rostest)
401
402 printlog("auto-starting new master")
403 p = create_master_process(
404 self.run_id, m.type, get_ros_root(), m.get_port(), self.num_workers,
405 self.timeout, master_logger_level=self.master_logger_level)
406 self.pm.register_core_proc(p)
407 success = p.start()
408 if not success:
409 raise RLException("ERROR: unable to auto-start master process")
410 timeout_t = time.time() + _TIMEOUT_MASTER_START
411 while not m.is_running() and time.time() < timeout_t:
412 time.sleep(0.1)
413
414 if not m.is_running():
415 raise RLException("ERROR: could not contact master [%s]"%m.uri)
416
417 printlog_bold("ROS_MASTER_URI=%s"%m.uri)
418
419
420
421 update_terminal_name(m.uri)
422
423
424 param_server = m.get()
425
426
427 self._check_and_set_run_id(param_server, self.run_id)
428
429 if self.server_uri:
430
431
432 hostname, port = rosgraph.network.parse_http_host_and_port(self.server_uri)
433 hostname = _hostname_to_rosname(hostname)
434 self.logger.info("setting /roslaunch/uris/%s__%s' to %s"%(hostname, port, self.server_uri))
435 param_server.setParam(_ID, '/roslaunch/uris/%s__%s'%(hostname, port),self.server_uri)
436
437 return not is_running
438
440 """
441 Initialize self.run_id to existing value or setup parameter
442 server with /run_id set to default_run_id
443 @param default_run_id: run_id to use if value is not set
444 @type default_run_id: str
445 @param param_server: parameter server proxy
446 @type param_server: xmlrpclib.ServerProxy
447 """
448 code, _, val = param_server.hasParam(_ID, '/run_id')
449 if code == 1 and not val:
450 printlog_bold("setting /run_id to %s"%run_id)
451 param_server.setParam('/roslaunch', '/run_id', run_id)
452 else:
453
454 code, _, val = param_server.getParam('/roslaunch', '/run_id')
455 if code != 1:
456
457
458 raise RLException("ERROR: unable to retrieve /run_id from parameter server")
459 if run_id != val:
460 raise RLException("run_id on parameter server does not match declared run_id: %s vs %s"%(val, run_id))
461
462
463
465 """
466 Launch a single L{Executable} object. Blocks until executable finishes.
467 @param e: Executable
468 @type e: L{Executable}
469 @raise RLException: if exectuable fails. Failure includes non-zero exit code.
470 """
471 try:
472
473 cmd = e.command
474 cmd = "%s %s"%(cmd, ' '.join(e.args))
475 print("running %s" % cmd)
476 local_machine = self.config.machines['']
477 env = setup_env(None, local_machine, self.config.master.uri)
478 retcode = subprocess.call(cmd, shell=True, env=env)
479 if retcode < 0:
480 raise RLException("command [%s] failed with exit code %s"%(cmd, retcode))
481 except OSError as e:
482 raise RLException("command [%s] failed: %s"%(cmd, e))
483
484
485
487 """
488 @raise RLException: if exectuable fails. Failure includes non-zero exit code.
489 """
490 exes = [e for e in self.config.executables if e.phase == PHASE_SETUP]
491 for e in exes:
492 self._launch_executable(e)
493
495 """
496 launch any core services that are not already running. master must
497 be already running
498 @raise RLException: if core launches fail
499 """
500 config = self.config
501 master = config.master.get()
502 tolaunch = []
503 for node in config.nodes_core:
504 node_name = rosgraph.names.ns_join(node.namespace, node.name)
505 code, msg, _ = master.lookupNode(_ID, node_name)
506 if code == -1:
507 tolaunch.append(node)
508 elif code == 1:
509 print("core service [%s] found" % node_name)
510 else:
511 print("WARN: master is not behaving well (unexpected return value when looking up node)", file=sys.stderr)
512 self.logger.error("ERROR: master return [%s][%s] on lookupNode call"%(code,msg))
513
514 for node in tolaunch:
515 node_name = rosgraph.names.ns_join(node.namespace, node.name)
516 name, success = self.launch_node(node, core=True)
517 if success:
518 print("started core service [%s]" % node_name)
519 else:
520 raise RLException("failed to start core service [%s]"%node_name)
521
523 """
524 Launch a single node locally. Remote launching is handled separately by the remote module.
525 If node name is not assigned, one will be created for it.
526
527 @param node Node: node to launch
528 @param core bool: if True, core node
529 @return obj, bool: Process handle, successful launch. If success, return actual Process instance. Otherwise return name.
530 """
531 self.logger.info("... preparing to launch node of type [%s/%s]", node.package, node.type)
532
533
534
535
536 if node.machine is None:
537 node.machine = self.config.machines['']
538 if node.name is None:
539 node.name = rosgraph.names.anonymous_name(node.type)
540
541 master = self.config.master
542 import roslaunch.node_args
543 try:
544 process = create_node_process(self.run_id, node, master.uri)
545 except roslaunch.node_args.NodeParamsException as e:
546 self.logger.error(e)
547 printerrlog("ERROR: cannot launch node of type [%s/%s]: %s"%(node.package, node.type, str(e)))
548 if node.name:
549 return node.name, False
550 else:
551 return "%s/%s"%(node.package,node.type), False
552
553 self.logger.info("... created process [%s]", process.name)
554 if core:
555 self.pm.register_core_proc(process)
556 else:
557 self.pm.register(process)
558 node.process_name = process.name
559 self.logger.info("... registered process [%s]", process.name)
560
561
562 success = process.start()
563 if not success:
564 if node.machine.name:
565 printerrlog("launch of %s/%s on %s failed"%(node.package, node.type, node.machine.name))
566 else:
567 printerrlog("local launch of %s/%s failed"%(node.package, node.type))
568 else:
569 self.logger.info("... successfully launched [%s]", process.name)
570 return process, success
571
573 """
574 Check for running node process.
575 @param node Node: node object to check
576 @return bool: True if process associated with node is running (launched && !dead)
577 """
578
579 return node.process_name and self.pm.has_process(node.process_name)
580
582 """
583 Same as spin() but only does one cycle. must be run from the main thread.
584 """
585 if not self.pm:
586 return False
587 return self.pm.mainthread_spin_once()
588
590 """
591 spin() must be run from the main thread. spin() is very
592 important for roslaunch as it picks up jobs that the process
593 monitor need to be run in the main thread.
594 """
595 self.logger.info("spin")
596
597
598
599 if not self.pm or not self.pm.get_active_names():
600 printlog_bold("No processes to monitor")
601 self.stop()
602 return
603 self.pm.mainthread_spin()
604
605 self.logger.info("process monitor is done spinning, initiating full shutdown")
606 self.stop()
607 printlog_bold("done")
608
610 """
611 Stop the launch and all associated processes. not thread-safe.
612 """
613 self.logger.info("runner.stop()")
614 if self.pm is not None:
615 printlog("shutting down processing monitor...")
616 self.logger.info("shutting down processing monitor %s"%self.pm)
617 self.pm.shutdown()
618 self.pm.join()
619 self.pm = None
620 printlog("... shutting down processing monitor complete")
621 else:
622 self.logger.info("... roslaunch runner has already been stopped")
623
625 """
626 Setup the state of the ROS network, including the parameter
627 server state and core services
628 """
629
630 self.config.assign_machines()
631
632 if self.remote_runner:
633
634 self.remote_runner.add_process_listener(self.listeners)
635
636
637 launched = self._launch_master()
638 if launched:
639 self._launch_core_nodes()
640
641
642
643
644
645 self._launch_setup_executables()
646
647
648 if not self.is_child:
649 self._load_parameters()
650
652 """
653 Run the launch. Depending on usage, caller should call
654 spin_once or spin as appropriate after launch().
655 @return ([str], [str]): tuple containing list of nodes that
656 successfully launches and list of nodes that failed to launch
657 @rtype: ([str], [str])
658 @raise RLException: if launch fails (e.g. run_id parameter does
659 not match ID on parameter server)
660 """
661 try:
662 self._setup()
663 succeeded, failed = self._launch_nodes()
664 return succeeded, failed
665 except KeyboardInterrupt:
666 self.stop()
667 raise
668
670 """
671 Run the test node. Blocks until completion or timeout.
672 @param test: test node to run
673 @type test: Test
674 @raise RLTestTimeoutException: if test fails to launch or test times out
675 """
676 self.logger.info("... preparing to run test [%s] of type [%s/%s]", test.test_name, test.package, test.type)
677 proc_h, success = self.launch_node(test)
678 if not success:
679 raise RLException("test [%s] failed to launch"%test.test_name)
680
681
682 timeout_t = time.time() + test.time_limit
683 pm = self.pm
684 while pm.mainthread_spin_once() and self.is_node_running(test):
685
686 if time.time() > timeout_t:
687 raise RLTestTimeoutException(
688 "max time [%ss] allotted for test [%s] of type [%s/%s]" %
689 (test.time_limit, test.test_name, test.package, test.type))
690 time.sleep(0.1)
691
692
693
694
695
697 """
698 API for remote running
699 """
705 """
706 Listen to events about remote processes dying. Not
707 threadsafe. Must be called before processes started.
708 @param l: listener
709 @type l: L{ProcessListener}
710 """
711 pass
712
714 """
715 Contact each child to launch remote nodes
716 @return: succeeded, failed
717 @rtype: [str], [str]
718 """
719 pass
720