▶ 使用信号量来进行线程间信息传递
● 代码
1 1 #include <stdio.h> 2 2 #include <pthread.h> 3 3 #include <semaphore.h> 4 4 #pragma comment(lib, "pthreadVC2.lib") 5 5 6 6 const int thread = 8, messageSize = 100; 7 7 char messageList[thread][messageSize]; // 全局信息列表 8 8 sem_t sem[thread]; // 各线程信号量，注意每个线程都有一个 9 9 1010 void* sendMessage(void* rank) 1111 { 1212 const long long localRank = (long long)rank, dest = (localRank + 1) % thread; 1313 int i; 1414 sprintf_s(messageList[dest], messageSize, "Hello from %2d to %2d.", localRank, dest); 1515 sem_post(&sem[dest]); // 解锁 dest 的信号量，因为上一行低吗已经完成了写入，注意每次执行完函数 sem_post() 后 sem[dest] 的值增加 1 1616 sem_wait(&sem[localRank]); // 等待自己编号的信号量解锁，注意每次执行完函数 sem_wait() 后 sem[localRank] 的值减小 1（但不会小于0） 1717 printf("Thread %2d > %s\n", localRank, messageList[localRank]); 1818 return nullptr; 1919 } 2020 2121 int main() 2222 { 2323 pthread_t pth[thread]; 2424 int i; 2525 long long list[thread]; 2626 2727 for (i = 0; i < thread; i++) 2828 { 2929 sem_init(&sem[i], 0, 0); // 依次初始化信号量 3030 list[i] = i; 3131 pthread_create(&pth[i], nullptr, sendMessage, (void *)list[i]); 3232 } 3333 for (i = 0; i < thread; i++) 3434 { 3535 pthread_join(pth[i], nullptr); 3636 sem_destroy(&sem[i]); // 销毁信号量 3737 } 3838 printf("\nfinish.\n"); 3939 getchar(); 4040 return 0; 4141 }
● 输出结果
1 1 Thread 1 > Hello from 0 to 1. 2 2 Thread 2 > Hello from 1 to 2. 3 3 Thread 3 > Hello from 2 to 3. 4 4 Thread 4 > Hello from 3 to 4. 5 5 Thread 5 > Hello from 4 to 5. 6 6 Thread 6 > Hello from 5 to 6. 7 7 Thread 0 > Hello from 7 to 0. 8 8 Thread 7 > Hello from 6 to 7. 9 9 1010 finish.
● 用到的定义，注意信号量不是由 phread.h 提供的，而是 semaphore.h
1 1 typedef struct sem_t_ * sem_t; 2 2 3 3 PTW32_DLLPORT int __cdecl sem_init(sem_t * sem, int pshared, unsigned int value); 4 4 // 初始化信号量，输入一个已经声明的信号量的指针，第二个参数不明，第三个参数为 0 表示初始化完成后信号量为上锁状态 5 5 6 6 PTW32_DLLPORT int __cdecl sem_destroy(sem_t * sem);// 销毁信号量 7 7 8 8 PTW32_DLLPORT int __cdecl sem_wait(sem_t * sem); // 等待信号量为解锁状态再向下运行 9 9 1010 PTW32_DLLPORT int __cdecl sem_post(sem_t * sem); // 解锁信号量
▶ 使用忙等待和互斥量来实现路障
● 代码
1 1 #include <stdio.h> 2 2 #include <pthread.h> 3 3 #pragma comment(lib, "pthreadVC2.lib") 4 4 5 5 const int thread = 8; 6 6 int count; 7 7 pthread_mutex_t pmt; 8 8 9 9 void* work(void* rank) 1010 { 1111 const long long localRank = (long long)rank, dest = (localRank + 1) % thread; 1212 pthread_mutex_lock(&pmt); // 进入读写区，上锁，计数器加一，解锁 1313 printf("Thread %2d reached the barrier.\n", localRank); fflush(stdout); 1414 count++; 1515 pthread_mutex_unlock(&pmt); 1616 while (count < thread); // 使用忙等待来等所有的线程都达到栅栏 1717 printf("Thread %2d passed the barrier.\n", localRank); fflush(stdout); 1818 return nullptr; 1919 } 2020 2121 int main() 2222 { 2323 pthread_t pth[thread]; 2424 int i; 2525 long long list[thread]; 2626 pthread_mutex_init(&pmt, nullptr); 2727 for (i = count = 0; i < thread; i++) 2828 { 2929 list[i] = i; 3030 pthread_create(&pth[i], nullptr, work, (void *)list[i]); 3131 } 3232 for (i = 0; i < thread; i++) 3333 pthread_join(pth[i], nullptr); 3434 pthread_mutex_destroy(&pmt); 3535 printf("\nfinish.\n"); 3636 getchar(); 3737 return 0; 3838 }
● 输出结果
1 1 Thread 1 reached the barrier. 2 2 Thread 5 reached the barrier. 3 3 Thread 2 reached the barrier. 4 4 Thread 3 reached the barrier. 5 5 Thread 4 reached the barrier. 6 6 Thread 0 reached the barrier. 7 7 Thread 6 reached the barrier. 8 8 Thread 7 reached the barrier. 9 9 Thread 5 passed the barrier. 1010 Thread 6 passed the barrier. 1111 Thread 3 passed the barrier. 1212 Thread 0 passed the barrier. 1313 Thread 1 passed the barrier. 1414 Thread 4 passed the barrier. 1515 Thread 2 passed the barrier. 1616 Thread 7 passed the barrier. 1717 1818 finish.
▶ 使用信号量来实现路障
● 代码
1 1 #include <stdio.h> 2 2 #include <pthread.h> 3 3 #include <semaphore.h> 4 4 #pragma comment(lib, "pthreadVC2.lib") 5 5 6 6 const int thread = 8; 7 7 int count; 8 8 sem_t sem_count, sem_barrier; 9 9 1010 void* work(void* rank) 1111 { 1212 const long long localRank = (long long)rank, dest = (localRank + 1) % thread; 1313 printf("Thread %2d reached the barrier.\n", localRank); fflush(stdout); 1414 sem_wait(&sem_count); // 等待允许访问计数器 count，注意执行完该语句时 sem_count 值减 1，自动上锁 1515 if (count == thread - 1) // 最后一个到达进入的线程 1616 { 1717 count = 0; // 计数器清零，以后可以重复使用 1818 sem_post(&sem_count); // 计数器解锁，sem_count 值加 1 1919 for (int i = 0; i < thread - 1; sem_post(&sem_barrier), i++);// 解锁整个栅栏， 2020 } // 每有一个线程通过后面的语句 sem_wait(&sem_barrier);， 2121 else // 前面到达的线程 // sem_barrier 的值就减 1，所以这里要为该变量加上 thread - 1 2222 { 2323 count++; // 计数器加一 2424 sem_post(&sem_count); // 解锁计数器 2525 sem_wait(&sem_barrier); // 等待栅栏解锁 2626 } 2727 printf("Thread %2d passed the barrier.\n", localRank); fflush(stdout); 2828 return nullptr; 2929 } 3030 3131 int main() 3232 { 3333 pthread_t pth[thread]; 3434 int i; 3535 long long list[thread]; 3636 3737 sem_init(&sem_count, 0, 1); // 计数器锁初始化为 1，开锁状态 3838 sem_init(&sem_barrier, 0, 0); // 栅栏初始化为 0，关锁状态 3939 for (i = count = 0; i < thread; i++) 4040 { 4141 list[i] = i; 4242 pthread_create(&pth[i], nullptr, work, (void *)list[i]); 4343 } 4444 for (i = 0; i < thread; i++) 4545 pthread_join(pth[i], nullptr); 4646 sem_destroy(&sem_count), sem_destroy(&sem_barrier); 4747 printf("\nfinish.\n"); 4848 getchar(); 4949 return 0; 5050 }
● 输出结果
1Thread 0 reached the barrier. 2Thread 3 reached the barrier. 3Thread 4 reached the barrier. 4Thread 2 reached the barrier. 5Thread 1 reached the barrier. 6Thread 5 reached the barrier. 7Thread 7 reached the barrier. 8Thread 6 reached the barrier. 9Thread 4 passed the barrier. 10Thread 5 passed the barrier. 11Thread 2 passed the barrier. 12Thread 7 passed the barrier. 13Thread 3 passed the barrier. 14Thread 1 passed the barrier. 15Thread 6 passed the barrier. 16Thread 0 passed the barrier. 17 18finish.
▶ 使用条件变量来实现路障
1 1 #include <stdio.h> 2 2 #include <pthread.h> 3 3 #include <semaphore.h> 4 4 #pragma comment(lib, "pthreadVC2.lib") 5 5 6 6 const int thread = 8; 7 7 int count; 8 8 pthread_mutex_t mutex; 9 9 pthread_cond_t cond; 1010 1111 void* work(void* rank) 1212 { 1313 const long long localRank = (long long)rank, dest = (localRank + 1) % thread; 1414 printf("Thread %2d reached the barrier.\n", localRank); fflush(stdout); 1515 pthread_mutex_lock(&mutex); // 上锁 1616 count++; 1717 if (count == thread) // 最后一个进入的线程 1818 { 1919 count = 0; // 计数器清零 2020 pthread_cond_broadcast(&cond); // 广播所有线程继续向下执行 2121 } 2222 else 2323 for (; pthread_cond_wait(&cond, &mutex) != 0;);// 等待其他线程 2424 pthread_mutex_unlock(&mutex); // 条件变量阻塞解除后会自动将互斥量上锁，需要手工解除 2525 2626 printf("Thread %2d passed the barrier.\n", localRank); fflush(stdout); 2727 return nullptr; 2828 } 2929 3030 int main() 3131 { 3232 pthread_t pth[thread]; 3333 int i; 3434 long long list[thread]; 3535 pthread_mutex_init(&mutex, nullptr); 3636 pthread_cond_init(&cond, nullptr); 3737 for (i = count = 0; i < thread; i++) 3838 { 3939 list[i] = i; 4040 pthread_create(&pth[i], nullptr, work, (void *)list[i]); 4141 } 4242 for (i = 0; i < thread; i++) 4343 pthread_join(pth[i], nullptr); 4444 pthread_mutex_destroy(&mutex); 4545 pthread_cond_destroy(&cond); 4646 printf("\nfinish.\n"); 4747 getchar(); 4848 return 0; 4949 }
● 输出结果
1Thread 0 reached the barrier. 2Thread 1 reached the barrier. 3Thread 2 reached the barrier. 4Thread 4 reached the barrier. 5Thread 5 reached the barrier. 6Thread 6 reached the barrier. 7Thread 7 reached the barrier. 8Thread 3 reached the barrier. 9Thread 3 passed the barrier. 10Thread 0 passed the barrier. 11Thread 1 passed the barrier. 12Thread 5 passed the barrier. 13Thread 4 passed the barrier. 14Thread 7 passed the barrier. 15Thread 2 passed the barrier. 16Thread 6 passed the barrier. 17 18finish.
● 用到的定义，pthread.h
1 1 typedef struct pthread_cond_t_ * pthread_cond_t; 2 2 3 3 PTW32_DLLPORT int PTW32_CDECL pthread_cond_init(pthread_cond_t * cond, const pthread_condattr_t * attr);// 初始化已经声明了的条件变量，第二个参数为属性指针 4 4 5 5 PTW32_DLLPORT int PTW32_CDECL pthread_cond_destroy(pthread_cond_t * cond); // 销毁条件变量 6 6 7 7 PTW32_DLLPORT int PTW32_CDECL pthread_cond_wait(pthread_cond_t * cond, pthread_mutex_t * mutex); // 阻塞线程以等待 signal 或 brocast 8 8 9 9 PTW32_DLLPORT int PTW32_CDECL pthread_cond_signal(pthread_cond_t * cond); // 解锁一个线程 1010 1111 PTW32_DLLPORT int PTW32_CDECL pthread_cond_broadcast(pthread_cond_t * cond); // 解锁所有的线程