資源競爭

當行程中的多個執行緒，同時讀取一塊記憶體資料，與此同時其中一個或多個執行緒修改了這塊記憶體資料，這樣就會導致不可預期的結果

因為執行緒不安全引起的錯誤往往非常難發現，因為這種現象是不能穩定復現的，

下面舉個例子：

#include <stdio.h>
#include <string.h>
#include <pthread.h>

int g = 0;
//100000000-200000000
void *pf(void *arg){
    int i;
    for(i=0;i<100000000;i++){
        g = g+1;//讀記憶體 加法指令 寫記憶體    
    }
}

int main(){
    pthread_t ids[2];

    int ret1 = pthread_create(&ids[0],NULL,pf,NULL);
    int ret2 = pthread_create(&ids[1],NULL,pf,NULL);
    if(ret1 != 0 || ret2 != 0){
        printf("pthread_create:%s\n",strerror(ret1!=0?ret1:ret2));
        return -1;
    }
    int i;
    for(i=0;i<2;i++){
        pthread_join(ids[i],NULL);
    }   
    printf("%d\n",g);

    return 0;    
}

輸出結果：

執行緒同步

當多個執行緒同時訪問其所共享的行程資源時，需要相互協調，以防止出現資料不一致、不完整的問題，這就叫執行緒同步，

1. 互斥鎖

特點：對于讀者和寫者來說，只要有一方獲取了鎖，另一方則不能繼續獲取，進而執行 臨界區代碼，
1.互斥鎖型別
1.普通鎖 （PTHREAD_MUTEX_NORMAL） 互斥鎖默認型別，當一個執行緒對一個普通鎖加鎖以后，其余請求該鎖的執行緒將形成一個 等待佇列，并在該鎖解鎖后按照優先級獲得它，這種鎖型別保證了資源分配的公平性，一個 執行緒如果對一個已經加鎖的普通鎖再次加鎖，將引發死鎖；對一個已經被其他執行緒加鎖的普 通鎖解鎖，或者對一個已經解鎖的普通鎖再次解鎖，將導致不可預期的后果，

2.檢錯鎖（PTHREAD_MUTEX_ERRORCHECK） 一個執行緒如果對一個已經加鎖的檢錯鎖再次加鎖，則加鎖操作回傳EDEADLK；對一個已 經被其他執行緒加鎖的檢錯鎖解鎖或者對一個已經解鎖的檢錯鎖再次解鎖，則解鎖操作回傳 EPERM；

3.嵌套鎖（PTHREAD_MUTEX_RECURSIVE） 該鎖允許一個執行緒在釋放鎖之前多次對它加鎖而不發生死鎖；其他執行緒要獲得這個鎖， 則當前鎖的擁有者必須執行多次解鎖操作；對一個已經被其他執行緒加鎖的嵌套鎖解鎖，或者 對一個已經解鎖的嵌套鎖再次解鎖，則解鎖操作回傳EPERM

4.默認鎖（PTHREAD_MUTEX_ DEFAULT） 一個執行緒如果對一個已經加鎖的默認鎖再次加鎖，或者雖一個已經被其他執行緒加鎖的默 認鎖解鎖，或者對一個解鎖的默認鎖解鎖，將導致不可預期的后果；這種鎖實作的時候可能 被映射成上述三種鎖之一；

//有兩種方法創建互斥鎖，靜態方式和動態方式，
//POSIX定義了一個宏 PTHREAD_MUTEX_INITIALIZER 來靜態初始化互斥鎖，方法如下： 
//pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER? 
//在LinuxThreads實作中，pthread_mutex_t是一個結構，而 PTHREAD_MUTEX_INITIALIZER則是一個結構常量， 
//動態方式是采用pthread_mutex_init()函式來初始化互斥鎖，API定義如下：
int pthread_mutex_init (pthread_mutex_t* mutex,const pthread_mutexattr_t* mutexattr);
//其中mutexattr用于指定互斥鎖屬性（見上），如果為NULL則使用預設屬性，PTHREAD_MUTEX_TIMED_NP，這是預設值，也就是普通鎖，

