作者 | 不可说

出品 | 汽车电子与软件

#01

MVBS

MVBS，全称为 Micro Vehicle Bus Service，是VBSLite工程的核心组件之一，理想的VCOS中面向MCU领域设计的通信中间件，实现资源受限环境下的高效业务互联互通；主要采用以数据为中心的发布-订阅（DCSP, Data-Centric Publish Subscribe）通信模型；并通过RTPS协议进行数据传输，提供低时延、高可靠的数据通信，同时支持若干必要的QoS策略，同时提供RPC（Remote Function Call）功能，支持请求-调用通信模型，构建支持多种通信模式的统一通信平台。

在MVBS的QoS策略中，理想实现了E2E来保证通信质量，使用的是CRC-32/P4算法。

#02

CRC-32/P4算法

CRC-32/P4是一种特定的32位循环冗余校验（CRC-32）算法，属于CRC校验算法的变体，通过特定的多项式和初始值设置，用于增强对数据传输或存储中错误的检测能力。以下是对CRC-32/P4的详细解析：

1. CRC-32基础

作用：CRC-32是一种广泛使用的错误检测码（EDC），通过生成一个32位的校验值，验证数据在传输或存储过程中是否出现错误。

原理：基于多项式除法，将数据视为一个长二进制数，与预定义的生成多项式进行模2除法，余数即为CRC校验值。

2. CRC-32/P4的特点

特定多项式：CRC-32/P4使用特定的生成多项式（如代码中的0xF4ACFB13U），与标准CRC-32（多项式为0x04C11DB7或0xEDB88320）不同，这使其适用于特定场景（如实时通信、嵌入式系统）。

初始值与异或值：

• 初始值：0xFFFFFFFFU，表示CRC寄存器在开始计算前的初始状态。

• 最终异或值：0xFFFFFFFFU，在计算完成后对CRC值进行异或操作，得到最终校验值。

查表法优化：通过预计算的查找表（如下面章节中代码中的crc_32P4_tab），加速CRC计算过程。

3. CRC-32/P4的应用场景

实时通信系统：如RTPS（实时发布-订阅协议），用于检测消息在传输过程中是否被篡改或损坏。

端到端（E2E）保护：在代码中，CRC-32/P4用于校验Reader/Writer ID、序列号（SN）和数据内容，确保数据的完整性和顺序性。

嵌入式系统：适用于资源受限的环境，因其计算效率高且错误检测能力强。

4. 与标准CRC-32的区别

多项式不同：标准CRC-32通常使用0x04C11DB7或0xEDB88320，而CRC-32/P4使用0xF4ACFB13U，导致校验值分布和错误检测能力不同。

应用场景不同：标准CRC-32适用于通用数据校验（如文件传输、网络通信），而CRC-32/P4针对特定协议或系统优化。

5. CRC-32/P4的优势

高效性：查表法使计算复杂度从O(n^2)降至O(n)，适合实时系统。

可靠性：能检测大多数突发错误和随机错误，错误漏检率低。

灵活性：通过调整多项式和初始值，可适应不同需求。

6. 代码中的实现细节

数据校验范围：包括输入数据、Reader/Writer ID（4字节）、序列号（8字节）和数据长度（4字节）。

序列号检查：结合计数器机制，确保消息顺序正确，防止乱序或重复。

错误处理：通过日志输出和状态码（如E2E_P04STATUS_ERROR）反馈校验结果。

#03

理想E2E核心功能解析与代码实现

1、CRC-32/P4计算

代码结构设计：

函数：e2e_calculate_crc32P4

作用：计算数据、Reader/WriterID、序列号（SN）和数据长度的CRC-32/P4校验值。

关键参数：

• data_ptr：待校验的数据指针。

• data_length：数据长度。

• reader_id/writer_id：通信双方的实体ID。

• sn：序列号（用于检测消息顺序）。

算法细节：

• 使用预计算的查表crc_32P4_tab加速CRC计算。

• 初始值为0xFFFFFFFF，多项式为0xF4ACFB13，最终结果异或0xFFFFFFFF。

• 校验范围包括：

1. 输入数据（逐字节）。

2. Reader/WriterID（各4字节）。

3. 序列号（8字节）。

4. 数据长度（4字节）。

代码实现思路：

1. 初始化参数

2. 检查输入有效性

避免空指针导致的未定义行为若data_ptr为NULL，直接返回默认值0。

