最近在學(xué)習(xí)OpenMP編程。記錄下學(xué)習(xí)過程中的學(xué)習(xí)資料和歷程。同時(shí)簡單實(shí)現(xiàn)了《并行程序設(shè)計(jì)導(dǎo)論》中5.8節(jié)的生產(chǎn)者消費(fèi)者問題（進(jìn)程線程經(jīng)典問題）。

OpenMP編程指南

參考資料：

《并行程序設(shè)計(jì)導(dǎo)論》第五章OpenMP

博客（知識(shí)點(diǎn)比書本總結(jié)全面些，里面有6篇，前4篇推薦學(xué)習(xí)）：

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

超算習(xí)題（提供了虛擬練習(xí)環(huán)境，不過相對(duì)簡單。有時(shí)間的推薦自己邊學(xué)邊敲代碼實(shí)現(xiàn)）

在線實(shí)訓(xùn)?www.easyhpc.net

以上就是本人在學(xué)習(xí)OpenMP過程涉及的資料。

生產(chǎn)者消費(fèi)者問題

理論以及基本操作學(xué)完，需要去找個(gè)問題用OpenMP實(shí)現(xiàn)（測試一下自己是不是真的會(huì)用了嘛！）。

本文根據(jù)《并行程序設(shè)計(jì)導(dǎo)論》第5章中5.8小節(jié)講解的生產(chǎn)者消費(fèi)者問題，用OpenMP實(shí)現(xiàn)了書中的偽代碼。接下來開始分析這個(gè)問題。

• 問題描述

針對(duì)典型的生產(chǎn)者和消費(fèi)者問題，使用OpenMP編程，在共享內(nèi)存系統(tǒng)上實(shí)現(xiàn)消息傳遞。

每一個(gè)線程有一個(gè)共享消息隊(duì)列。每個(gè)線程同時(shí)充當(dāng)生產(chǎn)者和消費(fèi)者角色

生產(chǎn)者：線程隨機(jī)產(chǎn)生整數(shù)“消息”和消息的目標(biāo)線程，發(fā)送消息給目標(biāo)線程

消費(fèi)者：線程從自己的消息隊(duì)列中取出消息并打印該消息

• 問題分析

數(shù)據(jù)競爭問題：當(dāng)有多個(gè)生產(chǎn)者向同一個(gè)消息隊(duì)列寫數(shù)據(jù)，存在數(shù)據(jù)競爭問題

數(shù)據(jù)競爭是主要解決的問題，同時(shí)書本還分析了很多其他問題。詳細(xì)見書本！

• 代碼實(shí)現(xiàn)

1、數(shù)據(jù)結(jié)構(gòu)

設(shè)計(jì)一個(gè)消息隊(duì)列結(jié)構(gòu)，生產(chǎn)者向隊(duì)尾發(fā)送數(shù)據(jù)，消費(fèi)者從隊(duì)頭取數(shù)據(jù)。數(shù)據(jù)結(jié)構(gòu)中包含隊(duì)頭和隊(duì)尾指針（用整數(shù)模擬，用數(shù)組模擬循環(huán)隊(duì)列）。為了提高并行度，隊(duì)列使用兩個(gè)鎖（隊(duì)頭鎖和隊(duì)尾鎖）實(shí)現(xiàn)互斥，以解決書中提到的數(shù)據(jù)競爭的問題。

//消息隊(duì)列 結(jié)構(gòu)體 struct MesgQueue{int *mesg;int enqueued, dequeued;omp_lock_t front_mutex, back_mutex; };

2、操作函數(shù)

初始化函數(shù)：初始化鎖，消息隊(duì)列的內(nèi)存分配，和相關(guān)的變量

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函數(shù)

發(fā)生消息函數(shù)：向目標(biāo)線程發(fā)送消息

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函數(shù)

接收消息函數(shù)：從消息隊(duì)列取消息并打印

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函數(shù)

終止函數(shù)：判斷線程是否完成任務(wù)

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 命令實(shí)際上，線程在發(fā)生消息依舊是串行的，因?yàn)槊看沃荒苡幸粋€(gè)線程執(zhí)行發(fā)生消息的代碼。而在我們的問題中，線程0發(fā)生消息給線程1，和線程2發(fā)生消息給線程3是允許同時(shí)進(jìn)行的。因此書中提到用鎖機(jī)制對(duì)每個(gè)線程的消息隊(duì)列進(jìn)行互斥。

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

1）critical和鎖運(yùn)行時(shí)（串行和并行差異）

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

使用omp critical命令時(shí)：

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

使用omp_lock鎖時(shí)：

可以看到，輸出簡直不忍直視，因?yàn)楫?dāng)線程之間發(fā)送的目標(biāo)線程不同時(shí)，可以同時(shí)執(zhí)行，因此輸出就很亂。而critical限制了每次只能有一個(gè)線程執(zhí)行發(fā)送消息的代碼。因此在實(shí)際運(yùn)用openmp時(shí)要分析清楚互斥關(guān)系，準(zhǔn)確使用好critical和鎖。不然可能程序性能不但沒有得到提升，甚至更差。

用書本的話總結(jié)就是：critical適用于代碼塊的互斥（粗粒度），鎖適用于需要互斥某個(gè)數(shù)據(jù)結(jié)構(gòu)（比如本文的每個(gè)線程的消息隊(duì)列），（細(xì)粒度）

2）critical和鎖時(shí)間差異。

顧名思義，肯定鎖的時(shí)間會(huì)更少，性能會(huì)更好。

在實(shí)驗(yàn)過程中，小數(shù)據(jù)量（線程數(shù)和需要發(fā)送消息數(shù)都比較小時(shí)），這兩個(gè)耗時(shí)相差不大。隨著數(shù)據(jù)量增大，對(duì)比會(huì)比較明顯（如下兩個(gè)圖）

omp critical:

鎖：

在實(shí)驗(yàn)中，還存在兩個(gè)問題：

1) 有時(shí)候程序會(huì)無法運(yùn)行出結(jié)果，如上面的critical圖中第二次運(yùn)行就沒有結(jié)果，程序自動(dòng)退出了emmm

2)當(dāng)把輸出都加上時(shí)（發(fā)送消息和接收消息的輸出），大部分實(shí)驗(yàn)鎖會(huì)比critical慢，不過這個(gè)應(yīng)該是輸出流造成的緩慢。

• 完整代碼

#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; }

歡迎交流指正！

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。