linux生产者消费者问题

Posted chenyibin1995

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了linux生产者消费者问题相关的知识,希望对你有一定的参考价值。

(多进程+共享内存+信号量)

一.分析

生产者和消费者问题是多个相互合作的进程之间的一种抽象。生产者和消费者之间的关系:
1.  对缓冲区的访问是互斥的。由于两者都会修改缓冲区,因此,一方修改缓冲区时,另一方不能修改,这就是互斥。
2.  一方的行为影响另一方。缓冲区不空,才能消费,何时不空?生产了就不空;缓冲区满,就不能生产,何时不满?消费了就不满。这是同步关系。
为了描述这种关系,一方面,使用共享内存代表缓冲区;另一方面,使用 互斥信号量 控制对缓冲区的访问,使用同步信号量描述两者的依赖关系。
 

二. 共享存储

共享存储是进程间通信的一种手段,通常,使用信号量同步或互斥访问共享存储。共享存储的原理是将进程的地址空间映射到一个共享存储段。在LINUX下,通过使用 shmget 函数创建或者获取共享内存。

1. 创建

1)不指定 KEY

// IPC_PRIVATE指出需要创建内存; 
//SHM_SIZE 指出字节大小; 
//SHM_MODE 指出访问权限字如 0600表示,用户可以读写该内存
int shmget(key_t IPC_PRIVATE,size_t SHM_SIZE,int SHM_MODE);

2)指定KEY

//如果SHM_KEY指向的共享存储已经存在,则返回共享存储的ID; 
//否则,创建共享存储并返回其ID
int  shmget(key_t SHM_KEY,size_t SHM_SIZE,int SHM_MODE);
 

2. 访问

方法一

只需要共享存储的 ID 就可以通过  shmat  函数获得共享存储所占用的实际地址。因此,可以在父进程的栈中用变量存放指向共享存储的指针,那么 fork 之后,子进程就可以很方便地通过这个指针访问共享存储了。

方法二

如果进程之间并没有父子关系,但是协商好了共享存储的 KEY , 那么在每个进程中,就可以通过 KEY 以及 shmget 函数获得共享存储的 I D , 进而通过 shmat 函数获得共享存储的实际地址,最后访问。
 
在我的实现中,我把生产者实现为父进程,消费者实现为子进程,并通过方法一实现进程之间共享内存。

三. 信号量集

信号量有两种原语 P 和 V ,P 锁定资源,V 释放资源。LINUX 下的使用信号量集合的接口特别复杂。我所用到的函数如下:

1. 创建或者获取信号量集合

// IPC_PRIVATE 表示创建信号量集, NUM_OF_SEM表示该集合中有多少信号量; FLAGS复杂不追究
semget(IPC_PRIVATE, NUM_OF_SEM, FLAGS );
// SEM_KEY 是 key_t 类型
//如果 SEM_KEY 代表的信号量集存在,则返回信号量集的ID
//如果不存在,则创建信号量集并返回ID
semget(SEM_KEY, NUM_OF_SEM,FLAGS);
 

2. 初始化信号量

创建的过程并未指定信号量的初始值,需要使用 semctl 函数指定。
semctl(int semSetId , int semIdx , int cmd, union semun su);
 
其中 semSetId 是指信号量集的 ID , semIdx 指信号量集中某个信号量的索引(从零开始), 如果是要设置信号量的值, 填 SETVAL 即可, 为了设置信号量的值,可以指定su.val为索要设置的值。
我在 UBUNTU 下使用 union semun 编译时总报错:
invalid use of undefined type ‘union semun’
 
据说是 Linux 下删除了 semun 的定义。可以通过自定义 semun 解决:
 
  1. #if defined(__GNU_LIBRARY__) && !defined(_SEM_SEMUN_UNDEFINED)  
  2.    
  3. #else   
  4.    
  5. union semun{  
  6.     int val;  
  7.     struct semid_ds *buf;  
  8.     unsigned short *array;  
  9. };  
  10. #endif  


 四.代码分解

1. 头文件

 
  1. #include "stdio.h"   //支持 printf  
  2. #include <sys/shm.h> //支持 shmget shmat 等  
  3. #include <sys/sem.h> //支持 semget   
  4. #include <stdlib.h>  //支持 exit  

2. 信号量

