Skip to content

Allow creating multiple channels #212

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: rolling
Choose a base branch
from
Open

Allow creating multiple channels #212

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
96b513b
Allow creating multiple channels
ShravanDeva5327 Sep 21, 2025
4ea8bfd
Add multiple channel tests
ShravanDeva5327 Sep 21, 2025
2407413
Fix code-style warnings
ShravanDeva5327 Sep 21, 2025
16b63d5
Fix ros2trace test failures
ShravanDeva5327 Sep 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion test_tracetools_launch/test/test_tracetools_launch/test_trace_action.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import shutil
import tempfile
import textwrap
from typing import Iterable
from typing import List
from typing import Optional
from typing import TextIO
Expand Down Expand Up @@ -116,8 +117,14 @@ def _check_trace_action(
perform_typed_substitution(context, action.append_trace, bool)
)
self.assertEqual(0, len(action.events_kernel))
# Covert events_ust to list of lists if needed
if all(not isinstance(x, Iterable) or isinstance(x, str) for x in events_ust):
events_ust = [events_ust]
self.assertEqual(
events_ust, [perform_substitutions(context, x) for x in action.events_ust])
events_ust,
[[perform_substitutions(context, x) for x in channel_events]
for channel_events in action.events_ust],
)
self.assertEqual(
subbuffer_size_ust,
perform_typed_substitution(context, action.subbuffer_size_ust, int)
Expand Down Expand Up @@ -617,6 +624,26 @@ def test_append_trace(self) -> None:

shutil.rmtree(tmpdir)

def test_multiple_channels(self) -> None:
tmpdir = tempfile.mkdtemp(prefix='TestTraceAction__test_multiple_channels')

action = Trace(
session_name='my-session-name',
base_path=tmpdir,
events_kernel=[],
syscalls=[],
events_ust=[
['ros2:*'],
['*'],
],
subbuffer_size_ust=524288,
subbuffer_size_kernel=1048576,
)
context = self._assert_launch_no_errors([action])
self._check_trace_action(action, context, tmpdir, events_ust=[['ros2:*'], ['*']])

shutil.rmtree(tmpdir)


if __name__ == '__main__':
unittest.main()
61 changes: 47 additions & 14 deletions tracetools_launch/tracetools_launch/action.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import re
import shlex
from typing import cast
from typing import Generator
from typing import Iterable
from typing import List
from typing import Mapping
Expand Down Expand Up @@ -154,8 +155,11 @@ def __init__(
:param events_ust: the list of ROS UST events to enable; if it's `None`, the default ROS
events are used for a normal session, and the default ROS initialization events are
used for the snapshot session in case of a dual session; if it's an empty list, no UST
events are enabled
:param events_kernel: the list of kernel events to enable
events are enabled; if a list of lists is provided, one channel per list is created
with the corresponding UST events enabled in each channel
:param events_kernel: the list of kernel events to enable; if a list of lists is provided,
one channel per list is created with the corresponding kernel events enabled in each
channel
Comment on lines +158 to +162
Copy link
Member

@christophebedard christophebedard Sep 24, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The type annotations for events_ust and events_kernel need to be updated; the typing-related test failures you see are probably related to this. However, it kind of highlights an issue I hadn't thought of, and I'm not quite sure how to solve it cleanly. I apologize for the long comment, but I think some context/explanation might be helpful.

Currently, events_ust is Optional[Iterable[SomeSubstitutionsType]]. It's kind of tricky, because SomeSubstitutionsType can be a single string/substitution, or it can be an iterable of strings/substitutions: https://github.com/ros2/launch/blob/1abf55ef6d3874ab5a4656e9ac0ea5d057c3c541/launch/launch/some_substitutions_type.py#L25-L30. If it's just a single string/substitution, then that's just 1 value, i.e., one event name. If it's an iterable, then it's still just 1 value: strings and substitutions (after being resolved) in the iterable will get concatenated into a single string. This is how substitutable inputs work.

Then, Iterable[SomeSubstitutionsType] means we have a list of values (where, again, each "value" can be a single string or an iterable of strings/substitutions that gets resolved and concatenated into a single value). normalize_to_list_of_substitutions() converts each SomeSubstitutionsType in that Iterable into a list of substitutions.

Since we want to support providing a list of this, we'd need something like List[Iterable[SomeSubstitutionsType]] (or Iterable[Iterable[SomeSubstitutionsType]]). The issue is that we don't know if it's a list of lists of individual values (single substitutions) or a list of lists that each represents a single value. Note that the output after normalization is definitely List[List[List[Substitution]]], though, like you did:

List[  # list of lists of events
    List[  # list of events
        List[Substitution]  # single event (potentially made up of multiple substitutions that get resolved and concatenated into a single value, i.e., a single event name)
    ]
]

To illustrate this, these two lists (before this PR) are functionally equivalent after normalization and substitution resolution:

events_ust_a = [
    'ros2:*',
    'ros2:rcl_publish',
    'ros2:rclcpp_publish',
]
events_ust_b = [
    'ros2:*',  # or ['ros2:*']
    ['ros2:rcl_publish'],
    ['ros2:', 'rclcpp_publish'],
]

With this PR, we'd like to do this:

events_ust_c = [
    [  # one list of events for a channel
        'ros2:*',
        ['ros2:rcl_publish'],
        ['ros2:', 'rclcpp_publish'],
    ],
    [  # another list of events for another channel
        'ros2:rmw_take',
        'ros2:rcl_take',
    ],
]

The problem is: How can you know if events_ust_c is a list of lists of events and not a list of lists of substitutions like events_ust_b? ['ros2:*', ['ros2:', 'rcl_publish']] is a single list of event names, but all(not isinstance(x, Iterable) or isinstance(x, (str, Generator)) for x in ['ros2:*', ['ros2:', 'rcl_publish']]) is False, meaning that the proposed code thinks it's actually a list of lists of events, when it's actually a single list of events.

An option might be to use sets to have separate lists of events, then you could do this:

if events_ust and not isinstance(events_ust, set):
    # Single list
     events_ust = [events_ust]
else:
    # Multiple lists
    events_ust = list(events_ust)

However, there's a fundamental flaw with this. We currently accept an Iterable for events_ust, so a single "list" of events might actually be a single set of events, since a set is an Iterable. I'm not quite sure how to solve this in a clean and backwards-compatible way, but a 100% safe option would be to have a separate input for multiple (channel-specific) lists of events, e.g., events_ust_channels. If events_ust_channels is used (not None and not empty), then events_ust is ignored. This way, there is no doubt about whether the input is a list of lists of events or a single list of events. I don't like having a separate option, but since this is a more "advanced" feature, it might be acceptable.

What do you think? Did I miss anything?

:param syscalls: the list of syscalls to enable
:param context_fields: the names of context fields to enable
if it's a list or a set, the context fields are enabled for both kernel and userspace;
Expand Down Expand Up @@ -194,8 +198,23 @@ def __init__(
if events_ust is None:
events_ust = names.DEFAULT_EVENTS_ROS if not self._dual_session \
else names.DEFAULT_INIT_EVENTS_ROS
self._events_ust = [normalize_to_list_of_substitutions(x) for x in events_ust]
self._events_kernel = [normalize_to_list_of_substitutions(x) for x in events_kernel]
# Convert to list of lists if a single list is provided
if events_ust and all(
not isinstance(x, Iterable) or isinstance(x, (str, Generator)) for x in events_ust
):
events_ust = [events_ust]
if events_kernel and all(
not isinstance(x, Iterable) or isinstance(x, (str, Generator)) for x in events_kernel
):
events_kernel = [events_kernel]
self._events_ust = [
[normalize_to_list_of_substitutions(x) for x in channel_events_ust]
for channel_events_ust in events_ust
]
self._events_kernel = [
[normalize_to_list_of_substitutions(x) for x in channel_events_kernel]
for channel_events_kernel in events_kernel
]
self._syscalls = [normalize_to_list_of_substitutions(x) for x in syscalls]
self._context_fields: Union[
Mapping[str, List[List[Substitution]]], List[List[Substitution]]
Expand Down Expand Up @@ -236,11 +255,11 @@ def trace_directory(self) -> Optional[str]:
return self._trace_directory

@property
def events_ust(self) -> List[List[Substitution]]:
def events_ust(self) -> List[List[List[Substitution]]]:
return self._events_ust

@property
def events_kernel(self) -> List[List[Substitution]]:
def events_kernel(self) -> List[List[List[Substitution]]]:
return self._events_kernel

@property
Expand Down Expand Up @@ -320,6 +339,9 @@ def _append_arg():
arg.append(sub)
if arg:
result_args.append(arg)

# Convert list of lists of substitutions to list of substitutions
result_args = [(arg for arg in sublist) for sublist in result_args]
return result_args

@classmethod
Expand Down Expand Up @@ -446,8 +468,14 @@ def execute(self, context: LaunchContext) -> List[Action]:
dual_session = perform_typed_substitution(context, self._dual_session, bool)
base_path = perform_substitutions(context, self._base_path)
append_trace = perform_typed_substitution(context, self._append_trace, bool)
events_ust = [perform_substitutions(context, x) for x in self._events_ust]
events_kernel = [perform_substitutions(context, x) for x in self._events_kernel]
events_ust = [
[perform_substitutions(context, x) for x in channel_events_ust]
for channel_events_ust in self._events_ust
]
events_kernel = [
[perform_substitutions(context, x) for x in channel_events_kernel]
for channel_events_kernel in self._events_kernel
]
syscalls = [perform_substitutions(context, x) for x in self._syscalls]
context_fields = (
{
Expand Down Expand Up @@ -516,18 +544,23 @@ def destroy(event: Event, context: LaunchContext) -> None:
context.register_event_handler(OnShutdown(on_shutdown=destroy))
return self._ld_preload_actions

def _get_ld_preload_actions(self, events_ust: List[str]) -> List[Action]:
def _get_ld_preload_actions(self, events_ust: List[List[str]]) -> List[Action]:
ld_preload_actions: List[Action] = []
# Add LD_PRELOAD actions if corresponding events are enabled
if self.has_libc_wrapper_events(events_ust):
if any(self.has_libc_wrapper_events(channel_events_ust)
for channel_events_ust in events_ust):
ld_preload_actions.append(LdPreload(self.LIB_LIBC_WRAPPER))
if self.has_pthread_wrapper_events(events_ust):
if any(self.has_pthread_wrapper_events(channel_events_ust)
for channel_events_ust in events_ust):
ld_preload_actions.append(LdPreload(self.LIB_PTHREAD_WRAPPER))
if self.has_dl_events(events_ust):
if any(self.has_dl_events(channel_events_ust)
for channel_events_ust in events_ust):
ld_preload_actions.append(LdPreload(self.LIB_DL))
# Warn if events match both normal AND fast profiling libs
has_fast_profiling_events = self.has_profiling_events(events_ust, True)
has_normal_profiling_events = self.has_profiling_events(events_ust, False)
has_fast_profiling_events = any(self.has_profiling_events(channel_events_ust, True)
for channel_events_ust in events_ust)
has_normal_profiling_events = any(self.has_profiling_events(channel_events_ust, False)
for channel_events_ust in events_ust)
# In practice, the first lib in the LD_PRELOAD list will be used, so the fast one here
if has_fast_profiling_events:
ld_preload_actions.append(LdPreload(self.LIB_PROFILE_FAST))
Expand Down
Loading
Loading