[博客翻译]机器人测试中的确定性：解决之道与快速执行

这里在Basis，我们正在构建一个专注于生产和测试的机器人框架。在这方面，测试中的确定性是我们的主要目标之一。本文将重点介绍Basis如何不仅在测试中实现确定性，而且利用它来以闪电般的速度运行你的测试。

以下是预览：

`replay`	`deterministic_replay`

背景

机器人技术中的确定性问题一直存在，无论是在实时运行还是测试中。本文旨在展示一种用于测试/模拟的解决方案。

主要问题包括：

ROS 和其他机器人框架没有任何方式能在测试模式下确定性地运行。
机器人和开发者工作站之间的代码运行速度不一致是一个问题。
CI 和机器人上的代码运行速度不一致是一个大问题，在CI硬件上节省成本或为加速测试时间而使用过多并行都会加剧这一问题。
在低层面上，即使在同一系统上运行相同的代码，由于传输层和调度的非确定性也会表现出不同的行为。

结果，创建机器人的集成测试很糟糕，并且现有的工具也效果不佳。CI 上的集成测试可能如下所示：

启动一些进程编排器（ROS Master等）
使用自定义启动文件启动机器人的某个子集
- 如果你幸运，有人在主启动文件中添加了一个测试标志
- 如果你不幸，有一个单独的测试启动文件可能已经过时，与实际运行在机器人上的不同步
以慢于实时的速度重放一些记录的数据试图避开传输/调度的非确定性
需要等待更久因为测试是以慢于实时速度运行的
你的测试失败了CI
希望测试清理能真正杀死所有被测试启动的进程
在你的开发桌面重复上述步骤
测试成功（或者随机性地没成功）
重新在CI上运行测试，等待更长时间
测试成功
感到无奈，放弃，并合并代码，如果main分支再次失败则责备仿真团队。

最终，这种情况会让工程团队忽视测试结果，并且不鼓励创建更大/更复杂的测试和模拟。当测试失败时，是因为框架的问题还是实际的机器人代码？特别是在安全关键环境中，重要的是排除测试中的随机因素。

Basis 的目标是每次都让测试给出相同的结果。未来还将包含其他类型的确定性（例如代码在重放时的运行时间和实时一致）。

非确定性的源头

非确定性来自多种源头。

目前Basis的主要焦点（也是这篇文章的重点）是非确定性中的传输层相关问题和调度相关的非确定性。

传输层上的非确定性包含网络堆栈中的不确定性，套接字处理顺序，两个消息同时到来时会发生什么等等。

调度非确定性包括线程相关的问题、测试执行之间的性能差异等。

还有其他的一些非确定性来源：

调用 random()
使用系统时钟
使用网络资源
数据竞争
编译器标志
宇宙射线
指针比较

这些我都曾在测试环境见过。

我们如何解决这个问题？

几个规则：

所有要以确定性运行的代码都是对消息或定时器的响应（在Basis术语中，一切都是在处理程序Handlers内运行）
不允许处理器之间使用侧通道通信。使用主题 -Basis支持仅运行时主题（原始C++结构体），因此请使用它们。
- 这不是严格的事实——同一单元中的处理程序会在同一个C++类中共享变量，这是一种侧通道通信。这对于复杂单元尤其是一个问题，该单元允许并发执行处理程序。我们最终会提供工具来帮助解决这种问题。
（与第2条共同考虑）机器人代码不知道订阅者和发布者的概念。它们从主题获取输入消息作为输入，并将输出消息发送到主题作为输出。
- 这并不能阻止关心这些概念的工具，它只是意味着这些工具不能作为测试的一部分确定性地运行。
- 这确实意味着外部触发的代码（传感器驱动）不会是确定性的，但这对于大多数测试不是问题。
所有处理程序都有与其相关的元数据描述其输入，输出和执行条件。
所有单元（可以理解为ROS节点）都可以动态加载并包含元数据
处理程序内部的代码是确定性的（当然如此）。

基于这些规则，可以构建一个计划程序，查看要运行的请求单元，查看任何要重放的数据的内容，并以正确（和确定的）顺序及正确的数据适当调用每个处理程序。Basis 是从零开始为支持这种用例构建的。

演示

让我们演示Basis是如何帮助实现确定性的。首先我们将记录一些现场运行的数据。然后我们将对该数据运行 replay 测试，找出测试过程中的问题。最后，我们将展示 deterministic_replay 解决这些问题的过程。