int pthread_mutex_lock (pthread_mutex_t* mutex);

int pthread_mutex_unlock (pthread_mutex_t* mutex);

int pthread_mutex_destroy (pthread_mutex_t* mutex);

1)互斥量被初始化為非鎖定狀態；
2)執行緒1呼叫pthread_mutex_lock函式，立即回傳，互斥量呈鎖定狀態；
3)執行緒2呼叫pthread_mutex_lock函式，阻塞等待；
4)執行緒1呼叫pthread_mutex_unlock函式，互斥量呈非鎖定狀態；
5)執行緒2被喚醒，從pthread_mutex_lock函式中回傳，互斥量呈鎖定狀態；

范例：

#include <stdio.h>
#include <string.h>
#include <pthread.h>

int g = 0;
pthread_mutex_t lock;
//100000000-200000000
void *pf(void *arg){
    int i;
    for(i=0;i<100000000;i++){
        pthread_mutex_lock(&lock);
        g = g+1;//讀記憶體 加法指令 寫記憶體   
        pthread_mutex_unlock(&lock);
    }
}

int main(){
    pthread_t ids[2];
    int ret = pthread_mutex_init(&lock,NULL);
    if(ret != 0){
        printf("pthread_mutex_init:%s\n",strerror(ret));   
    }
    int ret1 = pthread_create(&ids[0],NULL,pf,NULL);
    int ret2 = pthread_create(&ids[1],NULL,pf,NULL);
    if(ret1 != 0 || ret2 != 0){
        printf("pthread_create:%s\n",strerror(ret1!=0?ret1:ret2));
        return -1;
    }
    int i;
    for(i=0;i<2;i++){
        pthread_join(ids[i],NULL);
    }   
    printf("%d\n",g);
    pthread_mutex_destroy(&lock);
    return 0;    
}

2. 讀寫鎖

概念：
讀寫鎖實際是一種特殊的自旋鎖，這組鎖它把對共享資源的訪問者劃分成讀者和寫 者，讀者只對共享資源進行讀訪問，寫者則需要對共享資源進行寫操作，這種鎖相對于自旋 鎖而言，能提高并發性，因為在多處理器系統中，它允許同時有多個讀者來訪問共享資源， 最大可能的讀者數為實際的邏輯CPU數，寫者是排他性的，一個讀寫鎖同時只能有一個寫 者或多個讀者（與CPU數相關），但不能同時既有讀者又有寫者，

因為讀寫鎖保持期間也是搶占失效的，如果讀寫鎖當前沒有讀者，也沒有寫者，那么 寫者可以立刻獲得讀寫鎖，否則它必須自旋在那里，直到沒有任何寫者或讀者，如果讀寫鎖 沒有寫者，那么讀者可以立即獲得該讀寫鎖，否則讀者必須自旋在那里，直到寫者釋放該讀 寫鎖，

特性：
一次只有一個執行緒可以占有寫模式的讀寫鎖, 但是可以有多個執行緒同時占有讀模式的讀寫鎖

① 當讀寫鎖是寫加鎖（獨占）狀態時, 在這個鎖被解鎖之前, 所有試圖對這個鎖加鎖的執行緒 都會被阻塞

②當讀寫鎖在讀加鎖（共享）狀態時, 所有試圖以讀模式對它進行加鎖的執行緒都可以得到訪 問權，但是如果執行緒希望以寫模式對此鎖進行加鎖, 它必須直到所有的執行緒釋放鎖，并且讀 寫鎖通常會阻塞隨后的讀模式鎖請求, 這樣可以避免讀模式鎖長期占用, 而等待的寫模式鎖 請求長期阻塞.

