RingBuff在多核通訊之間的妙用

大家好，我是小麥，今天分享一篇RingBuff相關(guān)的文章。整體感覺很不錯。

前言

兩個核（分為主核和從核）之間進行通訊，一般使用共享內(nèi)存的形式進行。

核間通訊的機制，是以共享內(nèi)存為媒介，利用核間中斷來通知對方。通過核間對象的句柄進行具體的訪問和操作；

（1）多個核進行核間通訊時，首先由一個核創(chuàng)建一個核間對象，另外一個核通過名稱或索引定位到該對象的句柄，從而對核間通訊對象進行操作。

（2）核間中斷來通知對方，采用“硬件信號量”對資源進行臨界保護，再利用操作系統(tǒng)的信號量使得核間任務的通訊如同單核任務上的通訊。

1、定義

RingBuf又稱為Circular Buffer,分為兩種：分片形式的RingBuf、內(nèi)存分割形式的RingBuf。

2、特點

1. 由主核和從核定義好要進行數(shù)據(jù)共享內(nèi)存區(qū)域（地址、大?。?/li>
2. 環(huán)形緩沖區(qū)RingBuf用于兩個核之間通訊；
3. 環(huán)形緩沖區(qū)RingBuf的使用遵守嚴格的先進先出順序進行處理；
4. 環(huán)形緩沖區(qū)RingBuf是一項很好的技術(shù)，不用頻繁的分配內(nèi)存，而且在大多數(shù)情況下，內(nèi)存的反復使用也使得我們能用更少的內(nèi)存塊做更多的事。
5. 環(huán)形緩沖區(qū)RingBuf是一個先進先出的循環(huán)緩沖區(qū)，可以向通信程序提供對緩沖區(qū)的互斥訪問。
6. 環(huán)形緩沖區(qū)RingBuf是點對點的單向通訊；

3、環(huán)形緩沖區(qū)的實現(xiàn)原理

環(huán)形緩沖區(qū)通常有一個讀指針和一個寫指針。

讀指針指向環(huán)形緩沖區(qū)中可讀的數(shù)據(jù)，寫指針指向環(huán)形緩沖區(qū)中可寫的緩沖區(qū)。通過移動讀指針和寫指針就可以實現(xiàn)緩沖區(qū)的數(shù)據(jù)讀取和寫入。

由于Ringbuf具體實現(xiàn)上可以分為分片形式的RingBuf、內(nèi)存分割形式的RingBuf；下面分別講述這兩種實現(xiàn)方式。

分片形式的RingBuf

1 原理

分片形式的RingBuf首先在內(nèi)存堆中申請一塊兩核共享的內(nèi)存區(qū)域，然后將這塊內(nèi)存區(qū)域分割成固定大小的分片，然后使用相應的內(nèi)存管理結(jié)構(gòu)進行管理。如下圖所示：

2 RingBuf管理結(jié)構(gòu)體

typedef?struct
{
????u32?WrIndex;//寫入分片的索引值
????u32?RdIndex;//讀出分片的索引值
????u8??*MemBufAddr;//內(nèi)存堆的起始地址
????u32?MemShardingNum;//內(nèi)存分片數(shù)量
????u32?MemShardingSize;//每個內(nèi)存分片大小
????u32?InitDoneFlag;//初始化完成標志
????u32HandleSem;//RingBuf使用的信號量
????u32?SrcCpuID;//寫入端CPU?ID號
????u32?DstCpuID;//讀取端CPU?ID號
}RingBufMan;
3 RingBuf要實現(xiàn)的函數(shù)接口

1. 初始化RingBuf對象RingBufInit(), 由其中一個核根據(jù)Ringbuf配置參數(shù)創(chuàng)建RingBuf對象，創(chuàng)建完成后需要完成Ringbuf管理結(jié)構(gòu)體的初始化；

