C基础 读写锁中级剖析

引言

　　读写锁 是为了 解决, 大量 ''读'' 和 少量 ''写'' 的业务而设计的.

读写锁有3个特征:

　　1.当读写锁是写加锁状态时，在这个锁被解锁之前，所有试图对这个锁加锁的线程都会被阻塞
　　2.当读写锁在读加锁状态时，再以读模式对它加锁的线程都能得到访问权，但以写模式加锁的线程将会被阻塞
　　3.当读写锁在读加锁状态时，如果有线程试图以写模式加锁，读写锁通常会阻塞随后的读模式加锁

我们先举一段标准库构建的读写锁demo来了解读写锁api 的使用 .

pthread_rwlock.c

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

#define _INT_BZ     (13)

#define _INT_WTH    (2)

#define _INT_RTH    (10)

struct rwarg {

    pthread_t id;

    pthread_rwlock_t rwlock;    // 加锁用的

    int idx;                    // 指示buf中写到那了

    char buf[BUFSIZ];           // 存储临时数据

};

// 写线程, 主要随机写字符进去

void twrite(struct rwarg * arg);

// 读线程

void treads(struct rwarg * arg);

/*

 * 主函数测试线程读写逻辑

 * 少量写线程, 大量读线程测试

 */

int main(int argc, char * argv[]) {

    // 初始化定义需要使用的量. C99以上写法, 避免跨平台不同实现的警告问题, 感谢好人随性徜徉

    struct rwarg arg = { , .rwlock = PTHREAD_RWLOCK_INITIALIZER, , "" };

    int i;

    // 读线程跑起来

    for(i=; i<_INT_RTH; ++i)

        pthread_create((pthread_t *)&arg, NULL, (void * (*)(void *))treads, &arg);

    // 写线程再跑起来

    for(i=; i<_INT_WTH; ++i)

        pthread_create((pthread_t *)&arg, NULL, (void * (*)(void *))twrite, &arg);

    // 简单等待一下

    printf("sleep input enter:");

    getchar();

    return ;

}

// 写线程, 主要随机写字符进去

void

twrite(struct rwarg * arg) {

    pthread_detach(pthread_self());

    pthread_rwlock_wrlock(&arg->rwlock);

    while(arg->idx < _INT_BZ) {

        arg->buf[arg->idx] = 'a' + arg->idx;

        ++arg->idx;

    }

    pthread_rwlock_unlock(&arg->rwlock);

}

// 读线程

void

treads(struct rwarg * arg) {

    pthread_detach(pthread_self());

    while(arg->idx < _INT_BZ) {

        pthread_rwlock_rdlock(&arg->rwlock);

        puts(arg->buf);

        pthread_rwlock_unlock(&arg->rwlock);

    }

}

编译

gcc -Wall -ggdb2 -o pthread_rwlock.out pthread_rwlock.c -lpthread

执行的结果

linux上执行没反应, 这代码放在window 同样没有反应. 主要原因是 大量读加锁阻塞了写加锁.

预估主要原因是 pthread 实现的 rwlock 读写锁, 没有粗暴支持读写锁特性3 . 这也是其读写锁一个潜在bug(写锁没有阻塞后续的读锁).

是不是有些收获, 底层API有问题也不少的. 哈哈.

这里扯一点C基础语法 中 I 对于 美得 感受与写法. 希望大家有思考.

a) 指针一般 写法

int *piyo;

// 声明部分

int *heoo(int a, int *pi);

// 定义部分

int *

heoo(int a, int b) {

   ...

}

上面是一种通用写法.  缺点在 声明和 定义 不统一, 不协调, 不爽.

b) 指针仿照OOP 写法

int* piyo;

// 声明部分

int* heoo(int a, int* pi);

//定义部分

int*

heoo(int a, int b){

    ...

}

这种写法很好理解, 也很好看. 可惜这是C, (C++也是). 因为C出现比较早, 存在缺陷. 上面致命缺点是

int* piyo, *hoge;

特别丑.

c) 请用下面写法, 都是从无数别人代码中磨出来的.

int * piyo;

// 声明部分

int * heoo(int a, int * pi);

//定义部分

int *

heoo(int a, int b){

    ...

}

到这里扯淡结束了.  编程 希望是  实用->设计好->有美感->自然 . 而不是 屎一样的实现而妄想优雅的接口. 如果有一天能为自己写代码的话.

前言

