线程的同步与互斥

生产者消费者问题

编译指令

gcc -o pc pc.c -lpthread

#include "csapp.h"

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

pthread_cond_t cond = PTHREAD_COND_INITIALIZER; //条件变量
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;//互斥量

typedef struct node//链表结点
{
    int data;
    struct node* next;
}node, *node_p, **node_pp;

node_p getnode(int data)//创建结点
{
    node_p ret = malloc(sizeof(node));
    if(ret != NULL)
    {
        ret->data = data;
        ret->next = NULL;
        return ret;
    }
    return NULL;
}

void initlist(node_pp pphead)//初始化链表——创建头结点
{
    if(pphead == NULL)
        exit(1);
    *pphead = malloc(sizeof(node));
    (*pphead)->next = NULL;
}

int isEmpty(node_p phead)//链表判空，空返回1
{
    return (phead->next == NULL)?1:0;
}

void push(node_p phead, int data)//链表的头插
{
    node_p tmp = getnode(data);
    if(tmp != NULL)
    {
        tmp->next = phead->next;
        phead->next = tmp;
    }
}

void print(node_p phead)//打印链表结点的数据
{
    phead = phead->next;
    while(phead)
    {
        printf("%d  ", phead->data);
        phead = phead->next;
    }
}

void erase(node_p phead, int *x)//删除链表的数据
{

    if(isEmpty(phead))
        return;
    node_p cur = phead->next;
    phead->next = cur->next;
    *x = cur->data;
    free(cur);
}

void destroy(node_p phead)//销毁链表
{
    while(!isEmpty(phead))
    {
        int data;
        erase(phead, &data);
    }
    free(phead);
}

void *producer(void* arg)//生产者线程执行的函数
{
    node_p phead = (node_p)arg;
    while(1)
    {
        int data = rand()%1000;
        pthread_mutex_lock(&mutex);
        push(phead, data);
        printf("producer done, %d\n", data);
        pthread_mutex_unlock(&mutex);
        pthread_cond_signal(&cond);
        sleep(1);
    }
}

void* consumer(void* arg)//消费者线程执行的函数
{
    node_p phead = (node_p)arg;
    int data = 0;
    while(1)
    {
        pthread_mutex_lock(&mutex);
        while(isEmpty(phead))
        {
            printf("no product, please wait...\n");
            pthread_cond_wait(&cond, &mutex);
        }
        erase(phead, &data);
        printf("consumer done, %d\n", data);
        pthread_mutex_unlock(&mutex);
        usleep(100000);
    }
}
int main()
{
    node_p phead;
    initlist(&phead);

    srand((unsigned long)time(NULL));
    pthread_t t1, t2;
    pthread_create(&t1, NULL, producer, (void*)phead);//生产者线程
    pthread_create(&t2, NULL, consumer, (void*)phead);//消费者线程

    pthread_join(t1, NULL);//线程等待
    pthread_join(t2, NULL);

    destroy(phead);//释放链表
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    return 0;
}

读者写者问题

输入数据：

1 R 3 5
2 W 4 5
3 R 5 2
4 R 6 5
5 W 7 3

编译指令

gcc -o wr wr.c -lpthread

运行指令

./wr < data > out

大概需要运行20s。

#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <semaphore.h>
#include <stdio.h>

typedef struct {
  int tid;
  int delay;
  int last;
} Role;

enum {
  s_waiting,
  s_reading,
  s_writing
} state = s_waiting;

static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
sem_t Sig_read;
sem_t Sig_wrt;
int reader_count = 0;
int writer_count = 0;

time_t startTime;

// 读者进程
void* reader(void* argPtr) {
  Role role = *(Role*)argPtr;
  sleep(role.delay);
  printf("[%02.0lf秒]读者进程%d等待读取\n", 
    difftime(time(NULL), startTime), role.tid
  );

  pthread_mutex_lock(&mutex);
    ++reader_count;
    if(state == s_waiting 
    || (state == s_reading && writer_count == 0)) {
      sem_post(&Sig_read);
      state = s_reading;
    }
  pthread_mutex_unlock(&mutex);
  sem_wait(&Sig_read);

  printf("[%02.0lf秒]读者进程%d开始读取\n", 
    difftime(time(NULL), startTime), role.tid
  );

  // read
  sleep(role.last);

  pthread_mutex_lock(&mutex);
    --reader_count;
    if(writer_count != 0) {
      // 有读者在等待
      sem_post(&Sig_wrt);
      state = s_writing;
    } else if(reader_count == 0) {
      // 等待队列为空
      state = s_waiting;
    }
  pthread_mutex_unlock(&mutex);

  printf("[%02.0lf秒]读者进程%d读取结束\n", 
    difftime(time(NULL), startTime), role.tid
  );
  return NULL;
}

// 写者进程
void* writer(void* argPtr) {
  Role role = *(Role*)argPtr;
  sleep(role.delay);
  printf("[%02.0lf秒]写者进程%d等待写入\n", 
    difftime(time(NULL), startTime), role.tid
  );
  pthread_mutex_lock(&mutex);
  ++writer_count;
  if(state == s_waiting) {
    sem_post(&Sig_wrt);
    state = s_writing;
  }
  pthread_mutex_unlock(&mutex);

  sem_wait(&Sig_wrt);
  printf("[%02.0lf秒]写者进程%d开始写入\n", 
    difftime(time(NULL), startTime), role.tid
  );
  // write
  sleep(role.last);

  pthread_mutex_lock(&mutex);
    --writer_count;
    if(writer_count != 0) {
      sem_post(&Sig_wrt);
      state = s_writing;
    } else if(reader_count != 0) {
      for(int i=0; i!=reader_count; ++i) {
        sem_post(&Sig_read);
      }
      state = s_reading;
    } else {
      state = s_waiting;
    }
  pthread_mutex_unlock(&mutex);

  printf("[%02.0lf秒]写者进程%d写入结束\n", 
    difftime(time(NULL), startTime), role.tid
  );
  return NULL;
}

int main() {
  const int MAX_THREAD = 100;
  // 读写进程队列
  pthread_t tid[MAX_THREAD];
  Role role[MAX_THREAD];
  int tidEnd = 0;

  // 初始化信号量
  sem_init(&Sig_read, 0, 0);
  sem_init(&Sig_wrt, 0, 0);

  startTime = time(NULL);

  int arg_tid;
  int arg_delay;
  int arg_last;
  char arg_type;
  while(scanf("%d %c%d%d", &arg_tid, &arg_type, &arg_delay, &arg_last) == 4) {
    assert(tidEnd < MAX_THREAD);
    if(arg_type == 'R') {
      role[tidEnd].tid = arg_tid;
      role[tidEnd].delay = arg_delay;
      role[tidEnd].last = arg_last;
      pthread_create(tid + tidEnd, NULL, reader, role + tidEnd);
    } else {
      role[tidEnd].tid = arg_tid;
      role[tidEnd].delay = arg_delay;
      role[tidEnd].last = arg_last;
      pthread_create(tid + tidEnd, NULL, writer, role + tidEnd);
    }
    printf("[%02.0lf秒]创建进程%d\n", difftime(time(NULL), startTime), arg_tid);
    ++tidEnd;
  }

  for(int i=0; i!=tidEnd; ++i) {
    pthread_join(tid[i], NULL);
  }

  // 销毁信号量
  pthread_mutex_destroy(&mutex);
  sem_destroy(&Sig_read);
  sem_destroy(&Sig_wrt);
}