[[440988]]

 本文轉(zhuǎn)載自微信公眾號(hào)「非典型技術(shù)宅」，作者無知少年。轉(zhuǎn)載本文請(qǐng)聯(lián)系非典型技術(shù)宅公眾號(hào)。

前言

linux 開發(fā)應(yīng)該多少都聽過大名鼎鼎的 list.h ，其簡潔優(yōu)雅的設(shè)計(jì)，一個(gè)頭文件完成了一個(gè)高可用的鏈表。

但是 list.h 并不是線程安全的，在多線程的情況下使用，必須考慮多線程數(shù)據(jù)同步的問題。

然而。。。。

我在使用互斥鎖對(duì)鏈表的操作進(jìn)行保護(hù)之后，還是被坑了!

下面是把我坑了的 list_for_each_entry 和 list_for_each_entry_safe 兩個(gè)函數(shù)的詳細(xì)分析。

list.h 單線程使用

在 list.h 這個(gè)文件中有非常多值得學(xué)習(xí)的地方，比如其最經(jīng)典的 container_of 的實(shí)現(xiàn)。

在這里只介紹幾個(gè)常用的函數(shù)，然后重點(diǎn)分析在多線程使用時(shí)的碰到的坑。

鏈表初始化及添加節(jié)點(diǎn)

首先定義一個(gè)鏈表和鏈表節(jié)點(diǎn)，定義一個(gè)產(chǎn)品，其屬性為產(chǎn)品重量(weight)。

typedef struct product_s 
{ 
    struct list_head product_node; 
    uint32_t index; 
    uint32_t weight;   
}product_t; 
 
//初始化鏈表頭 
LIST_HEAD(product_list); 

生產(chǎn)者在生產(chǎn)完產(chǎn)品后，將產(chǎn)品加入鏈表，等待消費(fèi)者使用。

void producer(void) 
{ 
    product_t *product = malloc(sizeof(product_t)); 
 
    // 產(chǎn)品重量為 300 ± 10 
    product->weight = 290 + rand() % 20; 
     
    printf("product :%p, weight %d\n", product, product->weight); 
    list_add_tail(&product->product_node, &product_list); 
} 

遍歷鏈表

使用 list_for_each_entry 可以將鏈表進(jìn)行遍歷：

// 遍歷打印鏈表信息 
void print_produce_list(void) 
{ 
    product_t *product; 
    list_for_each_entry(product, &product_list, product_node) 
    { 
        printf("manufacture product :%p, weight %d\n", product, product->weight); 
    } 
} 

其具體實(shí)現(xiàn)是使用宏將 for 循環(huán)的初始條件和完成條件進(jìn)行了替換：

#define list_for_each_entry(pos, head, member)                \ 
    for (pos = list_first_entry(head, typeof(*pos), member);    \ 
         &pos->member != (head);                    \ 
         pos = list_next_entry(pos, member)) 

其中for循環(huán)的第一個(gè)參數(shù)將 pos = list_first_entry(head, typeof(*pos), member); 初始化為鏈表頭指向的第一個(gè)實(shí)體鏈表成員。

第二個(gè)參數(shù) &pos->member != (head) 為跳出條件，當(dāng)pos->member再次指向鏈表頭時(shí)跳出for循環(huán)。

for的第三個(gè)參數(shù)通過pos->member.next指針遍歷整個(gè)實(shí)體鏈表，當(dāng)pos->member.next再次指向我們的鏈表頭的時(shí)候跳出for循環(huán)。

但是 list_for_each_entry 不能在遍歷的循環(huán)體中刪除節(jié)點(diǎn)，因?yàn)樵谘h(huán)體中刪除鏈表節(jié)點(diǎn)后，當(dāng)前節(jié)點(diǎn)的前驅(qū)結(jié)點(diǎn)和后繼結(jié)點(diǎn)指針會(huì)被置空。

在for循環(huán)的第三個(gè)參數(shù)中，獲取下一個(gè)節(jié)點(diǎn)時(shí)，會(huì)發(fā)生非法指針訪問

“安全遍歷鏈表”

為了解決在遍歷鏈表過程中，無法刪除結(jié)點(diǎn)的問題，在 list.h 中提供了一個(gè)安全刪除節(jié)點(diǎn)的函數(shù)

// 刪除重量小于300的節(jié)點(diǎn) 
void remove_unqualified_produce(void) 
{ 
    product_t *product, *temp; 
    list_for_each_entry_safe(product, temp, &product_list, product_node) 
    { 
        // 移除重量小于300的產(chǎn)品 
        if (product->weight < 300) 
        { 
            printf("remove product :%p, weight %d\n", product, product->weight); 
            list_del(&product->product_node); 
            free(product); 
        } 
    } 
} 

