Add an event sequence API (ros2#4)

Signed-off-by: Michael Carroll <[email protected]>

ros2_profiling_demo/test/conftest.py

@@ -1,38 +1,11 @@
 import pytest
 
-import os
-import yaml
-
-from ros2profile.api.process import load_files, load_trace
+from ros2profile.api.process import load_mcap_data, load_event_graph
 
 def pytest_addoption(parser):
     parser.addoption("--input-dir", action="store")
 
 
-class ProfileDataModel():
-    def __init__(self, input_dir):
-        self.input_dir = input_dir
-
-        with open(os.path.join(input_dir, 'config.yaml'), 'r') as f:
-            self.config = yaml.load(f, Loader=yaml.SafeLoader)
-        self.trace_data = load_trace(input_dir)
-        self.mcap_data = load_files(input_dir)
-
-    def containers(self):
-        return self.config['containers']
-
-    def nodes(self):
-        return self.config['nodes']
-
-    def node_handle(self, node_name):
-        return self.trace_data.data.nodes[
-            self.trace_data.data.nodes.name == node_name
-        ].to_dict()
-
-    def cpu_memory_usage(self, container):
-        return self.mcap_data[container]['~/cpu_memory_usage']
-
-
 @pytest.fixture(scope='session')
 def input_dir(request):
     input_dir = request.config.option.input_dir
@@ -45,4 +18,12 @@ def profile_data(request):
     input_dir = request.config.option.input_dir
     if input_dir is None:
         pytest.skip()
-    return ProfileDataModel(input_dir)
+    return load_mcap_data(input_dir)
+
+@pytest.fixture(scope='session')
+def profile_event_graph(request):
+    input_dir = request.config.option.input_dir
+    if input_dir is None:
+        pytest.skip()
+    return load_event_graph(input_dir)
+

ros2_profiling_demo/test/test_event_seq.py

@@ -0,0 +1,42 @@
+import numpy as np
+
+from ros2profile.data.event_sequence import EventSequence
+
+def test_sequence(profile_event_graph):
+    graph = profile_event_graph
+    nodes = graph.nodes('arequipa')
+    assert len(nodes) == 1
+    arequipa = nodes[0]
+
+    subs = arequipa.subscriptions()
+    assert len(subs) == 1
+
+    arkansas_sub = subs[0]
+
+    sub_events = arkansas_sub.callback().events()
+    assert len(sub_events) > 0
+
+    events = []
+    for event in sub_events:
+        events.append(EventSequence(event))
+
+    # We are expecting 26 events in the chain
+    # Depending on which tracepoints are available
+    assert len(events[0].sequence) < 30
+
+    latencies = [e.latency() for e in events]
+
+    # Normalize to seconds
+    latencies = np.array(latencies)/1e9
+
+    # Assert mean latency is less than 1 ms
+    assert np.mean(latencies) < 1e-3
+
+    # Assert jitter is less than 100 microseconds
+    assert np.std(latencies) < 1e-4
+
+    # Assert max latency is less than 1 ms
+    assert np.max(latencies) < 1e-3
+
+    # Assert median latency is less than 1 ms
+    assert np.median(latencies) < 1e-3

ros2profile/ros2profile/data/event_sequence.py

@@ -0,0 +1,95 @@
+from typing import Any, List
+
+from .callback import CallbackEvent
+from .timer import Timer
+from .subscription import SubscriptionEvent
+from .publisher import PublishEvent
+
+class EventSequence():
+    def __init__(self, end_event, start_event=None):
+        self.start_event = start_event
+        self.end_event = end_event
+        self.sequence: List[Any] = []
+
+        self._build_sequence(self.end_event, self.start_event)
+
+    def latency(self):
+        return self.sequence[0]['timestamp'] - self.sequence[-1]['timestamp']
+
+    def _build_sequence(self, end_event, start_event=None):
+        event = end_event
+        self.sequence = []
+
+        while event and event != start_event:
+            if type(event) is CallbackEvent:
+                if type(event.source()) is Timer:
+                    self.sequence.append({
+                        'node': event.source().node().name(),
+                        'event': 'callback_end',
+                        'topic': 'timer',
+                        'timestamp': event.end()
+                    })
+                    self.sequence.append({
+                        'node': event.source().node().name(),
+                        'event': 'callback_start',
+                        'topic': 'timer',
+                        'timestamp': event.start()
+                    })
+                else:
+                    self.sequence.append({
+                        'node': event.source().node().name(),
+                        'event': 'callback_end',
+                        'topic': event.source().topic_name(),
+                        'timestamp': event.end()
+                    })
+                    self.sequence.append({
+                        'node': event.source().node().name(),
+                        'event': 'callback_start',
+                        'topic': event.source().topic_name(),
+                        'timestamp': event.start()
+                    })
+
+                event = event.trigger()
+            elif type(event) is SubscriptionEvent:
+                self.sequence.append({
+                    'node': event.source().node().name(),
+                    'event': 'rclcpp_take',
+                    'topic': event.source().topic_name(),
+                    'timestamp': event._rclcpp_init_time
+                })
+                self.sequence.append({
+                    'node': event.source().node().name(),
+                    'event': 'rcl_take',
+                    'topic': event.source().topic_name(),
+                    'timestamp': event._rcl_init_time
+                })
+                self.sequence.append({
+                    'node': event.source().node().name(),
+                    'event': 'rmw_take',
+                    'topic': event.source().topic_name(),
+                    'timestamp': event._rmw_init_time
+                })
+                event = event.trigger()
+            elif type(event) is PublishEvent:
+                self.sequence.append({
+                    'node': event.source().node().name(),
+                    'event': 'rclcpp_pub',
+                    'topic': event.source().topic_name(),
+                    'timestamp': event._rclcpp_init_time
+                })
+                self.sequence.append({
+                    'node': event.source().node().name(),
+                    'event': 'rcl_pub',
+                    'topic': event.source().topic_name(),
+                    'timestamp': event._rcl_init_time
+                })
+                self.sequence.append({
+                    'node': event.source().node().name(),
+                    'event': 'rmw_pub',
+                    'topic': event.source().topic_name(),
+                    'timestamp': event._rmw_init_time
+                })
+
+                event = event.trigger()
+            else:
+                break