realtime_support 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/realtime_support
版本：0.13.0（Humble），子包 2 个，构建类型 ament_cmake，许可证 Apache 2.0（tlsf_cpp 另含 LGPL 2.1，因底层 TLSF）。

realtime_support 仓库提供 ROS 2 实时性能支撑工具：面向 Linux（尤其 PREEMPT_RT 内核）的周期循环 测量库 rttest，以及将 TLSF 确定性内存分配器 接入 rclcpp 的 tlsf_cpp 包装。二者不实现 DDS/Executor 本身，而是帮助验证与控制 抖动（jitter）、缺页、调度优先级、堆分配延迟 等实时因素。

---

1. 总体认识

1.1 核心职责

子包	职责
rttest	周期性 clock_nanosleep 唤醒，测量 latency/jitter、pagefault，导出 CSV
tlsf_cpp	tlsf_heap_allocator — C++ STL/rclcpp 兼容的 TLSF 堆分配器

1.2 在 ROS 2 栈中的位置

关系	说明
tlsf_cpp → tlsf	调用 init_memory_pool / tlsf_malloc / tlsf_free（见 tlsf 源码分析）
tlsf_cpp → rclcpp	示例/测试将 TLSF 注入 Publisher/Subscription/Executor MemoryStrategy
rttest	独立 C API；pendulum demo 用于统计周期性能
非 ROS 核心路径	普通节点不依赖本仓库；仅实时调优/演示场景使用

平台限制：两个子包 CMake 均在 Windows / macOS / Android 上 skip，仅 Linux 构建。

---

2. 子包结构

realtime_support/
├── rttest/                         # 实时循环测量库 ★
│   ├── include/rttest/
│   │   ├── rttest.h                # C API（206 行）
│   │   ├── utils.hpp               # timespec 算术
│   │   └── math_utils.hpp          # 标准差计算
│   ├── src/rttest.cpp              # 实现（942 行）
│   ├── examples/example_loop.c     # 最小示例
│   ├── scripts/
│   │   ├── rttest_plot             # 绘图脚本
│   │   └── analyze.py            # 延迟统计
│   └── test/
└── tlsf_cpp/                       # TLSF C++ 包装 ★
    ├── include/tlsf_cpp/tlsf.hpp    # allocator（140 行）
    ├── example/allocator_example.cpp
    └── test/test_tlsf.cpp

包	版本	构建产物
rttest	0.13.0	共享库 librttest.so、头文件、rttest_plot
tlsf_cpp	0.13.0	INTERFACE 库（仅头文件）+ tlsf_allocator_example 可执行文件

---

3. 依赖关系

3.1 rttest

rttest
├── (无 ROS 依赖)
├── pthread / m / stdc++
└── Linux: mlockall, clock_nanosleep, sched_setscheduler, getrusage

package.xml 仅 ament_cmake + 测试依赖。

3.2 tlsf_cpp

tlsf_cpp
├── tlsf              # vendor 包，C TLSF 实现
├── rclcpp            # 示例与测试
├── rmw / std_msgs
└── rmw_implementation_cmake  # 测试多 RMW

---

4. rttest 详解

4.1 设计目标

为 固定周期任务（control loop、hard real-time thread）提供：

1. 绝对时间唤醒 — clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, ...)
2. Jitter 记录 — 期望唤醒时刻 vs 实际唤醒时刻（纳秒，可正可负）
3. 缺页统计 — 每轮 getrusage(RUSAGE_THREAD) 增量
4. 内存锁定 / 预缺页 — 减少运行时 page fault
5. SCHED_FIFO / SCHED_RR — 实时调度策略与优先级

面向 PREEMPT_RT 等低延迟 Linux 环境。

4.2 核心数据结构

struct rttest_params
{
  size_t iterations;
  struct timespec update_period;
  size_t sched_policy;
  int sched_priority;
  size_t stack_size;
  uint64_t prefault_dynamic_size;
  char * filename;
};

struct rttest_results
{
  size_t iteration;
  int64_t min_latency;
  int64_t max_latency;
  double mean_latency;
  double latency_stddev;
  size_t minor_pagefaults;
  size_t major_pagefaults;
};

内部 rttest_sample_buffer 按 iteration 存储 latency_samples、minor_pagefaults、major_pagefaults 向量。