共需要三个信号量:
第一个信号量用于限制生产者必须在缓冲区不满时才能生产,是同步信号量
第二个信号量用于限制消费者必须在缓冲区有产品时才消费,是同步信号量
第三个信号量用于限制生产者和消费者在访问缓冲区时必须互斥,是互斥信号量
 
创建信号量集合,semget
 
  1. if((semSetId = semget(IPC_PRIVATE,3,SEM_MODE)) < 0)  
  2. {  
  3.     perror("create semaphore failed");  
  4.     exit(1);  
  5. }  

初始化三个信号量,semctl,需要用到 union semun 
  1. union semun su;  
  2.   
  3.   
  4. su.val = N_BUFFER;  
  5. if(semctl(semSetId,0,SETVAL, su) < 0){  
  6.     perror("semctl failed");  
  7.     exit(1);  
  8. }  
  9. su.val = 0;  
  10. if(semctl(semSetId,1,SETVAL,su) < 0){  
  11.     perror("semctl failed");  
  12.     exit(1);  
  13. }  
  14. su.val = 1;  
  15. if(semctl(semSetId,2,SETVAL,su) < 0){  
  16.     perror("semctl failed");  
  17.     exit(1);  
  18. }  


 
封装对信号量集中的某个信号量的值的+1或者-1操作

  1.   
  2.   
  3. void waitSem(int semSetId,int semNum)  
  4. {  
  5.     struct sembuf sb;  
  6.     sb.sem_num = semNum;  
  7.     sb.sem_op = -1;  
  8.     sb.sem_flg = SEM_UNDO;  
  9.       
  10.       
  11.     if(semop(semSetId,&sb,1) < 0){  
  12.         perror("waitSem failed");  
  13.         exit(1);  
  14.     }  
  15. }  
  16. void sigSem(int semSetId,int semNum)  
  17. {  
  18.     struct sembuf sb;  
  19.     sb.sem_num = semNum;  
  20.     sb.sem_op = 1;  
  21.     sb.sem_flg = SEM_UNDO;  
  22.       
  23.       
  24.     if(semop(semSetId,&sb,1) < 0){  
  25.         perror("waitSem failed");  
  26.         exit(1);  
  27.     }  
  28. }  

3. 使用共享内存


  1.   
  2. struct ShM{  
  3.     int start;  
  4.     int end;  
  5. }* pSM;  
  6.   
  7.   
  8.   
  9. if((shmId = shmget(IPC_PRIVATE,SHM_SIZE,SHM_MODE)) < 0)  
  10. {  
  11.     perror("create shared memory failed");  
  12.     exit(1);  
  13. }  
  14.   
  15. pSM = (struct ShM *)shmat(shmId,0,0);  
  16.   
  17. pSM->start = 0;  
  18. pSM->end = 0;  


4. 生产过程

  1. while(1)  
  2. {  
  3.     waitSem(semSetId,0);  
  4.     waitSem(semSetId,2);  
  5.     produce();  
  6.     sigSem(semSetId,2);  
  7.     sleep(1);  
  8.     sigSem(semSetId,1);  
  9. }  

5. 消费过程

  1. while(1)  
  2. {  
  3.     waitSem(semSetId,1);  
  4.     waitSem(semSetId,2);  
  5.     consume();  
  6.     sigSem(semSetId,2);  
  7.     sigSem(semSetId,0);  
  8.     sleep(2);  
  9. }  
 