初始化和銷毀:
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr)?
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock)?
//成功則回傳0,出錯則回傳錯誤編號. 同互斥量以上,在釋放讀寫鎖占用的記憶體之前,需要先通過 pthread_rwlock_destroy對讀寫鎖進行清理作業, 釋放由init分配的資源. 

//讀和寫: 
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock)? 
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock)? 
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock)? 
//成功則回傳0,出錯則回傳錯誤編號.這3個函式分別實作獲取讀鎖,獲取寫鎖和釋放鎖的操作. 

//獲 取鎖的兩個函式是阻塞操作,同樣,非阻塞的函式為: 
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock)?
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock)?
//成功則回傳0,出錯則回傳錯誤編號.非阻塞的獲取鎖操作,如果可以獲取則回傳0,否則回傳 錯 誤的EBUSY

3. 自旋鎖

特點：輪詢忙等待，
自旋鎖是專為防止多處理器并發而引入的一種鎖，它在內核中大量應用于中斷處理等部 分（對于單處理器來說，防止中斷處理中的并發可簡單采用關閉中斷的方式，即在標志寄存 器中關閉/打開中斷標志位，不需要自旋鎖）
在單核cpu下不起作用：被自旋鎖保護的臨界區代碼執行時不能進行掛起狀態，會造成死 鎖
自旋鎖的初衷就是：在短期間內進行輕量級的鎖定，一個被爭用的自旋鎖使得請求它的 執行緒在等待鎖重新可用的期間進行自旋（特別浪費處理器時間），所以自旋鎖不應該被持有 時間過長，如果需要長時間鎖定的話, 最好使用信號量，

int pthread_spin_init(pthread_spinlock_t *lock, int pshared)?
// pshared可取以下屬性:
// PTHREAD_PROCESS_PRIVATE 
// PTHREAD_PROCESS_SHARED 

int pthread_spin_destroy(pthread_spinlock_t *lock)? 
int pthread_spin_lock(pthread_spinlock_t *lock)?

int pthread_spin_trylock(pthread_spinlock_t *lock)? 
int pthread_spin_unlock(pthread_spinlock_t *lock)?

4. 信號量

信號量是一個計數器，用于控制訪問有限共享資源的執行緒數，

#include <semaphore.h>

// 創建信號量
int sem_init (sem_t* sem, int pshared,unsigned int value);

sem     - 信號量ID，輸出，

pshared - 一般取0，表示呼叫行程的信號量，
          非0表示該信號量可以共享記憶體的方式，
          為多個行程所共享(Linux暫不支持)，

value   - 信號量初值，

// 信號量減1，不夠減即阻塞
int sem_wait (sem_t* sem);

// 信號量減1，不夠減即回傳-1，errno為EAGAIN
int sem_trywait (sem_t* sem);

// 信號量減1，不夠減即阻塞，
// 直到abs_timeout超時回傳-1，errno為ETIMEDOUT
int sem_timedwait (sem_t* sem,
    const struct timespec* abs_timeout);

struct timespec {
    time_t tv_sec;  // Seconds
    long   tv_nsec; // Nanoseconds [0 - 999999999]
};

// 信號量加1
int sem_post (sem_t* sem);

// 銷毀信號量
int sem_destroy (sem_t* sem);
    
int sval;
sem_getvalue (&g_sem, &sval); //獲取當前信號量，并存入sval中

范例:

#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>


int g = 0;
sem_t sem;
//100000000-200000000
void *pf(void *arg){
    int i;
    for(i=0;i<100000000;i++){
        sem_wait(&sem);//信號量-1 
        g = g+1;//讀記憶體 加法指令 寫記憶體   
        sem_post(&sem);//信號量+1
    }
}

