Linux同步技术之读写锁

Posted 2020-07-17 lm_123_456

互斥锁试图将想进入临界区的所有线程都阻塞住，但是有时候该临界区会涉及由这些线程共享的一个或多个数据的访问或更新，这时候我们就需要用到读写锁。

系统读写锁的分配规则：
（1）只要有没有线程持有给定的读写锁用于写，那么任意数量的线程可以持有该读写锁用于读。（系统规定写锁优先，但是可以更改为读锁优先）
（2）仅当没有线程持有某个读写锁用于读或用于写时，才能分配该读写锁用于写。

读写锁用于读称为共享锁，读写锁用于写称为独占锁。

读写锁的获取与释放：
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
该函数获取一个读出锁，如果对应的读写锁已由某个写入者持有，那就阻塞该调用线程。参数rwlock是读写锁变量，类型为pthread_rwlock_t。

int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);
该函数尝试获取一个读出锁或写入锁，但是如果该所不能马上获得，就返回一个EBUSY错误，但是不阻塞调用线程。

int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
该函数获取一个写入锁，如果对应的读写锁已由另一个写入者持有，或者由一个或多个读出者持有，就阻塞该调用线程。

int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
该函数释放一个读出锁或写入锁。

如果一个读写锁是静态分配的，即就是static pthread_rwlock_t rwlock，则其就初始化为PTHREAD_RWLOCK_INITIALIZER。

如果读写锁是动态分配的，那就需要调用如下函数进行初始化和释放！
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
读写锁的动态销毁函数。

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
const pthread_rwlockattr_t *restrict attr);
读写锁的动态初始化函数。第二个参数是属性，可以设置为NULL即就是使用默认属性，若不希望使用默认属性则可以通过调用系统的属性的初始化和销毁函数来设置和销毁属性。

以下是一个使用读写锁的实现读优先的例子：

//头文件
#pragma once

#include <unistd.h>
#include <stdio.h>
#include <pthread.h>
#include <errno.h>


typedef struct 
{
    pthread_mutex_t rw_mutex;
    pthread_cond_t  rw_condreaders;
    pthread_cond_t  rw_condwriters;
    int rw_magic;
    int rw_nwaitreaders;
    int rw_nwaitwriters;
    int rw_refcount;
}my_pthread_rwlock_t;

#define RW_MAGIC 0x19283746

#define MY_PTHREAD_RWLOCK_INITIALIZER {PTHREAD_MUTEX_INITIALIZER,\
                                   PTHREAD_COND_INITIALIZER,                                   PTHREAD_COND_INITIALIZER,                                   RW_MAGIC, 0, 0, 0}

typedef int my_pthread_rwlockattr_t;


int my_pthread_rwlock_init(my_pthread_rwlock_t *, my_pthread_rwlockattr_t *);
int my_pthread_rwlock_rdlock(my_pthread_rwlock_t *);
int my_pthread_rwlock_tryrdlock(my_pthread_rwlock_t *);
int my_pthread_rwlock_wrlock(my_pthread_rwlock_t *);
int my_pthread_rwlock_trywrlock(my_pthread_rwlock_t *);
int my_pthread_rwlock_unlock(my_pthread_rwlock_t *);
int my_pthread_rwlock_destroy(my_pthread_rwlock_t *);

void clean_up(void *arg);

//读写锁的初始化和销毁函数
#include "pthread_rwlock.h"

