C-exp-collection/exp1/task63.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;
}