3. 初始化CRC值

设置CRC计算的初始状态。需要将crc初始化为预定义的Crc_32P4_StartValue（0xFFFFFFFFU）。

4. 计算数据部分的CRC

校验输入数据的完整性。需要逐字节处理：遍历data_length字节的数据。根据公式更新crc：crc=((crc>>8)&0x00FFFFFFU)^crc_32P4_tab[(crc^*data_pointer)&0xFFU]。

就是将当前CRC右移8位，并与查表结果异或，更新CRC值。

5. 计算Reader/WriterID的CRC

校验通信实体ID的合法性。固定长度处理：分别对reader_id和writer_id的4字节数据执行与数据部分相同的查表计算。

6. 计算序列号（SN）的CRC

确保消息顺序的正确性。对sn的8字节数据执行查表计算（逻辑同数据部分）。

7. 计算数据长度的CRC

验证数据长度字段的合法性。对data_length的4字节数据执行查表计算。

8. 最终异或处理

标准化CRC结果；将最终CRC值与Crc_32P4_Xor（0xFFFFFFFFU）异或，得到最终校验码。

9. 返回结果

输出计算完成的CRC值。

代码流程图：

开始

├─ 初始化指针和查表

├─ 检查 data_ptr 是否为 NULL → 是 → 返回 0

├─ 初始化 crc = Crc_32P4_StartValue

├─ 计算数据部分的 CRC（逐字节）

├─ 计算 Reader ID 的 CRC（4 字节）

├─ 计算 Writer ID 的 CRC（4 字节）

├─ 计算 SN 的 CRC（8 字节）

├─ 计算 data_length 的 CRC（4 字节）

├─ crc ^= Crc_32P4_Xor

└─ 返回 crc

代码实现：

uint32_t e2e_calculate_crc32P4(const uint8_t* data_ptr,
uint32_t data_length,
const rtps_entity_id_t *reader_id,
const rtps_entity_id_t *writer_id,
struct rtps_sn *sn)
{
uint32_t crc                = 0; /* Default return value if NULL pointer */
const uint8_t* data_pointer     = data_ptr;
const uint8_t* reader_id_pointer    = (const uint8_t*) reader_id;
const uint8_t* writer_id_pointer    = (const uint8_t*) writer_id;
const uint8_t* sn_pointer       = (const uint8_t*) sn;
const uint8_t* data_length_pointer  = (const uint8_t*) &data_length;
static const uint32_t crc_32P4_tab[] = {
0x00000000U, 0x30850FF5U, 0x610A1FEAU, 0x518F101FU, 0xC2143FD4U,
//
省略查表内容
0x9F16DC99U
};
if (data_pointer != NULL) {
crc = Crc_32P4_StartValue;
for( uint32_t byte = 0; byte < data_length; byte++) {
crc = ((crc >> 8) & 0x00FFFFFFU) ^ crc_32P4_tab[(crc ^ *data_pointer) & 0xFFU];
data_pointer++;
}
for( uint32_t byte = 0; byte < 4U; byte++) {
crc = ((crc >> 8) & 0x00FFFFFFU) ^ crc_32P4_tab[(crc ^ *reader_id_pointer) & 0xFFU];
reader_id_pointer++;
}
for( uint32_t byte = 0; byte < 4U; byte++) {
crc = ((crc >> 8) & 0x00FFFFFFU) ^ crc_32P4_tab[(crc ^ *writer_id_pointer) & 0xFFU];
writer_id_pointer++;
}
for( uint32_t byte = 0; byte < 8U; byte++) {
crc = ((crc >> 8) & 0x00FFFFFFU) ^ crc_32P4_tab[(crc ^ *sn_pointer) & 0xFFU];
sn_pointer++;
}
for( uint32_t byte = 0; byte < 4U; byte++) {
crc = ((crc >> 8) & 0x00FFFFFFU) ^ crc_32P4_tab[(crc ^ *data_length_pointer) & 0xFFU];
 
data_length_pointer++;
}
crc = crc ^ Crc_32P4_Xor;
}
return crc;
}

2、端到端状态检查

代码结构设计：

函数：e2e_do_checkP04

作用：验证接收数据的CRC校验值和长度是否合法。

关键逻辑：

检查数据长度是否在配置的min_data_length和max_data_length范围内。

比较计算出的CRC（data_crc）与接收的CRC（header_src->crc）。