2. 將新創(chuàng)建的RingBuf對象加入到共享對象管理結(jié)構(gòu)體，便于集中管理(RingBufJoinShareQue())；

3. 查找相應的RingBuf對象ID（根據(jù)寫入端/讀取端CPU ID號在共享對象隊列中查找RingBufGetID()）；

4. 判空RingBufIsEmpty()

(ringBufMan->?WrIndex?==?ringBufMan->?RdIndex)


5. 判滿?RingBufIsFull()

Tmp?=?ringBufMan->?WrIndex?-?ringBufMan->?RdIndex? ?1

(Tmp?==0?||?Tmp?==?ringBufMan->?MemShardingNum)


6. 寫入RingBufPut(RingBufMan*,u8 *BufAddr,u32 LenPut)

pWr=ringBufMan->MemBufAddr ringBufMan->WrIndex*ringBufMan->MemShardingSize;

*(u32*)pWr?=?LenPut;

memcpy((void*)(?pWr sizeof(u32)?),(void*)pBufAddr,?LenPut);

ringBufMan->WrIndex=?(ringBufMan->WrIndex 1)%?ringBufMan->?MemShardingNum;


7. 讀取RingBufGet(RingBufMan*,u8 *BufAddr,u32 LenMaxGut,u8 *Len)


pRd=ringBufMan->MemBufAddr ringBufMan->RdIndex*ringBufMan->MemShardingSize;

LenValid=?*(u32*)pRd;

Lencp=min(LenValid,?LenMaxGut);

memcpy((void*)BufAddr?,(void*)(?pWr sizeof(u32)?),?Lencp);

ringBufMan->RdIndex=?(ringBufMan->RdIndex 1)%?ringBufMan->?MemShardingNum;

*Len=?Lencp;
當然，這里的Put\Get只是一個分片的讀寫，至于一包數(shù)據(jù)寫時需要多少個分片，讀時需要讀完幾個分片，需要根據(jù)數(shù)據(jù)包的大小具體的計算；

內(nèi)存分割形式的RingBuf

typedef?struct?{?

????????unsigned?char?*buffer;?
????????unsigned?int?size;?
????????unsigned?int?in;?
????????unsigned?int?out;?
????????spinlock_t?*lock;?
}kfifo?;
其中buffer指向存放數(shù)據(jù)的緩沖區(qū)，size是緩沖區(qū)的大小，in是寫指針下標，out是讀指針下標，lock是加到struct kfifo上的自旋鎖（上面說不加鎖不是這里的鎖），防止多個進程并發(fā)訪問此數(shù)據(jù)結(jié)構(gòu)。

• 當in==out時，說明緩沖區(qū)為空；
• 當(in-out)==size時，說明緩沖區(qū)已滿。

注：我們保有對應的讀寫指針，當?shù)谝慌鷶?shù)據(jù)（藍色）完成，第二批數(shù)據(jù)（紅色）會根據(jù)當前的寫指針位置繼續(xù)我們的數(shù)據(jù)操作，當達到最大的Buffer_Size時，會重新回到Buffer的開始端。

4、多個應用讀寫RingBuf情況下的處理

互斥鎖

在通常情況下，環(huán)形緩沖區(qū)的讀用戶僅僅會影響讀指針，而寫用戶僅僅會影響寫指針。如果僅僅有一個讀用戶和一個寫用戶，那么不需要添加互斥保護機制就可以保證數(shù)據(jù)的正確性。如果有多個讀寫用戶訪問環(huán)形緩沖區(qū)，那么必須添加互斥保護機制來確保多個用戶互斥訪問環(huán)形緩沖區(qū)。

互斥鎖可以采用兩個核共享的自旋鎖來實現(xiàn)，哪個核等到鎖，哪個核有權(quán)對RingBuf資源進行讀寫操作。

使用異步消息隊列AsyncMsgQ

AsyncMsgQ是基于RingBuf來實現(xiàn)的，用于主從核內(nèi)多個線程之間數(shù)據(jù)交換。