4.3 多线程模型

std::map<pthread_t, Rttest> rttest_instance_map;
pthread_t initial_thread_id = 0;

• 每 pthread 一个 Rttest 实例（注释：rttest can have one instance per thread）
• 首线程 rttest_init / rttest_read_args 后，其他线程调用 rttest_init_new_thread() 复制参数

C API 通过 pthread_self() 路由到当前线程实例。

4.4 典型使用流程

// 见 examples/example_loop.c
rttest_set_sched_priority(98, SCHED_RR);
rttest_read_args(argc, argv);      // 或 rttest_init(...)
rttest_lock_memory();              // mlockall
rttest_lock_and_prefault_dynamic(); // 堆预缺页
rttest_spin(callback, NULL);       // 周期循环
rttest_write_results();
rttest_finish();

4.5 spin 与 jitter 测量

int Rttest::spin_once(..., const size_t i)
{
  multiply_timespec(update_period, i, &wakeup_time);
  add_timespecs(start_time, &wakeup_time, &wakeup_time);
  clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &wakeup_time, NULL);
  clock_gettime(CLOCK_MONOTONIC, &current_time);

  this->record_jitter(&wakeup_time, &current_time, i);
  user_function(args);
  this->get_next_rusage(i);
  this->accumulate_statistics(i);
  return 0;
}

Jitter 符号：晚于 deadline 为正，早于 deadline 为负（record_jitter 中 parity）。

rttest_spin → spin_period：iterations=0 时无限循环（不保存完整 sample buffer，无法写文件）。

4.6 内存锁定与预缺页

API	实现要点
rttest_lock_memory()	mlockall(MCL_CURRENT | MCL_FUTURE)
rttest_prefault_stack()	alloca + memset 触摸栈页
rttest_lock_and_prefault_dynamic()	mallopt(M_TRIM_THRESHOLD,-1)、M_MMAP_MAX=0，循环 new char[] 直到无新 pagefault

动态预缺页默认上限 8GB（-d 可调），失败时恢复 malloc 参数并 munlockall。

4.7 命令行参数（rttest_read_args）

选项	含义	默认
-u	更新周期（s/ms/us/ns）	1ms
-i	迭代次数（≤0 为无限）	1000
-t	线程优先级	80
-s	调度策略 fifo / rr	SCHED_RR
-m	栈预缺页大小	1MB
-d	堆预缺页大小	8192MB
-f	结果输出文件名	不写文件

4.8 结果输出

write_results_file 写入文本 CSV：

iteration timestamp latency minor_pagefaults major_pagefaults

• scripts/analyze.py — 打印 min/max/mean latency、超 30µs 样本数
• scripts/rttest_plot — 安装到 bin，配合 -f 输出绘图

4.9 辅助头文件

文件	内容
utils.hpp	add_timespecs、subtract_timespecs、timespec_to_uint64
math_utils.hpp	calculate_stddev 模板

pendulum_control 等 demo 复用 utils.hpp 做时间运算。

---

5. tlsf_cpp 详解

5.1 设计目标

T TLSF（Two-Level Segregated Fit） 分配器具有 O(1) malloc/free 与 低碎片，适合实时线程。tlsf vendor 包提供 C 实现；tlsf_cpp 将其包装为 符合 std::allocator_traits 的 C++ 分配器，供 rclcpp 消息与 Executor 使用。

5.2 tlsf_heap_allocator

template<typename T, size_t DefaultPoolSize = 1024 * 1024>
struct tlsf_heap_allocator
{
  using value_type = T;

  explicit tlsf_heap_allocator(size_t size) { initialize(size); }
  tlsf_heap_allocator() { initialize(DefaultPoolSize); }

  size_t initialize(size_t size) {
    memory_pool = new char[pool_size];
    memset(memory_pool, 0, pool_size);
    init_memory_pool(pool_size, memory_pool);
    return pool_size;
  }

  T * allocate(size_t size) {
    T * ptr = static_cast<T *>(tlsf_malloc(size * sizeof(T)));
    if (ptr == NULL && size > 0) throw std::bad_alloc();
    return ptr;
  }

  void deallocate(T * ptr, size_t) { tlsf_free(ptr); }