int main(){
    pthread_t ids[2];
    int ret = sem_init(&sem,0,1);
    if(ret != 0){
        perror("sem_init");
    }
    int ret1 = pthread_create(&ids[0],NULL,pf,NULL);
    int ret2 = pthread_create(&ids[1],NULL,pf,NULL);
    if(ret1 != 0 || ret2 != 0){
        printf("pthread_create:%s\n",strerror(ret1!=0?ret1:ret2));
        return -1;
    }
    int i;
    for(i=0;i<2;i++){
        pthread_join(ids[i],NULL);
    }   
    printf("%d\n",g);
    sem_destroy(&sem);
    return 0;    
}

1)信號量APIs沒有宣告在pthread.h中，而是宣告在semaphore.h中，失敗也不回傳錯誤碼，而是回傳-1，同時設定errno，

2)互斥量任何時候都只允許一個執行緒訪問共享資源，而信號量則允許最多value個執行緒同時訪問共享資源，當value為1時，與互斥量等價，

范例：

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>

#define MAX_CONNS 5  // 最大連接數
#define MAX_USERS 50 // 最大用戶數

sem_t g_sem;

void* thread_user (void* arg) 
{
    pthread_t tid = pthread_self ();

    int sval;
    sem_getvalue (&g_sem, &sval);
    printf ("%lu執行緒：等待資料庫連接(還剩%d個空閑連接)...\n", tid, sval);

    sem_wait (&g_sem);
    sem_getvalue (&g_sem, &sval);
    printf ("%lu執行緒：獲得資料庫連接(還剩%d個空閑連接)！\n", tid, sval);

    usleep (1000000);

    sem_post (&g_sem);
    sem_getvalue (&g_sem, &sval);
    printf ("%lu執行緒：釋放資料庫連接(還剩%d個空閑連接)，\n", tid, sval);

    return NULL;
}

int main (void) 
{
    size_t i;
    pthread_t tids[MAX_USERS];
    int error;

    sem_init (&g_sem, 0, MAX_CONNS);

    for (i = 0; i < sizeof (tids) / sizeof (tids[0]); i++)
    {
        if ((error = pthread_create (&tids[i], NULL, thread_user,NULL)) != 0) 
        {
            fprintf (stderr, "pthread_create: %s\n", strerror (error));
            return -1;
        }
    }

    for (i = 0; i < sizeof (tids) / sizeof (tids[0]); i++)
    {
        if ((error = pthread_join (tids[i], NULL)) != 0) 
        {
            fprintf (stderr, "pthread_join: %s\n", strerror (error));
            return -1;
        }
    }

    sem_destroy (&g_sem);

    return 0;
}

5. 條件變數

生產者消費者模型
生產者：產生資料的執行緒，
消費者：使用資料的執行緒，

通過緩沖區隔離生產者和消費者，與二者直連相比，避免相互等待，提高運行效率，
生產快于消費，緩沖區滿，撐死，
消費快于生產，緩沖區空，餓死，

條件變數可以讓呼叫執行緒在滿足特定條件的情況下暫停，

int pthread_cond_init (pthread_cond_t* cond,const pthread_condattr_t* attr);
//亦可pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

// 使呼叫執行緒睡入條件變數cond，同時釋放互斥鎖mutex
int pthread_cond_wait (pthread_cond_t* cond,pthread_mutex_t* mutex);

int pthread_cond_timedwait (pthread_cond_t* cond,
    pthread_mutex_t* mutex,
    const struct timespec* abstime);

struct timespec {
    time_t tv_sec;  // Seconds
    long   tv_nsec; // Nanoseconds [0 - 999999999]
};

// 從條件變數cond中喚出一個執行緒，
// 令其重新獲得原先的互斥鎖
int pthread_cond_signal (pthread_cond_t* cond);

注意：被喚出的執行緒此刻將從pthread_cond_wait函式中回傳，
但如果該執行緒無法獲得原先的鎖，則會繼續阻塞在加鎖上，

// 從條件變數cond中喚出所有執行緒
int pthread_cond_broadcast (pthread_cond_t* cond);

int pthread_cond_destroy (pthread_cond_t* cond);