到这里我们按照上面读写锁的3条特性, 自己实现一个读写锁.  首先看数据结构

// init need 0

struct rwlock {

    int rlock;

    int wlock;

};

当我们需要使用读写锁时候只需要 struct rwlock lock = { 0 , 0 }; 很清洁

后面先在linux 使用gcc 提供的原子操作特性实现一个 读锁 实现

// 加读锁

static void rwlock_rlock(struct rwlock * lock) {

    for(;;) {

        // 看是否有人在试图读, 得到并防止代码位置优化

        while(lock->wlock)

            __sync_synchronize();

        __sync_add_and_fetch(&lock->rlock, );

        // 没有写占用, 开始读了

        if(!lock->wlock)

            break;

        // 还是有写, 删掉添加的读

        __sync_add_and_fetch(&lock->rlock, -);

    }

}

在加写锁时候, 先判断读锁是否没有人在使用了. __sync_synchronize 是为了防止进行代码位置优化. 后面逻辑是

开始加读锁, 但是在加读锁瞬间如果有写锁那么立马释放刚申请的读锁. 一切为读锁为核心设计.

对于写锁

// 加写锁

static void rwlock_wlock(struct rwlock * lock) {

    while(__sync_lock_test_and_set(&lock->wlock, ))

        ;

    // 等待读占用锁

    while(lock->rlock)

        __sync_synchronize();

}

只考虑读锁互相竞争, 竞争好了之后, 开始等待读锁. 好理解, 后面释放相对容易. 扯一点对于pthread标准库读写锁只有一个释放接口. 估计内存有状态保持. 才能保证释放.

使用方便, 但是性能不好. 读写锁实现耦合又大了. 这里实现

// 解写锁

static inline void rwlock_wunlock(struct rwlock * lock) {

    __sync_lock_release(&lock->wlock);

}

// 解读锁

static inline void rwlock_runlock(struct rwlock * lock) {

    __sync_add_and_fetch(&lock->rlock, -);

}

写锁 直接解锁到底, 读锁采用引用减少一处理. 看一个demo simple_rwlock.c

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

// init need 0

struct rwlock {

    int rlock;

    int wlock;

};

// 加读锁

static void rwlock_rlock(struct rwlock * lock) {

    for(;;) {

        // 看是否有人在试图读, 得到并防止代码位置优化

        while(lock->wlock)

            __sync_synchronize();

        __sync_add_and_fetch(&lock->rlock, );

        // 没有写占用, 开始读了

        if(!lock->wlock)

            break;

        // 还是有写, 删掉添加的读

        __sync_add_and_fetch(&lock->rlock, -);

    }

}

// 加写锁

static void rwlock_wlock(struct rwlock * lock) {

    while(__sync_lock_test_and_set(&lock->wlock, ))

        ;

    // 等待读占用锁

    while(lock->rlock)

        __sync_synchronize();

}

// 解写锁

static inline void rwlock_wunlock(struct rwlock * lock) {

    __sync_lock_release(&lock->wlock);

}

// 解读锁

static inline void rwlock_runlock(struct rwlock * lock) {

    __sync_add_and_fetch(&lock->rlock, -);

}

// ------------------------ 下面是业务代码 ----------------------------------

#define _INT_BZ     (13)

#define _INT_WTH    (2)

#define _INT_RTH    (10)

struct rwarg {

    pthread_t id;

    struct rwlock lock;    // 加锁用的

    int idx;                 // 指示buf中写道那了

    char buf[BUFSIZ];        // 存储临时数据

};

// 写线程, 主要随机写字符进去

void twrite(struct rwarg * arg);

// 读线程

void treads(struct rwarg * arg);

/*

 * 自己写读写锁底层

 */

int main(int argc, char * argv[]) {

    // 初始化定义需要使用的量

    struct rwarg arg = {  };

    int i;

    // 读线程跑起来

    for(i=; i<_INT_RTH; ++i)

        pthread_create((pthread_t *)&arg, NULL, (void * (*)(void *))treads, &arg);

    // 写线程再跑起来

    for(i=; i<_INT_WTH; ++i)

        pthread_create((pthread_t *)&arg, NULL, (void * (*)(void *))twrite, &arg);

    // 简单等待一下

    printf("sleep input enter:");

        getchar();

    return ;

}

// 写线程, 主要随机写字符进去

void

