linux 互斥锁和条件变量

草稿（编辑中）

版本1

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

#define maxitem  100000000
#define maxthread  100
#define min(x,y) ( x>y?y:x )

int nitem;

struct {
  pthread_mutex_t mutex;
  int buf[maxitem];
  int idx;
  int val;
}shared = {
  pthread_mutex_initializer
};

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

int main(int argc, char** argv){
  int i;
  int nthreads;
  int count[maxthread];

  pthread_t tid_produce[maxthread], tid_consume;

  if(argc != 3){
    printf("arg error\n");
    return 1;
  }

  nitem = min(maxitem,atoi(argv[1]));
  nthreads = min(maxthread, atoi(argv[2]));

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

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

  pthread_create(&tid_consume, null, consume, null);
  pthread_join(tid_consume, null);

  return 0;
}

void* produce(void* arg){
  while(1){
    pthread_mutex_lock(&shared.mutex);
    if(shared.idx >= nitem){
      pthread_mutex_unlock(&shared.mutex);
      return null;
    }
    shared.buf[shared.idx] = shared.val;
    shared.idx++;
    shared.val++;
    pthread_mutex_unlock(&shared.mutex);
    *((int*)arg) +=1;
  }
}

void* consume(void* arg){
  int i;
  for(i = 0; i < nitem; ++i){
    if(shared.buf[i] != i){
      printf("buf[%d] = %d\n", i, shared.buf[i]);
    }
  }
}

版本2

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

#define maxitem  100000000
#define maxthread  100
#define min(x,y) ( x>y?y:x )

int nitem;

struct {
  pthread_mutex_t mutex;
  int buf[maxitem];
  int idx;
  int val;
}shared = {
  pthread_mutex_initializer
};

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

int main(int argc, char** argv){
  int i;
  int nthreads;
  int count[maxthread];

  pthread_t tid_produce[maxthread], tid_consume;

  if(argc != 3){
    printf("arg error\n");
    return 1;
  }

  nitem = min(maxitem,atoi(argv[1]));
  nthreads = min(maxthread, atoi(argv[2]));

  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("cout[%d] = %d\n", i, count[i]);
  }
  pthread_join(tid_consume, null);

  return 0;
}

void* produce(void* arg){
  while(1){
    pthread_mutex_lock(&shared.mutex);
    if(shared.idx >= nitem){
      pthread_mutex_unlock(&shared.mutex);
      return null;
    }
    shared.buf[shared.idx] = shared.val;
    shared.idx++;
    shared.val++;
    pthread_mutex_unlock(&shared.mutex);
    *((int*)arg) +=1;
  }
}

void consume_wait(int i){
  while(1){
    pthread_mutex_lock(&shared.mutex);
    if(i < shared.idx){
      pthread_mutex_unlock(&shared.mutex);
      return;
    }
    pthread_mutex_unlock(&shared.mutex);
  }
}

void* consume(void* arg){
  int i;
  for(i = 0; i < nitem; ++i){
    consume_wait(i);
    if(shared.buf[i] != i){
      printf("buf[%d] = %d\n", i, shared.buf[i]);
    }
  }
  return null;
}

版本2

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

#define maxitem  100000000
#define maxthread  100
#define min(x,y) ( x>y?y:x )

int nitem;
int buf[maxitem];

struct {
  pthread_mutex_t mutex;
  int idx;
  int val;
} shared = {
  pthread_mutex_initializer
};

struct {
  pthread_mutex_t mutex;
  pthread_cond_t  cond;
  int nready;
} nready = {
  pthread_mutex_initializer,
  pthread_cond_initializer
};

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

int main(int argc, char** argv){
  int i;
  int nthreads;
  int count[maxthread];

  pthread_t tid_produce[maxthread], tid_consume;

  if(argc != 3){
    printf("arg error\n");
    return 1;
  }

  nitem = min(maxitem,atoi(argv[1]));
  nthreads = min(maxthread, atoi(argv[2]));

  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("cout[%d] = %d\n", i, count[i]);
  }
  pthread_join(tid_consume, null);

  return 0;
}

void* produce(void* arg){
  while(1){
    pthread_mutex_lock(&shared.mutex);
    if(shared.idx >= nitem){
      pthread_mutex_unlock(&shared.mutex);
      return null;
    }
    buf[shared.idx] = shared.val;
    shared.idx++;
    shared.val++;
    pthread_mutex_unlock(&shared.mutex);

    pthread_mutex_lock(&nready.mutex);
    if(nready.nready == 0){
      pthread_cond_signal(&nready.cond);
    }
    nready.nready++;
    pthread_mutex_unlock(&nready.mutex);

    *((int*) arg) += 1;
  }
}

void* consume(void* arg){
  int i;
  for(i = 0; i < nitem; ++i){
    pthread_mutex_lock(&nready.mutex);
    while(nready.nready == 0){
      pthread_cond_wait(&nready.cond, &nready.mutex);
    }
    nready.nready--;
    pthread_mutex_unlock(&nready.mutex);

    if(buf[i] != i){
      printf("buf[%d] = %d\n", i, buf[i]);
    }
  }
  printf("buf[%d] = %d\n", nitem-1, buf[nitem-1]);
}