四：线程 条件变量 pthread_cond_wait （九）

条件变量 函数：

初始化    ：pthread_cond_init
阻塞等待：  pthread_cond_wait
通知等待的第一个线程： pthread_cond_signal
指定时间内阻塞等待: pthread_cond_timewait
通知所有等待的线程：  pthread_cond_broadcast
销毁：       pthread_cond_destroy

分析 pthread_cond_wait 和 pthread_cond_signal  实现

    等待方式，都必须和一个互斥锁配合，以防止多个线程同时请求pthread_cond_wait()的竞争条件（Race Condition）。mutex互斥锁必须是普通锁（PTHREAD_MUTEX_TIMED_NP）或者适应锁（PTHREAD_MUTEX_ADAPTIVE_NP），且在调用pthread_cond_wait()前必须由本线程加锁（pthread_mutex_lock()），而在更新条件等待队列以前，mutex保持锁定状态，并在线程挂起进入等待前解锁。在条件满足从而离开pthread_cond_wait()之前，mutex将被重新加锁，以与进入pthread_cond_wait()前的加锁动作对应。 
  
  激发条件有两种形式，pthread_cond_signal()激活一个等待该条件的线程，存在多个等待线程时按入队顺序激活其中一个；而pthread_cond_broadcast()则激活所有等待线程。


实例：生产者 消费者 问题

    整个存储空间为 2 ，在任意时刻，最多只能有2个产品存放在临时空间。如果已经有2个产品存放在临时空间，将阻塞生产线程。同理，如果临时空间没有产品，显示需要阻塞消费线程。
    实现 生产 和 消费 的线程的同步

#include <stdio.h>    //printf
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#define BUFFER_SIZE 2

struct product_cons
{
    int buffer[BUFFER_SIZE]; //生产产品值
    pthread_mutex_t lock;    //互斥锁 volatile int
    int readpos, writepos;    //读写位置
    pthread_cond_t notempty;   //条件变量，非空
    pthread_cond_t notfull;    //非满
    /*  typedef struct{
        int volatile value;
            }pthread_cont_t;*/
};

void init(struct product_cons *p)
{
    pthread_mutex_init(&p->lock, NULL);  //互斥锁
    pthread_cond_init(&p->notempty, NULL);//条件变量
    pthread_cond_init(&p->notfull, NULL);  //条件变量
    p->readpos = 0;  //读写位置
    p->writepos = 0;
}

//存储 一个数据 到 bufferr
void put(struct product_cons *p, int data) //输入产品子函数
{
    pthread_mutex_lock(&p->lock);   //上锁

    /*等待，直到 buffer 不为 满*/
    while((p->writepos + 1) % BUFFER_SIZE == p->readpos)  //测试空间是否已满
    {
        printf("producer wait for not full\n");
        pthread_cond_wait(&p->notfull, &p->lock); //阻塞等待
        //这里，生产者 notfull 等待消费者 pthread_cond_signal(&p->notfull);信号
        //如果，消费者发送了 signal 信号，表示有了 空闲
    }

    p->buffer[p->writepos] = data;  //写数据
    p->writepos++;

    if(p->writepos >= BUFFER_SIZE)  //如果写到 尾部,返回
        p->writepos = 0;

    pthread_cond_signal(&p->notempty);  //发送有数据信号
    pthread_mutex_unlock(&p->lock);   //解锁

}

//读，移除 一个数据 从 buffer
int get(struct product_cons *p)
{
    int data;
    pthread_mutex_lock(&p->lock);

    /*等待，直到不为空*/
    while(p->writepos == p->readpos)
    {
        printf("consumer wait for not empty\n");
        pthread_cond_wait(&p->notempty,&p->lock);
    }

    /*读 一个 数据*/
    data = p->buffer[p->readpos];

    p->readpos++;
    if(p->readpos >= BUFFER_SIZE) //如果读到 尾
        p->readpos = 0;
    pthread_cond_signal(&p->notfull);
    pthread_mutex_unlock(&p->lock);
    return data;
}

#define OVER (-1)
struct product_cons buffer;

void *producer(void *data)   //子线程 ，生产
{
    int n;
    for(n = 1; n <=5; n++)  //生产 前5 个产品
    {
        printf("producer sleep 1 second ...\n"); //每一秒 生产 一个
        sleep(1);
        printf("put the %d product\n",n);
        put(&buffer,n);
    }
    for(n=6; n<10; n++)
    {
        printf("producer sleep 3 second...\n");
        sleep(3);
        printf("put the %d product",n);
        put(&buffer,n);
    }
    put(&buffer,OVER);
    printf("producer stopped\n");

    return NULL;

}

void *consumer(void *data)
{
    int d = 0;
    while(1)
    {
        printf("consumer sleep 2 second...\n");  //每2秒消费一个产品
        sleep(2);
        d = get(&buffer);
        printf("get the %d product\n",d);
        if(d ==OVER)
        {
            break;
        }
    }

    printf("consumer stopped\n");
    return NULL;
}

int main(int argc, char *argv[])
{
    pthread_t th_a,th_b;
    void *retval;

    init(&buffer);

    pthread_create(&th_a, NULL, producer, 0);
    pthread_create(&th_b, NULL, consumer, 0);

    pthread_join(th_a, &retval);
    pthread_join(th_b, &retval);

    return 0;
}