twrite(struct rwarg * arg) {

    pthread_detach(pthread_self());

    while(arg->idx < _INT_BZ) {

        rwlock_wlock(&arg->lock);

        arg->buf[arg->idx] = 'a' + arg->idx;

        ++arg->idx;

        rwlock_wunlock(&arg->lock);

    }

    puts("twrite is exit...");

}

// 读线程

void

treads(struct rwarg * arg) {

    pthread_detach(pthread_self());

    while(arg->idx < _INT_BZ) {

        rwlock_rlock(&arg->lock);

        puts(arg->buf);

        rwlock_runlock(&arg->lock);

    }

    puts("treads is exit...");

}

代码就是上面读写锁实现加上上面例子构建的. 编译命令

gcc -Wall -ggdb2 -o simple_rwlock.out simple_rwlock.out -lpthread

执行结果

得到我们想要的结果, 说明这个读写锁至少满足了业务, 比pthread 好用并且功能完整. 后面再看正文扩展成通用库.

正文

　　到这里前戏都已经快完毕, 需要到正文了.  通过上面细说, 我们能够封装一个跨平台的读写锁了. 看 scatom.c 

#ifndef _H_SIMPLEC_SCATOM

#define _H_SIMPLEC_SCATOM

/*

* 作者 : wz

*

* 描述 : 简单的原子操作,目前只考虑 VS(CL) 小端机 和 gcc

*         推荐用 posix 线程库

*/

// 如果 是 VS 编译器

#if defined(_MSC_VER)

#include <Windows.h>

//忽略 warning C4047: “==”:“void *”与“LONG”的间接级别不同

#pragma warning(disable:4047)

// v 和 a 都是 long 这样数据

#define ATOM_FETCH_ADD(v, a) InterlockedExchangeAdd((LONG volatile *)&(v), (LONG)(a))

#define ATOM_ADD_FETCH(v, a) InterlockedAdd((LONG volatile *)&(v), (LONG)(a))

#define ATOM_SET(v, a) InterlockedExchange((LONG volatile *)&(v), (LONG)(a))

#define ATOM_CMP(v, c, a) (c == InterlockedCompareExchange((LONG volatile *)&(v), (LONG)(a), (LONG)c))

/*

 *对于 InterlockedCompareExchange(v, c, a) 等价于下面

 *long tmp = v ; v == a ? v = c : ; return tmp;

 *

 *咱们的 ATOM_FETCH_CMP(v, c, a) 等价于下面

 *long tmp = v ; v == c ? v = a : ; return tmp;

 */

#define ATOM_FETCH_CMP(v, c, a) InterlockedCompareExchange((LONG volatile *)&(v), (LONG)(a), (LONG)c)

#define ATOM_LOCK(v) \

    while(ATOM_SET(v, )) \

        Sleep()

#define ATOM_UNLOCK(v) ATOM_SET(v, 0)

// 保证代码不乱序优化后执行

#define ATOM_SYNC()    MemoryBarrier()

// 否则 如果是 gcc 编译器

#elif defined(__GNUC__)

#include <unistd.h>

/*

 * type tmp = v ; v += a ; return tmp ;

 * type 可以是 8,16,32,64 bit的类型

 */

#define ATOM_FETCH_ADD(v, a) __sync_fetch_add_add(&(v), (a))

/*

 * v += a ; return v;

 */

#define ATOM_ADD_FETCH(v, a) __sync_add_and_fetch(&(v), (a))

/*

 * type tmp = v ; v = a; return tmp;

 */

#define ATOM_SET(v, a) __sync_lock_test_and_set(&(v), (a))

/*

 * bool b = v == c; b ? v=a : ; return b;

 */

#define ATOM_CMP(v, c, a) __sync_bool_compare_and_swap(&(v), (c), (a))

/*

 * type tmp = v ; v == c ? v = a : ;  return v;

 */

#define ATOM_FETCH_CMP(v, c, a) __sync_val_compare_and_swap(&(v), (c), (a))

/*

 *加锁等待,知道 ATOM_SET 返回合适的值

 *_INT_USLEEP 是操作系统等待纳秒数,可以优化,看具体操作系统

 *

 *使用方式

 * int lock = 0;

 * ATOM_LOCK(lock);

 *

 * // to do think ...

 *

 * ATOM_UNLOCK(lock);

 *

 */

#define _INT_USLEEP_LOCK (2)

#define ATOM_LOCK(v) \

    while(ATOM_SET(v, )) \

        usleep(_INT_USLEEP_LOCK)

