眾所周知，對(duì)共享數(shù)據(jù)的同步和保護(hù)基本上都要用到鎖。鎖的種類(lèi)也很多，如常用的互斥鎖，讀寫(xiě)鎖，信號(hào)量，自旋鎖等等。但用到鎖，就會(huì)有阻塞，無(wú)論阻塞時(shí)間長(zhǎng)短，都會(huì)對(duì)系統(tǒng)性能有一定影響，而這也恰恰是一些對(duì)實(shí)時(shí)性要求很高的產(chǎn)品所不允許的。這時(shí)，我們就會(huì)想到無(wú)鎖操作?？墒侨绾螌?shí)現(xiàn)無(wú)所操作呢？
下面我就根據(jù)個(gè)人經(jīng)歷來(lái)分享一下，也算是對(duì)學(xué)習(xí)知識(shí)的一個(gè)總結(jié)。

問(wèn)題背景
首先介紹下OpenEM。OpenEM 的全稱(chēng)是 Open Event Machine。是 TI 針對(duì)嵌入式應(yīng)用開(kāi)發(fā)的 multicore runtime system library。OpenEM 可以在多核上有效的調(diào)度，分發(fā)任務(wù)。它把任務(wù)調(diào)度給負(fù)載輕的核，進(jìn)而實(shí)現(xiàn)動(dòng)態(tài)的負(fù)載平衡。OpenEM 是基于 TI Keystone 系列芯片的 multicore Navigator 構(gòu)建的，具有開(kāi)銷(xiāo)小，效率高的特點(diǎn)。
OpenEM是一個(gè)事件(event)驅(qū)動(dòng)的調(diào)度機(jī)制,每次有事件分發(fā)給某個(gè)Core時(shí)，事件最終會(huì)被發(fā)送給這個(gè)Core的某個(gè)調(diào)度進(jìn)程，這個(gè)進(jìn)程則會(huì)調(diào)用相應(yīng)的callback函數(shù)處理事件。我的問(wèn)題就發(fā)生在這里。我的系統(tǒng)規(guī)定，每個(gè)事件處理的時(shí)間不能超過(guò)1ms，也就是調(diào)度進(jìn)程調(diào)用callback函數(shù)的執(zhí)行時(shí)間不能超過(guò)1ms，如果超過(guò)就會(huì)造成后續(xù)處理事件的delay,系統(tǒng)就會(huì)crash。

發(fā)生Delay的代碼：

//Event Scheduler Process
event_callback()
{   ...
    pthread_rwlock_rdlock(&(g_shmDataPtr->monLock));  //Delay occurs here
    ...
    pthread_rwlock_unlock(&(g_shmDataPtr->monLock));
    ...
}

g_shmDataPtr指向的共享內(nèi)存中存放了一些控制數(shù)據(jù)，我在多個(gè)進(jìn)程中都會(huì)進(jìn)行上述讀操作。

//Process_Write
function_wr()
{   ...
    pthread_rwlock_wrlock(&(g_shmDataPtr->monLock)); 
    ...
    pthread_rwlock_unlock(&(g_shmDataPtr->monLock));
    ...
}

但是我只在Process_Write進(jìn)程中對(duì)這塊數(shù)據(jù)進(jìn)行寫(xiě)操作。而且通常情況下該操作只會(huì)在系統(tǒng)啟動(dòng)時(shí)發(fā)生，當(dāng)然如果有用戶(hù)主動(dòng)發(fā)送修改請(qǐng)求時(shí)，function_wr()也會(huì)被調(diào)用。
開(kāi)始時(shí)，我懷疑是rdlock發(fā)生了阻塞，可能是之前有人發(fā)送了修改數(shù)據(jù)的請(qǐng)求而上了寫(xiě)鎖導(dǎo)致的。于是我改用不會(huì)阻塞的trylock接口：

if (0 != pthread_rwlock_tryrdlock(&(g_shmDataPtr->monLock)))
{ return 0;/*沒(méi)有獲取鎖返回成功，不影響系統(tǒng)功能*/}

可是系統(tǒng)跑了一段時(shí)間后測(cè)試還是失敗。于是我測(cè)試了tryrdlock函數(shù)的執(zhí)行時(shí)間，結(jié)果發(fā)現(xiàn)光是這個(gè)函數(shù)的執(zhí)行時(shí)間有時(shí)候就有200us+?？磥?lái)用鎖是沒(méi)辦法滿(mǎn)足現(xiàn)在系統(tǒng)對(duì)性能的要求了。
于是我上網(wǎng)查了關(guān)于無(wú)鎖編程的一些資料，發(fā)現(xiàn)可以用原子量來(lái)進(jìn)行數(shù)據(jù)的更新管理。這種方法很像是我們的代碼的版本控制，修改和讀取的數(shù)據(jù)總是存在兩個(gè)不同的地方。于是我就把這種方法叫做版本控制，最終的實(shí)現(xiàn)方案如下：

//Event Scheduler Process
Atomic32* g_ActiveDataIndexPtr;   /*指向原子變量的全局指針*/
DataStruct * g_aaTraceCtrlDataPtr; /*新開(kāi)辟共享內(nèi)存可以存放兩份數(shù)據(jù), 索引為0，1 */
event_callback()
{   ...
    u32 actIndex = AtomicRead32(g_ActiveDataIndexPtr);
    DataStruct *currentDataPtr = &g_aaTraceCtrlDataPtr[actIndex];
    ...
}

在讀數(shù)據(jù)的地方，直接讀取現(xiàn)在有效的數(shù)據(jù)索引。即使剛讀完有效索引，索引就更新了，我們也只是使用了老的正確的數(shù)據(jù)。當(dāng)下次再讀時(shí)就可以讀到更新后的數(shù)據(jù)了。

//Process_Write
function_wr()
{   ...
    u32 currectActiveIndex =  AtomicRead32(g_ActiveDataIndexPtr);
    u32 backUpIndex = currectActiveIndex ? 0 : 1;
    /*write new date in backup shm at first*/
    ...
    /*after date already updated to backup shm, switch backup to active*/
    AtomicSet32(g_ActiveDataIndexPtr,backUpIndex );
    ...
}

在寫(xiě)數(shù)據(jù)的時(shí)候，我們先在空閑的備份共享內(nèi)存中更新數(shù)據(jù)，這樣別的進(jìn)程還在Active的共享內(nèi)存中讀數(shù)據(jù)，所以不會(huì)有任何沖突。等數(shù)據(jù)更新完，再把備份數(shù)據(jù)的索引變成有效索引，這樣之后再被讀取就用的是新的數(shù)據(jù)了。

這樣就解決了性能的問(wèn)題，但是缺點(diǎn)就是需要多一倍的控制數(shù)據(jù)存儲(chǔ)空間。

當(dāng)然，以上只是我的方法。如果文中有任何問(wèn)題，請(qǐng)大家指出，共同學(xué)習(xí)，共同進(jìn)步。：）