设计：丢失与录制消息数量的统计

背景

Rosbag2 的当前实现不提供关于实时或后处理过程中丢失和录制消息数量的任何统计。更具体地说，当前实现只会在当前 bag 文件关闭时向日志打印 Rosbag2 写入器丢弃的消息数量——即写入器无法跟上传入消息、消息因内部缓冲区已满而被丢弃的情况。 然而，它不提供任何关于 DDS 传输层丢失消息数量的信息。而且这些信息在 bag 文件本身或录制期间的运行时中都不可用。 这是一个问题，因为它使评估录制数据的质量和识别录制过程的潜在问题变得困难。

本设计提议为 Rosbag2 添加一个新功能，提供实时和后处理期间丢失和录制消息数量以及每个话题录制字节数的统计。

用例

1. 实时监控录制性能。
   1. 从录制开始起，按话题统计的丢失消息数和写入字节数的总体统计，应可通过服务请求实时获取。
   2. 当发生消息丢失时，按话题统计的丢失消息数和写入字节数的增量统计应发布到特定话题上。
2. 按话题统计的丢失消息数和写入到当前 bag 文件的字节数统计，应在录制结束时保存到元数据中。

设计提案与实现细节考虑

1. 传输层消息丢失跟踪

为实现收集 RMW 传输层丢失消息数量统计的功能，我们可以在 rosbag2_transport::recorder 中创建 GenericSubscription 时通过提供 rclcpp::SubscriptionOptions 参数，在 rclcpp::SubscriptionOptions::event_callbacks 中指定 message_lost_callback。message_lost_callback 将提供一个对 rclcpp::QOSMessageLostInfo 的引用，它映射到 rmw_message_lost_status_s 结构体。

  typedef struct RMW_PUBLIC_TYPE rmw_message_lost_status_s
  {
    /// Total number of messages lost.
    size_t total_count;
    /// Number of messages lost since last callback.
    size_t total_count_change;
  } rmw_message_lost_status_t;

2. Recorder 侧消息丢失跟踪

为实现收集 Rosbag2 recorder 侧丢失消息数量统计的功能，需要在 rosbag2_cpp::writer 类中添加一个新的 messages_lost_callback。当 rosbag2_cpp::writer 无法跟上传入消息且内部缓冲区已满时，将调用 messages_lost_callback。messages_lost_callback 将有一个参数作为返回值，定义为对 rosbag2_cpp::MessageLostInfo 结构的 std::vector 的常量引用，该结构包含以下字段：

• topic_name (string)：丢失消息所在话题的名称。
• num_messages_lost (uint64_t)：丢失的消息数量。

messages_lost_callback 应定义在 rosbag2_cpp::bag_events 命名空间中，并添加到 WriterEventCallbacks 结构体中，类似于 write_split_callback。 此外，还需要添加一个新的存储写入方法，即 storage->write(std::vector<SerializedBagMessage>)，它返回一个向量或无序映射，包含每个话题未写入的消息数量。

3. 统计更新速率的 CLI 选项

需要向 Rosbag2 recorder 添加一个 CLI 选项，以不超过指定速率在预定义话题上发布统计。

• 该 CLI 选项可以命名为 --statistics-update-rate，定义为整数值。
• 它用于指定发布统计的最大更新速率（次/秒）。默认值为 1 Hz。值为 0 将禁用统计发布。

4. 消息丢失通知器

需要向 rosbag2_transport::recorder 添加一个新的 MessagesLostNotifier 类，当发生消息丢失时，在预定义话题上发布按话题统计的丢失消息数量的增量统计。

• 为实现这一点，Rosbag2 recorder 应满足以下要求：
  1. 消息丢失事件应从 Rosbag2 recorder 发布在专用话题上。例如 /events/rosbag2_messages_lost。
  2. 消息丢失事件应包含按话题统计的丢失消息数量。消息类型可以命名为 rosbag2_interfaces::msg::MessagesLostEvent，定义在 rosbag2_interfaces 包中。消息类型将类似于 rosbag2_interfaces::msg::WriteSplitEvent 消息类型，并应包含以下字段：
     • node_name (string)：正在录制的节点名称。
     • 以下字段的数组：
     • topic_name (string)：丢失消息所在话题的名称。
     • messages_lost_in_transport (uint64_t)：自上次事件以来 DDS 传输层丢失的消息数量。
     • messages_lost_in_recorder (uint64_t)：自上次事件以来 Rosbag2 recorder 中丢失的消息数量。
  3. 为避免过度的资源消耗，消息丢失事件的发布频率不得超过用户指定的事件更新速率。
  4. 消息丢失事件不得包含丢失消息数为零的话题。要获取已录制和丢失消息数量的完整统计，用户应使用第 5 节「获取统计的新服务请求」中描述的 GetRecorderStatistics 服务请求。
  5. 消息丢失事件应在单独的线程中发布，以避免阻塞录制过程。
  6. MessagesLostNotifier 将从第 1 节「传输层消息丢失跟踪」中描述的 rclcpp::SubscriptionOptions::event_callbacks 中指定的 message_lost_callback 和第 2 节「Recorder 侧消息丢失跟踪」中描述的 messages_lost_callback 获取信息。

5. 获取统计的新服务请求

需要向 rosbag2_transport::recorder 添加一个新的服务请求，提供自录制开始以来每个话题的丢失和录制消息数量以及录制字节数的统计。

