/*
 *Produce/Consume using multithread
 */
 
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>
#include <assert.h>    //断言，在条件错误时执行
#include <time.h>
#include <unistd.h>
#include <errno.h>

#define BSIZE    30
#define PRODUCE_EXIT    '~'            //produce thread exit flag
#define CONSUME_EXIT    '@'            //consume thread exit flag
#define PRODUCE_NUMBER     3            //produce thread number
#define CONSUME_NUMBER    5            //consume thread number

typedef struct buffer_t{
    char     buf[BSIZE];
    int     occupied;            //product number
    int     nextin;
    int     nextout;
    pthread_mutex_t     mutex;
    pthread_cond_t        less;
    pthread_cond_t    more;
}buffer_t;

pthread_once_t once = PTHREAD_ONCE_INIT;
buffer_t ware;

/*initialization*/
void init()
{
    memset(&ware, 0, sizeof(ware));
    pthread_mutex_init(&ware.mutex, NULL);
    pthread_cond_init(&ware.less, NULL);
    pthread_cond_init(&ware.more, NULL);
}

void destroy(void)
{
    pthread_mutex_destroy(&ware.mutex);
    pthread_cond_destroy(&ware.less);
    pthread_cond_destroy(&ware.more);
}

/*the produce thread*/
void *threadProduce(void *arg)
{
    char c = 0;
    struct timespec tt;
    int result;
   
    while(c != PRODUCE_EXIT)
    {
        //read product from keyboard
        c = getchar();
        if(c == '\n')
            continue;
        //lock_mutex
        assert(pthread_mutex_lock(&ware.mutex) == 0);
        //condition wait
        tt.tv_sec = time(NULL) + 1;
        tt.tv_nsec = 0;
        if(ware.occupied >= BSIZE){
            result = pthread_cond_timedwait(&ware.more, &ware.mutex, &tt);
            if(result = ETIMEDOUT){
                assert( pthread_mutex_unlock(&ware.mutex) == 0);
                continue;
            }else if(result != 0){
                printf("pthread_cond_timedwait error!\n");
                assert(pthread_mutex_unlock(&ware.mutex) == 0);
                pthread_exit(NULL);
            }
        }
        //insert produce into buf
        ware.buf[ware.nextin] = c;
        ware.occupied++;
        ware.nextin = (ware.nextin + 1)%BSIZE;
        printf("produce [tid=%u] [%c].\n", pthread_self(), c);
        //aware the consume thread
        pthread_cond_signal(&ware.less);
        //unlock mutex
        assert(pthread_mutex_unlock(&ware.mutex) == 0);
        sleep(1);
    }
    printf("produce [tid=%u] EXIT.\n", pthread_self());
    pthread_exit(NULL);
}

//consume thread
void *threadConsume(void *arg)
{
    char c = 0;
    struct timespec tt;
    int result;
   
    while(c != CONSUME_EXIT){
        assert(pthread_mutex_lock(&ware.mutex) == 0);
        tt.tv_sec = time(NULL) + 1;
        tt.tv_nsec = 0;
        if(ware.occupied <= 0){
            result = pthread_cond_timedwait(&ware.less, &ware.mutex, &tt);
            if(result ==ETIMEDOUT){
                assert(pthread_mutex_unlock(&ware.mutex) == 0);
                continue;
            }else if(result != 0){
                printf("pthread_cond_timedwait error!\n");
                assert(pthread_mutex_unlock(&ware.mutex) == 0);
                pthread_exit(NULL);
            }
        }
            c = ware.buf[ware.nextout];
            ware.occupied--;
            ware.nextout = (ware.nextout + 1)%BSIZE;
            printf("comsue [tid=%u] [%c].\n", pthread_self(), c);
            pthread_cond_signal(&ware.more);
            assert(pthread_mutex_unlock(&ware.mutex) == 0);
            sleep(1);
    }
        printf("consume [tid=%u] EXIT.\n", pthread_self());
        pthread_exit(NULL);
}

int main()
{
    pthread_t tid[PRODUCE_NUMBER+CONSUME_NUMBER];
    int i;
    pthread_once(&once, init);
    for(i=0; i < PRODUCE_NUMBER; i++){
        if(pthread_create(&tid[i], NULL, threadProduce, NULL)){
            perror("pthread_create error.\n");
            destroy();
            exit(-1);
        }
    }
    for(; i < PRODUCE_NUMBER+CONSUME_NUMBER; i++){
        if(pthread_create(&tid[i], NULL, threadConsume, NULL)){
            perror("pthread_create error.\n");
            destroy();
            exit(-1);
        }
    }
     for(i=0; i<PRODUCE_NUMBER+CONSUME_NUMBER; i++)
    {
        pthread_join(tid[i], NULL);
    }
    destroy();
   
    pthread_exit(NULL);
}

自顶