五. 代码全文

 
  1. #include "stdio.h"  
  2. #include <sys/shm.h>  
  3. #include <sys/sem.h>  
  4. #include <stdlib.h>  
  5. #define SHM_SIZE (1024*1024)  
  6. #define SHM_MODE 0600  
  7. #define SEM_MODE 0600  
  8.   
  9. #if defined(__GNU_LIBRARY__) && !defined(_SEM_SEMUN_UNDEFINED)  
  10.    
  11. #else   
  12.    
  13. union semun{  
  14.     int val;  
  15.     struct semid_ds *buf;  
  16.     unsigned short *array;  
  17. };  
  18. #endif  
  19.   
  20. struct ShM{  
  21.     int start;  
  22.     int end;  
  23. }* pSM;  
  24.   
  25. const int N_CONSUMER = 3;  
  26. const int N_BUFFER = 5;  
  27. int shmId = -1,semSetId=-1;  
  28. union semun su;  
  29.   
  30.   
  31.   
  32. void waitSem(int semSetId,int semNum)  
  33. {  
  34.     struct sembuf sb;  
  35.     sb.sem_num = semNum;  
  36.     sb.sem_op = -1;  
  37.     sb.sem_flg = SEM_UNDO;  
  38.       
  39.       
  40.     if(semop(semSetId,&sb,1) < 0){  
  41.         perror("waitSem failed");  
  42.         exit(1);  
  43.     }  
  44. }  
  45. void sigSem(int semSetId,int semNum)  
  46. {  
  47.     struct sembuf sb;  
  48.     sb.sem_num = semNum;  
  49.     sb.sem_op = 1;  
  50.     sb.sem_flg = SEM_UNDO;  
  51.       
  52.       
  53.     if(semop(semSetId,&sb,1) < 0){  
  54.         perror("waitSem failed");  
  55.         exit(1);  
  56.     }  
  57. }  
  58.   
  59. void produce()  
  60. {  
  61.     int last = pSM->end;  
  62.     pSM->end = (pSM->end+1) % N_BUFFER;  
  63.     printf("生产 %d\\n",last);  
  64. }  
  65.   
  66. void consume()  
  67. {  
  68.     int last = pSM->start;  
  69.     pSM->start = (pSM->start + 1)%N_BUFFER;  
  70.     printf("消耗 %d\\n",last);  
  71. }  
  72.   
  73. void init()  
  74. {  
  75.       
  76.     if((shmId = shmget(IPC_PRIVATE,SHM_SIZE,SHM_MODE)) < 0)  
  77.     {  
  78.         perror("create shared memory failed");  
  79.         exit(1);  
  80.     }  
  81.     pSM = (struct ShM *)shmat(shmId,0,0);  
  82.     pSM->start = 0;  
  83.     pSM->end = 0;  
  84.       
  85.       
  86.       
  87.       
  88.       
  89.   
  90.     if((semSetId = semget(IPC_PRIVATE,3,SEM_MODE)) < 0)  
  91.     {  
  92.         perror("create semaphore failed");  
  93.         exit(1);  
  94.     }  
  95.       
  96.     su.val = N_BUFFER;  
  97.     if(semctl(semSetId,0,SETVAL, su) < 0){  
  98.         perror("semctl failed");  
  99.         exit(1);  
  100.     }  
  101.     su.val = 0;  
  102.     if(semctl(semSetId,1,SETVAL,su) < 0){  
  103.         perror("semctl failed");  
  104.         exit(1);  
  105.     }  
  106.     su.val = 1;  
  107.     if(semctl(semSetId,2,SETVAL,su) < 0){  
  108.         perror("semctl failed");  
  109.         exit(1);  
  110.     }  
  111. }  
  112. int main()  
  113. {  
  114.     int i = 0,child = -1;  
  115.     init();  
  116.       
  117.     for(i = 0; i < N_CONSUMER; i++)  
  118.     {  
  119.         if((child = fork()) < 0)  
  120.         {  
  121.             perror("the fork failed");  
  122.             exit(1);  
  123.         }  
  124.         else if(child == 0)  
  125.         {  
  126.             printf("我是第 %d 个消费者子进程,PID = %d\\n",i,getpid());  
  127.             while(1)  
  128.             {  
  129.                 waitSem(semSetId,1);  
  130.                 waitSem(semSetId,2);  
  131.                 consume();  
  132.                 sigSem(semSetId,2);  
  133.                 sigSem(semSetId,0);  
  134.                 sleep(2);  
  135.             }  
  136.             break;  
  137.         }  
  138.     }  
  139.       
  140.       
  141.       
  142.     if(child > 0)  
  143.     {  
  144.         while(1)  
  145.         {  
  146.             waitSem(semSetId,0);  
  147.             waitSem(semSetId,2);  
  148.             produce();  
  149.             sigSem(semSetId,2);  
  150.             sleep(1);  
  151.             sigSem(semSetId,1);  
  152.         }  
  153.     }  
  154.     return 0;  
  155. }  
  156. 技术分享图片










以上是关于linux生产者消费者问题的主要内容,如果未能解决你的问题,请参考以下文章

linux生产者消费者问题

在linux下用c语言实现用多进程同步方法演示“生产者-消费者”问题

Linux线程编程之生产者消费者问题

LINUX多线程(生产者消费者模型,POXIS信号量)

[Linux]生产者与消费者 三种模型 C

Linux多线程——生产者消费者模型