• 该服务可以命名为 rosbag2_interfaces::srv::GetRecorderStatistics，定义在 rosbag2_interfaces 包中。
• rosbag2_interfaces::srv::GetRecorderStatistics 服务请求应返回以下字段：
  • node_name (string)：正在录制的节点名称。
  • 以下字段的数组：
    • topic_name (string)：正在录制的话题名称。
    • messages_lost_in_transport (uint64_t)：每个话题在 DDS 传输层丢失的消息数量。
    • messages_lost_in_recorder (uint64_t)：每个话题在 Rosbag2 recorder 中丢失的消息数量。
    • messages_recorded (uint64_t)：该话题录制的消息数量。
    • bytes_written (uint64_t)：该话题写入的字节数。
    • messages_rate (float)：该话题的平均消息速率，单位 Hz。
• 为了在 GetRecorderStatistics 回调中提供 rosbag2_interfaces::srv::RecorderStatistics 消息所需的信息，rosbag2_cpp::writer 类应新增一个 getter 方法 get_total_topics_statistics，返回内部 total_topics_stat_ 变量收集的信息，该变量是一个以话题名称为键的 std::unordered_map。关于 total_topics_stat_ 的更多细节，请参考第 6 节「在元数据中存储统计」。

6. 在元数据中存储统计

需要在 bag 文件中添加新的元数据字段，在关闭当前 bag 文件时存储按话题统计的丢失消息数和写入字节数。 为实现这一点，应完成以下工作：

1. 需要在 rosbag2_storage 包中添加一个新的 rosbag2_storage::TopicStatistics 结构体。rosbag2_storage::TopicStatistics 结构体将包含以下字段：
   • topic_name (string)：话题名称。
   • messages_recorded (uint64_t)：该话题录制的消息数量。
   • messages_lost_in_transport (uint64_t)：该话题在 DDS 传输层丢失的消息数量。
   • messages_lost_in_recorder (uint64_t)：该话题在 Rosbag2 recorder 中丢失的消息数量。
   • bytes_written (uint64_t)：写入 bag 文件的该话题字节数。
   • messages_rate (float)：该话题的平均消息速率，单位 Hz。
2. rosbag2_storage::TopicInformation 结构体应修改为包含新的 rosbag2_storage::TopicStatistics 结构体。size_t message_count 字段应被移除，因为它已被 rosbag2_storage::TopicStatistics::messages_recorded 取代。
3. rosbag2_storage::FileInformation 结构体应修改为包含新字段 topics_statistics，它是 rosbag2_storage::TopicStatistics 结构体的向量。size_t message_count 字段应被移除，因为它已被 rosbag2_storage::TopicStatistics::messages_recorded 取代。
4. rosbag2_storage::BagMetadata::version 应增加到 10，因为我们在改变数据结构的内容。这将允许我们在元数据 yaml 解析器中与之前的版本保持向后兼容。
5. 为了能够累积所需的统计，rosbag2_cpp::writers::SequentialWriter 类应扩展为包含两个以 topic_id 为键、以 rosbag2_storage::TopicStatistics 结构体为值的 std::unordered_map。
6. 一个可能命名为 total_topics_stat_ 的 std::unordered_map 将用于跟踪自录制开始以来的丢失消息数量，另一个可能命名为 bag_topics_stat_ 的 std::unordered_map 将用于跟踪自当前 bag 文件录制开始以来的丢失消息数量。 注意，在同一个录制会话中可以有多个 bag 文件。
7. total_topics_stat_ 应取代 rosbag2_cpp::writers::SequentialWriter 类中的 topics_names_to_info_ 映射。TopicMetadata 应作为一个以 topic_id 为键、以 TopicMetadata 为值的 std::unordered_map 单独存储。
8. rosbag2_storage::FileInformation::topics_statistics 字段应在关闭当前 bag 文件时用 bag_topics_stat_ 映射中的数据填充。
9. 为了保持 messages_lost_in_transport 的最新状态，需要在 rosbag2_cpp::writer 类中添加一个新的 on_messages_lost_in_transport 方法，当 message_lost_callback 被触发时调用。on_messages_lost_in_transport 方法将接收对 rclcpp::QOSMessageLostInfo 结构体和 topic_name 的常量引用，并更新 RecorderStatistics 类中 total_topics_stat_ 和 bag_topics_stat_ 映射中的 messages_lost_in_transport 字段。

7. ros2 bag info 命令扩展

需要扩展 ros2 bag info 命令，通过从保存的元数据中读取信息，包含按话题统计的丢失消息数量的统计。 该统计应可通过 --verbose CLI 选项获得。

实施阶段

这不应该一次性整体实现。实现应专注于小的、渐进式的、具有可靠测试且易于审查的 PR。以下是建议的操作顺序。

1. 第 1 节「传输层消息丢失跟踪」
   • 第 2 节「Recorder 侧消息丢失跟踪」
2. 第 4 节「消息丢失通知器」
3. 第 3 节「统计更新速率的 CLI 选项」
4. 第 6 节「在元数据中存储统计」
5. 第 5 节「获取统计的新服务请求」
6. 第 7 节「ros2 bag info 命令扩展」

注意：第 2 步独立于第 4-6 步，可以独立或并行实现。此外，第 1 步和第 2 步直到”消息丢失通知器”的实现很可能可以移植回 Kilted 和 Jazzy ROS 2 发行版，因为它们不太可能需要任何 API/ABI 破坏性变更。第 5 步和第 6 步相互独立，可以并行实现。但是，它们依赖于第 4 步。