生产者消费者模型模版

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

void *producer(void *);
void *consumer(void *);


pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  notFull =PTHREAD_COND_INITIALIZER;
pthread_cond_t  notEmpty=PTHREAD_COND_INITIALIZER;

#define NELEMENT 1000

int full=0;
int empty=1;
int list[NELEMENT];
int ipos=0;
int stop=0;
int stop2=0;
int npro=0;
int ncon=0;
int ioffs;

int iadd=0;
int iremove=0;
int ns=3501;


unsigned char *buf;
segy tr1,tr2,tr3;
int main(int argc,char *argv[])
{

    pthread_t tid1,tid2;
    
    
    buf=(unsigned char*)malloc( (sizeof(float)*ns+240)*NELEMENT );
        
    
    pthread_create(&tid2,NULL,consumer,NULL);
    pthread_create(&tid1,NULL,producer,NULL);
    
    
    
    
    
    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);
    
    printf("npro %d ncon=%d\n",npro,ncon);
    
}

/* Simulate Read */
void *producer(void *arg)
{
    int noffs=1;
    int ioffl1=1000,ioffl2=1010;
    
    int ioffs,ioffl;
    
    char str[1024];
    FILE *fp;
    

    for (ioffs=0;ioffs<noffs;++ioffs)
    {
        for (ioffl=ioffl1;ioffl<=ioffl2;++ioffl)
        {
            sprintf(str,"/home/yc/20150708yongle/data0/off%d/line%d",ioffs,ioffl);
            fp=fopen(str,"rb");
            if (fp==NULL) continue;
            
            while(fgettr(fp,&tr1))
            {
                pthread_mutex_lock(&mutex);
                
                while(full)
                {
                    pthread_cond_wait(&notFull,&mutex);
                }
                
                empty=0;
                list[iadd]=ioffs;
                memcpy(&buf[(sizeof(float)*ns+240)*iadd],&tr1,(sizeof(float)*ns+240));
                
                iadd++;
                
                
                
                iadd=iadd%NELEMENT;
                
                if (iadd==iremove)
                    full=1;
                else
                    full=0;
                
                npro++;
                
                pthread_mutex_unlock(&mutex);
                pthread_cond_signal(&notEmpty);
            
            }
            fclose(fp);
        }
        /* poison pill 1 */
/*        pthread_mutex_lock(&mutex);
        while(full)
        {
            pthread_cond_wait(&notFull,&mutex);
        }
        
        empty=0;
        list[iadd]=-1;
        iadd++;
        iadd=iadd%NELEMENT;
        
        if (iadd==iremove)
            full=1;
        else
            full=0;
        
        pthread_mutex_unlock(&mutex);
        pthread_cond_signal(&notEmpty);
*/        
    }
    /* poison pill 2 */
    pthread_mutex_lock(&mutex);
    while(full)
    {
        pthread_cond_wait(&notFull,&mutex);
    }
    
    empty=0;
    list[iadd]=-2;
    iadd++;
    iadd=iadd%NELEMENT;
    
    if (iadd==iremove)
        full=1;
    else
        full=0;
    
    pthread_mutex_unlock(&mutex);
    pthread_cond_signal(&notEmpty);
    
}

/* Simulate Process */
void *consumer(void *arg)
{
    int first=1;
    int it;
    
    FILE *fp;
    char str[1024];
    
    int ioffs=0;
    int ioffsold=0;
    
    int flag=0;
    while(1)
    {
        while (1)
        {
            pthread_mutex_lock(&mutex);
            
            while(empty)
            {
                pthread_cond_wait(&notEmpty,&mutex);
            }
            
            full=0;
            
            flag=0;
            /* poison pill */
            if (list[iremove]==-1)
            {
                flag=1;
            }
            
            list[iremove]=0;
            memcpy(&tr2,&buf[(sizeof(float)*ns+240)*iremove],(sizeof(float)*ns+240));
            
            iremove++;
            iremove=iremove%NELEMENT;
            
            if (iremove==iadd)
                empty=1;
            else
                empty=0;
            if (flag==0)
            ncon++;
            
            
            pthread_mutex_unlock(&mutex);
            pthread_cond_signal(&notFull);
            
            if (flag==1)
            {
                break;
            }
        }
        
        tr3.ns=ns;
        tr3.dt=1000;
        sprintf(str,"off%d.su",ioffsold);
        ioffs++;
        fp=fopen(str,"wb");
        
        fputtr(fp,&tr3);
        
        fclose(fp);
        
    }
}