记录一些数据

下面是一个想要做一些工作的Basis单元示例。这个单元将接收一条消息，等待100毫秒（执行“工作”），然后退出，允许其他回调继续执行。

threading_model:
  single
cpp_includes:
  - simple_pub_sub.pb.h
handlers:
  OnChatter:
    sync:
      type: all
    inputs:
      /chatter:
        type: protobuf:StringMessage

OnChatter::Output simple_sub::OnChatter(const OnChatter::Input &input) {
  // Convert protobuf nanoseconds into a basis timestamp
  auto send_stamp = basis::core::MonotonicTime::FromNanoseconds(input.chatter->send_stamp());
  BASIS_LOG_INFO("OnChatter: {} {}", input.chatter->message(), send_stamp.ToSeconds());

  // Calculate delay between "now" and when the message was sent
  basis::core::Duration delay = input.time - send_stamp;
  if (delay > basis::core::Duration::FromSeconds(0.2)) {
    BASIS_LOG_WARN("/chatter delayed by {:.2f}s - queueing has occured", delay.ToSeconds());
  }
  constexpr int work_time_ms = 2000;
  BASIS_LOG_INFO("Doing {} ms worth of work", work_time_ms);
  std::this_thread::sleep_for(std::chrono::milliseconds(work_time_ms));

  return OnChatter::Output();
}

这个单元和另一个按1Hz频率在 /chatter 生产的单元一起运行将会得到类似的控制台输出：

[124998.122650826] [launch] [info] 录制 (异步) 到 /tmp/demo_124998.122377076.mcap
[124998.126956742] [launch] [info] 运行带有 2 个组件的进程
[124998.130765492] [launch] [info] 启动带有组件 /opt/basis/unit/simple_sub.unit.so 的线程
[124998.134928909] [launch] [info] 启动带有组件 /opt/basis/unit/simple_pub.unit.so 的线程
[124999.137999076] [/simple_pub] [info] PublishAt1Hz
[124999.138074076] [/simple_sub] [info] OnChatter: Hello, world! 124999.138065618
[124999.138080410] [/simple_sub] [info] 处理 100 毫秒的工作
[125000.136988785] [/simple_pub] [info] PublishAt1Hz
[125000.137143077] [/simple_sub] [info] OnChatter: Hello, world! 125000.137124493
[125000.137151452] [/simple_sub] [info] 处理 100 毫秒的工作
[125001.136178202] [/simple_pub] [info] PublishAt1Hz
[125001.136319869] [/simple_sub] [info] OnChatter: Hello, world! 125001.136305744
[125001.136327452] [/simple_sub] [info] 处理 100 毫秒的工作
[125002.137870703] [/simple_pub] [info] PublishAt1Hz
[125002.138054536] [/simple_sub] [info] OnChatter: Hello, world! 125002.137991619
...

回放数据

非常好，但假设这是一个录制在机器人上的数据。我们可能会运行 replay /tmp/demo_124998.122377076.mcap，并伴随着 simple_sub。如果两个环境中的硬件相似，我们可能会看到这样的结果：

[125458.039238910] [launch] [info] 运行带有 1 个组件的进程
[125458.040486243] [launch] [info] 启动带有组件 /opt/basis/unit/simple_sub.unit.so 的线程
[124998.307061284] [launch] [info] simple_sub 检测到重放重置，正在重启 ...

（请注意这里的时间跳变，我们得到了一个新的模拟时间步）

