课设二 使用IPC机制实现“生产者-过滤者-消费者”问题

@

目录

• 要求
• 知识梳理
  • 生产者-过滤者-消费者问题
  • 信号量和信号量集
  • 信号量的pv操作
  • 共享内存

要求

学习Linux进程间通信机制，使用信号量和共享内存实现进程同步问题“生产者-过滤者-消费者”问题。具体要求：
1.创建信号量集，实现同步互斥信号量。
2.创建共享内存，模拟存放产品的公共缓冲池。
3.创建并发进程，实现进程对共享缓冲池的并发操作。生产者生产产品后放入缓冲池，过滤者取出产品，对产品过滤（可自己设计，如对产品值进行+1）再放入缓冲池，消费者将过滤后的产品取走。可创建一个或两个缓冲池。

知识梳理

生产者-过滤者-消费者问题

生产者消费者问题（英语：Producer-consumer problem），也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

关键在于：生产者放入产品之后才解锁了消费者消费产品的过程，消费者拿出产品之后才解锁了生产者生产产品的过程。

信号量和信号量集

信号量(Semaphore)，有时被称为信号灯，是在多线程环境下使用的一种设施，是可以用来保证两个或多个关键代码段不被并发调用。在进入一个关键代码段之前，线程必须获取一个信号量；一旦该关键代码段完成了，那么该线程必须释放信号量。
当利用信号量机制解决了单个资源的互斥访问后，我们讨论如何控制同时需要多个资源的互斥访问。信号量集是指同时需要多个资源时的信号量操作。

信号量的pv操作

信号量是个被保护的量，只有P、V操作和信号量初始化操作才能访问和改变它的值。
P操作和V操作定义如下，其中S为信号量值:

共享内存

共享存储

共享存储操作适得两个或两个以上的进程可以共用一段物理内存（一般情况下，两个进程的数据区是完全独立的，父进程用fork创建子进程后，子进程会复制父进程数据到自己的数据区）。

（1）创建共享内存

include<sys/shm.h>

int shmget(key_t key,size_t size, int permflags);

参数key是共享内存的标识，size是共享内存段的最小字节数，permflags是访问权限，值的设置同semget一样。

（2）共享内存的控制

include<sys/shm.h>

int shmctl(int shmid, int command, struct shmid_ds *shm_stat);

该函数semctl相似，command可设为IPC_STAT,IPC_SET,IPC_RMID。参数shm_stat指向存放属性的结构体，具体内容请参考手册。

（3）共享内存的附接和断开

＃include<sys/shm.h>

void *shmat(int shmid, const void *addr, int shmflags);

int shmdt(const void *addr);

由于两个函数需指出进程地址空间中的地址，因此比较复杂。简化的方法是将shmat中的地址设为NULL。

**P (S)

{
S=S-1；
若S<0,将该进程状态置为等待状态，然后将该进程的PCB插入相应的S信号量等待 队列末尾，直到有其他进程在S上执行V操作为止；
}**

V (S)
{
S = S + 1 ；
若S<=0,释放在S信号量队列中等待的一个进程，将其状态改变为就绪态，并将其插 入就绪队列；然后，执行本操作的进程继续执行；
}

代码实现

#include<sys/sem.h>
#include<sys/ipc.h>
#include<sys/types.h>
#include<errno.h>
#include<stdlib.h>
#include<stdio.h>
#include<unistd.h>

#include <stddef.h>
#include <sys/shm.h>
#include <string.h>

#define TEXT_SIZE 1024


union union_semun {
int val;
struct semid_ds *buf;
unsigned short *array;
}ctl_arg;

/*struct sembuf {
unsigned short sem_num;
short sem_op;
short sem_flg;
};*/

struct databuf {
int written;
char text[TEXT_SIZE];
};

static int shmid, semid_full, semid_empty, semid_zuse;

void initshm(struct databuf**p) {
//u521bu5efau5171u4eabu5185u5b58
shmid = shmget((key_t)4321, sizeof(struct databuf), 0600 | IPC_CREAT);
if (shmid == -1) {
printf("shmget failed!
");
return;
}
//u8fdeu63a5u5230u5f53u524du8fdbu7a0buff0cu8fd4u56deu6307u5411u7b2cu4e00u4e2au5b57u8282u7684u6307u9488
void *shm = shmat(shmid, NULL, 0);
if (shm == (void *)-1) {
printf("shmat failed!
");
return;
}
printf("
memory attached at %d
", (int)shm);

shm = *p;

}

void setdata(struct databuf**p) {

}
int initsem(key_t semkey) {
//u521bu5efau4fe1u53f7u91cfu96c6uff0cu5e76u4e14u521du59cbu5316
printf("
initsem running...");
int tmp = semget(semkey, 1, 0600 | IPC_CREAT | IPC_EXCL);
if (tmp == -1) {
printf("semget create failed!");
return;
}
printf("semid %d created successful", tmp);
return tmp;
}
void remobj(void) {
if (semctl(semid_empty, IPC_RMID, 0) == -1)	printf("
remctl remobj error
");
else printf("
remctl delete success");
if (semctl(semid_full, IPC_RMID, 0) == -1)	printf("
remctl remobj error
");
else printf("
remctl delete success");
}

int P(int semid) {
struct sembuf p_buf;
p_buf.sem_num = 0;
p_buf.sem_op = -1;
p_buf.sem_flg = SEM_UNDO;
if (semop(semid, &p_buf, 1) == -1) {
perror("p(semid) failed");
exit(1);
}
return 1;
}

int V(int semid) {
struct sembuf p_buf;
p_buf.sem_num = 0;
p_buf.sem_op = 1;
p_buf.sem_flg = SEM_UNDO;
if (semop(semid, &p_buf, 1) == -1) {
perror("p(semid) failed");
exit(1);
}
return 1;
}

void producter() {
//while (1) {
//if (P(semid)) {
P(semid_empty);
P(semid_zuse);
struct databuf *p = NULL;
//u6d4bu8bd5uff0cu653eu5165u6570u636e
void *shm = shmat(shmid, NULL, 0);
if (shm == (void *)-1) {
printf("producter shmat failed!
");
}
printf("
producter get shm scuess!");

p = (struct databuf *)shm;
//printf("
please input something:");
strcpy(p->text, "Apple");
printf("
Get data success!");
sleep(10);
V(semid_zuse);
V(semid_full);
//}
//}
}

void filter() {
while (1) {
printf("
filter running...");
sleep(10);
}
}

void consumer() {
/*while(1){

}*/
//u6d4bu8bd5uff0cu8bfbu51fau6570u636e
while (1) {
//if (P(semid)) {
P(semid_full);
P(semid_zuse);
struct databuf *p = NULL;

void *shm = shmat(shmid, 0, 0);
if (shm == (void *)-1) {
printf("consumer shmat failed!
");
}
printf("
comsumer get shm scuess!");

p = (struct databuf *)shm;
printf("
You wirte is %s", p->text);
V(semid_zuse);
V(semid_empty);

}
}

int main() {
key_t semkey, shmkey;
struct databuf *buf;
initshm(&buf);
semid_full = initsem(116);
semid_empty = initsem(117);
semid_zuse = initsem(118);


pid_t fpid = fork();

if (fpid < 0) {
printf("error in fork!");
}
else if (fpid == 0) {
producter();
}
else {
sleep(5);
pid_t fpid2 = fork();
if (fpid2 < 0)	printf("error in fork!");
else if (fpid2 == 0)	filter();
else	consumer();
}

remobj();

}