//读写锁的初始化
int my_pthread_rwlock_init(my_pthread_rwlock_t *rw, my_pthread_rwlockattr_t *attr)
{
    if(attr != NULL){    //属性只有设置了NULL，若不为NULL则返回错误
        return(EINVAL);
    }

    int result;
    if((result = pthread_mutex_init(&rw->rw_mutex, NULL)) != 0){   //互斥锁的初始化，成功返回0，失败返回非0
        return result;
    }
    if((result = pthread_cond_init(&rw->rw_condreaders, NULL)) != 0){  //初始化读条件变量，成功返回0，失败时销毁互斥量
        pthread_mutex_destroy(&rw->rw_mutex);
    }

    if((result = pthread_cond_init(&rw->rw_condwriters, NULL)) != 0){   //初始化写条件变量，成功返回0，失败时销毁互斥量和读条件变量
        pthread_cond_destroy(&rw->rw_condreaders);
        pthread_mutex_destroy(&rw->rw_mutex);
    }

    rw->rw_nwaitreaders = 0;   //初始化等待读锁的线程数
    rw->rw_nwaitwriters = 0;   //初始化等待写锁的线程数
    rw->rw_refcount = 0;       //初始化持有锁的线程数
    rw->rw_magic = RW_MAGIC;   //初始化魔法数（魔法数可有可无，该数只是用来作判断）


    return 0;
}

//读写锁的销毁
int my_pthread_rwlock_destroy(my_pthread_rwlock_t *rw)
{
    if(rw->rw_magic != RW_MAGIC){
        return (EINVAL);
    }

    if(rw->rw_refcount != 0 || rw->rw_nwaitreaders != 0 || rw->rw_nwaitwriters != 0){   //若持有锁的线程数不为0，或等待读锁的线程数不为0，或等待写锁的线程数不为0，就返回BUSY;
        return (EBUSY);
    }

    pthread_mutex_destroy(&rw->rw_mutex);   //销毁互斥锁
    pthread_cond_destroy(&rw->rw_condreaders);  //销毁读条件变量
    pthread_cond_destroy(&rw->rw_condwriters);  //销毁写条件变量

    rw->rw_magic = 0;   //魔法数赋0
    return 0;
}

//系统默认的写锁优先实现
int my_pthread_rint my_pthread_rwlock_rdlock(my_pthread_rwlock_t *rw)
{
    int result;

    if(rw->rw_magic != RW_MAGIC){
        return (EINVAL);
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){    //mutex_lock 保护临界区域
        return result;
    }

    while(rw->rw_refcount < 0 || rw->rw_nwaitwriters > 0){   //（写优先）有线程持有锁，或是写等待的线程数大于0，即有写线程在等待锁时，读线程数增加
        rw->rw_nwaitreaders++;
        result = pthread_cond_wait(&rw->rw_condreaders, &rw->rw_mutex);    //等待读的线程被阻塞，当某时收到信号时，等待读的线程被唤醒， 等待读的线程数减少
        rw->rw_nwaitreaders--;

        if(result != 0){
            break;
        }
    } 

    if(result == 0){
        rw->rw_refcount++;     //持有锁的读线程数增加，读锁是共享锁，同时可有多个线程持有读锁
    }

    pthread_mutex_unlock(&rw->rw_mutex);   

    return result;
}


int my_pthread_rwlock_wrlock(my_pthread_rwlock_t *rw)
{
    int result;

    if(rw->rw_magic != RW_MAGIC){
        return EINVAL;
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){
        return result;
    }

    while(rw->rw_refcount != 0){  //说明有线程正在持有锁，所以等待加写锁的线程数增加
        rw->rw_nwaitwriters++;
        result = pthread_cond_wait(&rw->rw_condwriters, &rw->rw_mutex);
        rw->rw_nwaitwriters--;

        if(result != 0){
            break;
        }
    }

    if(result == 0){   //正常退出循环或者未进入循环
        rw->rw_refcount = -1;  //有一个线程持有写锁，写锁是独占锁，每次只能有一个线程持有写锁
    }

    pthread_mutex_unlock(&rw->rw_mutex);
    return result;
}
#endif



