Pthreads 信號量，路障，條件變量

▶ 使用信號量來進行線程間信息傳遞

● 代碼

 1 #include <stdio.h>
 2 #include <pthread.h>
 3 #include <semaphore.h>
 4 #pragma comment(lib, "pthreadVC2.lib")
 5 
 6 const int thread = 8, messageSize = 100;
 7 char messageList[thread][messageSize];  // 全局信息列表
 8 sem_t sem[thread];                      // 各線程信號量，注意每一個線程都有一個
 9 
10 void* sendMessage(void* rank)
11 {
12     const long long localRank = (long long)rank, dest = (localRank + 1) % thread;
13     int i; 
14     sprintf_s(messageList[dest], messageSize, "Hello from %2d to %2d.", localRank, dest);
15     sem_post(&sem[dest]);               // 解鎖 dest 的信號量，由於上一行低嗎已經完成了寫入，注意每次執行完函數 sem_post() 後 sem[dest] 的值增長 1
16     sem_wait(&sem[localRank]);          // 等待本身編號的信號量解鎖，注意每次執行完函數 sem_wait() 後 sem[localRank] 的值減少 1（但不會小於0）
17     printf("Thread %2d > %s\n", localRank, messageList[localRank]);
18     return nullptr;
19 }
20 
21 int main()
22 {    
23     pthread_t pth[thread];
24     int i;
25     long long list[thread];
26     
27     for (i = 0; i < thread; i++)
28     {
29         sem_init(&sem[i], 0, 0);        // 依次初始化信號量
30         list[i] = i;
31         pthread_create(&pth[i], nullptr, sendMessage, (void *)list[i]);
32     }
33     for (i = 0; i < thread; i++)
34     {
35         pthread_join(pth[i], nullptr);
36         sem_destroy(&sem[i]);           // 銷燬信號量
37     }
38     printf("\nfinish.\n");
39     getchar();
40     return 0;
41 }

● 輸出結果

 1 Thread  1 > Hello from  0 to  1.
 2 Thread  2 > Hello from  1 to  2.
 3 Thread  3 > Hello from  2 to  3.
 4 Thread  4 > Hello from  3 to  4.
 5 Thread  5 > Hello from  4 to  5.
 6 Thread  6 > Hello from  5 to  6.
 7 Thread  0 > Hello from  7 to  0.
 8 Thread  7 > Hello from  6 to  7.
 9 
10 finish.

● 用到的定義，注意信號量不是由 phread.h 提供的，而是 semaphore.h

 1 typedef struct sem_t_ * sem_t;
 2 
 3 PTW32_DLLPORT int __cdecl sem_init(sem_t * sem, int pshared, unsigned int value);
 4     // 初始化信號量，輸入一個已經聲明的信號量的指針，第二個參數不明，第三個參數爲 0 表示初始化完成後信號量爲上鎖狀態
 5 
 6 PTW32_DLLPORT int __cdecl sem_destroy(sem_t * sem);// 銷燬信號量
 7 
 8 PTW32_DLLPORT int __cdecl sem_wait(sem_t * sem);   // 等待信號量爲解鎖狀態再向下運行
 9 
10 PTW32_DLLPORT int __cdecl sem_post(sem_t * sem);   // 解鎖信號量

 

▶ 使用忙等待和互斥量來實現路障

● 代碼

 1 #include <stdio.h>
 2 #include <pthread.h>
 3 #pragma comment(lib, "pthreadVC2.lib")
 4 
 5 const int thread = 8;
 6 int count;
 7 pthread_mutex_t pmt;
 8 
 9 void* work(void* rank)
