作者 | 不可說出品 | 汽車電子與軟件

#01MVBS

MVBS，全稱為 Micro Vehicle Bus Service，是VBSLite工程的核心組件之一，理想的VCOS中面向MCU領域設計的通信中間件，實現(xiàn)資源受限環(huán)境下的高效業(yè)務互聯(lián)互通；主要采用以數(shù)據(jù)為中心的發(fā)布-訂閱（DCSP, Data-Centric Publish Subscribe）通信模型；并通過RTPS協(xié)議進行數(shù)據(jù)傳輸，提供低時延、高可靠的數(shù)據(jù)通信，同時支持若干必要的QoS策略，同時提供RPC（Remote Function Call）功能，支持請求-調(diào)用通信模型，構建支持多種通信模式的統(tǒng)一通信平臺。 在MVBS的QoS策略中，理想實現(xiàn)了E2E來保證通信質量，使用的是CRC-32/P4算法。

#02CRC-32/P4算法

CRC-32/P4是一種特定的32位循環(huán)冗余校驗（CRC-32）算法，屬于CRC校驗算法的變體，通過特定的多項式和初始值設置，用于增強對數(shù)據(jù)傳輸或存儲中錯誤的檢測能力。以下是對CRC-32/P4的詳細解析：

1. CRC-32基礎

作用：CRC-32是一種廣泛使用的錯誤檢測碼（EDC），通過生成一個32位的校驗值，驗證數(shù)據(jù)在傳輸或存儲過程中是否出現(xiàn)錯誤。

原理：基于多項式除法，將數(shù)據(jù)視為一個長二進制數(shù)，與預定義的生成多項式進行模2除法，余數(shù)即為CRC校驗值。

2. CRC-32/P4的特點

特定多項式：CRC-32/P4使用特定的生成多項式（如代碼中的0xF4ACFB13U），與標準CRC-32（多項式為0x04C11DB7或0xEDB88320）不同，這使其適用于特定場景（如實時通信、嵌入式系統(tǒng)）。

初始值與異或值：

初始值：0xFFFFFFFFU，表示CRC寄存器在開始計算前的初始狀態(tài)。

最終異或值：0xFFFFFFFFU，在計算完成后對CRC值進行異或操作，得到最終校驗值。

查表法優(yōu)化：通過預計算的查找表（如下面章節(jié)中代碼中的crc_32P4_tab），加速CRC計算過程。

3. CRC-32/P4的應用場景

實時通信系統(tǒng)：如RTPS（實時發(fā)布-訂閱協(xié)議），用于檢測消息在傳輸過程中是否被篡改或損壞。

端到端（E2E）保護：在代碼中，CRC-32/P4用于校驗Reader/Writer ID、序列號（SN）和數(shù)據(jù)內(nèi)容，確保數(shù)據(jù)的完整性和順序性。

嵌入式系統(tǒng)：適用于資源受限的環(huán)境，因其計算效率高且錯誤檢測能力強。

4. 與標準CRC-32的區(qū)別

多項式不同：標準CRC-32通常使用0x04C11DB7或0xEDB88320，而CRC-32/P4使用0xF4ACFB13U，導致校驗值分布和錯誤檢測能力不同。

應用場景不同：標準CRC-32適用于通用數(shù)據(jù)校驗（如文件傳輸、網(wǎng)絡通信），而CRC-32/P4針對特定協(xié)議或系統(tǒng)優(yōu)化。

5. CRC-32/P4的優(yōu)勢

高效性：查表法使計算復雜度從O(n2)降至O(n)，適合實時系統(tǒng)。

可靠性：能檢測大多數(shù)突發(fā)錯誤和隨機錯誤，錯誤漏檢率低。

靈活性：通過調(diào)整多項式和初始值，可適應不同需求。

6. 代碼中的實現(xiàn)細節(jié)

數(shù)據(jù)校驗范圍：包括輸入數(shù)據(jù)、Reader/Writer ID（4字節(jié)）、序列號（8字節(jié)）和數(shù)據(jù)長度（4字節(jié)）。

序列號檢查：結合計數(shù)器機制，確保消息順序正確，防止亂序或重復。

錯誤處理：通過日志輸出和狀態(tài)碼（如E2E_P04STATUS_ERROR）反饋校驗結果。