返回状态码（如E2E_P04STATUS_OK或E2E_P04STATUS_ERROR）。

关键数据结构

rtps_entity_id_t：通信实体（Reader/Writer）的唯一标识。

rtps_sn：序列号，用于消息顺序跟踪。

structe2e_p04_cfg：E2E配置参数（如min_data_length、max_data_length、max_delta_counter）。

状态返回

日志输出：通过pr_err打印错误信息（如数据长度不匹配、CRC校验失败、序列号异常）。

状态码：

E2E_P04STATUS_OK：校验通过。

E2E_P04STATUS_ERROR：CRC或长度不匹配。

E2E_P04STATUS_NODATAAVAILABLE：数据长度超出配置范围。

E2E_P04STATUS_REPEATED/E2E_P04STATUS_OKSOMELOST/E2E_P04STATUS_WRONGSEQUENCE：序列号异常。

代码实现思路：

1.初始化参数

2.检查数据长度合法性，确保数据长度在配置的min_data_length和max_data_length范围内。

步骤：

• 条件判断：

若header_src->length不满足profile04_cfg.min_data_length<=header_src->length<=profile04_cfg.max_data_length，则：输出错误日志（E2E_P04STATTUS_NODATAAVAILABLE）并直接返回错误状态码E2E_P04STATTUS_NODATAAVAILABLE。

• 数据长度匹配检查：

若data_size（实际接收的数据长度）不等于header_src->length（配置的期望长度），则：输出错误日志（E2E_P04STATUS_ERROR）、返回错误状态码E2E_P04STATUS_ERROR。

3. 计算并验证CRC值

通过CRC-32/P4算法校验数据完整性。

• 调用CRC计算函数：

使用e2e_calculate_crc32P4计算接收数据（pdata）、Reader/WriterID（reader_id、writer_id）、序列号（sn）和数据长度的CRC值。

• 比对CRC值：

若header_src->crc（接收的CRC）不等于data_crc（计算的CRC）：设置e2e_status为E2E_P04STATUS_ERROR、输出错误日志（包含接收的CRC和计算的CRC）。否则，保持e2e_status为E2E_P04STATUS_OK。

4. 返回校验结果，输出最终的校验状态。

代码实现流程：

开始

│

├─ 初始化 data_crc 和 e2e_status

├─ 检查数据长度是否在 [min_data_length, max_data_length] 范围内 → 不满足 → 返回 NODATAAVAILABLE

├─ 检查 data_size 是否等于 header_src->length → 不满足 → 返回 ERROR

├─ 调用 e2e_calculate_crc32P4 计算 CRC

├─ 比对 header_src->crc 和 data_crc → 不匹配 → 设置 e2e_status = ERROR

└─ 返回 e2e_status

代码实现：

uint32_t e2e_do_checkP04(const struct e2e_header *header_src, const uint8_t *pdata,
rtps_entity_id_t *reader_id, rtps_entity_id_t *writer_id, struct rtps_sn *sn,
struct e2e_p04_cfg profile04_cfg, uint16_t data_size)
{
uint32_t data_crc           = 0xFFFFFFFFU;
enum E2E_P04CheckStatusType e2e_status  = E2E_P04STATUS_OK;
 
if (!(profile04_cfg.min_data_length <= header_src->length) ||
!(profile04_cfg.max_data_length >= header_src->length)) {
 
pr_err(ERR_FAULT, "Error status:E2E_P04STATTUS_NODATAAVAILABLE!"
"[header_src->length:%u]\n",header_src->length);
return (uint32_t)E2E_P04STATTUS_NODATAAVAILABLE;
}
 
if (!(data_size == header_src->length)) {
 
pr_err(ERR_FAULT, "Error status:E2E_P04STATUS_ERROR! 
[data_size:%u]"
"[header_src->length:%u]\n", data_size, header_src->length);
return (uint32_t)E2E_P04STATUS_ERROR;
}
/* crc verification */
 
data_crc = e2e_calculate_crc32P4(pdata, header_src->length,
reader_id, writer_id, sn);
 
if (header_src->crc != data_crc) {
e2e_status = E2E_P04STATUS_ERROR;
 
pr_err(ERR_FAULT, "Error status:E2E_P04STATUS_ERROR! 
"
"[header_src->crc:%u] [data_crc:%u]\n",header_src->crc, data_crc);
} else {
e2e_status = E2E_P04STATUS_OK;
}
return (uint32_t)e2e_status;
}