范例：

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t notempty = PTHREAD_COND_INITIALIZER;//不為空的條件  
pthread_cond_t notfull = PTHREAD_COND_INITIALIZER;//不為滿的條件

#define CAPCITY 20
char stock[CAPCITY];
int size = 0;

void show(const char *who,const char *con,char s){
    int i;
    for(i=0;i<size;i++){
        printf("%c",stock[i]);    
    }
    printf("  %s %c(%s)\n",con,s,who);
}

//生產者執行緒
void *producer(void *arg){
    const char *who = (char*)arg;    
    for(;;){
        pthread_mutex_lock(&mutex);
        //被喚醒之后 需要重新判斷倉庫是否滿了
        while(size >= CAPCITY){//倉庫滿了,不能進行生產
            printf("倉庫滿了,%s執行緒不能進行生產!\n",who);
            //釋放鎖 要等待 倉庫不滿的信號 
            pthread_cond_wait(&notfull,&mutex);//阻塞釋放鎖
            //被喚醒之后 重新獲得鎖
            printf("有人進行了消費,%s執行緒可以進行生產了!\n",who);
        }
        char prod = 'A'+rand()%26;
        show(who,"<--",prod);
        stock[size++] = prod;
        pthread_cond_signal(&notempty);
        pthread_mutex_unlock(&mutex);
        usleep ((rand () % 100) * 1000);
    }
    return NULL;
}
//消費者執行緒
void *customer(void *arg){
    const char *who = (const char *)arg;
    for(;;){
        pthread_mutex_lock(&mutex);
        while(size == 0){
            printf("倉庫空了,%s消費者等待!\n",who);
            pthread_cond_wait(&notempty,&mutex);
            printf("倉庫有貨物了,%s可以進行消費了!\n",who);
        }
        
        char prod = stock[--size];
        show(who,"-->",prod);

        pthread_cond_signal(&notfull);
        pthread_mutex_unlock(&mutex);
        usleep ((rand () % 100) * 1000);
    }
    return NULL;
}


int main(){
    pthread_t pid[2],cid[2];
    
    pthread_create(&pid[0],NULL,producer,"生產者1號");
    pthread_create(&pid[1],NULL,producer,"生產者2號");
    

    pthread_create(&cid[0],NULL,customer,"消費者1號");
    pthread_create(&cid[1],NULL,customer,"消費者2號");


    getchar();
    return 0;    
}

6. 屏障

屏障（barrier)是用戶協調多個執行緒并行作業的同步機制，屏障允許每個執行緒等待，直到 所有的合作執行緒都到達某一點，然后從該點繼續執行，pthread_join函式就是一種屏障，允 許一個執行緒等待，直到另一個執行緒退出，
但屏障物件的概念更廣，允許任意數量的執行緒等待，直到所有的執行緒完成處理作業，而 執行緒不需要退出，所有的執行緒達到屏障后可以接著作業，

int pthread_barrier_init(pthread_barrier_t *barrier,const pthread_barrrierattr_t *attr,unsigned int count) 

int pthread_barrier_destroy(pthread_barrier_t *barrier)?

int pthread_barrier_wait(pthread_barrier_t *barrier)

范例：

#include <stdio.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>

pthread_barrier_t barr;
void pf(void *arg){
    int x = (int)(arg);
    x = (x+2)*3;//6 9 12
    printf("%lu run....\n",pthread_self());
    sleep(x);
    printf("%lu weak!\n",pthread_self());
    pthread_barrier_wait(&barr);
    printf("%lu go on...\n",pthread_self());
}

int main(){
    pthread_t ids[3];
    pthread_barrier_init(&barr,NULL,3);
    int i;
    for(i=0;i<3;i++){
        pthread_create(&ids[i],NULL,pf,(void*)i);    
    }
    
    getchar();
    pthread_barrier_destroy(&barr);
    return 0;    
}