#03理想E2E核心功能解析與代碼實現(xiàn)

1、CRC-32/P4計算

代碼結構設計：

函數(shù)：e2e_calculate_crc32P4

作用：計算數(shù)據(jù)、Reader/WriterID、序列號（SN）和數(shù)據(jù)長度的CRC-32/P4校驗值。

關鍵參數(shù)：

data_ptr：待校驗的數(shù)據(jù)指針。

data_length：數(shù)據(jù)長度。

reader_id/writer_id：通信雙方的實體ID。

sn：序列號（用于檢測消息順序）。

算法細節(jié)：

使用預計算的查表crc_32P4_tab加速CRC計算。

初始值為0xFFFFFFFF，多項式為0xF4ACFB13，最終結果異或0xFFFFFFFF。

校驗范圍包括：

1. 輸入數(shù)據(jù)（逐字節(jié)）。2. Reader/WriterID（各4字節(jié)）。3. 序列號（8字節(jié)）。4. 數(shù)據(jù)長度（4字節(jié)）。 代碼實現(xiàn)思路：

1. 初始化參數(shù)

2. 檢查輸入有效性

避免空指針導致的未定義行為若data_ptr為NULL，直接返回默認值0。

3. 初始化CRC值

設置CRC計算的初始狀態(tài)。需要將crc初始化為預定義的Crc_32P4_StartValue（0xFFFFFFFFU）。

4. 計算數(shù)據(jù)部分的CRC

校驗輸入數(shù)據(jù)的完整性。需要逐字節(jié)處理：遍歷data_length字節(jié)的數(shù)據(jù)。根據(jù)公式更新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字節(jié)數(shù)據(jù)執(zhí)行與數(shù)據(jù)部分相同的查表計算。

6. 計算序列號（SN）的CRC

確保消息順序的正確性。對sn的8字節(jié)數(shù)據(jù)執(zhí)行查表計算（邏輯同數(shù)據(jù)部分）。

7. 計算數(shù)據(jù)長度的CRC

驗證數(shù)據(jù)長度字段的合法性。對data_length的4字節(jié)數(shù)據(jù)執(zhí)行查表計算。

8. 最終異或處理

標準化CRC結果；將最終CRC值與Crc_32P4_Xor（0xFFFFFFFFU）異或，得到最終校驗碼。

9. 返回結果

輸出計算完成的CRC值。 代碼流程圖：

開始

├─ 初始化指針和查表├─ 檢查 data_ptr 是否為 NULL → 是 → 返回 0 ├─ 初始化 crc = Crc_32P4_StartValue├─ 計算數(shù)據(jù)部分的 CRC（逐字節(jié)）├─ 計算 Reader ID 的 CRC（4 字節(jié)）├─ 計算 Writer ID 的 CRC（4 字節(jié)）├─ 計算 SN 的 CRC（8 字節(jié)）├─ 計算 data_length 的 CRC（4 字節(jié)）├─ crc ^= Crc_32P4_Xor└─ 返回 crc 代碼實現(xiàn)：

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; 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,//省略查表內(nèi)容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、端到端狀態(tài)檢查

代碼結構設計：

函數(shù)：e2e_do_checkP04

作用：驗證接收數(shù)據(jù)的CRC校驗值和長度是否合法。

關鍵邏輯：

檢查數(shù)據(jù)長度是否在配置的min_data_length和max_data_length范圍內(nèi)。比較計算出的CRC（data_crc）與接收的CRC（header_src->crc）。返回狀態(tài)碼（如E2E_P04STATUS_OK或E2E_P04STATUS_ERROR）。 關鍵數(shù)據(jù)結構

rtps_entity_id_t：通信實體（Reader/Writer）的唯一標識。

rtps_sn：序列號，用于消息順序跟蹤。

structe2e_p04_cfg：E2E配置參數(shù)（如min_data_length、max_data_length、max_delta_counter）。 狀態(tài)返回

日志輸出：通過pr_err打印錯誤信息（如數(shù)據(jù)長度不匹配、CRC校驗失敗、序列號異常）。

狀態(tài)碼：

E2E_P04STATUS_OK：校驗通過。

E2E_P04STATUS_ERROR：CRC或長度不匹配。

E2E_P04STATUS_NODATAAVAILABLE：數(shù)據(jù)長度超出配置范圍。

E2E_P04STATUS_REPEATED/E2E_P04STATUS_OKSOMELOST/E2E_P04STATUS_WRONGSEQUENCE：序列號異常。 代碼實現(xiàn)思路：

1.初始化參數(shù)

2.檢查數(shù)據(jù)長度合法性，確保數(shù)據(jù)長度在配置的min_data_length和max_data_length范圍內(nèi)。

步驟：

條件判斷：

若header_src->length不滿足profile04_cfg.min_data_length<=header_src->length<=profile04_cfg.max_data_length，則：輸出錯誤日志（E2E_P04STATTUS_NODATAAVAILABLE）并直接返回錯誤狀態(tài)碼E2E_P04STATTUS_NODATAAVAILABLE。

數(shù)據(jù)長度匹配檢查：

若data_size（實際接收的數(shù)據(jù)長度）不等于header_src->length（配置的期望長度），則：輸出錯誤日志（E2E_P04STATUS_ERROR）、返回錯誤狀態(tài)碼E2E_P04STATUS_ERROR。

3. 計算并驗證CRC值

通過CRC-32/P4算法校驗數(shù)據(jù)完整性。

調(diào)用CRC計算函數(shù)：

使用e2e_calculate_crc32P4計算接收數(shù)據(jù)（pdata）、Reader/WriterID（reader_id、writer_id）、序列號（sn）和數(shù)據(jù)長度的CRC值。

比對CRC值：

若header_src->crc（接收的CRC）不等于data_crc（計算的CRC）：設置e2e_status為E2E_P04STATUS_ERROR、輸出錯誤日志（包含接收的CRC和計算的CRC）。否則，保持e2e_status為E2E_P04STATUS_OK。

4. 返回校驗結果，輸出最終的校驗狀態(tài)。 代碼實現(xiàn)流程：

開始

│├─ 初始化 data_crc 和 e2e_status├─ 檢查數(shù)據(jù)長度是否在 [min_data_length, max_data_length] 范圍內(nèi) → 不滿足 → 返回 NODATAAVAILABLE├─ 檢查 data_size 是否等于 header_src->length → 不滿足 → 返回 ERROR├─ 調(diào)用 e2e_calculate_crc32P4 計算 CRC├─ 比對 header_src->crc 和 data_crc → 不匹配 → 設置 e2e_status = ERROR└─ 返回 e2e_status

代碼實現(xiàn)：

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;} 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、序列號自增計數(shù)

代碼結構設計：

函數(shù)：e2e_increment_counter

作用：遞增一個16 位無符號整數(shù)計數(shù)器。

關鍵邏輯：

掩碼的作用：MAX_P04_COUNTER_MASK 確保計數(shù)器在達到最大值后重新從 0 開始（模運算效果）。

無符號整數(shù)處理：使用uint16_t 和 1u 避免符號擴展和溢出問題。 代碼實現(xiàn)思路：

1. 參數(shù)解析

獲取待遞增的計數(shù)器指針。輸入?yún)?shù)counter是一個指向uint16_t的指針，表示需要遞增的計數(shù)器。

2. 遞增計數(shù)器

將計數(shù)器值加1。

3. 應用掩碼限制范圍

確保計數(shù)器值不超過MAX_P04_COUNTER_MASK（通常為0xFFFF或更?。?。

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

MAX_P04_COUNTER_MASK是一個掩碼，用于截斷高位，限制計數(shù)器范圍。

例如，若MAX_P04_COUNTER_MASK=0xFFFF，則計數(shù)器在0x0000到0xFFFF之間循環(huán)。

4. 更新計數(shù)器值

將計算后的新值寫回計數(shù)器。通過*counter=...將遞增并掩碼后的值賦給原計數(shù)器。

代碼流程：

開始

├─ 解引用 counter 獲取當前值├─ 計算新值：current_value + 1├─ 應用掩碼：new_value & MAX_P04_COUNTER_MASK└─ 將新值寫回 counter 代碼實現(xiàn)：

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

4、序列號計數(shù)器檢查

代碼結構設計：

函數(shù)：e2e_check_counter

