最近在學習OpenMP編程。記錄下學習過程中的學習資料和歷程。同時簡單實現了《并行程序設計導論》中5.8節的生產者消費者問題（進程線程經典問題）。

OpenMP編程指南

參考資料：

《并行程序設計導論》第五章OpenMP

博客（知識點比書本總結全面些，里面有6篇，前4篇推薦學習）：

OpenMP編程指南 - 周偉明的多核、測試專欄 - CSDN博客?blog.csdn.net

超算習題（提供了虛擬練習環境，不過相對簡單。有時間的推薦自己邊學邊敲代碼實現）

在線實訓?www.easyhpc.net

以上就是本人在學習OpenMP過程涉及的資料。

生產者消費者問題

理論以及基本操作學完，需要去找個問題用OpenMP實現（測試一下自己是不是真的會用了嘛！）。

本文根據《并行程序設計導論》第5章中5.8小節講解的生產者消費者問題，用OpenMP實現了書中的偽代碼。接下來開始分析這個問題。

• 問題描述

針對典型的生產者和消費者問題，使用OpenMP編程，在共享內存系統上實現消息傳遞。

每一個線程有一個共享消息隊列。每個線程同時充當生產者和消費者角色

生產者：線程隨機產生整數“消息”和消息的目標線程，發送消息給目標線程

消費者：線程從自己的消息隊列中取出消息并打印該消息

• 問題分析

數據競爭問題：當有多個生產者向同一個消息隊列寫數據，存在數據競爭問題

數據競爭是主要解決的問題，同時書本還分析了很多其他問題。詳細見書本！

• 代碼實現

1、數據結構

設計一個消息隊列結構，生產者向隊尾發送數據，消費者從隊頭取數據。數據結構中包含隊頭和隊尾指針（用整數模擬，用數組模擬循環隊列）。為了提高并行度，隊列使用兩個鎖（隊頭鎖和隊尾鎖）實現互斥，以解決書中提到的數據競爭的問題。

//消息隊列 結構體 struct MesgQueue{int *mesg;int enqueued, dequeued;omp_lock_t front_mutex, back_mutex; };

2、操作函數

初始化函數：初始化鎖，消息隊列的內存分配，和相關的變量

void init(MesgQueue* MQ){MQ->mesg = new int[send_max];MQ->dequeued = 0;MQ->enqueued = 0;omp_init_lock(&(MQ->front_mutex));omp_init_lock(&(MQ->back_mutex)); }

3、Send_msg函數

發生消息函數：向目標線程發送消息

