192 lines
4.4 KiB
C
192 lines
4.4 KiB
C
/*
|
||
编写一个多线程程序task63.c，创建k个生产者线程和m个消费者线程，
|
||
每个生产者线程产生若干个随机数，通过由N个单元构成的缓冲区，发送给消费者线程，
|
||
进行输出显示，产生Pthread信号量实现生产者/消费者线程间同步，并且设计一种方案对程序正确性进行验证。
|
||
提示：一种非严谨的简单验证方案是，将生产者线程产生的所有随机数相加得到一个和，消费者线程接收到的所有随机数相加得到另一个和，验证两个和是否一致来来验证程序正确性
|
||
|
||
*/
|
||
#include <pthread.h>
|
||
#include <time.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <unistd.h>
|
||
#include <semaphore.h>
|
||
|
||
#define POISON -1
|
||
|
||
typedef struct {
|
||
int *buf;
|
||
int n;
|
||
int inpos;
|
||
int outpos;
|
||
sem_t mutex;
|
||
sem_t slots;
|
||
sem_t items;
|
||
} sbuf_t;
|
||
|
||
long long produced_sum = 0;
|
||
long long consumed_sum = 0;
|
||
sem_t sum_mutex;
|
||
int total_items;
|
||
|
||
void sbuf_init(sbuf_t *sp, int n)
|
||
{
|
||
sp->buf = (int *)malloc(n*sizeof(int));
|
||
sp->n = n;
|
||
sp->inpos = 0;
|
||
sp->outpos = 0;
|
||
sem_init(&sp->mutex, 0, 1);
|
||
sem_init(&sp->slots, 0, n);
|
||
sem_init(&sp->items, 0, 0);
|
||
}
|
||
|
||
void sbuf_deinit(sbuf_t *sp)
|
||
{
|
||
free(sp->buf);
|
||
sem_destroy(&sp->mutex);
|
||
sem_destroy(&sp->slots);
|
||
sem_destroy(&sp->items);
|
||
}
|
||
|
||
void sbuf_insert(sbuf_t *sp, int item)
|
||
{
|
||
sem_wait(&sp->slots);
|
||
sem_wait(&sp->mutex);
|
||
sp->buf[sp->inpos] = item;
|
||
sp->inpos = (sp->inpos + 1) % sp->n;
|
||
sem_post(&sp->mutex);
|
||
sem_post(&sp->items);
|
||
}
|
||
|
||
int sbuf_remove(sbuf_t *sp)
|
||
{
|
||
sem_wait(&sp->items);
|
||
sem_wait(&sp->mutex);
|
||
int item = sp->buf[sp->outpos];
|
||
sp->outpos = (sp->outpos + 1) % sp->n;
|
||
sem_post(&sp->mutex);
|
||
sem_post(&sp->slots);
|
||
return item;
|
||
}
|
||
|
||
|
||
typedef struct {
|
||
sbuf_t *sp;
|
||
int num_items;
|
||
int id;
|
||
} producer_arg_t;
|
||
|
||
typedef struct
|
||
{
|
||
sbuf_t *sp;
|
||
int id;
|
||
} consumer_arg_t;
|
||
|
||
|
||
void *producer(void *arg)
|
||
{
|
||
producer_arg_t *pa = (producer_arg_t *)arg;
|
||
for (int i = 0; i < pa->num_items; i++)
|
||
{
|
||
int val = rand() % 1000;
|
||
sbuf_insert(pa->sp,val);
|
||
sem_wait(&sum_mutex);
|
||
produced_sum += val;
|
||
sem_post(&sum_mutex);
|
||
printf("[Producer %d] produced %d\n",pa->id,val);
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
void *consumer(void *arg)
|
||
{
|
||
consumer_arg_t *ca = (consumer_arg_t *)arg;
|
||
while(1){
|
||
int item = sbuf_remove(ca->sp);
|
||
if(item == POISON){
|
||
sbuf_insert(ca->sp, POISON);
|
||
break;
|
||
}
|
||
sem_wait(&sum_mutex);
|
||
consumed_sum += item;
|
||
sem_post(&sum_mutex);
|
||
printf("[Consumer %d] consumed %d\n",ca->id,item);
|
||
}
|
||
|
||
return NULL;
|
||
}
|
||
|
||
int main(int argc, char *argv[])
|
||
{
|
||
if (argc != 4)
|
||
{
|
||
fprintf(stderr,"Usage: %s <k_producers> <m_consumers> <N_buffer>\n",argv[0]);
|
||
return 1;
|
||
}
|
||
|
||
int k = atoi(argv[1]);
|
||
int m = atoi(argv[2]);
|
||
int N = atoi(argv[3]);
|
||
total_items = 0;
|
||
|
||
srand(time(NULL));
|
||
sem_init(&sum_mutex, 0, 1);
|
||
|
||
sbuf_t buf;
|
||
sbuf_init(&buf,N);
|
||
|
||
|
||
pthread_t *producers = malloc(k*sizeof(pthread_t));
|
||
pthread_t *consumers = malloc(m*sizeof(pthread_t));
|
||
producer_arg_t *pargs = malloc(k*sizeof(producer_arg_t));
|
||
consumer_arg_t *cargs = malloc(m*sizeof(consumer_arg_t));
|
||
|
||
for (int i = 0; i < k; i++)
|
||
{
|
||
pargs[i].sp = &buf;
|
||
pargs[i].num_items = rand() % 5 + 1; // 每个生产者产生1-10个随机数
|
||
pargs[i].id = i + 1;
|
||
total_items += pargs[i].num_items;
|
||
pthread_create(&producers[i],NULL,producer,&pargs[i]);
|
||
}
|
||
|
||
for (int i = 0; i < m; i++)
|
||
{
|
||
cargs[i].sp = &buf;
|
||
cargs[i].id = i + 1;
|
||
pthread_create(&consumers[i],NULL,consumer,&cargs[i]);
|
||
}
|
||
|
||
for (int i = 0; i < k; i++)
|
||
{
|
||
pthread_join(producers[i],NULL);
|
||
}
|
||
|
||
sbuf_insert(&buf, POISON);
|
||
|
||
for (int i = 0; i < m; i++)
|
||
{
|
||
pthread_join(consumers[i],NULL);
|
||
}
|
||
|
||
printf("\n=== Verifcation Results ===\n");
|
||
printf("Produced sum: %lld\n", produced_sum);
|
||
printf("Consumed sum: %lld\n", consumed_sum);
|
||
if (produced_sum == consumed_sum)
|
||
{
|
||
printf("Verification successful: produced sum matches consumed sum.\n");
|
||
}
|
||
else
|
||
{
|
||
printf("Verification failed: produced sum does not match consumed sum.\n");
|
||
}
|
||
|
||
free(producers);
|
||
free(consumers);
|
||
free(pargs);
|
||
free(cargs);
|
||
sbuf_deinit(&buf);
|
||
sem_destroy(&sum_mutex);
|
||
|
||
return 0;
|
||
} |