线程同步——条件变量

/*
 * 线程同步——条件变量
 * 生产者——消费者
 * Lzy 2011-6-19
 */

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

int x = 0;                //定义全局变量
pthread_mutex_t    mutex = PTHREAD_MUTEX_INITIALIZER;    //静态初始化互斥量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;    //静态初始化条件变量

void *producer(void *arg)
{
    while(1)
    {
        pthread_mutex_lock(&mutex);    //对互斥量上锁

        while(5-x > 0)
        {
            x++;
            printf("Producing: %d\n",x);
            sleep(1);
        }
        if(x == 5)
            pthread_cond_signal(&cond);        //激活等待的线程

        pthread_mutex_unlock(&mutex);    //对互斥量解锁
        sleep(1);
    }

    pthread_exit(NULL);
}

void *consumer(void *arg)
{
    while(1)
    {
        pthread_mutex_lock(&mutex);    //对互斥量上锁
        pthread_cond_wait(&cond, &mutex);    //条件等待

        while(x > 0)
        {
            x--;
            printf("Consumer: %d\n",x);
            sleep(1);
        }

        pthread_mutex_unlock(&mutex);    //对互斥量解锁
        sleep(1);
    }

    pthread_exit(NULL);
}

int main(void)
{
    pthread_t tidpro, tidcon;

    pthread_create(&tidpro, NULL, producer, NULL);        //创建生产者线程
    pthread_create(&tidcon, NULL, consumer, NULL);        //创建消费者线程

    pthread_exit(NULL);        //等待线程退出

    return 0;
}

/*
 * 生产者——消费者模型
 * 使用多线程实现生产者与消费者模型，产品缓冲区使用循环队列
 * Lzy 2011-6-19
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>
#include <errno.h>

#define PRODUCE_EXIT    '~'    //produce thread exit flag
#define CONSUME_EXIT    '@'    //consume thread exit flag
#define BSIZE             20
#define PRODUCER_NUM    3        //定义3个生产者
#define CONSUMER_NUM    3        //定义5个消费者

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;

buffer_t ware;
pthread_once_t once = PTHREAD_ONCE_INIT;

void init(void)
{
    bzero(&ware, sizeof(ware));                //清0
    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);        //注销生产者条件变量
}

void *producer(void *arg)            //生产者线程运行函数
{
    char c;
    struct timespec tt;

    while(1)
    {
        if((c = getchar()) != '\n')
        {
            if(c == PRODUCE_EXIT)
                break;                //结束循环

            pthread_mutex_lock(&ware.mutex);    //互斥锁上锁

            if(ware.occupied >= BSIZE)            //空间已满
            {
                tt.tv_sec = time(NULL) + 1;        //等待时间为3秒
                tt.tv_nsec = 0;

                if(pthread_cond_timedwait(&ware.more, &ware.mutex, &tt) == ETIMEDOUT)    //阻塞
                {
                    pthread_mutex_unlock(&ware.mutex);    //互斥锁解锁
                    continue;
                }
            }

            ware.buf[ware.nextin] = c;    //写入字符
            ware.nextin = (ware.nextin + 1) % BSIZE;
            ware.occupied++;            //总数量加一

            printf("produce: [tid=%u] [%c].\n",pthread_self(), c);

            pthread_cond_signal(&ware.less);    //激活消费者线程
            pthread_mutex_unlock(&ware.mutex);    //互斥锁解锁
        }
        sleep(1);
    }

    printf("produce [tid=%u] EXIT.\n", pthread_self());
    pthread_exit(NULL);
}

void *consumer(void *arg)            //消费者线程运行函数
{
    char c;
    struct timespec tt;

    while(1)
    {
        pthread_mutex_lock(&ware.mutex);    //互斥锁上锁

        if(ware.occupied <= 0)            //空间已空
        {
            tt.tv_sec = time(NULL) + 1;        //等待时间为3秒
            tt.tv_nsec = 0;

            if(pthread_cond_timedwait(&ware.less, &ware.mutex, &tt) == ETIMEDOUT)    //阻塞
            {
                pthread_mutex_unlock(&ware.mutex);    //互斥锁解锁
                continue;
            }
        }

        c = ware.buf[ware.nextout];    //输出字符
        ware.nextout = (ware.nextout + 1) % BSIZE;
        ware.occupied--;            //总数量减一

        printf("consumer: [tid=%u] [%c].\n",pthread_self(), c);

        pthread_cond_signal(&ware.more);    //激活生产者线程
        pthread_mutex_unlock(&ware.mutex);    //互斥锁解锁

        if(c == CONSUME_EXIT)
            break;

        sleep(1);
    }

    printf("consume [tid=%u] EXIT.\n", pthread_self());
    pthread_exit(NULL);
}

int main(void)
{
    int i;

    pthread_t tid[PRODUCER_NUM + CONSUMER_NUM];        //定义线程描述符
    pthread_once(&once, init);                        //控制初始化例程只被执行一次

    for(i = 0; i < PRODUCER_NUM; i++)
        pthread_create(&tid[i], NULL, producer, NULL);        //创建生产者线程

    for(i = 0; i < CONSUMER_NUM; i++)
        pthread_create(&tid[i], NULL, consumer, NULL);        //创建消费者线程

    pthread_exit(NULL);            //等待所有线程退出
    destroy();                    //注销

    return 0;
}