void Send_msg(){int mesg = rand();int dest = rand() % thread_count;//#pragma omp criticalomp_set_lock(&Msg[omp_get_thread_num()].back_mutex);Enqueue(dest, mesg);omp_unset_lock(&Msg[omp_get_thread_num()].back_mutex); }

4、Try_receive函數

接收消息函數：從消息隊列取消息并打印

void Try_receive(){int cur_p = omp_get_thread_num();int queue_size = Msg[cur_p].enqueued - Msg[cur_p].dequeued;if(queue_size == 0) return;else if(queue_size == 1){//#pragma omp criticalDequeue(cur_p);}else{Dequeue(cur_p);} }

5、Done函數

終止函數：判斷線程是否完成任務

int Done(){int cur_p = omp_get_thread_num();int queue_size = Msg[cur_p].enqueued - Msg[cur_p].dequeued;if(queue_size == 0 && done_sending == thread_count) return 1;else return 0; }

完整代碼在文尾！

• 性能分析

書本中提到用omp critical 命令實際上，線程在發生消息依舊是串行的，因為每次只能有一個線程執行發生消息的代碼。而在我們的問題中，線程0發生消息給線程1，和線程2發生消息給線程3是允許同時進行的。因此書中提到用鎖機制對每個線程的消息隊列進行互斥。

因此作此性能分析，觀察用critical和鎖時程序的不同，以及相同數據量時間上的差異。

1）critical和鎖運行時（串行和并行差異）

為了更好的感受用critical和鎖時線程之間的串行和并行操作，在發生消息時添加了輸出

使用omp critical命令時：

可以看到，輸出井然有序，線程之間先后發送消息。

使用omp_lock鎖時：

可以看到，輸出簡直不忍直視，因為當線程之間發送的目標線程不同時，可以同時執行，因此輸出就很亂。而critical限制了每次只能有一個線程執行發送消息的代碼。因此在實際運用openmp時要分析清楚互斥關系，準確使用好critical和鎖。不然可能程序性能不但沒有得到提升，甚至更差。

用書本的話總結就是：critical適用于代碼塊的互斥（粗粒度），鎖適用于需要互斥某個數據結構（比如本文的每個線程的消息隊列），（細粒度）

2）critical和鎖時間差異。

顧名思義，肯定鎖的時間會更少，性能會更好。

在實驗過程中，小數據量（線程數和需要發送消息數都比較小時），這兩個耗時相差不大。隨著數據量增大，對比會比較明顯（如下兩個圖）

omp critical:

鎖：

在實驗中，還存在兩個問題：

1) 有時候程序會無法運行出結果，如上面的critical圖中第二次運行就沒有結果，程序自動退出了emmm

2)當把輸出都加上時（發送消息和接收消息的輸出），大部分實驗鎖會比critical慢，不過這個應該是輸出流造成的緩慢。

• 完整代碼

#include <iostream> #include <stdio.h> #include <stdlib.h> #include <math.h> #include <omp.h> #include <time.h>using namespace std;int thread_count; int send_max; int done_sending;struct MesgQueue{int *mesg;int enqueued, dequeued;omp_lock_t front_mutex, back_mutex; }; struct MesgQueue* Msg;void Enqueue(int dest, int mesg){int cur_p = omp_get_thread_num();// printf("Thread %d send message %d to %d success!n",cur_p, mesg, dest);Msg[dest].mesg[Msg[dest].enqueued] = mesg;Msg[dest].enqueued++; }void Dequeue(int dest){// printf("Thread %d receive message %d success!n", dest, Msg[dest].mesg[Msg[dest].dequeued]);Msg[dest].dequeued++; }void Send_msg(){int mesg = rand();int dest = rand() % thread_count;//#pragma omp criticalomp_set_lock(&Msg[omp_get_thread_num()].back_mutex);Enqueue(dest, mesg);omp_unset_lock(&Msg[omp_get_thread_num()].back_mutex); }void Try_receive(){int cur_p = omp_get_thread_num();int queue_size = Msg[cur_p].enqueued - Msg[cur_p].dequeued;if(queue_size == 0) return;else if(queue_size == 1){//#pragma omp criticalDequeue(cur_p);}else{Dequeue(cur_p);} }int Done(){int cur_p = omp_get_thread_num();int queue_size = Msg[cur_p].enqueued - Msg[cur_p].dequeued;if(queue_size == 0 && done_sending == thread_count) return 1;else return 0; }void init(MesgQueue* MQ){MQ->mesg = new int[send_max];MQ->dequeued = 0;MQ->enqueued = 0;omp_init_lock(&(MQ->front_mutex));omp_init_lock(&(MQ->back_mutex)); }void destroy(){delete [] Msg; }int main(int argc, char* argv[]){if(argc != 2) printf("Error Command!n"), exit(0);thread_count = strtol(argv[1], NULL, 10);printf("thread_count = %d, Input the number of message:n", thread_count);cin>>send_max;Msg = new MesgQueue[thread_count];srand((unsigned)time(NULL));int sent_msgs, i;clock_t s = clock(); #pragma omp parallel num_threads(thread_count) { #pragma omp forfor(i=0; i< thread_count; ++i) init(&Msg[omp_get_thread_num()]);#pragma omp barrier#pragma omp for private(sent_msgs)for(i=0; i<thread_count; ++i){for(sent_msgs = 0; sent_msgs < send_max; ++sent_msgs){Send_msg();Try_receive();}// printf("thread %d send message done!n", omp_get_thread_num());#pragma omp atomicdone_sending++;while (!Done()){Try_receive();}} }destroy();clock_t e = clock();printf("Running time is: %dmsn", e-s);return 0; }

歡迎交流指正！

總結

以上是生活随笔為你收集整理的openmp并行编程_OpenMP实现生产者消费者问题的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。