10 {
11     const long long localRank = (long long)rank, dest = (localRank + 1) % thread;    
12     pthread_mutex_lock(&pmt);   // 進入讀寫區，上鎖，計數器加一，解鎖
13     printf("Thread %2d reached the barrier.\n", localRank); fflush(stdout);
14     count++;
15     pthread_mutex_unlock(&pmt);
16     while (count < thread);     // 使用忙等待來等全部的線程都達到柵欄
17     printf("Thread %2d passed the barrier.\n", localRank); fflush(stdout);
18     return nullptr;
19 }
20 
21 int main()
22 {    
23     pthread_t pth[thread];
24     int i;
25     long long list[thread];
26     pthread_mutex_init(&pmt, nullptr);
27     for (i = count = 0; i < thread; i++)
28     {
29         list[i] = i;
30         pthread_create(&pth[i], nullptr, work, (void *)list[i]);
31     }
32     for (i = 0; i < thread; i++)
33         pthread_join(pth[i], nullptr);
34     pthread_mutex_destroy(&pmt);
35     printf("\nfinish.\n");
36     getchar();
37     return 0;
38 }

● 輸出結果

 1 Thread  1 reached the barrier.
 2 Thread  5 reached the barrier.
 3 Thread  2 reached the barrier.
 4 Thread  3 reached the barrier.
 5 Thread  4 reached the barrier.
 6 Thread  0 reached the barrier.
 7 Thread  6 reached the barrier.
 8 Thread  7 reached the barrier.
 9 Thread  5 passed the barrier.
10 Thread  6 passed the barrier.
11 Thread  3 passed the barrier.
12 Thread  0 passed the barrier.
13 Thread  1 passed the barrier.
14 Thread  4 passed the barrier.
15 Thread  2 passed the barrier.
16 Thread  7 passed the barrier.
17 
18 finish.

 

▶ 使用信號量來實現路障

● 代碼

 1 #include <stdio.h>
 2 #include <pthread.h>
 3 #include <semaphore.h>
 4 #pragma comment(lib, "pthreadVC2.lib")
 5 
 6 const int thread = 8;
 7 int count;
 8 sem_t sem_count, sem_barrier;
 9 
10 void* work(void* rank)
11 {
12     const long long localRank = (long long)rank, dest = (localRank + 1) % thread;        
13     printf("Thread %2d reached the barrier.\n", localRank); fflush(stdout);
14     sem_wait(&sem_count);       // 等待容許訪問計數器 count，注意執行完該語句時 sem_count 值減 1，自動上鎖
15     if (count == thread - 1)    // 最後一個到達進入的線程
16     {
17         count = 0;              // 計數器清零，之後能夠重複使用
18         sem_post(&sem_count);   // 計數器解鎖，sem_count 值加 1
19         for (int i = 0; i < thread - 1; sem_post(&sem_barrier), i++);// 解鎖整個柵欄，
20     }                                                                // 每有一個線程經過後面的語句 sem_wait(&sem_barrier);，
21     else                        // 前面到達的線程                     // sem_barrier 的值就減 1，因此這裏要爲該變量加上 thread - 1 
22     {
23         count++;                // 計數器加一
24         sem_post(&sem_count);   // 解鎖計數器
25         sem_wait(&sem_barrier); // 等待柵欄解鎖
26     }
27     printf("Thread %2d passed the barrier.\n", localRank); fflush(stdout);    
28     return nullptr;
29 }
30 
31 int main()
32 {    
33     pthread_t pth[thread];
34     int i;
35     long long list[thread];
36     
37     sem_init(&sem_count, 0, 1);     // 計數器鎖初始化爲 1，開鎖狀態
38     sem_init(&sem_barrier, 0, 0);   // 柵欄初始化爲 0，關鎖狀態
39     for (i = count = 0; i < thread; i++)
40     {
41         list[i] = i;
42         pthread_create(&pth[i], nullptr, work, (void *)list[i]);
43     }
44     for (i = 0; i < thread; i++)
45         pthread_join(pth[i], nullptr);
46     sem_destroy(&sem_count), sem_destroy(&sem_barrier);
47     printf("\nfinish.\n");
48     getchar();
49     return 0;
50 }

● 輸出結果