[124998.977061284] [launch] [info] 运行带有 1 个组件的进程
[124998.977061284] [launch] [info] 启动带有组件 /opt/basis/unit/simple_sub.unit.so 的线程
[125001.137061284] [/simple_sub] [info] OnChatter: Hello, world! 125001.137061284
[125001.137061284] [/simple_sub] [info] 处理 100 毫秒的工作
[125002.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125002.147061284
[125002.147061284] [/simple_sub] [info] 处理 100 毫秒的工作
[125003.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125003.147061284
[125003.147061284] [/simple_sub] [info] 处理 100 毫秒的工作
...

问题 1 - 缺失数据，非确定性

重新运行几次后，我们发现了第一个问题 - 有时我们错过了测试的第一个消息，并且看到输出如下：

[125002.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125002.147061284

用 mcap-cli 快速检查一下发现 ...

basis@aee413836118:/basis/demos/simple_pub_sub/build$ ~/mcap-linux-arm64 cat --json /tmp/demo_124998.122377076.mcap --topics /chatter
{"topic":"/chatter","sequence":0,"log_time":124999.138044993,"publish_time":124999.138044993,"data":{"sendStamp":"124999137959660", "message":"Hello, world!"}}
{"topic":"/chatter","sequence":0,"log_time":125000.137092327,"publish_time":125000.137092327,"data":{"sendStamp":"125000136919993", "message":"Hello, world!"}}
{"topic":"/chatter","sequence":0,"log_time":125001.136257244,"publish_time":125001.136257244,"data":{"sendStamp":"125001136105286", "message":"Hello, world!"}}
...

有几个消息从测试中丢失了。即使我们最完整的重放也缺少 124999 和 125000。

问题 2 - 性能差异可能导致非确定性

如果我们的测试环境变慢（过载 CI、不同的/没有 GPU、更少的 CPU 核心）。在这种情况下，假设处理需要 2000 毫秒，并且设置 constexpr int work_time_ms = 2000;。

再运行一次相同的重放测试。

[125002.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125002.137789828
[125002.147061284] [/simple_sub] [info] 处理 2000 毫秒的工作
[125004.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125003.137948037
[125004.147061284] [/simple_sub] [warning] /chatter 延迟了 1.01 秒 - 队列已溢出
[125004.147061284] [/simple_sub] [info] 处理 2000 毫秒的工作
[125006.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125004.13672012
[125006.147061284] [/simple_sub] [warning] /chatter 延迟了 2.01 秒 - 队列已溢出
[125006.147061284] [/simple_sub] [info] 处理 2000 毫秒的工作
[125008.147061284] [/simple_sub] [info] OnChatter: Hello, world! 125005.136193412
[125008.147061284] [/simple_sub] [warning] /chatter 延迟了 3.01 秒 - 队列已溢出
[125008.147061284] [/simple_sub] [info] 处理 2000 毫秒的工作

看这样 - 第二个消息不是在 125003.137948037 被处理，而是在 125004.147061284 被处理，比预期晚了一秒钟（或者说是正好预期到了增加的一秒钟的阻塞...）。延迟只会继续增加，pub/sub 系统的队列一般不会被配置成无限大小，这将导致消息丢失和不正确的定时。

如果原始的运行情况是这样的：

在我们的人工减速下，它现在看起来像这样：

注意队列 - 当 OnChatter 正在执行时，我们无法处理更多的消息，但重放系统对此并不知情并且继续发布消息。

问题 3 - 集成测试速度慢，笨重，并且与单元测试框架不兼容

如果每个消息都需要 100 毫秒来处理，并且我们重放 10 条消息，你期望这个测试会花多长时间？通过常规集成测试：10 秒钟，用于 1 秒钟的工作。

假设我们的持续集成（CI）硬件更快，能够在 10 毫秒内处理一条消息 - 那么测试仍然需要 10 秒来运行，总共只需 100 毫秒的工作。

除此之外，我们如何将此测试与 gtest 集成呢？我们必须构建某种框架，管理进程的分叉或启动每一部分，并配备超时机制、子子孙孙进程的管理等。

如果我们希望程序性地创建消息（避免使用磁盘中的序列化数据），或者创建单条消息并检查系统在每个步骤之后的输出，我们该如何实现呢？使用 ROS 你有两种选择：

启动一个 ROS Master，启动你关心的节点，将你的测试初始化为一个 ROS 节点，发布消息，一直旋转直到（也许）收到响应。（别忘了为你的测试关闭并行性！）
将你的 ROS 节点拆分为库，手动在各个节点的业务逻辑之间传递消息。（祝你好运能让这些与 launch 文件同步）

解决方案

让我们用 basis 的 确定性 回放器重新运行这个测试。

deterministic_replay /tmp/demo_124998.122377076.mcap /basis/demos/simple_pub_sub/launch_single_process.yaml --disable_unit demo:simple_pub
[129000.980036858] [replayer] [info] 发现组件 /simple_sub simple_sub 位于 /opt/basis/unit/simple_sub.unit.so
[129000.982984441] [replayer] [info] 开始确定性回放...
[129000.984189191] [replayer] [info] 回放主题 /chatter
[129000.984319524] [replayer] [info] 回放主题 /log
[129000.984376316] [replayer] [info] 从 124998.127061284 开始回放
[129000.985144524] [replayer] [info] 组件 /simple_sub 已初始化
[124999.238044993] [/simple_sub] [info] OnChatter: Hello, world! 124999.13795966
[124999.238044993] [/simple_sub] [info] 执行 2000 毫秒的工作
[125000.237092327] [/simple_sub] [info] OnChatter: Hello, world! 125000.136919993
[125000.237092327] [/simple_sub] [info] 执行 2000 毫秒的工作
[125001.236257244] [/simple_sub] [info] OnChatter: Hello, world! 125001.136105286

太好了 - 我们甚至抓到了最初几次被丢弃的消息。现在调度器会等待那些应该完成但尚未在现实时间内完成的处理器（就模拟时间而言）。

我们可以看到，在 OnChatter 执行期间没有执行更多工作（如回放消息），因为处理器只应占据 100 毫秒的模拟时间。虽然我们在实际运行速度上慢了一倍，但这比用了不准确的数据要好很多。

问题 1，问题 2 - 解决了。

至于问题 3……还记得文章开头提到的那个预览 GIF 吗？

让我们把工作时间降到 10 毫秒。因为 deterministic_replay 知道代码应该何时运行，它也知道代码何时不是正在运行。我们可以尽可能快的速度运行代码，忽略回放数据以 1 Hz 发布的事实。这使得可以在闪电般快速的情况下进行集成测试。

（其他数据回放系统通常有一些形式的速率倍增命令 —— 但这种设置难以针对所有条件调整）。

`replay`	`deterministic_replay`

一切都可以在一个进程中以测试模式运行，这意味着仅在需要与可视化工具或其他工具通信时才需要协调器（主）进程（单元测试不需要）。可以将 deterministic_replay 链接到一个单元测试中，然后运行一个 launch 文件（甚至是一个程序性创建的组件列表），推送消息，查询输出消息和节点内部状态，修改飞行中的消息等。无需额外进程。

更进一步

利用这种强大的功能，我们可以做更多的事情：

查找如果其中一条消息晚于预期到达会发生什么，或者测试与时间相关的不同错误路径。你在围绕时间的降级状态下究竟测试了多少？
deterministic_replay 的默认行为是总是按照相同顺序重新运行在同一时间发生事件。如果改为反转呢？或者用种子随机化？加上发送时间抖动随机化，可以长时间测试与时间相关的 bug。
测试那些可能实施起来很昂贵的“假设”情况。假设你的感知负责人说他能在一个季度内用两个工程师将感知栈加快 15%。这在理论上听起来很好，但更快的感知栈真的能带来更好的机器人性能吗？在花费一个季度的工作之前，可以用承诺的时间运行集成测试套件，代替实际工作。
lldb -- deterministic_replayer ... 只需简单运行即可，并且可以正确暂停所有组件，无需处理分叉模式的问题。

这样的功能还没有建立？

这类工具以前曾被人请求过：

未回答的问题：answers.ros.org 帖子
ROS1 文档：“对于像 turtlesim 这样的节点，当处理命令消息时较小的时间变化可能会微妙地改变行为，用户不应期望完全模仿的行为。”
各种其他与 ROS 相关的问题帖子
在公司内部与 Gazebo/Applied Intuition 内部工具等集成的工程师

各个团队曾尝试过解决这个问题：

BOSCH 关于 ROS 运行时确定性的幻灯片（这里没有涵盖运行时确定性，但 basis 确实使其更简单）
Flow 框架 https://github.com/ZebraDevs/flow_ros?tab=readme-ov-file （这似乎主要是在 ROS 发布者和机器人业务逻辑中间的一个中介 - 我赞同，架构良好。但确定性的回放在哪儿呢？）
ROS2 DEF - 来自乌尔姆大学给 ROS2 的一个附加系统。虽有勇气的努力，但已被放弃且未完成。由于 ROS2 的架构，无法完全发挥作用。
几家大型机器人公司不得不使用各种策略在公司内部实现这一点，程度不一。不幸的是，这些都是非常特定于公司的，他们不太可能将其公开或出售。

有趣的是，Applied Intuition 完全回避了这个问题，认为在 ADAS 中的确定性是好的，但并没有给出如何在集成时实现它的答案。