读写信号量,读写锁,顺序锁
#ifdef CONFIG_RWSEM_PI struct rt_rw_semaphore { struct rt_mutex lock; int read_depth; #ifdef CONFIG_DEBUG_LOCK_ALLOC struct lockdep_map dep_map; #endif }; #define up_read(rwsem) PICK_FUNC_1ARG(struct rw_semaphore, struct rt_rw_semaphore, anon_up_read, rt_up_read, rwsem)
先来看读写信号量。
如果是不配置CONFIG_RWSEM_PI,则
__down_read的实现为:
if (sem->activity >= 0 && list_empty(&sem->wait_list)) { /* granted */ sem->activity++; goto out;
即原始的read是判断activity字段是否大于等于0,那么什么时候小于0呢
在__down_write中,
if (sem->activity == 0 && list_empty(&sem->wait_list)) { /* granted */ sem->activity = -1; goto out;
关于wait_list字段还需要说明一下,为什么判断时要加一句wait_list的判断?
1)考虑以下情况,如果已经有很多reader在读,突然来了一个writer,那么writer会被挂起,
从而wait_list不为空。如果这个时候再来reader,那么由于sem->activity仍然大于0,则需要判断wait_list是否为空,
从而指示当前是否有写者在等待,最后挂起读者。所以wait_list既可能有reader也可能有writer。
这是关于reader端判断wait_list的必要性。
2)那么在写者端的wait_list是否空的判断是什么情况下生效?
即某时候sem->activity为0,但sem->wait_list不为空。
也就是说,写加锁成功仅仅在没有读者,并且也没有其他写者的情况下才生效。
再来看up_read,释放读锁,即减少sem->activity计数,如果计数为0,则说明临界区所有读者都释放锁了,
如果此时wait_list还不为空,说明有写者在等,则唤醒写者,在唤醒写者的流程里,有一句
sem->activity = -1;
即强行把activity计数改为-1 然后再唤醒写者,指示着写者获取到了此锁。
为什么说up_read里唤醒的一定是写者?
首先,等待队列是先入先出的,第一个被读者阻塞的,一定是写者。
既等待队列可能类似:
WRRRRRWRWWWR...
也就是说,读者尝试唤醒时的队列,一定是写者打头的。
而对于up_write,释放写锁时,则可能既唤醒写者,也唤醒读者。
则写者唤醒的策略为,连续尽可能多的唤醒读者直到下一个写者为止。
写者释放锁,唤醒时,检查wait_list的第一个成员,如果是写者,就把锁交给他,然后唤醒一个写者;
如果是读者,则一直唤醒到下一个写者之前的所有读者。
可以看出linux内核设计的是多么巧妙。
再来看读写锁
经典的读写锁,读者优先,写者次之,核心之思想是rwlock_t结构体,有一个字段,初始化时为0x1000000,读者加锁的时候,
就尝试减去1, 只要非负就加锁成功。
写者尝试加锁的时候,就减去0x1000000,只要非负(其实就是只要为0)就加锁成功。
写:
static inline void __raw_write_lock(raw_rwlock_t *rw) { asm volatile(LOCK_PREFIX " subl %1,(%0) " //先减去0x1000000 "jz 1f " //如果等于0,说明没有读写者,直接成功 "call __write_lock_failed " "1: " ::LOCK_PTR_REG (rw), "i" (RW_LOCK_BIAS) : "memory"); }
.align .globl __write_lock_failed __write_lock_failed: " LOCK "addl $" RW_LOCK_BIAS_STR ",(%eax) //把0x1000000加回来。 1: rep; nop cmpl $" RW_LOCK_BIAS_STR ",(%eax) //lock是否等于0x1000000? jne 1b //如果不相等,说明有读者或者是有写者抓到锁了。 " LOCK "subl $" RW_LOCK_BIAS_STR ",(%eax) //相等,则尝试再次抓锁(减去0x1000000) jnz __write_lock_failed //unlikely抓锁失败,跳到__write_lock_failed继续。 ret
static inline void __raw_read_lock(raw_rwlock_t *rw) { asm volatile(LOCK_PREFIX " subl $1,(%0) " temp = rw-1 "jns 1f " if(temp>0) goto 1; "call __read_lock_failed " "1: " ::LOCK_PTR_REG (rw) : "memory"); } /* rdi: pointer to rwlock_t */ ENTRY(__read_lock_failed) CFI_STARTPROC LOCK_PREFIX incl (%rdi) rw = rw+1 //恢复之前减去的1 1: rep nop cmpl $1,(%rdi) if(rw - 1<0) flag = 1 else flag = 0 js 1b if(rw<1) goto 1 继续尝试抓锁 LOCK_PREFIX decl (%rdi) rw - 1 > 0了,这里再次尝试减rw的值,rw=rw-1 js __read_lock_failed unlikely rw < 0 ,跳到__read_lock_failed,否则likely加锁成功 ret CFI_ENDPROC END(__read_lock_failed)
可以看出,读者并不能如读写信号量一样,感知是否有写者在等待资源,写者很容易发生饥饿。
如果想把读写锁改成写者优先,就是说,当读者发现有写者被pending时,读者不去获取锁,而是等待写者获取到锁之后把pend位清空。
接下来是顺序锁。
对于读写锁来说,读者与读者可同时运行,读者与写者互斥,写者与写者互斥。
但顺序锁的话,则可以让读者与写者同时访问临界区(虽然是同时访问,但一旦发生同时访问,读者必须重读)
在顺序锁的应用里,写者之间需要用另外一把spinlock或者mutex来互斥,或者逻辑上就不要设计
写者的同步。
/* * Sequence counter only version assumes that callers are using their * own mutexing. */
读者则实际上是个轮询的过程,读者的实现思想就是,从进入临界区前,到最后退出临界区,是否
写者进入过临界区?如果有的话,就重读取一遍。
具体如何实现呢?
读者如何知道写者位于临界区? 可以用一个flag来指示,写者每次进入临界区,就将flag置1,退出时,就置0。
考虑这种情况,读者先进入临界区,获取flag=0,然后写者进入临界区,置flag=1,接着写者退出,置flag=0,最后读者退出
临界区,发现flag=0,认为是没有写者动过手脚,于是本次读取的数据都是修改前的了,很可能无效。
我们需要一个框架,来指示写者“进入过”临界区,很明显就需要一个计数。
因此改成,写者退出时,不置flag为0,而是继续加1。因此,一旦flag为奇数,那么写者一定位于临界区,
而如果读者进入临界区之前读取的值,没有读者退出临界区读的flag大,则说明写者曾经写过临界区,读者需要重新读取。
自然,读锁的实现如下:
int read_seq_begin(seq_t* lock) { if(lock->flag&1) //写者在临界区 return lock->flag-1; return lock->flag; } int read_seq_retry(seq_t* lock,int flag) { return (lock->flag!=flag) }
读锁的使用:
do{ flag = read_seq_begin(lock); /* read critical section */ }while(read_seq_retry(lock,flag);
实际情况是, read_seq_begin 应该写成 while(lock->flag&1) cpu_relax();
这样的话,就不用再进入read critical section去做无用功了。
声明:该文观点仅代表作者本人,牛骨文系教育信息发布平台,牛骨文仅提供信息存储空间服务。
- 上一篇: 原子操作与 x86 上的 lock 指令前缀
- 下一篇: 基于 Redis 的分布式锁