1. AsyncMsgQ是基于RingBuf，實現(xiàn)了兩個核之間兩個線程之間的通訊，通過MsgAttr來表示消息來自哪個核的哪個線程ID；

2. 可以將AsyncMsgQ與線程ID進行綁定，并在收發(fā)端設置一個守護線程，根據(jù)接收到的MsgAttr中的目的線程ID進行消息分發(fā)，進而將消息分發(fā)到不同的AsyncMsgQ隊列中，一包消息接收完成之后可以調(diào)用AsyncMsgQ已經(jīng)掛載好的回調(diào)函數(shù)，進一步對消息進行解析或使用信號量將消息發(fā)送到任務。

多個應用程序讀寫還可以在RingBuf的基礎(chǔ)上實現(xiàn)消息隊列，消息隊列通過管理結(jié)構(gòu)體記錄消息的port，保證寫入到ringbuf時數(shù)據(jù)寫入/讀取的原子性。

除了保證寫入/讀取的原子性操作，還有一個問題就是，若核1中有多個應用程序以臨界訪問的形式向RingBuf中寫數(shù)據(jù)，那么另外一個核0如何知道是哪個應用程序?qū)懭氲絉ingBuf？

為了實現(xiàn)Ringbuf這種攜帶數(shù)據(jù)的功能，我們可以對寫入RingBuf的每一條消息進行標識。

例如，在每一包消息的頭部增加一個數(shù)據(jù)結(jié)構(gòu)用于表示該報數(shù)據(jù)來自哪個核的哪個應用，又去往哪個核的哪個應用，包含了多少數(shù)據(jù)，使用了多少分片等等信息。

typedef?struct
{
????u32?MsgShardingNum;//本消息包占用的Ringbuf分片數(shù)
????u32?MsgLen;//本消息包的長度
????u32?SrcCpuID;//源端CPU?ID號
????u32?DstCpuID;//目的端CPU?ID號
????u32?SrcThreadID;//源端線程ID
????u32?DstThreadID;//目的端線程ID

}MsgAttr;
同時為了表示每個分片所屬消息ID及有效數(shù)據(jù)大小，還涉及了以下結(jié)構(gòu)：

typedef?struct
{
????u16ShardingID;//分片ID，用于識別是第幾個分片
????u16ShardingLen;//分片有效數(shù)據(jù)長度

}ShardingAttr;
RingBuf在內(nèi)存中的分布如下：

AsyncMsgQ主要的接口函數(shù)：

1. 初始化AsyncMsgQ
2. AsyncMsgQ通道申請

申請未被使用的AsyncMsgQ通道，并同ThreadID綁定，同時，設置相應的消息接收完成回調(diào)函數(shù)

3. 數(shù)據(jù)發(fā)送

計算發(fā)送數(shù)據(jù)所需要的RingBuf分片數(shù)，將數(shù)據(jù)拷貝到分片，并設置相應的MsgAttr\?ShardingAttr

4. 數(shù)據(jù)接收

包括輪詢接收，信號量阻塞接收，根據(jù)接收到的MsgAttr進行數(shù)據(jù)解析和分發(fā)；

參考：

https://en.wikipedia.org/wiki/Circular_buffer

原文鏈接：https://blog.csdn.net/u010961173/article/details/79839450

—— The End?——
推薦好文??
BLDC驅(qū)動入門最簡教程

2021-07-01

深究：預編譯指令的點點滴滴

2021-06-13

工作五年反思

2021-06-05

Linux系統(tǒng)是如何用虛擬內(nèi)存來欺騙應用程序的？

2021-05-25

原創(chuàng)不易，歡迎轉(zhuǎn)發(fā)、留言、點贊、分享給你的朋友，感謝您的支持！

長按識別二維碼關(guān)注我

你點的每個好看，我都認真當成了喜歡