// 对ATOM_LOCK 解锁, 当然 直接调用相当于 v = 0;

#define ATOM_UNLOCK(v) __sync_lock_release(&(v))

// 保证代码不乱序

#define ATOM_SYNC()    __sync_synchronize()

#endif // !_MSC_VER && !__GNUC__

#ifndef _STRUCT_RWLOCK

#define _STRUCT_RWLOCK

/*

 * 这里构建simple write and read lock

 * struct rwlock need zero.

 */

 // init need all is 0

struct rwlock {

    int rlock;

    int wlock;

};

// add read lock

static void rwlock_rlock(struct rwlock * lock) {

    for (;;) {

        // 看是否有人在试图读, 得到并防止代码位置优化

        while (lock->wlock)

            ATOM_SYNC();

        ATOM_ADD_FETCH(lock->rlock, );

        // 没有写占用, 开始读了

        if (!lock->wlock)

            break;

        // 还是有写, 删掉添加的读

        ATOM_ADD_FETCH(lock->rlock, -);

    }

}

// add write lock

static void rwlock_wlock(struct rwlock * lock) {

    ATOM_LOCK(lock->wlock);

    // 等待读占用锁

    while (lock->rlock)

        ATOM_SYNC();

}

// unlock write

static inline void rwlock_wunlock(struct rwlock * lock) {

    ATOM_UNLOCK(lock->wlock);

}

// unlock read

static inline void rwlock_runlock(struct rwlock * lock) {

    ATOM_ADD_FETCH(lock->rlock, -);

}

#endif // !_STRUCT_RWLOCK

#endif // !_H_SIMPLEC_SCATOM

我们是可以直接在window上测试, 对于window上线程模型同样采用 pthread for win32.

测试文件 sc_template/sc_console_template/main/test_atom_rwlock.c

#include <stdio.h>

#include <scatom.h>

#include <pthread.h>

#define _INT_BZ     (13)

#define _INT_WTH    (2)

#define _INT_RTH    (10)

struct rwarg {

    pthread_t id;

    struct rwlock lock;    // 加锁用的

    int idx;                 // 指示buf中写道那了

    char buf[BUFSIZ];        // 存储临时数据

};

// 写线程, 主要随机写字符进去

void twrite(struct rwarg * arg);

// 读线程

void treads(struct rwarg * arg);

/*

* 自己写读写锁底层

*/

int main(int argc, char * argv[]) {

    // 初始化定义需要使用的量

    int i;

    struct rwarg arg = {  };

    // 读线程跑起来

    for (i = ; i<_INT_RTH; ++i)

        pthread_create((pthread_t *)&arg, NULL, (void * (*)(void *))treads, &arg);

    // 写线程再跑起来

    for (i = ; i<_INT_WTH; ++i)

        pthread_create((pthread_t *)&arg, NULL, (void * (*)(void *))twrite, &arg);

    // 简单等待一下

    printf("sleep input enter:");

    getchar();

    return ;

}

// 写线程, 主要随机写字符进去

void

twrite(struct rwarg * arg) {

    pthread_detach(pthread_self());

    while (arg->idx < _INT_BZ) {

        rwlock_wlock(&arg->lock);

        arg->buf[arg->idx] = 'a' + arg->idx;

        ++arg->idx;

        rwlock_wunlock(&arg->lock);

    }

    puts("twrite is exit...");

}

// 读线程

void

treads(struct rwarg * arg) {

    pthread_detach(pthread_self());

    while (arg->idx < _INT_BZ) {

        rwlock_rlock(&arg->lock);

        puts(arg->buf);

        rwlock_runlock(&arg->lock);

    }

    puts("treads is exit...");

}

F7 -> Ctrl + F5 运行结果

一切正常. 后面将代码放入linux 上测试一下. 先看下面目录结构

编译命令

gcc -Wall -ggdb2 -I. -o test_atom_rwlock.out test_atom_rwlock.c -lpthread

执行结果 也是一切正常

这里 -I是为了附加指定的查找路径, -ggdb2 启用宏调试等级.  代码中存在 struct rwarg arg = { 0 };

其实使用了C初始化特性, 标注的按照标注的初始化, 未标注的直接按照零初始化.

后记

　　错误是难免欢迎吐糙~~  (●￣(ｴ)￣●)

　　烂泥　http://music.163.com/#/song?id=411314656