diff --git a/.gitignore b/.gitignore index b793570..d3e032b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,5 @@ # Ignore all pycache files **/__pycache__/** + +# Ignore generated framework-ceiling benchmark results +ceiling_benchmark_results/ diff --git a/README.md b/README.md index 55c1571..8ca76a1 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,33 @@ The benchmark generates: - Perf profiling reports (if enabled) - Symlinks to latest results for easy access +## Framework-Ceiling Microbenchmarks + +The repository also includes focused CPU-only microbenchmarks for measuring executor dispatch and +minimal intra-process message-passing ceilings. They use continuously ready `rclcpp::Waitable` +sources: + +- `int64_ceiling_benchmark`: minimal `std_msgs/msg/Int64` source-to-sink flows +- `scheduler_ceiling_benchmark`: executor dispatch with no message transport + +After building and sourcing the workspace, use the YAML-driven runner to sweep EventsCBGExecutor +worker threads, flows/operators, and repeated runs: + +```bash +python3 src/ros2_framework_perf/scripts/run_ceiling_benchmarks.py \ + --config src/ros2_framework_perf/config/framework_ceiling.yaml +``` + +Summarize repeated-run medians and validate executor-dispatch invariants: + +```bash +python3 src/ros2_framework_perf/scripts/summarize_ceiling_results.py \ + ceiling_benchmark_results/ +``` + +See [Framework-Ceiling Microbenchmarks](docs/framework_ceiling.md) for methodology, metrics, and +limitations. + # Running the benchmark ## Environment diff --git a/config/framework_ceiling.yaml b/config/framework_ceiling.yaml new file mode 100644 index 0000000..ed65959 --- /dev/null +++ b/config/framework_ceiling.yaml @@ -0,0 +1,28 @@ +# Framework-ceiling microbenchmarks use continuously ready waitables and minimal callback work. +schema_version: 1 +output_directory: ceiling_benchmark_results +repetitions: 5 +executor: events_cbg + +message_passing: + enabled: true + messages_per_flow: 100000 + timeout_seconds: 60 + matrix: + - {threads: 1, flows: 1} + - {threads: 2, flows: 2} + - {threads: 4, flows: 4} + - {threads: 8, flows: 8} + - {threads: 2, flows: 8} + +scheduler: + enabled: true + operations_per_operator: 100000 + timeout_seconds: 60 + matrix: + - {threads: 1, operators: 1} + - {threads: 2, operators: 2} + - {threads: 4, operators: 4} + - {threads: 8, operators: 8} + - {threads: 16, operators: 16} + - {threads: 2, operators: 8} diff --git a/docs/framework_ceiling.md b/docs/framework_ceiling.md new file mode 100644 index 0000000..ed6d9b2 --- /dev/null +++ b/docs/framework_ceiling.md @@ -0,0 +1,134 @@ +# ROS 2 Framework-Ceiling Microbenchmarks + +The framework-ceiling microbenchmarks complement the configurable application-graph benchmark. +They intentionally remove graph processing and rich message metadata to isolate two lower-level +limits: + +1. Executor-dispatched, intra-process message throughput and callback latency using + `std_msgs/msg/Int64`. +2. Executor dispatch throughput with no message transport or application work. + +Both benchmarks use continuously ready, guard-condition-backed `rclcpp::Waitable` instances. + +## Why a separate microbenchmark path? + +The configurable `EmitterNode` graph benchmark measures application-like message journeys. Its +custom payload metadata, lifecycle nodes, composable container, graph stages, and instrumentation +are useful parts of that measurement. + +The ceiling benchmarks answer a narrower question: when application work is nearly zero, how fast +can an executor dispatch ready work and move a tiny intra-process message? Keeping these paths +separate makes the source of a throughput limit easier to identify. + +The benchmarks do not modify ROS 2. They use public `rclcpp` APIs, including +`EventsCBGExecutor`, `Waitable`, and `GuardCondition`. + +## Benchmarks + +### Int64 message passing + +Each flow consists of one source node and one sink node in the same process: + +```text +always-ready source waitable -> publish Int64 timestamp -> subscription callback +``` + +The source waitable publishes one message per executor dispatch and retriggers itself until the +configured message count is reached. The sink computes latency from the timestamp sampled +immediately before `publish()` to the beginning of the subscription callback. + +The primary metrics are: + +- `source_publish_msg_s` +- `throughput_msg_s` +- average, minimum, and maximum callback latency +- source-to-sink drain lag + +### Scheduler dispatch + +Each scheduler node owns one always-ready waitable. Every executor dispatch increments a counter +and retriggers the waitable until the configured operation count is reached. There is no publisher, +subscription, or payload. + +The primary metric is `throughput_ops_s`. + +### Executor-dispatch invariant + +Every result includes waitable trace counters. A successful run requires: + +```text +waitable_execute_count == published_messages +``` + +for message passing, or: + +```text +waitable_execute_count == total_operations +``` + +for scheduler dispatch. This confirms the measured work ran from executor-dispatched +`Waitable::execute()` calls rather than directly from the initial trigger or guard-condition +callback. + +## Run the matrix + +Build the workspace as described in the repository README, then: + +```bash +source /opt/ros/rolling/setup.bash +source install/setup.bash + +python3 src/ros2_framework_perf/scripts/run_ceiling_benchmarks.py \ + --config src/ros2_framework_perf/config/framework_ceiling.yaml +``` + +The default YAML runs five repetitions of each matrix cell with `EventsCBGExecutor`. Edit or copy +the YAML to change thread counts, flow/operator counts, operation counts, or repetitions. + +Validate a configuration without executing it: + +```bash +python3 src/ros2_framework_perf/scripts/run_ceiling_benchmarks.py \ + --config src/ros2_framework_perf/config/framework_ceiling.yaml \ + --dry-run +``` + +Run only one benchmark family: + +```bash +python3 src/ros2_framework_perf/scripts/run_ceiling_benchmarks.py \ + --config src/ros2_framework_perf/config/framework_ceiling.yaml \ + --benchmark message_passing +``` + +## Summarize results + +The runner writes one JSON file per repetition and matrix cell. Summarize repeated runs and validate +result completeness and executor-dispatch invariants with: + +```bash +python3 src/ros2_framework_perf/scripts/summarize_ceiling_results.py \ + ceiling_benchmark_results/ +``` + +The summarizer writes `summary.json` and `summary.csv`. It exits nonzero if any run is incomplete, +an invariant fails, or a required metric is missing. + +## Methodology and limitations + +- These are saturation tests. Sources remain continuously ready and do not represent an + application-selected publish rate. +- The message-passing benchmark uses an 8-byte `Int64` timestamp and intra-process communication. + It does not measure DDS serialization, networking, cross-process IPC, large payload transfer, GPU + inference, or application callback work. +- The scheduler benchmark measures a minimal counter operation. It is useful for executor + comparison, not as an application throughput prediction. +- More threads do not guarantee higher throughput. Queue synchronization, cache contention, and + competition between continuously ready source work and sink callbacks can dominate. +- Latency begins immediately before `publish()`. Source scheduling delay before that timestamp is + intentionally outside the latency interval. +- Performance results depend on the ROS distribution, `rclcpp` version, build type, compiler, + hardware, kernel, CPU configuration, and system load. Record these inputs and compare + repeated-run medians rather than isolated runs. +- Rolling changes continuously. For reproducible published results, record the container image + digest and package versions used for a run. diff --git a/ros2_framework_perf/CMakeLists.txt b/ros2_framework_perf/CMakeLists.txt index 2c91282..eb57810 100644 --- a/ros2_framework_perf/CMakeLists.txt +++ b/ros2_framework_perf/CMakeLists.txt @@ -52,11 +52,51 @@ target_link_libraries(emitter_node ) rclcpp_components_register_nodes(emitter_node "ros2_framework_perf::EmitterNode") +ament_auto_add_executable(int64_ceiling_benchmark + src/int64_ceiling_benchmark.cpp +) +target_compile_definitions(int64_ceiling_benchmark + PRIVATE + ROS2_FRAMEWORK_PERF_BUILD_TYPE="${CMAKE_BUILD_TYPE}" +) + +ament_auto_add_executable(scheduler_ceiling_benchmark + src/scheduler_ceiling_benchmark.cpp +) +target_compile_definitions(scheduler_ceiling_benchmark + PRIVATE + ROS2_FRAMEWORK_PERF_BUILD_TYPE="${CMAKE_BUILD_TYPE}" +) + if(BUILD_TESTING) + find_package(ament_cmake_pytest REQUIRED) find_package(ament_lint_auto REQUIRED) + find_package(Python3 REQUIRED COMPONENTS Interpreter) ament_lint_auto_find_test_dependencies() find_package(launch_testing_ament_cmake REQUIRED) + ament_add_pytest_test( + test_ceiling_tools + test/test_ceiling_tools.py + ) + + add_test( + NAME test_int64_ceiling_smoke + COMMAND + Python3::Interpreter + ${CMAKE_CURRENT_SOURCE_DIR}/test/verify_ceiling_executable.py + $ + message_passing + ) + add_test( + NAME test_scheduler_ceiling_smoke + COMMAND + Python3::Interpreter + ${CMAKE_CURRENT_SOURCE_DIR}/test/verify_ceiling_executable.py + $ + scheduler + ) + # Add launch test add_launch_test( test/test_emitter_launch.py @@ -67,4 +107,4 @@ if(BUILD_TESTING) endif() -ament_auto_package(INSTALL_TO_SHARE ../scripts launch) \ No newline at end of file +ament_auto_package(INSTALL_TO_SHARE ../config ../docs ../scripts launch) diff --git a/ros2_framework_perf/include/ros2_framework_perf/ceiling_benchmark_utils.hpp b/ros2_framework_perf/include/ros2_framework_perf/ceiling_benchmark_utils.hpp new file mode 100644 index 0000000..351ccb1 --- /dev/null +++ b/ros2_framework_perf/include/ros2_framework_perf/ceiling_benchmark_utils.hpp @@ -0,0 +1,222 @@ +// SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +// Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// SPDX-License-Identifier: Apache-2.0 +// SPDX-Generated-By: Cursor + +#ifndef ROS2_FRAMEWORK_PERF__CEILING_BENCHMARK_UTILS_HPP_ +#define ROS2_FRAMEWORK_PERF__CEILING_BENCHMARK_UTILS_HPP_ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "rcl/wait.h" +#include "rclcpp/guard_condition.hpp" +#include "rclcpp/rclcpp.hpp" +#include "rclcpp/waitable.hpp" + +namespace ros2_framework_perf +{ +namespace ceiling_benchmark +{ + +inline int64_t EnvInt(const char * name, const int64_t default_value) +{ + const char * value = std::getenv(name); + if (value == nullptr || std::string(value).empty()) { + return default_value; + } + return std::stoll(value); +} + +inline uint64_t NowNs() +{ + return static_cast( + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count()); +} + +inline std::string JsonEscape(const std::string & value) +{ + std::string result; + result.reserve(value.size()); + for (const char character : value) { + switch (character) { + case '"': + result += "\\\""; + break; + case '\\': + result += "\\\\"; + break; + case '\n': + result += "\\n"; + break; + case '\r': + result += "\\r"; + break; + case '\t': + result += "\\t"; + break; + default: + result += character; + break; + } + } + return result; +} + +struct WaitableTraceCounts +{ + uint64_t trigger{0}; + uint64_t on_ready_callback{0}; + uint64_t add_to_wait_set{0}; + uint64_t is_ready{0}; + uint64_t is_ready_true{0}; + uint64_t take_data{0}; + uint64_t take_data_by_entity_id{0}; + uint64_t execute{0}; +}; + +class AlwaysReadyWaitable : public rclcpp::Waitable +{ +public: + explicit AlwaysReadyWaitable(std::function on_execute) + : on_execute_(std::move(on_execute)), + guard_condition_(std::make_shared()) + { + } + + size_t get_number_of_ready_guard_conditions() override + { + return 1; + } + + void add_to_wait_set(rcl_wait_set_t & wait_set) override + { + add_to_wait_set_count_.fetch_add(1, std::memory_order_relaxed); + const rcl_ret_t result = rcl_wait_set_add_guard_condition( + &wait_set, &guard_condition_->get_rcl_guard_condition(), &wait_set_guard_condition_index_); + if (result != RCL_RET_OK) { + throw std::runtime_error("Failed to add ceiling benchmark guard condition to wait set"); + } + } + + bool is_ready(const rcl_wait_set_t & wait_set) override + { + is_ready_count_.fetch_add(1, std::memory_order_relaxed); + const bool ready = + wait_set.guard_conditions[wait_set_guard_condition_index_] != nullptr; + if (ready) { + is_ready_true_count_.fetch_add(1, std::memory_order_relaxed); + } + return ready; + } + + std::shared_ptr take_data() override + { + take_data_count_.fetch_add(1, std::memory_order_relaxed); + return nullptr; + } + + std::shared_ptr take_data_by_entity_id(size_t) override + { + take_data_by_entity_id_count_.fetch_add(1, std::memory_order_relaxed); + return nullptr; + } + + std::vector> get_timers() const override + { + return {}; + } + + void execute(const std::shared_ptr &) override + { + execute_count_.fetch_add(1, std::memory_order_relaxed); + on_execute_(); + } + + void set_on_ready_callback(std::function callback) override + { + guard_condition_->set_on_trigger_callback( + [this, callback](size_t number_of_events) { + on_ready_callback_count_.fetch_add(1, std::memory_order_relaxed); + callback(number_of_events, 0); + }); + } + + void clear_on_ready_callback() override + { + guard_condition_->set_on_trigger_callback(nullptr); + } + + void Trigger() + { + trigger_count_.fetch_add(1, std::memory_order_relaxed); + guard_condition_->trigger(); + } + + WaitableTraceCounts TraceCounts() const + { + return WaitableTraceCounts{ + trigger_count_.load(std::memory_order_relaxed), + on_ready_callback_count_.load(std::memory_order_relaxed), + add_to_wait_set_count_.load(std::memory_order_relaxed), + is_ready_count_.load(std::memory_order_relaxed), + is_ready_true_count_.load(std::memory_order_relaxed), + take_data_count_.load(std::memory_order_relaxed), + take_data_by_entity_id_count_.load(std::memory_order_relaxed), + execute_count_.load(std::memory_order_relaxed)}; + } + +private: + std::function on_execute_; + std::shared_ptr guard_condition_; + size_t wait_set_guard_condition_index_{0}; + std::atomic trigger_count_{0}; + std::atomic on_ready_callback_count_{0}; + std::atomic add_to_wait_set_count_{0}; + std::atomic is_ready_count_{0}; + std::atomic is_ready_true_count_{0}; + std::atomic take_data_count_{0}; + std::atomic take_data_by_entity_id_count_{0}; + std::atomic execute_count_{0}; +}; + +inline void Accumulate( + WaitableTraceCounts & total, + const WaitableTraceCounts & value) +{ + total.trigger += value.trigger; + total.on_ready_callback += value.on_ready_callback; + total.add_to_wait_set += value.add_to_wait_set; + total.is_ready += value.is_ready; + total.is_ready_true += value.is_ready_true; + total.take_data += value.take_data; + total.take_data_by_entity_id += value.take_data_by_entity_id; + total.execute += value.execute; +} + +} // namespace ceiling_benchmark +} // namespace ros2_framework_perf + +#endif // ROS2_FRAMEWORK_PERF__CEILING_BENCHMARK_UTILS_HPP_ diff --git a/ros2_framework_perf/package.xml b/ros2_framework_perf/package.xml index 01f851d..72cd1f7 100644 --- a/ros2_framework_perf/package.xml +++ b/ros2_framework_perf/package.xml @@ -32,6 +32,7 @@ SPDX-License-Identifier: Apache-2.0 ament_cmake_auto + rcl rclcpp rclcpp_action rclcpp_components @@ -42,6 +43,9 @@ SPDX-License-Identifier: Apache-2.0 yaml-cpp message_filters + python3-yaml + + ament_cmake_pytest ament_lint_auto ament_lint_common diff --git a/ros2_framework_perf/src/int64_ceiling_benchmark.cpp b/ros2_framework_perf/src/int64_ceiling_benchmark.cpp new file mode 100644 index 0000000..51834a5 --- /dev/null +++ b/ros2_framework_perf/src/int64_ceiling_benchmark.cpp @@ -0,0 +1,417 @@ +// SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +// Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// SPDX-License-Identifier: Apache-2.0 +// SPDX-Generated-By: Cursor + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "rclcpp/executors/events_cbg_executor/events_cbg_executor.hpp" +#include "rclcpp/rclcpp.hpp" +#include "ros2_framework_perf/ceiling_benchmark_utils.hpp" +#include "std_msgs/msg/int64.hpp" + +#ifndef ROS2_FRAMEWORK_PERF_BUILD_TYPE +#define ROS2_FRAMEWORK_PERF_BUILD_TYPE "unknown" +#endif + +namespace +{ + +using Int64 = std_msgs::msg::Int64; +using ros2_framework_perf::ceiling_benchmark::Accumulate; +using ros2_framework_perf::ceiling_benchmark::AlwaysReadyWaitable; +using ros2_framework_perf::ceiling_benchmark::EnvInt; +using ros2_framework_perf::ceiling_benchmark::JsonEscape; +using ros2_framework_perf::ceiling_benchmark::NowNs; +using ros2_framework_perf::ceiling_benchmark::WaitableTraceCounts; + +class SourceNode : public rclcpp::Node +{ +public: + SourceNode( + const size_t index, + const int64_t message_count, + const int64_t qos_depth) + : Node( + "ceiling_source_" + std::to_string(index), + rclcpp::NodeOptions().use_intra_process_comms(true)), + message_count_(message_count) + { + const auto qos = rclcpp::QoS(static_cast(qos_depth)); + publisher_ = create_publisher( + "ceiling_flow_" + std::to_string(index), qos); + callback_group_ = create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive); + waitable_ = std::make_shared([this]() {PublishOnce();}); + get_node_waitables_interface()->add_waitable(waitable_, callback_group_); + } + + void Kick() + { + waitable_->Trigger(); + } + + uint64_t Published() const + { + return published_.load(std::memory_order_relaxed); + } + + uint64_t FirstPublishNs() const + { + return first_publish_ns_.load(std::memory_order_relaxed); + } + + uint64_t LastPublishNs() const + { + return last_publish_ns_.load(std::memory_order_relaxed); + } + + WaitableTraceCounts TraceCounts() const + { + return waitable_->TraceCounts(); + } + +private: + void PublishOnce() + { + const uint64_t previous = published_.load(std::memory_order_relaxed); + if (previous >= static_cast(message_count_)) { + return; + } + + const uint64_t publish_ns = NowNs(); + if (previous == 0) { + first_publish_ns_.store(publish_ns, std::memory_order_release); + } + + Int64 message; + message.data = static_cast(publish_ns); + publisher_->publish(message); + last_publish_ns_.store(publish_ns, std::memory_order_release); + + const uint64_t published = + published_.fetch_add(1, std::memory_order_acq_rel) + 1; + if (published < static_cast(message_count_)) { + waitable_->Trigger(); + } + } + + const int64_t message_count_; + rclcpp::Publisher::SharedPtr publisher_; + rclcpp::CallbackGroup::SharedPtr callback_group_; + std::shared_ptr waitable_; + std::atomic published_{0}; + std::atomic first_publish_ns_{0}; + std::atomic last_publish_ns_{0}; +}; + +class SinkNode : public rclcpp::Node +{ +public: + SinkNode( + const size_t index, + const int64_t message_count, + const int64_t qos_depth) + : Node( + "ceiling_sink_" + std::to_string(index), + rclcpp::NodeOptions().use_intra_process_comms(true)), + message_count_(message_count) + { + const auto qos = rclcpp::QoS(static_cast(qos_depth)); + subscription_ = create_subscription( + "ceiling_flow_" + std::to_string(index), qos, + [this](const Int64 & message) {Receive(message);}); + } + + bool Complete() const + { + return complete_.load(std::memory_order_acquire); + } + + uint64_t Received() const + { + return received_.load(std::memory_order_relaxed); + } + + uint64_t FirstReceiveNs() const + { + return first_receive_ns_.load(std::memory_order_relaxed); + } + + uint64_t LastReceiveNs() const + { + return last_receive_ns_.load(std::memory_order_relaxed); + } + + uint64_t TotalLatencyNs() const + { + return total_latency_ns_.load(std::memory_order_relaxed); + } + + uint64_t MinLatencyNs() const + { + return min_latency_ns_.load(std::memory_order_relaxed); + } + + uint64_t MaxLatencyNs() const + { + return max_latency_ns_.load(std::memory_order_relaxed); + } + +private: + static void UpdateMin(std::atomic & target, const uint64_t value) + { + uint64_t current = target.load(std::memory_order_relaxed); + while (value < current && + !target.compare_exchange_weak(current, value, std::memory_order_relaxed)) + { + } + } + + static void UpdateMax(std::atomic & target, const uint64_t value) + { + uint64_t current = target.load(std::memory_order_relaxed); + while (value > current && + !target.compare_exchange_weak(current, value, std::memory_order_relaxed)) + { + } + } + + void Receive(const Int64 & message) + { + const uint64_t receive_ns = NowNs(); + const uint64_t previous = received_.fetch_add(1, std::memory_order_acq_rel); + if (previous == 0) { + first_receive_ns_.store(receive_ns, std::memory_order_release); + } + last_receive_ns_.store(receive_ns, std::memory_order_release); + + const uint64_t latency_ns = + receive_ns - static_cast(message.data); + total_latency_ns_.fetch_add(latency_ns, std::memory_order_relaxed); + UpdateMin(min_latency_ns_, latency_ns); + UpdateMax(max_latency_ns_, latency_ns); + + if (previous + 1 >= static_cast(message_count_)) { + complete_.store(true, std::memory_order_release); + } + } + + const int64_t message_count_; + rclcpp::Subscription::SharedPtr subscription_; + std::atomic received_{0}; + std::atomic first_receive_ns_{0}; + std::atomic last_receive_ns_{0}; + std::atomic total_latency_ns_{0}; + std::atomic min_latency_ns_{UINT64_MAX}; + std::atomic max_latency_ns_{0}; + std::atomic complete_{false}; +}; + +} // namespace + +int main(int argc, char ** argv) +{ + rclcpp::init(argc, argv); + + try { + const int64_t thread_count = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_THREADS", 1); + const int64_t flow_count = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_FLOWS", 1); + const int64_t messages_per_flow = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_MESSAGES", 100000); + const int64_t qos_depth = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_QOS_DEPTH", messages_per_flow); + const int64_t timeout_seconds = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_TIMEOUT_SEC", 60); + + if (thread_count <= 0 || flow_count <= 0 || messages_per_flow <= 0 || + qos_depth <= 0 || timeout_seconds <= 0) + { + throw std::runtime_error( + "threads, flows, messages, QoS depth, and timeout must be greater than zero"); + } + + auto executor = std::make_shared( + rclcpp::ExecutorOptions{}, static_cast(thread_count)); + std::vector> sources; + std::vector> sinks; + sources.reserve(static_cast(flow_count)); + sinks.reserve(static_cast(flow_count)); + + for (int64_t index = 0; index < flow_count; ++index) { + auto sink = std::make_shared( + static_cast(index), messages_per_flow, qos_depth); + auto source = std::make_shared( + static_cast(index), messages_per_flow, qos_depth); + executor->add_node(sink); + executor->add_node(source); + sinks.push_back(std::move(sink)); + sources.push_back(std::move(source)); + } + + for (const auto & source : sources) { + source->Kick(); + } + + std::atomic stop_monitor{false}; + std::thread monitor( + [&]() { + const auto deadline = + std::chrono::steady_clock::now() + + std::chrono::seconds(timeout_seconds); + while (!stop_monitor.load(std::memory_order_relaxed) && + std::chrono::steady_clock::now() < deadline) + { + if (std::all_of( + sinks.begin(), sinks.end(), + [](const auto & sink) {return sink->Complete();})) + { + executor->cancel(); + return; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + executor->cancel(); + }); + + try { + executor->spin(); + } catch (...) { + stop_monitor.store(true, std::memory_order_relaxed); + executor->cancel(); + monitor.join(); + throw; + } + stop_monitor.store(true, std::memory_order_relaxed); + executor->cancel(); + monitor.join(); + + uint64_t published = 0; + uint64_t received = 0; + uint64_t first_publish_ns = UINT64_MAX; + uint64_t last_publish_ns = 0; + uint64_t last_receive_ns = 0; + uint64_t total_latency_ns = 0; + uint64_t min_latency_ns = UINT64_MAX; + uint64_t max_latency_ns = 0; + bool complete = true; + WaitableTraceCounts trace_counts; + + for (const auto & source : sources) { + published += source->Published(); + if (source->FirstPublishNs() > 0) { + first_publish_ns = std::min(first_publish_ns, source->FirstPublishNs()); + } + last_publish_ns = std::max(last_publish_ns, source->LastPublishNs()); + Accumulate(trace_counts, source->TraceCounts()); + } + + for (const auto & sink : sinks) { + received += sink->Received(); + last_receive_ns = std::max(last_receive_ns, sink->LastReceiveNs()); + total_latency_ns += sink->TotalLatencyNs(); + min_latency_ns = std::min(min_latency_ns, sink->MinLatencyNs()); + max_latency_ns = std::max(max_latency_ns, sink->MaxLatencyNs()); + complete = complete && sink->Complete(); + } + + const double source_duration_seconds = + first_publish_ns != UINT64_MAX && last_publish_ns > first_publish_ns ? + static_cast(last_publish_ns - first_publish_ns) / 1e9 : 0.0; + const double end_to_end_duration_seconds = + first_publish_ns != UINT64_MAX && last_receive_ns > first_publish_ns ? + static_cast(last_receive_ns - first_publish_ns) / 1e9 : 0.0; + const double source_throughput = + source_duration_seconds > 0.0 ? + static_cast(published) / source_duration_seconds : 0.0; + const double throughput = + end_to_end_duration_seconds > 0.0 ? + static_cast(received) / end_to_end_duration_seconds : 0.0; + const double average_latency_ns = + received > 0 ? + static_cast(total_latency_ns) / static_cast(received) : 0.0; + const double min_latency_us = + min_latency_ns == UINT64_MAX ? + 0.0 : static_cast(min_latency_ns) / 1e3; + const double drain_lag_ms = + last_receive_ns > last_publish_ns ? + static_cast(last_receive_ns - last_publish_ns) / 1e6 : 0.0; + const uint64_t expected = + static_cast(messages_per_flow) * + static_cast(flow_count); + const bool waitable_invariant = + trace_counts.execute == published; + complete = complete && published == expected && received == expected && + waitable_invariant; + + std::ostringstream output; + output << "{" + << "\"benchmark\":\"int64_message_passing\"," + << "\"ready_mode\":\"waitable\"," + << "\"executor\":\"events_cbg\"," + << "\"build_type\":\"" + << JsonEscape(ROS2_FRAMEWORK_PERF_BUILD_TYPE) << "\"," + << "\"threads\":" << thread_count << "," + << "\"flows\":" << flow_count << "," + << "\"messages_per_flow\":" << messages_per_flow << "," + << "\"expected_messages\":" << expected << "," + << "\"published_messages\":" << published << "," + << "\"received_messages\":" << received << "," + << "\"complete\":" << (complete ? "true" : "false") << "," + << "\"waitable_invariant\":" + << (waitable_invariant ? "true" : "false") << "," + << "\"waitable_trigger_count\":" << trace_counts.trigger << "," + << "\"waitable_on_ready_callback_count\":" + << trace_counts.on_ready_callback << "," + << "\"waitable_take_data_by_entity_id_count\":" + << trace_counts.take_data_by_entity_id << "," + << "\"waitable_execute_count\":" << trace_counts.execute << "," + << "\"source_publish_msg_s\":" << source_throughput << "," + << "\"throughput_msg_s\":" << throughput << "," + << "\"avg_latency_us\":" << average_latency_ns / 1e3 << "," + << "\"min_latency_us\":" << min_latency_us << "," + << "\"max_latency_us\":" + << static_cast(max_latency_ns) / 1e3 << "," + << "\"drain_lag_ms\":" << drain_lag_ms << "," + << "\"source_duration_s\":" << source_duration_seconds << "," + << "\"duration_s\":" << end_to_end_duration_seconds + << "}"; + + std::cout << output.str() << std::endl; + rclcpp::shutdown(); + return complete ? 0 : 1; + } catch (const std::exception & error) { + std::cerr << "{\"benchmark\":\"int64_message_passing\"," + << "\"complete\":false,\"error\":\"" + << JsonEscape(error.what()) << "\"}" << std::endl; + } + + if (rclcpp::ok()) { + rclcpp::shutdown(); + } + return 1; +} diff --git a/ros2_framework_perf/src/scheduler_ceiling_benchmark.cpp b/ros2_framework_perf/src/scheduler_ceiling_benchmark.cpp new file mode 100644 index 0000000..56ae2fe --- /dev/null +++ b/ros2_framework_perf/src/scheduler_ceiling_benchmark.cpp @@ -0,0 +1,267 @@ +// SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +// Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// SPDX-License-Identifier: Apache-2.0 +// SPDX-Generated-By: Cursor + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "rclcpp/executors/events_cbg_executor/events_cbg_executor.hpp" +#include "rclcpp/rclcpp.hpp" +#include "ros2_framework_perf/ceiling_benchmark_utils.hpp" + +#ifndef ROS2_FRAMEWORK_PERF_BUILD_TYPE +#define ROS2_FRAMEWORK_PERF_BUILD_TYPE "unknown" +#endif + +namespace +{ + +using ros2_framework_perf::ceiling_benchmark::Accumulate; +using ros2_framework_perf::ceiling_benchmark::AlwaysReadyWaitable; +using ros2_framework_perf::ceiling_benchmark::EnvInt; +using ros2_framework_perf::ceiling_benchmark::JsonEscape; +using ros2_framework_perf::ceiling_benchmark::NowNs; +using ros2_framework_perf::ceiling_benchmark::WaitableTraceCounts; + +class SchedulerNode : public rclcpp::Node +{ +public: + SchedulerNode(const size_t index, const int64_t operation_count) + : Node("ceiling_scheduler_" + std::to_string(index)), + operation_count_(operation_count) + { + callback_group_ = create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive); + waitable_ = std::make_shared([this]() {ExecuteOnce();}); + get_node_waitables_interface()->add_waitable(waitable_, callback_group_); + } + + void Kick() + { + waitable_->Trigger(); + } + + bool Complete() const + { + return complete_.load(std::memory_order_acquire); + } + + uint64_t Count() const + { + return count_.load(std::memory_order_relaxed); + } + + uint64_t FirstOperationNs() const + { + return first_operation_ns_.load(std::memory_order_relaxed); + } + + uint64_t LastOperationNs() const + { + return last_operation_ns_.load(std::memory_order_relaxed); + } + + WaitableTraceCounts TraceCounts() const + { + return waitable_->TraceCounts(); + } + +private: + void ExecuteOnce() + { + const uint64_t previous = count_.fetch_add(1, std::memory_order_acq_rel); + if (previous == 0) { + first_operation_ns_.store(NowNs(), std::memory_order_release); + } + + if (previous + 1 >= static_cast(operation_count_)) { + last_operation_ns_.store(NowNs(), std::memory_order_release); + complete_.store(true, std::memory_order_release); + return; + } + + waitable_->Trigger(); + } + + const int64_t operation_count_; + rclcpp::CallbackGroup::SharedPtr callback_group_; + std::shared_ptr waitable_; + std::atomic count_{0}; + std::atomic first_operation_ns_{0}; + std::atomic last_operation_ns_{0}; + std::atomic complete_{false}; +}; + +} // namespace + +int main(int argc, char ** argv) +{ + rclcpp::init(argc, argv); + + try { + const int64_t thread_count = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_THREADS", 1); + const int64_t operator_count = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_OPERATORS", 1); + const int64_t operations_per_operator = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_OPERATIONS", 100000); + const int64_t timeout_seconds = + EnvInt("ROS2_FRAMEWORK_PERF_CEILING_TIMEOUT_SEC", 60); + + if (thread_count <= 0 || operator_count <= 0 || + operations_per_operator <= 0 || timeout_seconds <= 0) + { + throw std::runtime_error( + "threads, operators, operations, and timeout must be greater than zero"); + } + + auto executor = std::make_shared( + rclcpp::ExecutorOptions{}, static_cast(thread_count)); + std::vector> nodes; + nodes.reserve(static_cast(operator_count)); + + for (int64_t index = 0; index < operator_count; ++index) { + auto node = std::make_shared( + static_cast(index), operations_per_operator); + executor->add_node(node); + nodes.push_back(std::move(node)); + } + + for (const auto & node : nodes) { + node->Kick(); + } + + std::atomic stop_monitor{false}; + std::thread monitor( + [&]() { + const auto deadline = + std::chrono::steady_clock::now() + + std::chrono::seconds(timeout_seconds); + while (!stop_monitor.load(std::memory_order_relaxed) && + std::chrono::steady_clock::now() < deadline) + { + if (std::all_of( + nodes.begin(), nodes.end(), + [](const auto & node) {return node->Complete();})) + { + executor->cancel(); + return; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + executor->cancel(); + }); + + try { + executor->spin(); + } catch (...) { + stop_monitor.store(true, std::memory_order_relaxed); + executor->cancel(); + monitor.join(); + throw; + } + stop_monitor.store(true, std::memory_order_relaxed); + executor->cancel(); + monitor.join(); + + uint64_t total_operations = 0; + uint64_t first_operation_ns = UINT64_MAX; + uint64_t last_operation_ns = 0; + bool complete = true; + WaitableTraceCounts trace_counts; + + for (const auto & node : nodes) { + total_operations += node->Count(); + if (node->FirstOperationNs() > 0) { + first_operation_ns = + std::min(first_operation_ns, node->FirstOperationNs()); + } + last_operation_ns = + std::max(last_operation_ns, node->LastOperationNs()); + complete = complete && node->Complete(); + Accumulate(trace_counts, node->TraceCounts()); + } + + const uint64_t expected_operations = + static_cast(operations_per_operator) * + static_cast(operator_count); + const bool waitable_invariant = + trace_counts.execute == total_operations; + complete = complete && total_operations == expected_operations && + waitable_invariant; + + const double duration_seconds = + first_operation_ns != UINT64_MAX && + last_operation_ns > first_operation_ns ? + static_cast(last_operation_ns - first_operation_ns) / 1e9 : 0.0; + const double throughput = + duration_seconds > 0.0 ? + static_cast(total_operations) / duration_seconds : 0.0; + + std::ostringstream output; + output << "{" + << "\"benchmark\":\"scheduler_dispatch\"," + << "\"ready_mode\":\"waitable\"," + << "\"executor\":\"events_cbg\"," + << "\"build_type\":\"" + << JsonEscape(ROS2_FRAMEWORK_PERF_BUILD_TYPE) << "\"," + << "\"threads\":" << thread_count << "," + << "\"operators\":" << operator_count << "," + << "\"operations_per_operator\":" << operations_per_operator << "," + << "\"expected_operations\":" << expected_operations << "," + << "\"total_operations\":" << total_operations << "," + << "\"complete\":" << (complete ? "true" : "false") << "," + << "\"waitable_invariant\":" + << (waitable_invariant ? "true" : "false") << "," + << "\"waitable_trigger_count\":" << trace_counts.trigger << "," + << "\"waitable_on_ready_callback_count\":" + << trace_counts.on_ready_callback << "," + << "\"waitable_add_to_wait_set_count\":" + << trace_counts.add_to_wait_set << "," + << "\"waitable_is_ready_count\":" << trace_counts.is_ready << "," + << "\"waitable_is_ready_true_count\":" + << trace_counts.is_ready_true << "," + << "\"waitable_take_data_count\":" << trace_counts.take_data << "," + << "\"waitable_take_data_by_entity_id_count\":" + << trace_counts.take_data_by_entity_id << "," + << "\"waitable_execute_count\":" << trace_counts.execute << "," + << "\"throughput_ops_s\":" << throughput << "," + << "\"duration_s\":" << duration_seconds + << "}"; + + std::cout << output.str() << std::endl; + rclcpp::shutdown(); + return complete ? 0 : 1; + } catch (const std::exception & error) { + std::cerr << "{\"benchmark\":\"scheduler_dispatch\"," + << "\"complete\":false,\"error\":\"" + << JsonEscape(error.what()) << "\"}" << std::endl; + } + + if (rclcpp::ok()) { + rclcpp::shutdown(); + } + return 1; +} diff --git a/ros2_framework_perf/test/test_ceiling_tools.py b/ros2_framework_perf/test/test_ceiling_tools.py new file mode 100644 index 0000000..2ec2c1d --- /dev/null +++ b/ros2_framework_perf/test/test_ceiling_tools.py @@ -0,0 +1,302 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# SPDX-Generated-By: Cursor + +"""Tests for the framework-ceiling matrix and summary tools.""" + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +import yaml + + +REPOSITORY_ROOT = Path(__file__).parents[2] + + +def load_script(name): + """Import a repository script by filename.""" + path = REPOSITORY_ROOT / 'scripts' / name + spec = importlib.util.spec_from_file_location(path.stem, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +runner = load_script('run_ceiling_benchmarks.py') +summarizer = load_script('summarize_ceiling_results.py') + + +def test_default_config_builds_expected_runs(): + config = runner.load_config( + REPOSITORY_ROOT / 'config' / 'framework_ceiling.yaml') + runs = list(runner.build_runs(config, 'all', repetitions=1)) + + expected_cells = ( + len(config['message_passing']['matrix']) + + len(config['scheduler']['matrix']) + ) + assert len(runs) == expected_cells + + +def test_parse_result_uses_final_json_line(): + result = runner.parse_result( + 'diagnostic output\n{"complete":true,"waitable_invariant":true}\n') + assert result['complete'] is True + assert result['waitable_invariant'] is True + + +def test_load_results_rejects_unknown_benchmark(tmp_path): + result_path = tmp_path / 'unknown.json' + result_path.write_text( + json.dumps({'benchmark': 'unknown_benchmark'}), + encoding='utf-8') + + with pytest.raises(ValueError, match='unknown ceiling benchmark'): + summarizer.load_results(tmp_path) + + +def test_summary_calculates_medians_and_validates_invariant(tmp_path): + base = { + 'benchmark': 'scheduler_dispatch', + 'executor': 'events_cbg', + 'threads': 1, + 'operators': 1, + 'run_index': 1, + 'complete': True, + 'waitable_invariant': True, + 'duration_s': 1.0, + } + throughputs = [100.0, 300.0, 200.0] + for index, throughput in enumerate(throughputs): + result = { + **base, + 'run_index': index + 1, + 'throughput_ops_s': throughput, + } + (tmp_path / f'run-{index}.json').write_text( + json.dumps(result), encoding='utf-8') + + rows, errors = summarizer.summarize( + summarizer.load_results(tmp_path)) + + assert errors == [] + assert len(rows) == 1 + assert rows[0]['run_count'] == 3 + assert rows[0]['median_throughput_ops_s'] == 200.0 + assert rows[0]['all_waitable_invariants'] is True + + +def test_summary_reports_failed_invariant(tmp_path): + result = { + 'benchmark': 'int64_message_passing', + 'executor': 'events_cbg', + 'threads': 1, + 'flows': 1, + 'run_index': 1, + 'complete': True, + 'waitable_invariant': False, + 'source_publish_msg_s': 100.0, + 'throughput_msg_s': 100.0, + 'avg_latency_us': 1.0, + 'min_latency_us': 0.5, + 'max_latency_us': 2.0, + 'drain_lag_ms': 0.0, + 'duration_s': 1.0, + } + (tmp_path / 'run.json').write_text( + json.dumps(result), encoding='utf-8') + + rows, errors = summarizer.summarize( + summarizer.load_results(tmp_path)) + + assert rows[0]['all_waitable_invariants'] is False + assert 'median_throughput_msg_s' not in rows[0] + assert any('waitable invariant failed' in error for error in errors) + summarizer.print_summary(rows) + + +def test_timeout_output_text_decodes_bytes(): + assert runner.timeout_output_text(b'partial output') == 'partial output' + assert runner.timeout_output_text(None) == '' + + +def test_expected_result_validation_reports_missing_run(tmp_path): + config = { + 'schema_version': 1, + 'executor': 'events_cbg', + 'repetitions': 2, + 'message_passing': { + 'enabled': True, + 'messages_per_flow': 100, + 'matrix': [{'threads': 1, 'flows': 1}], + }, + 'scheduler': {'enabled': False}, + } + (tmp_path / 'framework_ceiling.yaml').write_text( + yaml.safe_dump(config), encoding='utf-8') + result = { + 'benchmark': 'int64_message_passing', + 'executor': 'events_cbg', + 'threads': 1, + 'flows': 1, + 'run_index': 1, + } + result_path = tmp_path / 'run-1.json' + result_path.write_text(json.dumps(result), encoding='utf-8') + result['_path'] = str(result_path) + + errors = summarizer.validate_expected_results(tmp_path, [result]) + + assert any('missing configured run' in error for error in errors) + + +@pytest.mark.parametrize('result_overrides, result_count', [ + ({}, 2), + ({'threads': 2}, 1), +]) +def test_expected_result_validation_reports_duplicate_or_unexpected_run( + tmp_path, result_overrides, result_count): + config = { + 'schema_version': 1, + 'executor': 'events_cbg', + 'repetitions': 1, + 'message_passing': { + 'enabled': True, + 'messages_per_flow': 100, + 'matrix': [{'threads': 1, 'flows': 1}], + }, + 'scheduler': {'enabled': False}, + } + (tmp_path / 'framework_ceiling.yaml').write_text( + yaml.safe_dump(config), encoding='utf-8') + result = { + 'benchmark': 'int64_message_passing', + 'executor': 'events_cbg', + 'threads': 1, + 'flows': 1, + 'run_index': 1, + '_path': str(tmp_path / 'run.json'), + **result_overrides, + } + + errors = summarizer.validate_expected_results( + tmp_path, [result.copy() for _ in range(result_count)]) + + assert any( + 'unexpected or duplicate run' in error for error in errors) + + +def test_missing_metric_prints_unavailable_instead_of_raising(capsys): + rows = [{ + 'benchmark': 'scheduler_dispatch', + 'executor': 'events_cbg', + 'threads': 1, + 'operators': 1, + 'run_count': 1, + }] + + summarizer.print_summary(rows) + + assert 'metrics unavailable' in capsys.readouterr().out + + +def test_invalid_result_set_suppresses_medians(): + rows = [{ + 'benchmark': 'scheduler_dispatch', + 'median_throughput_ops_s': 100.0, + 'median_duration_s': 1.0, + }] + + summarizer.suppress_medians(rows) + + assert 'median_throughput_ops_s' not in rows[0] + assert 'median_duration_s' not in rows[0] + + +def test_main_writes_effective_selected_config( + tmp_path, monkeypatch, capsys): + config = { + 'schema_version': 1, + 'output_directory': str(tmp_path / 'unused'), + 'repetitions': 5, + 'executor': 'events_cbg', + 'message_passing': { + 'enabled': True, + 'messages_per_flow': 100, + 'matrix': [{'threads': 1, 'flows': 1}], + }, + 'scheduler': { + 'enabled': True, + 'operations_per_operator': 100, + 'matrix': [{'threads': 1, 'operators': 1}], + }, + } + config_path = tmp_path / 'config.yaml' + config_path.write_text(yaml.safe_dump(config), encoding='utf-8') + output_directory = tmp_path / 'results' + monkeypatch.setattr(runner, 'run_one', lambda *args: True) + monkeypatch.setattr( + sys, 'argv', + [ + 'run_ceiling_benchmarks.py', + '--config', str(config_path), + '--benchmark', 'message_passing', + '--repetitions', '2', + '--output-directory', str(output_directory), + ]) + + assert runner.main() == 0 + + effective_config = yaml.safe_load( + (output_directory / 'framework_ceiling.yaml').read_text( + encoding='utf-8')) + assert effective_config['repetitions'] == 2 + assert effective_config['message_passing']['enabled'] is True + assert effective_config['scheduler']['enabled'] is False + assert 'summarize_ceiling_results.py' in capsys.readouterr().out + + +def test_main_rejects_configuration_with_no_runs( + tmp_path, monkeypatch): + config = { + 'schema_version': 1, + 'output_directory': str(tmp_path / 'unused'), + 'repetitions': 1, + 'executor': 'events_cbg', + 'message_passing': {'enabled': False}, + 'scheduler': {'enabled': False}, + } + config_path = tmp_path / 'config.yaml' + config_path.write_text(yaml.safe_dump(config), encoding='utf-8') + output_directory = tmp_path / 'results' + monkeypatch.setattr( + sys, 'argv', + [ + 'run_ceiling_benchmarks.py', + '--config', str(config_path), + '--output-directory', str(output_directory), + ]) + + with pytest.raises(SystemExit) as error: + runner.main() + + assert error.value.code == 2 + assert not output_directory.exists() diff --git a/ros2_framework_perf/test/verify_ceiling_executable.py b/ros2_framework_perf/test/verify_ceiling_executable.py new file mode 100644 index 0000000..226e067 --- /dev/null +++ b/ros2_framework_perf/test/verify_ceiling_executable.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 + +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# SPDX-Generated-By: Cursor + +"""Run a small ceiling benchmark and validate its JSON contract.""" + +import json +import os +import subprocess +import sys + + +def main(): + """Run the selected executable with a 100-operation workload.""" + if len(sys.argv) != 3: + print( + 'usage: verify_ceiling_executable.py EXECUTABLE ' + '{message_passing|scheduler}', + file=sys.stderr) + return 2 + + executable, benchmark = sys.argv[1:] + environment = os.environ.copy() + environment.update({ + 'ROS2_FRAMEWORK_PERF_CEILING_THREADS': '1', + 'ROS2_FRAMEWORK_PERF_CEILING_TIMEOUT_SEC': '10', + }) + if benchmark == 'message_passing': + environment.update({ + 'ROS2_FRAMEWORK_PERF_CEILING_FLOWS': '1', + 'ROS2_FRAMEWORK_PERF_CEILING_MESSAGES': '100', + }) + expected_benchmark = 'int64_message_passing' + elif benchmark == 'scheduler': + environment.update({ + 'ROS2_FRAMEWORK_PERF_CEILING_OPERATORS': '1', + 'ROS2_FRAMEWORK_PERF_CEILING_OPERATIONS': '100', + }) + expected_benchmark = 'scheduler_dispatch' + else: + print(f'unknown benchmark: {benchmark}', file=sys.stderr) + return 2 + + process = subprocess.run( + [executable], + env=environment, + text=True, + capture_output=True, + timeout=15, + check=False, + ) + if process.returncode != 0: + print(process.stdout, file=sys.stderr) + print(process.stderr, file=sys.stderr) + return 1 + + result = None + for line in reversed(process.stdout.splitlines()): + candidate = line.strip() + if candidate.startswith('{') and candidate.endswith('}'): + result = json.loads(candidate) + break + assert result is not None + assert result['benchmark'] == expected_benchmark + assert result['complete'] is True + assert result['waitable_invariant'] is True + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/scripts/run_ceiling_benchmarks.py b/scripts/run_ceiling_benchmarks.py new file mode 100644 index 0000000..79b2976 --- /dev/null +++ b/scripts/run_ceiling_benchmarks.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 + +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# SPDX-Generated-By: Cursor + +"""Run YAML-defined ROS 2 framework-ceiling benchmark matrices.""" + +import argparse +import copy +import json +import os +import shlex +import subprocess +import sys +from datetime import datetime +from pathlib import Path + +import yaml + + +BENCHMARKS = { + 'message_passing': { + 'executable': 'int64_ceiling_benchmark', + 'result_benchmark': 'int64_message_passing', + 'count_key': 'messages_per_flow', + 'dimension_key': 'flows', + 'count_env': 'ROS2_FRAMEWORK_PERF_CEILING_MESSAGES', + 'dimension_env': 'ROS2_FRAMEWORK_PERF_CEILING_FLOWS', + }, + 'scheduler': { + 'executable': 'scheduler_ceiling_benchmark', + 'result_benchmark': 'scheduler_dispatch', + 'count_key': 'operations_per_operator', + 'dimension_key': 'operators', + 'count_env': 'ROS2_FRAMEWORK_PERF_CEILING_OPERATIONS', + 'dimension_env': 'ROS2_FRAMEWORK_PERF_CEILING_OPERATORS', + }, +} + + +def load_config(path): + """Load and minimally validate a ceiling benchmark configuration.""" + with path.open('r', encoding='utf-8') as config_file: + config = yaml.safe_load(config_file) + + if not isinstance(config, dict) or config.get('schema_version') != 1: + raise ValueError('framework ceiling config must use schema_version: 1') + if config.get('executor') != 'events_cbg': + raise ValueError( + 'framework ceiling config executor must be events_cbg') + if int(config.get('repetitions', 0)) <= 0: + raise ValueError( + 'framework ceiling repetitions must be greater than zero') + + for benchmark_name, definition in BENCHMARKS.items(): + benchmark_config = config.get(benchmark_name, {}) + if not benchmark_config.get('enabled', False): + continue + if int(benchmark_config.get(definition['count_key'], 0)) <= 0: + raise ValueError( + f'{benchmark_name}.{definition["count_key"]} must be greater ' + 'than zero') + if not benchmark_config.get('matrix'): + raise ValueError(f'{benchmark_name}.matrix must not be empty') + for cell in benchmark_config['matrix']: + if int(cell.get('threads', 0)) <= 0: + raise ValueError( + f'{benchmark_name} matrix threads must be greater than ' + 'zero') + if int(cell.get(definition['dimension_key'], 0)) <= 0: + raise ValueError( + f'{benchmark_name} matrix {definition["dimension_key"]} ' + 'must be greater than zero') + return config + + +def parse_result(output): + """Return the final JSON object printed by a benchmark executable.""" + for line in reversed(output.splitlines()): + candidate = line.strip() + if candidate.startswith('{') and candidate.endswith('}'): + return json.loads(candidate) + raise ValueError('benchmark did not print a JSON result') + + +def timeout_output_text(output): + """Normalize TimeoutExpired output for JSON serialization.""" + if isinstance(output, bytes): + return output.decode(errors='replace') + return output or '' + + +def build_runs(config, selected_benchmark, repetitions): + """Yield benchmark command metadata in interleaved repetition order.""" + benchmark_names = ( + [selected_benchmark] + if selected_benchmark != 'all' + else list(BENCHMARKS)) + for repetition in range(1, repetitions + 1): + for benchmark_name in benchmark_names: + benchmark_config = config.get(benchmark_name, {}) + if not benchmark_config.get('enabled', False): + continue + definition = BENCHMARKS[benchmark_name] + for cell in benchmark_config['matrix']: + yield ( + repetition, benchmark_name, benchmark_config, + definition, cell, config['executor']) + + +def run_one(run, output_directory, dry_run): + """Run one matrix cell and write its JSON result.""" + ( + repetition, benchmark_name, benchmark_config, definition, cell, + executor, + ) = run + threads = int(cell['threads']) + dimension = int(cell[definition['dimension_key']]) + timeout_seconds = int(benchmark_config.get('timeout_seconds', 60)) + + environment = os.environ.copy() + environment.update({ + 'ROS2_FRAMEWORK_PERF_CEILING_THREADS': str(threads), + definition['count_env']: str( + benchmark_config[definition['count_key']]), + definition['dimension_env']: str(dimension), + 'ROS2_FRAMEWORK_PERF_CEILING_TIMEOUT_SEC': str(timeout_seconds), + }) + command = [ + 'ros2', 'run', 'ros2_framework_perf', definition['executable'], + ] + result_path = output_directory / ( + f'{benchmark_name}_{executor}_threads-{threads}_' + f'{definition["dimension_key"]}-{dimension}_run-{repetition}.json') + + print( + f'[{repetition}] {benchmark_name}: executor={executor} ' + f'threads={threads} {definition["dimension_key"]}={dimension}') + if dry_run: + print( + ' env ' + + ' '.join( + f'{key}={environment[key]}' for key in environment + if key.startswith('ROS2_FRAMEWORK_PERF_CEILING_')) + + ' ' + ' '.join(command)) + return True + + try: + process = subprocess.run( + command, + env=environment, + text=True, + capture_output=True, + timeout=timeout_seconds + 15, + check=False, + ) + return_code = process.returncode + try: + result = parse_result(process.stdout + '\n' + process.stderr) + except (ValueError, json.JSONDecodeError) as error: + result = { + 'benchmark': definition['result_benchmark'], + 'complete': False, + 'error': str(error), + 'stdout': process.stdout, + 'stderr': process.stderr, + } + except subprocess.TimeoutExpired as error: + return_code = None + result = { + 'benchmark': definition['result_benchmark'], + 'complete': False, + 'error': f'process timed out after {error.timeout} seconds', + 'stdout': timeout_output_text(error.stdout), + 'stderr': timeout_output_text(error.stderr), + } + + result.setdefault('executor', executor) + result.setdefault('threads', threads) + result.setdefault(definition['dimension_key'], dimension) + result.setdefault('waitable_invariant', False) + result['run_index'] = repetition + result['process_return_code'] = return_code + result_path.write_text( + json.dumps(result, indent=2, sort_keys=True) + '\n', + encoding='utf-8') + + valid = ( + return_code == 0 and + result.get('complete') is True and + result.get('waitable_invariant') is True + ) + if not valid: + print(f' FAILED: {result_path}', file=sys.stderr) + return valid + + +def main(): + """Run the configured benchmark matrices.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + '--config', + type=Path, + default=Path('config/framework_ceiling.yaml'), + help='YAML matrix configuration') + parser.add_argument( + '--benchmark', + choices=['all', *BENCHMARKS], + default='all', + help='run one benchmark family or all enabled families') + parser.add_argument( + '--repetitions', + type=int, + help='override repetitions from YAML') + parser.add_argument( + '--output-directory', + type=Path, + help=( + 'write results to this directory instead of a timestamped ' + 'directory')) + parser.add_argument( + '--dry-run', + action='store_true', + help='validate and print commands without executing them') + args = parser.parse_args() + + config = load_config(args.config) + repetitions = ( + args.repetitions + if args.repetitions is not None + else int(config['repetitions']) + ) + if repetitions <= 0: + parser.error('--repetitions must be greater than zero') + + runs = list(build_runs(config, args.benchmark, repetitions)) + if not runs: + parser.error('configuration selects no enabled benchmark runs') + + if args.output_directory: + output_directory = args.output_directory + else: + timestamp = datetime.now().strftime('%Y%m%d-%H%M%S') + output_directory = Path(config['output_directory']) / timestamp + + if not args.dry_run: + output_directory.mkdir(parents=True, exist_ok=False) + effective_config = copy.deepcopy(config) + effective_config['repetitions'] = repetitions + if args.benchmark != 'all': + for benchmark_name in BENCHMARKS: + if benchmark_name != args.benchmark: + benchmark_config = effective_config.setdefault( + benchmark_name, {}) + benchmark_config['enabled'] = False + (output_directory / 'framework_ceiling.yaml').write_text( + yaml.safe_dump(effective_config, sort_keys=False), + encoding='utf-8') + + success = True + for run in runs: + success = run_one(run, output_directory, args.dry_run) and success + + if not args.dry_run: + print(f'Results written to {output_directory}') + summarizer = Path(__file__).with_name( + 'summarize_ceiling_results.py') + print( + 'Summarize with: ' + f'python3 {shlex.quote(str(summarizer))} ' + f'{shlex.quote(str(output_directory))}') + return 0 if success else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/scripts/summarize_ceiling_results.py b/scripts/summarize_ceiling_results.py new file mode 100644 index 0000000..f8d59e7 --- /dev/null +++ b/scripts/summarize_ceiling_results.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python3 + +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# SPDX-Generated-By: Cursor + +"""Validate ceiling benchmark results and calculate repeated-run medians.""" + +import argparse +import csv +import json +import sys +from collections import Counter +from pathlib import Path +from statistics import median + +import yaml + + +BENCHMARK_FIELDS = { + 'int64_message_passing': { + 'config_name': 'message_passing', + 'dimension': 'flows', + 'metrics': [ + 'source_publish_msg_s', + 'throughput_msg_s', + 'avg_latency_us', + 'min_latency_us', + 'max_latency_us', + 'drain_lag_ms', + 'duration_s', + ], + }, + 'scheduler_dispatch': { + 'config_name': 'scheduler', + 'dimension': 'operators', + 'metrics': [ + 'throughput_ops_s', + 'duration_s', + ], + }, +} + + +def load_results(directory): + """Load individual benchmark result files from a directory.""" + results = [] + for path in sorted(directory.glob('*.json')): + if path.name == 'summary.json': + continue + with path.open('r', encoding='utf-8') as result_file: + result = json.load(result_file) + benchmark = result.get('benchmark') + if benchmark not in BENCHMARK_FIELDS: + raise ValueError( + f'{path}: unknown ceiling benchmark {benchmark!r}') + result['_path'] = str(path) + results.append(result) + if not results: + raise ValueError( + f'no ceiling benchmark result JSON files found in {directory}') + return results + + +def validate_expected_results(directory, results): + """Validate result identities against the effective run configuration.""" + config_path = directory / 'framework_ceiling.yaml' + if not config_path.exists(): + return [] + + with config_path.open('r', encoding='utf-8') as config_file: + config = yaml.safe_load(config_file) + if not isinstance(config, dict): + raise ValueError(f'invalid ceiling benchmark config: {config_path}') + + repetitions = int(config.get('repetitions', 0)) + if repetitions <= 0: + raise ValueError( + f'ceiling benchmark config has invalid repetitions: {config_path}') + executor = config.get('executor') + + expected = Counter() + for benchmark, definition in BENCHMARK_FIELDS.items(): + benchmark_config = config.get(definition['config_name'], {}) + if not benchmark_config.get('enabled', False): + continue + for cell in benchmark_config.get('matrix', []): + for run_index in range(1, repetitions + 1): + expected[( + benchmark, + executor, + int(cell['threads']), + int(cell[definition['dimension']]), + run_index, + )] += 1 + + actual = Counter() + errors = [] + for result in results: + definition = BENCHMARK_FIELDS[result['benchmark']] + required = [ + 'executor', 'threads', definition['dimension'], 'run_index', + ] + missing = [field for field in required if field not in result] + if missing: + errors.append( + f'{result["_path"]}: cannot identify configured run; ' + f'missing fields {", ".join(missing)}') + continue + actual[( + result['benchmark'], + result['executor'], + int(result['threads']), + int(result[definition['dimension']]), + int(result['run_index']), + )] += 1 + + for identity, count in sorted((expected - actual).items()): + errors.append(f'missing configured run {identity} (count={count})') + for identity, count in sorted((actual - expected).items()): + errors.append( + f'unexpected or duplicate run {identity} (count={count})') + return errors + + +def summarize(results): + """Group runs by matrix cell and calculate medians.""" + grouped = {} + validation_errors = [] + + for result in results: + benchmark = result['benchmark'] + definition = BENCHMARK_FIELDS[benchmark] + dimension_key = definition['dimension'] + required = ['executor', 'threads', dimension_key, 'run_index'] + missing = [field for field in required if field not in result] + if missing: + validation_errors.append( + f'{result["_path"]}: missing fields {", ".join(missing)}') + continue + + key = ( + benchmark, + result['executor'], + int(result['threads']), + int(result[dimension_key]), + ) + grouped.setdefault(key, []).append(result) + + if result.get('complete') is not True: + validation_errors.append( + f'{result["_path"]}: complete is not true') + if result.get('waitable_invariant') is not True: + validation_errors.append( + f'{result["_path"]}: waitable invariant failed') + + rows = [] + for key, group in sorted(grouped.items()): + benchmark, executor, threads, dimension = key + definition = BENCHMARK_FIELDS[benchmark] + group_valid = all( + item.get('complete') is True and + item.get('waitable_invariant') is True + for item in group) + row = { + 'benchmark': benchmark, + 'executor': executor, + 'threads': threads, + definition['dimension']: dimension, + 'run_count': len(group), + 'all_complete': all( + item.get('complete') is True for item in group), + 'all_waitable_invariants': all( + item.get('waitable_invariant') is True for item in group), + } + for metric in definition['metrics']: + values = [ + float(item[metric]) for item in group if metric in item + ] + if len(values) != len(group): + validation_errors.append( + f'{benchmark}/{executor}/{threads}/{dimension}: ' + f'missing metric {metric}') + continue + if group_valid: + row[f'median_{metric}'] = median(values) + rows.append(row) + + return rows, validation_errors + + +def write_csv(path, rows): + """Write summary rows with the union of their columns.""" + fieldnames = [] + for row in rows: + for field in row: + if field not in fieldnames: + fieldnames.append(field) + with path.open('w', encoding='utf-8', newline='') as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + + +def suppress_medians(rows): + """Remove performance medians when the result set is invalid.""" + for row in rows: + for field in list(row): + if field.startswith('median_'): + del row[field] + + +def print_summary(rows): + """Print a compact human-readable summary.""" + for row in rows: + dimension_key = BENCHMARK_FIELDS[row['benchmark']]['dimension'] + if row['benchmark'] == 'int64_message_passing': + throughput = row.get('median_throughput_msg_s') + latency = row.get('median_avg_latency_us') + primary = ( + f'{throughput:,.0f} msg/s, {latency:.3f} us avg' + if throughput is not None and latency is not None + else 'metrics unavailable') + else: + throughput = row.get('median_throughput_ops_s') + primary = ( + f'{throughput:,.0f} ops/s' + if throughput is not None + else 'metrics unavailable') + print( + f'{row["benchmark"]}: executor={row["executor"]} ' + f'threads={row["threads"]} {dimension_key}={row[dimension_key]} ' + f'runs={row["run_count"]} median={primary}') + + +def main(): + """Validate and summarize a result directory.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('result_directory', type=Path) + args = parser.parse_args() + + try: + results = load_results(args.result_directory) + rows, validation_errors = summarize(results) + validation_errors.extend( + validate_expected_results(args.result_directory, results)) + if validation_errors: + suppress_medians(rows) + except ( + OSError, + ValueError, + json.JSONDecodeError, + yaml.YAMLError, + ) as error: + print(f'error: {error}', file=sys.stderr) + return 1 + + summary = { + 'result_directory': str(args.result_directory), + 'result_count': len(results), + 'validation_passed': not validation_errors, + 'validation_errors': validation_errors, + 'groups': rows, + } + summary_json_path = args.result_directory / 'summary.json' + summary_csv_path = args.result_directory / 'summary.csv' + summary_json_path.write_text( + json.dumps(summary, indent=2, sort_keys=True) + '\n', + encoding='utf-8') + write_csv(summary_csv_path, rows) + print_summary(rows) + print(f'JSON summary: {summary_json_path}') + print(f'CSV summary: {summary_csv_path}') + + if validation_errors: + for error in validation_errors: + print(f'validation error: {error}', file=sys.stderr) + return 1 + return 0 + + +if __name__ == '__main__': + sys.exit(main())