int my_pthread_rwlock_unlock(my_pthread_rwlock_t *rw)
{
    int result;
    if(rw->rw_magic != RW_MAGIC){
        return EINVAL;
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){
        return result;
    }

    if(rw->rw_refcount > 0){   //说明有线程持有读锁，解锁是持有读锁的线程数减少
        rw->rw_refcount--;
    }else if(rw->rw_refcount == -1){   //说明有一个线程持有写锁，解锁完成后持有锁的线程数为0，因为每次只会有一个线程持有写锁
        rw->rw_refcount = 0;
    }else{    //没有线程持有锁，解锁失败
        printf("unlock error.\n");
    }


    //处理等待锁的线程
    if(rw->rw_nwaitwriters > 0){   //先处理等待写的线程，因为是写锁优先
        if(rw->rw_refcount == 0){   //若当前没有线程持有锁，则给等待写锁的线程发送一个信号，单播信号，每次只能有一个线程持有写锁
            result = pthread_cond_signal(&rw->rw_condwriters);
        }
    }else if(rw->rw_nwaitreaders > 0){   //再处理等待读的线程，
        result = pthread_cond_broadcast(&rw->rw_condreaders);  //读锁是共享锁，所以以广播的方式给等待读的线程发送信号，唤醒所有等待读的线程
    }

    pthread_mutex_unlock(&rw->rw_mutex);

    return result;
}wlock_rdlock(my_pthread_rwlock_t *rw)
{
    int result;

    if(rw->rw_magic != RW_MAGIC){
        return (EINVAL);
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){    //mutex_lock 保护临界区域
        return result;
    }

    while(rw->rw_refcount < 0 || rw->rw_nwaitwriters > 0){   //（写优先）有线程持有锁，或是写等待的线程数大于0，即有写线程在等待锁时，读线程数增加
        rw->rw_nwaitreaders++;
        result = pthread_cond_wait(&rw->rw_condreaders, &rw->rw_mutex);    //等待读的线程被阻塞，当某时收到信号时，等待读的线程被唤醒， 等待读的线程数减少
        rw->rw_nwaitreaders--;

        if(result != 0){
            break;
        }
    } 

    if(result == 0){
        rw->rw_refcount++;     //持有锁的读线程数增加，读锁是共享锁，同时可有多个线程持有读锁
    }

    pthread_mutex_unlock(&rw->rw_mutex);   

    return result;
}


int my_pthread_rwlock_wrlock(my_pthread_rwlock_t *rw)
{
    int result;

    if(rw->rw_magic != RW_MAGIC){
        return EINVAL;
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){
        return result;
    }

    while(rw->rw_refcount != 0){  //说明有线程正在持有锁，所以等待加写锁的线程数增加
        rw->rw_nwaitwriters++;
        result = pthread_cond_wait(&rw->rw_condwriters, &rw->rw_mutex);
        rw->rw_nwaitwriters--;

        if(result != 0){
            break;
        }
    }

    if(result == 0){   //正常退出循环或者未进入循环
        rw->rw_refcount = -1;  //有一个线程持有写锁，写锁是独占锁，每次只能有一个线程持有写锁
    }

    pthread_mutex_unlock(&rw->rw_mutex);
    return result;
}
#endif

int my_pthread_rwlock_unlock(my_pthread_rwlock_t *rw)
{
    int result;
    if(rw->rw_magic != RW_MAGIC){
        return EINVAL;
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){
        return result;
    }

    if(rw->rw_refcount > 0){   //说明有线程持有读锁，解锁是持有读锁的线程数减少
        rw->rw_refcount--;
    }else if(rw->rw_refcount == -1){   //说明有一个线程持有写锁，解锁完成后持有锁的线程数为0，因为每次只会有一个线程持有写锁
        rw->rw_refcount = 0;
    }else{    //没有线程持有锁，解锁失败
        printf("unlock error.\n");
    }


    //处理等待锁的线程
    if(rw->rw_nwaitwriters > 0){   //先处理等待写的线程，因为是写锁优先
        if(rw->rw_refcount == 0){   //若当前没有线程持有锁，则给等待写锁的线程发送一个信号，单播信号，每次只能有一个线程持有写锁
            result = pthread_cond_signal(&rw->rw_condwriters);
        }
    }else if(rw->rw_nwaitreaders > 0){   //再处理等待读的线程，
        result = pthread_cond_broadcast(&rw->rw_condreaders);  //读锁是共享锁，所以以广播的方式给等待读的线程发送信号，唤醒所有等待读的线程
    }

    pthread_mutex_unlock(&rw->rw_mutex);

    return result;
}

