前面一系列练习已经把进程控制、线程、进程间通信的大概知识过了一遍，现在进入综合练习，首先练习经典问题：生产者和消费者问题
1.问题概述
多个生产/消费者在有界缓冲上操作。它利用N个字节的共享内存作为有界循环缓冲区，利用写一字符模拟放一个产品，利用读一字符模拟消费一个产品。当缓冲区空时消费者应阻塞睡眠，而当缓冲区满时生产者应当阻塞睡眠。一旦缓冲区中有空单元，生产者进程就向空单元中入写字符,并报告写的内容和位置。一旦缓冲区中有未读过的字符，消费者进程就从该单元中读出字符，并报告读取位置。生产者不能向同一单元中连续写两次以上相同的字符，消费者也不能从同一单元中连续读两次以上相同的字符。
2.问题分析
首先看阶层，有两个，分别是生产者和消费者，他们之间的缓冲区是共享内存，首先想到一点：System V共享内存实现这一个缓冲区；又因为缓冲区是临界资源，所以要用一个互斥信号量实现；生产者和消费者要采用PV信号量操作实现进程同步
因为要求多个进程能同步，所以进程访问缓冲的指针也需要共享内存实现
大致框图：
生产者->[位置指针]--------（共享内存）缓冲-------[位置指针]->消费者
3.原语
进程：Producer - 生产者进程，Consumer - 消费者进程
共有的数据结构：
buffer: array [0..k-1] of integer;
in,out: 0..k-1;
— in记录第一个空缓冲区，out记录第一个不空的缓冲区
prod_key(缓冲区空的个数),cons_key(缓冲区满的个数),mutex(临界区): semaphore;
— prod_key控制缓冲区不满,cons_key控制缓冲区不空,mutex保护临界区；
初始化prod_key=k,cons_key=0,mutex=1
producer（生产者进程）：
Item_Type item;
{
while (true)
{
produce(&item);
p(prod_key);
p(mutex);
buffer[in]:=item;
in:=(in+1) mod k;
v(mutex);
v(cons_key);
}
}
consumer（消费者进程）：
Item_Type item;
{
while (true)
{
p(cons_key);
p(mutex);
item:=buffer[out];
out:=(out+1) mod k;
v(mutex);
v(prod_key);
consume(&item);
}
}
4.IPC操作函数
我们编写一个函数实现对IPC 信息队列、共享内存、信号量的包装，以备接下来更好的编写
程序
头文件ipc.h：声明ipc操作函数和一些变量
|
/*Filename : ipc.h*/
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/msg.h>
#define BUFSZ 256
//建立或获取ipc 的一组函数的原型说明
int get_ipc_id(char *proc_file,key_t key);
char *set_shm(key_t shm_key,int shm_num,int shm_flag);
int set_msq(key_t msq_key,int msq_flag);
int set_sem(key_t sem_key,int sem_val,int sem_flag);
int down(int sem_id);
int up(int sem_id);
/*信号灯控制用的共同体*/
typedef union semuns
{
int val;
} Sem_uns;
/* 消息结构体*/
typedef struct msgbuf
{
long mtype;
char mtext[1];
} Msg_buf;
//生产消费者共享缓冲区即其有关的变量
key_t buff_key;
int buff_num;
char *buff_ptr;
//生产者放产品位置的共享指针
key_t pput_key;
int pput_num;
int *pput_ptr;
//消费者取产品位置的共享指针
key_t cget_key;
int cget_num;
int *cget_ptr;
//生产者有关的信号量
key_t prod_key; //IPC key标识
int prod_sem; //生产者同步信号量
//消费者有关的信号量
key_t cons_key; //IPC key标识
int cons_sem; //消费者同步信号量
int sem_val;
int sem_flg;
int shm_flg;
/*互斥信号量*/
key_t mtx_key; //IPC key标识
int mtx_sem;
|
ipc.c:这里包装了一些ipc的操作函数，包括信息队列、共享内存、信号量的创建/获得，及PV操作
这里逐一分析它们的实现过程：（在注释中）
|
/*ipc.c*/
#include "ipc.h"
/*
* get_ipc_id() 从/proc/sysvipc/文件系统中获取IPC 的id 号
* pfile: 对应/proc/sysvipc/目录中的IPC 文件分别为
* msg-消息队列,sem-信号量,shm-共享内存
* key: 对应要获取的IPC 的id 号的键值
*/
int get_ipc_id(char *proc_file,key_t key)
{
FILE *pf;
int i,j;
char line[BUFSZ],colum[BUFSZ];
if((pf = fopen(proc_file,"r")) == NULL)
{
perror("Proc file not open");
exit(EXIT_FAILURE);
}
fgets(line, BUFSZ,pf);
while(!feof(pf))
{
i = j = 0;
fgets(line, BUFSZ,pf);
while(line[i] == ' ')
i++;
while(line[i] !=' ')
colum[j++] = line[i++];
colum[j] = '\0';
if(atoi(colum) != key)
continue;
j=0;
while(line[i] == ' ')
i++;
while(line[i] !=' ')
colum[j++] = line[i++];
colum[j] = '\0';
i = atoi(colum);
fclose(pf);
return i;
} fclose(pf);
return -1;
}
/*
* 信号灯上的P/V 操作
* sem_id:信号灯数组标识符
* sem_num:信号灯数组下标
* buf:操作信号灯的结构
*/
/*请求 P操作*/
int P_operation(int sem_id)
{
struct sembuf buf;
buf.sem_op = -1;
buf.sem_num = 0;
buf.sem_flg = SEM_UNDO;
if((semop(sem_id,&buf,1)) <0)
{
perror("down error ");
exit(EXIT_FAILURE);
} return EXIT_SUCCESS;
}
/*释放 V操作*/
int V_operation(int sem_id)
{
struct sembuf buf;
buf.sem_op = 1;
buf.sem_num = 0;
buf.sem_flg = SEM_UNDO;
if((semop(sem_id,&buf,1)) <0)
{
perror("up error ");
exit(EXIT_FAILURE);
} return EXIT_SUCCESS;
}
/*
* set_sem 函数建立一个具有n 个信号灯的信号量
* 如果建立成功，返回一个信号灯数组的标识符sem_id
* 输入参数：
* sem_key 信号灯数组的键值
* sem_val 信号灯数组中信号灯的个数
* sem_flag 信号灯数组的存取权限
*/
/*实现过程:*建立信号灯->设置信号灯初值->返回ID*/
int set_sem(key_t sem_key,int sem_val,int sem_flg)
{
int sem_id;
Sem_uns sem_arg;
//测试由sem_key 标识的信号灯数组是否已经建立
if((sem_id = get_ipc_id("/proc/sysvipc/sem",sem_key)) < 0 )
{
//semget 新建一个信号灯,其标号返回到sem_id
if((sem_id = semget(sem_key,1,sem_flg)) < 0)
{
perror("semaphore create error");
exit(EXIT_FAILURE);
}
//设置信号灯的初值
sem_arg.val = sem_val;
if(semctl(sem_id,0,SETVAL,sem_arg) <0)
{
perror("semaphore set error");
exit(EXIT_FAILURE);
}
}
return sem_id;
}
/*
* set_shm 函数建立一个具有n 个字节的共享内存区
* 如果建立成功，返回一个指向该内存区首地址的指针shm_buf
* 输入参数：
* shm_key 共享内存的键值
* shm_val 共享内存字节的长度
* shm_flag 共享内存的存取权限
*/
/*实现过程:创建共享内存->映射到进程的指针并返回*/
char * set_shm(key_t shm_key,int shm_num,int shm_flg)
{
int i,shm_id;
char * shm_buf;
//测试由shm_key 标识的共享内存区是否已经建立
if((shm_id = get_ipc_id("/proc/sysvipc/shm",shm_key)) < 0 )
{
//shmget 新建一个长度为shm_num 字节的共享内存,其标号返回到shm_id
if((shm_id = shmget(shm_key,shm_num,shm_flg)) <0)
{
perror("shareMemory set error");
exit(EXIT_FAILURE);
}
//shmat 将由shm_id 标识的共享内存附加给指针shm_buf
if((shm_buf = (char *)shmat(shm_id,0,0)) < (char *)0)
{
perror("get shareMemory error");
exit(EXIT_FAILURE);
}
/*共享内存区初始化*/
for(i=0; i<shm_num; i++)
shm_buf[i] = 0; //初始为0
}
//shm_key 标识的共享内存区已经建立,将由shm_id 标识的共享内存附加给指针shm_buf
if((shm_buf = (char *)shmat(shm_id,0,0)) < (char *)0)
{
perror("get shareMemory error");
exit(EXIT_FAILURE);
}
return shm_buf;
}
/*
* set_msq 函数建立一个消息队列
* 如果建立成功，返回一个消息队列的标识符msq_id
* 输入参数：
* msq_key 消息队列的键值
* msq_flag 消息队列的存取权限
*/
/*实现过程:创建消息队列->返回ID*/
int set_msq(key_t msq_key,int msq_flg)
{
int msq_id;
//测试由msq_key 标识的消息队列是否已经建立
if((msq_id = get_ipc_id("/proc/sysvipc/msg",msq_key)) < 0 )
{
//msgget 新建一个消息队列,其标号返回到msq_id
if((msq_id = msgget(msq_key,msq_flg)) < 0)
{
perror("messageQueue set error");
exit(EXIT_FAILURE);
}
}
return msq_id;
}
|
5.生产者程序实现：
首先建立（已存在时为打开）一系列的信号量和共享内存，接着就按照操作原语去实现了，代码如下：
|
/*Filename : producer.c*/
#include "ipc.h"
int main(int argc,char *argv[])
{
int rate;
//可在在命令行第一参数指定一个进程睡眠秒数，以调解进程执行速度
if(argv[1] != NULL)
rate = atoi(argv[1]);
else
rate = 3; //不指定为3 秒
//共享内存使用的变量,其中键值任给,但是注意键值的
//唯一性,在另外的文件中要用同一共享内存也要采用统一键值
buff_key = 101;//缓冲区任给的键值
buff_num = 8;//缓冲区任给的长度
pput_key = 102;//生产者放产品指针的键值
pput_num = 1; //指针数
shm_flg = IPC_CREAT | 0644;//共享内存读写权限
//获取缓冲区使用的共享内存，buff_ptr 指向缓冲区首地址
buff_ptr = (char *)set_shm(buff_key,buff_num,shm_flg);
//获取生产者放产品位置指针pput_ptr
pput_ptr = (int *)set_shm(pput_key,pput_num,shm_flg);
//信号量使用的变量
prod_key = 201;//生产者同步信号灯键值
mtx_key = 202;//互斥信号灯键值
cons_key = 301;//消费者同步信号灯键值
sem_flg = IPC_CREAT | 0644;
//生产者同步信号灯初值设为缓冲区最大可用量
sem_val = buff_num;
//获取生产者同步信号灯，引用标识存prod_sem
prod_sem = set_sem(prod_key,sem_val,sem_flg);
//消费者初始无产品可取，同步信号灯初值设为0
sem_val = 0;
//获取消费者同步信号灯，引用标识存cons_sem
cons_sem = set_sem(cons_key,sem_val,sem_flg);
//生产者互斥信号灯初值为1
sem_val = 1;
//获取互斥信号灯，引用标识存mtx_sem
mtx_sem = set_sem(mtx_key,sem_val,sem_flg);
//循环执行模拟生产者不断放产品
while(1)
{
//如果缓冲区满则生产者阻塞
P_operation(prod_sem);
//如果有进程进入，本生产者阻塞
P_operation(mtx_sem);
//用写一字符的形式模拟生产者放产品，报告本进程号和放入的字符及存放的位置
buff_ptr[*pput_ptr] = 'A'+ *pput_ptr;
//挂起rate秒
sleep(rate);
printf("%d producer put: %c to Buffer[%d]\n",getpid(),buff_ptr[*pput_ptr],*pput_ptr);
//存放位置循环下移
*pput_ptr = (*pput_ptr+1) % buff_num;
//唤醒阻塞的进程
V_operation(mtx_sem);
//唤醒阻塞的消费者
V_operation(cons_sem);
}
return EXIT_SUCCESS;
}
|
6.消费者程序实现
如同生产者上述，代码如下：
|
/* consumer.c*/
#include "ipc.h"
int main(int argc,char *argv[])
{
int rate;
//可在在命令行第一参数指定一个进程睡眠秒数，以调解进程执行速度
if(argv[1] != NULL)
rate = atoi(argv[1]);
else
rate = 3; //不指定为3 秒
//共享内存使用的变量
buff_key = 101; //缓冲区任给的键值
buff_num = 8; //缓冲区任给的长度
cget_key = 103; //消费者取产品指针的键值
cget_num = 1; //指针数
shm_flg = IPC_CREAT | 0644; //共享内存读写权限
//获取缓冲区使用的共享内存，buff_ptr 指向缓冲区首地址
buff_ptr = (char *)set_shm(buff_key,buff_num,shm_flg);
//获取消费者取产品指针，cget_ptr 指向索引地址
cget_ptr = (int *)set_shm(cget_key,cget_num,shm_flg);
//信号量使用的变量
prod_key = 201; //生产者同步信号灯键值
mtx_key = 202; //互斥信号灯键值
cons_key = 301; //消费者同步信号灯键值
sem_flg = IPC_CREAT | 0644; //信号灯操作权限
//生产者同步信号灯初值设为缓冲区最大可用量
sem_val = buff_num;
//获取生产者同步信号灯，引用标识存prod_sem
prod_sem = set_sem(prod_key,sem_val,sem_flg);
//消费者初始无产品可取，同步信号灯初值设为0
sem_val = 0;
//获取消费者同步信号灯，引用标识存cons_sem
cons_sem = set_sem(cons_key,sem_val,sem_flg);
//消费者互斥信号灯初值为1
sem_val = 1;
//获取互斥信号灯，引用标识存mtx_sem
mtx_sem = set_sem(mtx_key,sem_val,sem_flg);
//循环执行模拟消费者不断取产品
while(1){
//如果无产品消费者阻塞
P_operation(cons_sem);
//如果有进程进入,本消费者阻塞
P_operation(mtx_sem);
//用读一字符的形式模拟消费者取产品，报告本进程号和获取的字符及读取的位置
sleep(rate);
printf("%d consumer get: %c fromBuffer[%d]\n",getpid(),buff_ptr[*cget_ptr],*cget_ptr);
//读取位置循环下移
*cget_ptr = (*cget_ptr+1) % buff_num;
//唤醒阻塞的进程
V_operation(mtx_sem);
//唤醒阻塞的生产者
V_operation(prod_sem);
}
return EXIT_SUCCESS;
}
|
7.编写Makefile
由于本项目工程有多个文件，所以需要Makefile来方便编译
|
hdrs = ipc.h
opts = -g -c
c_src = consumer.c ipc.c
c_obj = consumer.o ipc.o
p_src = producer.c ipc.c
p_obj = producer.o ipc.o
all: producer consumer
consumer: $(c_obj)
gcc $(c_obj) -o consumer
consumer.o: $(c_src) $(hdrs)
gcc $(opts) $(c_src)
producer: $(p_obj)
gcc $(p_obj) -o producer
producer.o: $(p_src) $(hdrs)
gcc $(opts) $(p_src)
clean:
rm consumer producer *.o
|
8.编译：
$ make
运行时打开多个终端窗口，输入
$./producer 1
另一个窗口输入：
$./consumer 1
.......
这时可以看到同步过程
阅读(2182) | 评论(0) | 转发(0) |
