1. 实验内容与要求
2. 实验问题分析
生产者进程中存在3个生产者线程、一个管道线程以及一个20个slots的缓冲区。生产者线程负责向缓冲区写入一个数据，管道线程负责从缓冲区读出一个数据，当生产者将缓冲区写满时要等待管道线程将缓冲区中的数据拿出才可以继续写入。这里可以使用full-empty-mutex锁加上共享一个长度20的循环队列实现，这保证了同一时间只有一个线程在对缓冲区进行读/写操作。
在另一边，消费者进程类似地存在3个消费者线程、一个管道线程以及一个20个slots的缓冲区。其中消费者线程负责从缓冲区读出一个数据，管道线程负责向缓冲区写入一个数据，当缓冲区被写满后管道线程需要等待消费者线程读取数据后才能继续向缓冲区写入。这里同样可以使用full-empty-mutex锁加上共享一个长度20的循环队列实现。
当生产者的管道线程读取到生产者缓冲区中的数据后，会将数据传送到管道中。消费者的管道线程在获取到读写锁后，以FIFO的方式从管道中读取数据放入缓冲区。值得注意的一点是，当消费者的缓冲区被写满后，消费者的管道线程将无法获得对缓冲区的写入锁，管道也将被堵塞。
生产者线程和消费者线程生产和消费数据有一定的时间间隔，此间隔按照负指数分布且存在于读写缓冲区之前，即生产者和消费者线程在休息了一定时间后才开始争取读写锁。这点体现了多线程的作用。
3. 实验环境
操作系统环境：VMware + Ubuntu 20.04
编程语言： C语言
4.功能流程图
5.代码实现
缓冲区数据结构
使用循环队列来完成缓冲区20个slot的定义，并创建空指针用于之后的共享内存。需要注意server和client两个进程间缓冲区不共享，创建共享内存时应取不同的名字。这里以client.c中为例。
//client.c
#define BUFFER_SIZE 20//缓冲区slot数
/定义缓冲区
typedef struct buffer_struct{
int rear;//尾指针，用于写入
int front;//头指针，用于读出
int buffer[BUFFER_SIZE];//长度20的int数组
}buffer_struct;
void *ptr;//创建空指针，用于访问共享内存
int main(int argc, char *argv[])
{
//声明缓冲区对象并清零
buffer_struct bf;
memset(&bf, 0, sizeof(buffer_struct));
//......
//创建共享内存
int shm_fd = shm_open("client_buf", O_CREAT | O_RDWR, 0666);//server.c中”server_buf”
ftruncate(shm_fd, sizeof(buffer_struct));
ptr = mmap(0, sizeof(buffer_struct), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
//......
}
定义锁
定义full-empty-mutex锁结构，full表示有多少个满的缓冲区，empty表示有多少个空的缓冲区，mutex为标准的读写锁。使用sem_t对buffer上锁时需要注意锁在server.c与client.c两个线程之间不共享，命名不能一致。具体的使用在具体的线程函数中展示，下面以client.c为例展示锁的定义与初始化
#define NUM_THREADS 3//线程数
//定义锁
sem_t *full;
sem_t *empty;
sem_t *mutex;
int main(int argc, char *argv[])
{
//......
//打开具名信号量并初始化
full = sem_open("cfull", O_CREAT, 0666, 0);//server.c中为”sfull”
empty = sem_open("cempty", O_CREAT, 0666, 0);//server.c中为”sempty”
mutex = sem_open("cmutex", O_CREAT, 0666, 0);//server.c中为”smutex”
sem_init(full, 1, 0);//初始时有0个满的slot
sem_init(empty, 1, BUFFER_SIZE);//20个空位对应20个slot
sem_init(mutex, 1, 1);//读写锁，一次只能有一个线程读写
//......
//创建3个服务器和一个管道
pthread_t tid[NUM_THREADS];
pthread_t tid0;
pthread_attr_t attr[NUM_THREADS];
pthread_attr_t attr0;
for (int i = 0; i < NUM_THREADS; i++){
pthread_attr_init(&attr[i]);
pthread_create(&tid[i], &attr[i], client, &lambda_c);//生产者线程
//server.c中为pthread_create(&tid[i], &attr[i], server, &lambda_s);
}
pthread_attr_init(&attr0);
pthread_create(&tid0, &attr0, pipe_func, &lambda_c);//管道线程,server.c中改&lambda_s
for (int j = 0; j < NUM_THREADS; j++){
pthread_join(tid[j], NULL);
}
pthread_join(tid0, NULL);
return 0;
}时间函数
设x为随机变量，λ为参数，则令x符合负指数分布的概率密度函数和概率分布函数为
其中概率分布函数是概率密度函数的积分。则x有x=-ln(1-F(x;lambda))/lambda，其中可以用rand()函数生成F(x;λ)，因为其是在(0,1)上的随机分布。
//符合负指数分布的时间函数
double sleep_time(double lambda_c){
double r;
r = ((double)rand() / RAND_MAX);
while(r == 0 || r == 1){
r = ((double)rand() / RAND_MAX);
}
r = (-1 / lambda_c) * log(1-r);
return r;
}在定义了管道int pipef;和读入lambda_c或在server.c中的lambda_s后，我们进入四个主要函数的讲解。在int main中读入参数
int main(int argc, char *argv[])
{
//......
if (argc != 2){
printf("The number of supplied arguments are false.\n");
return -1;
}
if (atof(argv[1]) < 0){
printf("The lambda entered should be greater than 0.\n");
return -1;
}
double lambda_c = atof(argv[1]);//读取lambda p转化为数字
//......
}client.c中的生产者函数
/生产者函数
void *client(void *temp){
double lambda_c = *(double *)temp;
do{
double interval_time = lambda_c;
unsigned int sleepTime = (unsigned int)sleep_time(interval_time);
sleep(sleepTime);//生产数据的间隔
int item = rand()%999+1;//生产一个1-999之间的数
buffer_struct *shm_ptr = ((buffer_struct *)ptr);//指向共享的缓冲区struct
sem_wait(empty);//empty--
sem_wait(mutex);//mutex--
printf("Sleep Time: %d s | Producing the data %d to buffer[%d] by thread %ld in process %d.\n", sleepTime, item,shm_ptr->rear,gettid(), getpid());
shm_ptr->buffer[shm_ptr->rear] = item;//向缓冲区输入item
shm_ptr->rear = (shm_ptr->rear+1) % BUFFER_SIZE;//尾指针后移
sem_post(mutex);//mutex++
sem_post(full);//full++
}while(1);
pthread_exit(0);
}client.c中的管道函数
void *pipe_func(void *temp)
{
int *rec = (int *)malloc(sizeof(int));
pipef = open("./pipe_func", O_WRONLY);//以只写的方式打开管道文件
if(pipef < 0)
{
printf("open pipef error is %s\n", strerror(errno));
*rec = -1;
pthread_exit((void *)rec);
return NULL;
}
do{
sem_wait(full);//full--
sem_wait(mutex);//mutex--
buffer_struct *shm_ptr = ((buffer_struct *)ptr);//指向共享的缓冲区struct
int item = shm_ptr->buffer[shm_ptr->front];//从头指针读出
char str[5];
sprintf(str+strlen(str), "%d", item);//将item转化为string形式
write(pipef, str, sizeof(str));//向管道写入item的string
shm_ptr->front = (shm_ptr->front+1) % BUFFER_SIZE;//头指针后移
memset(str, 0, sizeof(str));//string空间清零
sem_post(mutex);//mutex++
sem_post(empty);//empty++
}while(1);
pthread_exit(0);
}server.c中的管道函数
void *pipe_func(void *temp)
{
int *rec = (int *)malloc(sizeof(int));
pipef = open("./pipe_func",O_RDONLY);//以只读的方式打开管道文件
if(pipef < 0)//判断管道文件是否存在
{
printf("open pipef error is %s\n", strerror(errno));
*rec = -1;
pthread_exit((void *)rec);
return NULL;
}
do{
int item;
char str[5];
if(read(pipef, str, sizeof(str)) > 0)//读取str。read函数在管道为空时自动拥塞，不进行读取
{
sem_wait(empty);//empty--
sem_wait(mutex);//mutex--
buffer_struct *shm_ptr = ((buffer_struct *)ptr);//指向共享的缓冲区struct
sscanf(str,"%d",&item);//将str转化为int形式的item
shm_ptr->buffer[shm_ptr->rear] = item;//将item写入缓冲区
shm_ptr->rear = (shm_ptr->rear+1) % BUFFER_SIZE;//尾指针后移
memset(str, 0, sizeof(str));//清除str中的内容
sem_post(mutex);//mutex++
sem_post(full);//full++
}
}while(1);
}server.c中的消费者函数
void *server(void *temp){
double lambda_s = *(double *)temp;
do{
double interval_time = lambda_s;
unsigned int sleepTime;
sleepTime = (unsigned int)sleep_time(interval_time);
sleep(sleepTime);//消费一段时间
buffer_struct *shm_ptr = ((buffer_struct *)ptr);//指向共享的缓冲区struct
sem_wait(full);//full--
sem_wait(mutex);//mutex--
int item = shm_ptr->buffer[shm_ptr->front];//读取缓冲区内容
printf("Sleep Time: %d s | Consuming the data %d from the buffer[%d] by the thread %ld in the process %d.\n", sleepTime, item, shm_ptr->front, gettid(), getpid());
shm_ptr->front = (shm_ptr->front+1) % BUFFER_SIZE;//头指针后移
sem_post(mutex);//mutex++
sem_post(empty);//empty++
}while(1);
}6.实验过程与结果
根据实验要求5.运行的时候，开两个窗口，一个./client lambda_c，另一个./server lambda_s，要求测试不同的参数组合，打印结果。
当λc=λs=0.3时，写入速度与读取速度相似。client先启动，server的读取与client的写入相差的数量虽有波动但较为固定。
当λc=0.5>λs=0.1时，写入速度比读取速度快。当写入较多时client的缓冲区已满，写入速度有所下降。
当λc=0.1<λs=0.5时，写入速度比读取速度慢，几乎一有写入就会被读取。