3、序列号自增计数

代码结构设计：

函数：e2e_increment_counter

作用：递增一个16 位无符号整数计数器。

关键逻辑：

掩码的作用：MAX_P04_COUNTER_MASK 确保计数器在达到最大值后重新从 0 开始（模运算效果）。

无符号整数处理：使用uint16_t 和 1u 避免符号扩展和溢出问题。

代码实现思路：

1. 参数解析

获取待递增的计数器指针。输入参数counter是一个指向uint16_t的指针，表示需要递增的计数器。

2. 递增计数器

将计数器值加1。

3. 应用掩码限制范围

确保计数器值不超过MAX_P04_COUNTER_MASK（通常为0xFFFF或更小）。

按位与操作：(*counter+1u)&MAX_P04_COUNTER_MASK。

• MAX_P04_COUNTER_MASK是一个掩码，用于截断高位，限制计数器范围。

• 例如，若MAX_P04_COUNTER_MASK=0xFFFF，则计数器在0x0000到0xFFFF之间循环。

4. 更新计数器值

将计算后的新值写回计数器。通过*counter=...将递增并掩码后的值赋给原计数器。

代码流程：

开始

├─ 解引用 counter 获取当前值

├─ 计算新值：current_value + 1

├─ 应用掩码：new_value & MAX_P04_COUNTER_MASK

└─ 将新值写回 counter

代码实现：

void e2e_increment_counter(uint16_t *counter)
{
*counter = (uint16_t)((*counter + 1u) & MAX_P04_COUNTER_MASK);
}

4、序列号计数器检查

代码结构设计：

函数：e2e_check_counter

作用：检测序列号（SN）的连续性，防止消息丢失或乱序。

关键逻辑：

• 首次接收时初始化计数器（e2e_first_rcv）。

• 后续接收时计算计数器差值（counter_delta）：

• counter_delta==1：正常。

• 1<counter_delta<=max_delta_counter：部分丢失（E2E_P04STATUS_OKSOMELOST）。

• counter_delta<=0：重复消息（E2E_P04STATUS_REPEATED）。

• counter_delta>max_delta_counter：严重乱序（E2E_P04STATUS_WRONGSEQUENCE）。

代码思路：

该函数e2e_check_counter用于检查通信系统中消息序列号（SN）的连续性，确保消息顺序正确且无丢失或重复。

1. 检查远程Writer状态

确保Writer在线且可提供计数器信息。若wproxy==NULL（Writer离线），输出错误日志并直接返回。

2. 检查Reader的E2E功能状态

若Reader未启用E2E，跳过后续检查。调用reader_e2e_enabled(r)检查Reader是否启用E2E。若未启用，设置rcc->e2e_status=E2E_P04STATUS_OK和rcc->counter=0，直接返回。

3. 处理首次接收消息

初始化首次接收的计数器。若wproxy->e2e_first_rcv为true（首次接收）：

• 设置e2e_first_rcv=false，标记非首次接收。

• 记录当前序列号到wproxy->last_e2e_count，直接返回。

4. 处理CRC或数据长度错误

若之前校验失败，仅更新计数器，不修改状态。若rcc->e2e_status!=E2E_P04STATUS_OK（校验失败）：更新wproxy->last_e2e_count为当前序列号，直接返回。

5. 计算序列号差值

检测消息顺序是否正常。

a) 计算差值：counter_delta=current_sn-last_e2e_count。

current_sn通过rtps_sn_to_int64(&rcc->cc.sn)转换得到。

b) 判断差值范围：

• 正常（counter_delta==1）：消息顺序正确，设置e2e_status=E2E_P04STATUS_OK。

• 部分丢失（1<counter_delta<=max_delta_counter+1）：设置e2e_status=E2E_P04STATUS_OKSOMELOST，输出错误日志。

• 重复（counter_delta<=0）：设置e2e_status=E2E_P04STATUS_REPEATED，输出错误日志。

• 严重乱序（counter_delta>max_delta_counter+1）：设置e2e_status=E2E_P04STATUS_WRONGSEQUENCE，输出错误日志。

6. 更新计数器

记录当前序列号用于下一次比较：将wproxy->last_e2e_count更新为当前序列号current_sn。

代码流程：

开始

├─ 检查 wproxy 是否为 NULL → 是 → 输出错误日志并返回