作用：檢測序列號（SN）的連續(xù)性，防止消息丟失或亂序。

關鍵邏輯：

首次接收時初始化計數(shù)器（e2e_first_rcv）。

后續(xù)接收時計算計數(shù)器差值（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）。

代碼思路：

該函數(shù)e2e_check_counter用于檢查通信系統(tǒng)中消息序列號（SN）的連續(xù)性，確保消息順序正確且無丟失或重復。

1. 檢查遠程Writer狀態(tài)

確保Writer在線且可提供計數(shù)器信息。若wproxy==NULL（Writer離線），輸出錯誤日志并直接返回。

2. 檢查Reader的E2E功能狀態(tài)

若Reader未啟用E2E，跳過后續(xù)檢查。調(diào)用reader_e2e_enabled(r)檢查Reader是否啟用E2E。若未啟用，設置rcc->e2e_status=E2E_P04STATUS_OK和rcc->counter=0，直接返回。

3. 處理首次接收消息

初始化首次接收的計數(shù)器。若wproxy->e2e_first_rcv為true（首次接收）：

設置e2e_first_rcv=false，標記非首次接收。

記錄當前序列號到wproxy->last_e2e_count，直接返回。

4. 處理CRC或數(shù)據(jù)長度錯誤

若之前校驗失敗，僅更新計數(shù)器，不修改狀態(tài)。若rcc->e2e_status!=E2E_P04STATUS_OK（校驗失?。焊聎proxy->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. 更新計數(shù)器

記錄當前序列號用于下一次比較：將wproxy->last_e2e_count更新為當前序列號current_sn。

代碼流程：

開始

├─ 檢查 wproxy 是否為 NULL → 是 → 輸出錯誤日志并返回├─ 檢查 Reader 是否啟用 E2E → 未啟用 → 設置狀態(tài)為 OK，返回├─ 檢查是否為首次接收 → 是 → 初始化 last_e2e_count，返回├─ 檢查 e2e_status 是否為 OK → 不是 → 更新 last_e2e_count，返回├─ 計算 counter_delta = current_sn - last_e2e_count├─ 判斷 counter_delta 范圍：│ ├─ counter_delta == 1 → 設置狀態(tài)為 OK│ ├─ 1 < counter_delta <= max_delta_counter + 1 → 設置狀態(tài)為 OKSOMELOST，輸出日志│ ├─ counter_delta <= 0 → 設置狀態(tài)為 REPEATED，輸出日志│ └─ counter_delta > max_delta_counter + 1 → 設置狀態(tài)為 WRONGSEQUENCE，輸出日志└─ 更新 last_e2e_count 為 current_sn

代碼實現(xiàn)：

void e2e_check_counter(struct reader *r, struct reader_cache_change *rcc, struct writer_proxy *wproxy){if (wproxy == NULL) { pr_err(ERR_FAULT, "The remote writer has offline, can't get counter and e2e_status from writer");return;}if (!(reader_e2e_enabled(r))) {rcc->e2e_status = (uint32_t)E2E_P04STATUS_OK;rcc->counter = 0;return;}if (wproxy->e2e_first_rcv) {wproxy->e2e_first_rcv= false;wproxy->last_e2e_count= rtps_sn_to_int64(&rcc->cc.sn);return;}if (rcc->e2e_status != (uint32_t)E2E_P04STATUS_OK) {wproxy->last_e2e_count= rtps_sn_to_int64(&rcc->cc.sn);return;}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小  結

上文中分析的代碼實現(xiàn)了一個完整的端到端數(shù)據(jù)完整性保護機制，適用于對實時性和可靠性要求高的分布式系統(tǒng)。其核心是通過CRC-32/P4校驗和序列號跟蹤，確保數(shù)據(jù)的正確性和順序性。 這套E2E校驗算法被理想VCOS應用與DDS中的RTPS協(xié)議。代碼中使用查表法加速CRC計算。代碼健壯性也比較好，覆蓋了數(shù)據(jù)長度、CRC、序列號等多維度校驗。 不過也有幾個可以改進的方向：

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

可配置性：CRC多項式和初始值可通過配置文件動態(tài)設置。

線程安全：若在多線程環(huán)境中使用，需增加對共享數(shù)據(jù)（如last_e2e_count）的鎖保護。