Thread  0 reached the barrier.
Thread  3 reached the barrier.
Thread  4 reached the barrier.
Thread  2 reached the barrier.
Thread  1 reached the barrier.
Thread  5 reached the barrier.
Thread  7 reached the barrier.
Thread  6 reached the barrier.
Thread  4 passed the barrier.
Thread  5 passed the barrier.
Thread  2 passed the barrier.
Thread  7 passed the barrier.
Thread  3 passed the barrier.
Thread  1 passed the barrier.
Thread  6 passed the barrier.
Thread  0 passed the barrier.

finish.

 

▶ 使用條件變量來實現路障

 1 #include <stdio.h>
 2 #include <pthread.h>
 3 #include <semaphore.h>
 4 #pragma comment(lib, "pthreadVC2.lib")
 5 
 6 const int thread = 8;
 7 int count;
 8 pthread_mutex_t mutex;
 9 pthread_cond_t cond;
10 
11 void* work(void* rank)
12 {
13     const long long localRank = (long long)rank, dest = (localRank + 1) % thread;        
14     printf("Thread %2d reached the barrier.\n", localRank); fflush(stdout);
15     pthread_mutex_lock(&mutex);         // 上鎖
16     count++;
17     if (count == thread)                // 最後一個進入的線程
18     {
19         count = 0;                      // 計數器清零
20         pthread_cond_broadcast(&cond);  // 廣播全部線程繼續向下執行
21     }
22     else
23         for (; pthread_cond_wait(&cond, &mutex) != 0;);// 等待其餘線程
24     pthread_mutex_unlock(&mutex);       // 條件變量阻塞解除後會自動將互斥量上鎖，須要手工解除    
25     
26     printf("Thread %2d passed the barrier.\n", localRank); fflush(stdout);    
27     return nullptr;
28 }
29 
30 int main()
31 {    
32     pthread_t pth[thread];
33     int i;
34     long long list[thread];
35     pthread_mutex_init(&mutex, nullptr);
36     pthread_cond_init(&cond, nullptr);
37     for (i = count = 0; i < thread; i++)
38     {
39         list[i] = i;
40         pthread_create(&pth[i], nullptr, work, (void *)list[i]);
41     }
42     for (i = 0; i < thread; i++)
43         pthread_join(pth[i], nullptr);
44     pthread_mutex_destroy(&mutex);
45     pthread_cond_destroy(&cond);
46     printf("\nfinish.\n");
47     getchar();
48     return 0;
49 }

● 輸出結果

Thread  0 reached the barrier.
Thread  1 reached the barrier.
Thread  2 reached the barrier.
Thread  4 reached the barrier.
Thread  5 reached the barrier.
Thread  6 reached the barrier.
Thread  7 reached the barrier.
Thread  3 reached the barrier.
Thread  3 passed the barrier.
Thread  0 passed the barrier.
Thread  1 passed the barrier.
Thread  5 passed the barrier.
Thread  4 passed the barrier.
Thread  7 passed the barrier.
Thread  2 passed the barrier.
Thread  6 passed the barrier.

finish.

● 用到的定義，pthread.h

 1 typedef struct pthread_cond_t_ * pthread_cond_t;
 2 
 3 PTW32_DLLPORT int PTW32_CDECL pthread_cond_init(pthread_cond_t * cond, const pthread_condattr_t * attr);// 初始化已經聲明瞭的條件變量，第二個參數爲屬性指針
 4 
 5 PTW32_DLLPORT int PTW32_CDECL pthread_cond_destroy(pthread_cond_t * cond);                              // 銷燬條件變量
 6 
 7 PTW32_DLLPORT int PTW32_CDECL pthread_cond_wait(pthread_cond_t * cond, pthread_mutex_t * mutex);        // 阻塞線程以等待 signal 或 brocast
 8 
 9 PTW32_DLLPORT int PTW32_CDECL pthread_cond_signal(pthread_cond_t * cond);                               // 解鎖一個線程
10 
11 PTW32_DLLPORT int PTW32_CDECL pthread_cond_broadcast(pthread_cond_t * cond);                            // 解鎖全部的線程