├─ 检查 Reader 是否启用 E2E → 未启用 → 设置状态为 OK，返回

├─ 检查是否为首次接收 → 是 → 初始化 last_e2e_count，返回

├─ 检查 e2e_status 是否为 OK → 不是 → 更新 last_e2e_count，返回

├─ 计算 counter_delta = current_sn - last_e2e_count

├─ 判断 counter_delta 范围：

│ ├─ counter_delta == 1 → 设置状态为 OK

│ ├─ 1 < counter_delta <= max_delta_counter + 1 → 设置状态为 OKSOMELOST，输出日志

│ ├─ counter_delta <= 0 → 设置状态为 REPEATED，输出日志

│ └─ counter_delta > max_delta_counter + 1 → 设置状态为 WRONGSEQUENCE，输出日志

└─ 更新 last_e2e_count 为 current_sn

代码实现：

void e2e_check_counter(struct reader *r, struct reader_cache_change *rcc, struct writer_proxy *wproxy)
{
if (wproxy == NULL) { /* the remote has offline */
pr_err(ERR_FAULT, "The remote writer has offline, can't get counter and e2e_status from writer");
return;
}
/* reader not enable e2e */
if (!(reader_e2e_enabled(r))) {
rcc->e2e_status = (uint32_t)E2E_P04STATUS_OK;
rcc->counter = 0;
return;
}
/* no need to check deltcount */
if (wproxy->e2e_first_rcv) {
wproxy->e2e_first_rcv= false;
wproxy->last_e2e_count= rtps_sn_to_int64(&rcc->cc.sn);
return;
}
/* e2e check failed of crc or datalength..... */
if (rcc->e2e_
status != (uint32_t)E2E_P04STATUS_OK) {
wproxy->last_e2e_count
= rtps_sn_to_int64(&rcc->cc.sn);
return;
}
/* check delt count of e2e */
uint16_t counter_delta = (((int16_t)rtps_sn_to_int64(&rcc->cc.sn)) - wproxy->last_e2e_count);
if (counter_delta <= (r->attr->ep_attr.e2e.e2e_p04_max_delta_counter + 1U)) {
if (counter_delta > 0U) {
if (counter_delta == 1U) {
rcc->e2e_status = (uint32_t)E2E_P04STATUS_OK;
} else {
rcc->e2e_status = (uint32_t)E2E_P04STATUS_OKSOMELOST;
pr_err(ERR_FAULT, "Error status:E2E_P04STATUS_OKSOMELOST! 
"
"[profile04_cfg.max_delta_counter:%hu] [counter_delta:%hu]\n",
r->attr->ep_attr.e2e.e2e_p04_max_delta_counter, counter_delta);
}
} else {
rcc->e2e_status = (uint32_t)E2E_P04STATUS_REPEATED;
pr_err(ERR_FAULT, "Error status:E2E_P04STATUS_REPEATED! 
"
"[profile04_cfg.max_delta_counter:%hu] [counter_delta:%hu]\n",
r->attr->ep_attr.e2e.e2e_p04_max_delta_counter, counter_delta);
}
} else {
rcc->e2e_status = (uint32_t)E2E_P04STATUS_WRONGSEQUENCE;
pr_err(ERR_FAULT, "Error status:E2E_P04STATUS_WRONGSEQUENCE! 
"
"[profile04_cfg.max_delta_counter:%hu] [counter_delta:%hu]\n",
r->attr->ep_attr.e2e.e2e_p04_max_delta_counter, counter_delta);
}
wproxy->last_e2e_count
= rtps_sn_to_int64(&rcc->cc.sn);
return;
}

#04

小 结

上文中分析的代码实现了一个完整的端到端数据完整性保护机制，适用于对实时性和可靠性要求高的分布式系统。其核心是通过CRC-32/P4校验和序列号跟踪，确保数据的正确性和顺序性。

这套E2E校验算法被理想VCOS应用与DDS中的RTPS协议。代码中使用查表法加速CRC计算。代码健壮性也比较好，覆盖了数据长度、CRC、序列号等多维度校验。

不过也有几个可以改进的方向：

• 安全性：CRC并非加密哈希，如需防篡改可改用SHA-256等算法。

• 可配置性：CRC多项式和初始值可通过配置文件动态设置。

• 线程安全：若在多线程环境中使用，需增加对共享数据（如last_e2e_count）的锁保护。

/ END /