其實(shí)現(xiàn)是使用一個(gè)中間變量，在開始每次開始執(zhí)行循環(huán)體前，將當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)保存到中間變量，從而實(shí)現(xiàn)"安全"遍歷

#define list_for_each_entry_safe(pos, n, head, member)            \ 
    for (pos = list_first_entry(head, typeof(*pos), member),    \ 
        n = list_next_entry(pos, member);            \ 
         &pos->member != (head);                    \ 
         pos = n, n = list_next_entry(n, member)) 

多線程中使用list.h

上面我們?cè)谥骶€程里面創(chuàng)建了產(chǎn)品，并放入到鏈表中并，并過濾了重量小于300的產(chǎn)品。

后面我們?cè)诙嗑€程中對(duì)產(chǎn)品進(jìn)行消費(fèi)(兩個(gè)線程同時(shí)消費(fèi)鏈表的數(shù)據(jù)，使用完成后刪除并釋放結(jié)點(diǎn))。

這里的邏輯和單線程中的差不多，同樣是遍歷鏈表，然后從鏈表中刪除節(jié)點(diǎn)。不同的是，由于list.h自身沒有帶鎖，所以需要使用互斥鎖將鏈表的操作進(jìn)行保護(hù)。

于是很自然的有了下面的代碼

void * consumer(void *arg) 
{ 
    product_t *product, *temp; 
 
    // 使用互斥鎖對(duì)鏈表進(jìn)行保護(hù) 
    pthread_mutex_lock(&producer_mutex); 
    list_for_each_entry_safe(product, temp, &product_list, product_node) 
    { 
        list_del(&product->product_node); 
        printf("consume product :%p, weight %d, consumer :%p\n", product, product->weight, (void *)pthread_self()); 
        pthread_mutex_unlock(&producer_mutex); 
 
        // 睡一會(huì)，防止太快了 
        usleep(10*1000); 
        free(product); 
        pthread_mutex_lock(&producer_mutex); 
    } 
    pthread_mutex_unlock(&producer_mutex); 
     
    return NULL; 
} 

在上面的這段代碼中，在對(duì)鏈表操作時(shí)，使用互斥鎖對(duì)鏈表進(jìn)行了保護(hù)，使同時(shí)只能有一個(gè)線程訪問鏈表。

不過你以為這樣就好了嘛，如果時(shí)這樣，這篇文章就沒存在的必要了。。。

[[440989]]

在兩個(gè)線程同時(shí)遍歷時(shí)，即便是加了鎖之后，數(shù)據(jù)訪問也不安全。

在遍歷使用的 list_for_each_entry_safe 宏中，使用了一個(gè)零時(shí)變量對(duì)保存著當(dāng)前鏈表的下一個(gè)節(jié)點(diǎn)。

但是多線程訪問鏈表時(shí)，有可能零時(shí)變量保存的節(jié)點(diǎn)，被另一個(gè)線程刪除了，所以訪問的時(shí)候又是 Segmentation fault

[[440990]]

后記

原因找到了，也就好辦了。以至于解決方法嘛，我是使用一個(gè)全局的零時(shí)變量，將需要?jiǎng)h除節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)保存起來，手動(dòng)實(shí)現(xiàn)多線程的"安全刪除"。

// 消費(fèi)者 
void * consumer(void *arg) 
{ 
    product_t *product; 
 
    // 使用互斥鎖對(duì)鏈表進(jìn)行保護(hù) 
    pthread_mutex_lock(&producer_mutex); 
    list_for_each_entry(product, &product_list, product_node) 
    { 
        temp = list_next_entry(product, product_node); 
        list_del(&product->product_node); 
        printf("consume product :%p, weight %d, consumer :%p\n", product, product->weight, (void *)pthread_self()); 
        pthread_mutex_unlock(&producer_mutex); 
 
        // 睡一會(huì)，防止太快 
        usleep(10*1000); 
        free(product); 
 
        pthread_mutex_lock(&producer_mutex); 
        if(temp != NULL){ 
            product = list_prev_entry(temp, product_node); 
        } 
    } 
    pthread_mutex_unlock(&producer_mutex); 
 
    return NULL; 
} 

一個(gè)晚上找到了這個(gè)bug，然后又花了一個(gè)晚上記錄下來這個(gè)bug。

據(jù)說 klist.h 是 list.h 的線程安全版本，后面花時(shí)間在研究一下去，今天就先睡了。。。