//自己设置的读锁优先的实现
int my_pthread_rwlock_unlock(my_pthread_rwlock_t *rw)  //解锁过程
{
    int result;
    if(rw->rw_magic != RW_MAGIC){
        return EINVAL;
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){   //互斥锁加锁与解锁之间形成临界区
        return result;
    }

    if(rw->rw_refcount > 0){  //说明持有读锁的线程数 > 0
        rw->rw_refcount--;   //将读锁的数量减一
    }else if(rw->rw_refcount == -1){  //说明有一个线程持有读锁
        rw->rw_refcount = 0;  //解锁之后持有锁的线程数为0
    }else{  //没有线程持有锁，解锁失败
        printf("unlock error.\n");
    }
    //处理等待加读写锁的线程
    if(rw->rw_nwaitreaders > 0){   //若等待加读锁的线程数>0,就广播发送信号（读锁是共享锁即每次可有多个线程加上读锁），使所有等待加读锁的线程全部加锁
        result = pthread_cond_broadcast(&rw->rw_condreaders);
    }else if(rw->rw_nwaitwriters > 0){   //若等到加写锁的线程>0，并且持有锁的线程的个数为0，就单播发送信号，使得等待加写锁的线程中的一个线程被唤醒
        if(rw->rw_refcount == 0){
            result = pthread_cond_signal(&rw->rw_condwriters);
        }
    }

    pthread_mutex_unlock(&rw->rw_mutex);  //解锁互斥量

    return result;
}

int my_pthread_rwlock_rdlock(my_pthread_rwlock_t *rw)   //读锁实现
{
    int result;

    if(rw->rw_magic != RW_MAGIC){
        return (EINVAL);
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){  //加锁互斥量与解锁互斥量形成临界区
        return result;
    }
    //读锁优先的实现
    while(rw->rw_refcount != 0){    //说明有线程持有锁
        rw->rw_nwaitreaders++;   //所以等待加读锁的线程数增加
        pthread_cleanup_push(clean_up, NULL);
        result = pthread_cond_wait(&rw->rw_condreaders, &rw->rw_mutex);  //等待加读锁的线程被阻塞，当某时收到加读锁信号的时候，等待加读锁的线程数减少
        pthread_cleanup_pop(0);
        rw->rw_nwaitreaders--;

        if(result != 0){
            break;
        }
    }


    if(result == 0){  //正常退出循环或者未进入while循环
        rw->rw_refcount++;    //所有等待读锁的线程加锁（读锁是共享锁，同时可有多个线程加锁成功）
    }

    pthread_mutex_unlock(&rw->rw_mutex);   //解锁互斥量                  
    return result;
}

//读锁优先下的写锁实现
int my_pthread_rwlock_wrlock(my_pthread_rwlock_t *rw)
{
    int result;

    if(rw->rw_magic != RW_MAGIC){
        return EINVAL;
    }

    if((result = pthread_mutex_lock(&rw->rw_mutex)) != 0){
        return result;
    }

    while(rw->rw_refcount > 0 || rw->rw_nwaitreaders > 0){   //当持有锁的线程数>0并且等待加读锁的线程数>0时，等待写锁的线程数加一
        rw->rw_nwaitwriters++;
        pthread_cleanup_push(clean_up, NULL);
        result = pthread_cond_wait(&rw->rw_condwriters, &rw->rw_mutex);  //等待加写锁的线程被阻塞，当某时收到可以加写锁的信号的时候，等待加写锁的一个线程被唤醒
        pthread_cleanup_pop(0);
        rw->rw_nwaitwriters--;

        if(result != 0){
            break;
        }
    } 

    if(result == 0){   //while循环正常退出或者未进入循环
        rw->rw_refcount = -1;   //一个线程持有写锁（写锁是独占锁，同一时刻只有一个线程持有写锁）
    }

    pthread_mutex_unlock(&rw->rw_mutex);   //解锁互斥量
    return result;
}