生产者与消费者学习笔记（二）

#include "unipc.h"

#define MAXNITIEMS 1000000
#define MAXNTHREADS 100

int nitems;

/*critical region*/
struct {
    pthread_mutex_t mutex; /*mutex*/
    int buff[MAXNITIEMS]; /*shared data array*/
    int nput;
    int nval;
} shared = {
    PTHREAD_MUTEX_INITIALIZER,{0},0,0
};


int
set_concurrency(int level)
{
#ifdef  HAVE_THR_SETCONCURRENCY_PROTO
    int thr_setconcurrency(int);

    return(thr_setconcurrency(level));
#else
    return(0);
#endif
}

void
Set_concurrency(int level)
{
    if (set_concurrency(level) != 0){
        printf("set_concurrency error\n");
    }
}

void *produce(void *),*consume(void*);



void *produce(void *arg)
{
    for(;;) {
        pthread_mutex_lock(&shared.mutex);
        if(shared.nput >= nitems) {
            pthread_mutex_unlock(&shared.mutex);
            return NULL;
        }
        shared.buff[shared.nput++] = shared.nval++;
        pthread_mutex_unlock(&shared.mutex);
        (*(int*)arg)++;
    }
    return NULL;
}

void consume_wait(int i)
{
    for(;;) {
        pthread_mutex_lock(&shared.mutex);
        if(i < shared.nput) {
            pthread_mutex_unlock(&shared.mutex);
            return;
        }
        pthread_mutex_unlock(&shared.mutex);
    }
}

void *consume(void *arg)
{
    int i;
    for(i = 0;i < nitems;i++) {
        consume_wait(i);
        if(i != shared.buff[i]) {
            printf("i = %d,buffer[i] = %d\n",i,shared.buff[i]);
        }
    }
    return NULL;
}
int min(int a, int b)
{
    if(a < b) {
        return a;
    }else {
        return b;
    }
}
int main(int argc, char *argv[])
{
    int i, nthreads,count[MAXNTHREADS];
    pthread_t tid_produce[MAXNTHREADS],tid_consume;


    if(argc != 3) {
        printf("usage: produces2 <#item> <#threads>\n");
        return -1;
    }

    nitems = min(atoi(argv[1]),MAXNITIEMS);
    nthreads = min(atoi(argv[2]),MAXNTHREADS);

    //Set_concurrency(nthreads+1);

    for(i = 0; i < nthreads; i++)
    {
        count[i] = 0;
        pthread_create(&tid_produce[i],NULL,produce,&count[i]);
    }

    pthread_create(&tid_consume,NULL,consume,NULL);

    for(i = 0; i < nthreads; i++) {
        pthread_join(tid_produce[i],NULL);
        printf("thread %zu, count=%d\n",tid_produce[i],count[i]);
    }


    pthread_join(tid_consume,NULL);

    exit(0);
}