  char * memory_pool;
  size_t pool_size;
};

要点：

• 默认池大小 1MB（模板参数 DefaultPoolSize）
• 同一 memory_pool 的 allocator 实例 operator== 为 true（可互相 deallocate）
• 析构时 destroy_memory_pool

5.3 与 rclcpp 集成模式

example/allocator_example.cpp 展示完整注入链：

注入点	作用
PublisherOptionsWithAllocator	发布路径分配
SubscriptionOptionsWithAllocator	订阅路径分配
MessageMemoryStrategy	消息 buffer
AllocatorMemoryStrategy	Executor wait/collect 路径
AllocRebind + custom deleter	publish(std::move(unique_ptr)) 正确释放

启动参数 intra 等可切换 use_intra_process_comms，对比进程内/跨进程分配行为。

5.4 测试

test/test_tlsf.cpp 通过 call_for_each_rmw_implementation 对每个 RMW 跑 gtest，验证 TLSF 下 pub/sub + spin 正常。

---

6. 下游消费者

包	使用
demos/pendulum_control	rttest_read_args、rttest_lock_and_prefault_dynamic、tlsf_heap_allocator、发布 pendulum_msgs/RttestResults
（示例）	tlsf_allocator_example 二进制

pendulum_demo.cpp 是 rttest + tlsf_cpp + rclcpp 集成的参考实现：实时线程跑 pendulum 控制，同时统计 rttest 结果并通过 ROS topic 输出。

---

7. 实时实践要点（与本仓库相关）

技术	rttest API	目的
内存锁定	rttest_lock_memory	防止进程内存被 swap
栈预缺页	rttest_prefault_stack	避免栈 growth 缺页
堆预缺页	rttest_lock_and_prefault_dynamic	避免控制循环中 malloc 缺页
实时调度	rttest_set_sched_priority	FIFO/RR 高优先级
确定性分配	tlsf_heap_allocator	替代 glibc malloc 的无界延迟
周期测量	rttest_spin	量化 jitter 与 pagefault

注意：rclcpp 默认仍使用系统分配器；要使用 TLSF 需 显式 配置 allocator 与 memory strategy（如 pendulum / allocator_example）。

---

8. 构建与安装

8.1 rttest

• 输出 librttest（SHARED）
• 安装 include/rttest/、bin/rttest_plot
• 非 ament 环境支持纯 CMake 安装（rttestConfig.cmake.in）

8.2 tlsf_cpp

• INTERFACE 库，无 .so；依赖 tlsf::tlsf
• 安装头文件 include/tlsf_cpp/tlsf.hpp
• 构建 tlsf_allocator_example → lib/tlsf_cpp/tlsf_allocator_example

---

9. 数据路径：rttest 一次迭代

---

10. 与相关包对照

包	层级	关系
tlsf	C 分配器	tlsf_cpp 底层
rclcpp	客户端库	MemoryStrategy / 消息 allocator 注入点
rcpputils	C++ 工具	无直接关系
realtime_support	实时支撑	测量 + TLSF 包装

---

11. 推荐阅读顺序

1. rttest/README.md — CLI 与使用说明
2. rttest/examples/example_loop.c — 最小集成
3. rttest/src/rttest.cpp — spin_once、lock_and_prefault_dynamic、read_args
4. rttest/include/rttest/rttest.h — 完整 C API
5. tlsf_cpp/include/tlsf_cpp/tlsf.hpp — allocator 接口
6. tlsf_cpp/example/allocator_example.cpp — rclcpp 注入示例
7. demos/pendulum_control/src/pendulum_demo.cpp — 生产级 demo 集成
8. tlsf 源码分析 — 底层池化算法

---

12. 小结

realtime_support 是 ROS 2 实时性工具链，含两个互补子包：

• rttest — Linux 周期唤醒 ** instrumentation**：jitter、pagefault、调度、内存锁定；942 行 C++ 实现 + C 头 API
• tlsf_cpp — 将 TLSF 接入 rclcpp allocator / MemoryStrategy，实现可预测的堆分配

二者均 非默认 ROS 运行时依赖，主要用于 PREEMPT_RT 验证、pendulum 演示、实时内存策略实验。排查控制循环延迟时，应结合 rttest 输出与是否启用 TLSF 分配器一并分析。