碼上快樂

相關內容    簡體    繁體

linux 互斥鎖和條件變量

本文轉載自   查看原文   2019-06-17 17:32   1361    c/ linux/ 多線程/ c/c++

為什么有條件變量？

請參看一個線程等待某種事件發生

注意：本文是linux c版本的條件變量和互斥鎖(mutex)，不是C++的。

mutex ： mutual exclusion(相互排斥)

1，互斥鎖的初始化，有以下2種方式。

  • 調用方法的初始化：互斥鎖是用malloc動態分配，或者分配在內存共享區的時候使用。
  • 不調用方法的初始化：靜態分配的時候使用。
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
const pthread_mutexattr_t *restrict attr);
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  • 返回值：成功0；失敗errno

2，互斥鎖的銷毀

int pthread_mutex_destroy(pthread_mutex_t *mutex);
  • 返回值：成功0；失敗errno
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

3，加鎖和解鎖

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
  • pthread_mutex_lock：加鎖。如果是沒有加鎖的狀態，則加鎖並返回不阻塞。如果是已經被加鎖的狀態，這阻塞在這里，並一直等待，直到解鎖。
  • pthread_mutex_trylock：嘗試去加鎖。如果是沒有加鎖的狀態，則加鎖並返回不阻塞。果是已經被加鎖的狀態，則不阻塞，立即返回，返回值為EBUSY。

4，條件變量的2個函數

int pthread_cond_wait(pthread_cond_t *restrict cond,
           pthread_mutex_t *restrict mutex);
int pthread_cond_signal(pthread_cond_t *cond);

  • pthread_cond_wait：

    • 調用此函數時點的處理：

      1，給互斥鎖解鎖。

      2，把調用此函數的線程投入睡眠，直到另外某個線程就本條件變量調用pthread_cond_signal。

    • 被喚醒后的處理：返回前重新給互斥鎖加鎖。

  • pthread_cond_signal：喚醒調用pthread_cond_wait函數的線程

條件變量通常用於生產者和消費者模式。

什么是生成者和消費者模式？

版本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：所有生產者線程和消費者線程都是並行執行的。這時會有個問題，就是消費者線程被先執行的情況下，生產者線程還沒有生產數據，這時消費者線程就只能循環給互斥鎖解鎖又上鎖。這成為輪轉（spinning)或者輪詢(polling)，是一種多CPU時間的浪費。我們也可以睡眠很短的一段時間，但是不知道睡多久。這時，條件變量就登場了。

#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;
}

版本3：所有生產者線程和消費者線程都是並行執行的。解決版本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]);
}

關於互斥鎖和條件變量的最佳實踐：

1，把要多個線程共享的數據定義和互斥鎖定義在一個結構體里。

2，把條件變量，互斥鎖，和臨界條件定義在一個結構體里。

3，在①的地方，最后不要用if，理由是，pthread_cond_wait返回后，有可能另一個消費者線程把它消費掉了，所以要再次測試相應的條件成立與否，防止發生虛假的(spurious)喚醒。各種線程都應該試圖最大限度減少這些虛假喚醒，但是仍有可能發生。

4，注意②處的代碼pthread_cond_signal，設想一下最壞的情況，調用該函數后，另外一個等待的線程立即被喚醒，所以被喚醒的pthread_cond_wait函數要立即加鎖，但是調用pthread_cond_signal函數的線程還沒有執行到③處的pthread_mutex_unlock，所以被喚醒的線程又立即終止了。所以為了避免這種情況發生，把②處的代碼pthread_cond_signal放在③處的下一行。

參考下面的偽代碼：

int flag;    
pthread_mutex_lock(&nready.mutex);
int = nready.nready == 0);
nready.nready++;
pthread_mutex_unlock(&nready.mutex);

if(flag){
  pthread_cond_signal(&nready.cond);
}

c/c++ 學習互助QQ群：877684253

本人微信：xiaoshitou5854


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 詳解linux互斥鎖 pthread_mutex和條件變量pthread_cond 再談互斥鎖與條件變量！（終於搞清楚了啊！！！！！） 信號量、互斥鎖和條件變量的區別 線程同步：何時互斥鎖不夠，還需要條件變量? linux c 線程間同步（通信）的幾種方法--互斥鎖，條件變量，信號量，讀寫鎖 C++ 並發編程之互斥鎖和條件變量的性能比較 C++ 並發編程之互斥鎖和條件變量的性能比較 node源碼詳解（七） —— 文件異步io、線程池【互斥鎖、條件變量、管道、事件對象】 Golang 鎖和條件變量 Linux 自旋鎖，互斥量（互斥鎖），讀寫鎖
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM