linux中的并发控制(2)

System linux,驱动

中断屏蔽
原子操作
自旋锁
1. 读-复制-更新
信号量
互斥体
完成量(Completion)

在上一章节讲述了程序执行及linux中存在的并发与竞态问题，同时从编译和执行的角度分析了对于并发与竞态问题的影响及消除这些影响所采取的一些技术，这章将重点讨论linux在解决并发与竞态问题所采取的途径。

中断屏蔽

CPU一般都具有开中断和关中断的功能，这项功能用在竞态问题的解决上表现为保证执行过程不被中断处理程序所打断，和避免某些竞态条件的发生。归根到底就是因为内核中实现进程调度以及单核处理器上的并发状态都是依赖于中断技术实现的，因此关闭中断就杜绝了并发状态，从而不存在竞态。

如下为使用中断屏蔽技术消除竞态的示例：

local_irq_disable(); //关中断
...
critical section        //临界区
...
local_irq_enable(); //开中断

但这对于单核CPU有效是因为在单核CPU下存在的并发是由进程调度而产生的宏观上的并发。但在SMP架构下，还存在多核之间的并发，显然中断屏蔽只能消除单核的并发状态，而不能消除多核的并发状态。因此在SMP架构下单就中断屏蔽技术而言并不能消除竞态问题，但这在许多单片系统上却是一个实现简单且有效的竞态消除的途径。

原子操作

原子操作可以保证对一个整型数据的修改是排他性的。在linux中对于原子操作相关接口的实现中依赖与CPU底层的原子操作，因此其实现与CPU的架构密切相关。

如linux中的实现，在ia64架构下，原子操作的实现依赖于fetch-and-add指令：

#define atomic_sub_return(i,v)
({                                  
    int __ia64_asr_i = (i);                 
    __ia64_atomic_const(i)                  
        ? ia64_fetch_and_add(-__ia64_asr_i, &(v)->counter)
        : ia64_atomic_sub(__ia64_asr_i, v);
})

而在arm架构下，原子操作的实现则使用ldrex和strex指令：

static inline void atomic_##op(int i, atomic_t *v)
{                   
    unsigned long tmp;                      
    int result;                     
    prefetchw(&v->counter);             
    __asm__ __volatile__("@ atomic_" #op "\n"
"1: ldrex   %0, [%3]\n"                 
"   " #asm_op " %0, %0, %4\n"               
"   strex   %1, %0, [%3]\n"             
"   teq %1, #0\n"                       
"   bne 1b"                         
    : "=&r" (result), "=&r" (tmp), "+Qo" (v->counter)
    : "r" (&v->counter), "Ir" (i)
    : "cc");
}

ldrex与strex配对使用，可以让总线监控ldrex与strex之间有无其它实体存取该地址，如有并发访问，执行strex时，第一个寄存器的值被设置为1，存储行为不成功；如无并发访问，执行strex时，定一个寄存器的值被设置为0，存储行为成功。显然这一特性能够实现多核间对同一地址的并发访问的控制。

借此，linux中实现了诸如atormic_xxx_xxx的整型原子操作和xxx_xxx_bit的位原子弹操作的接口。

自旋锁

自旋锁是一种典型的对临界资源互斥访问的手段，其名称来源于其工作方式。为获取自旋锁，CPU上需循环执行一个原子操作，该原子操作测试并设置一个内存变量，若测试成功则获取锁；如失败，则一直循环。

自旋锁使用了原子操作指令和内存屏障指令。同时自旋锁可以保证临界区不受别的CPU和本CPU内的进程所抢占，但得到锁的代码路径在执行临界区时仍可能受中断和底半部的影响。为了消除这种影响，自旋锁又通常会与关中断配合使用。

linux中关于自旋锁的API:

spinlock_t lock; //定义

spin_lock_init(&lock); //初始化

spin_lock(&lock); //获取锁，可禁止进程间并发，中断中使用
spin_trylock(&lock); //非忙等
spin_unlock(&lock); //释放锁

spin_lock_irq(&lock); //获取锁，禁止中断影响，非中断中使用
spin_unlock_irq(&lock);

spin_lock_irqsave(&lock);//获取锁，禁止中断影响，非中断中使用
spin_unlock_irqrestore(&lock);

spin_lock_bh(&lock);//获取锁，禁止下半部影响，非中断中使用
spin_unlock_bh(&lock);

针对自旋锁的特点及访存的特点，对于自旋锁进行了优化，从而产生了读写自旋锁和顺序锁。

读写自旋锁

读写自旋锁的特点是在保留自旋锁特点的情况下允许读操作的并发。

rwlock_t lock;
rwlock_init(&lock); //初始化

//读锁定
void read_lock(rwlock_t *lock);
void read_lock_irq(rwlock_t *lock);
void read_lock_irqsave(rwlock_t *lock, unsigned long flags);
void read_lock_bh(rwlock_t *lock);
//读释放
void read_unlock(rwlock_t *lock);
void read_unlock_irq(rwlock_t *lock);
void read_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
void read_unlock_bh(rwlock_t *lock);

//写锁定
void write_lock(rwlock_t *lock);
void write_lock_irq(rwlock_t *lock);
void write_lock_irqsave(rwlock_t *lock, unsigned long flags);
void write_lock_bh(rwlock_t *lock);
int write_trylock(rwlock_t *lock);

//写释放
void write_unlock(rwlock_t *lock);
void write_unlock_irq(rwlock_t *lock);
void write_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
void write_unlock_bh(rwlock_t *lock);

顺序锁

顺序锁是对读写锁的一种优化，若使用顺序锁，读执行单元不会被写执行单元阻塞，反之亦然。其互斥性仅体现在写执行单元之间。

seqlock_t lock;

//写锁定
void write_seqlock(seqlock_t *lock);
void write_seqlock_irq(seqlock_t *lock);
void write_seqlock_irqsave(seqlock_t *lock, unsigned long flags);
void write_seqlock_bh(seqlock_t *lock);
int write_tryseqlock(seqlock_t *lock);

//写释放
void write_sequnlock(seqlock_t *lock);
void write_sequnlock_irq(seqlock_t *lock);
void write_sequnlock_irqrestore(seqlock_t *lock, unsigned long flags);
void write_sequnlock_bh(seqlock_t *lock);

//读开始
unsigned read_seqbegin(const seqlock_t *lock);
unsigned read_seqbegin_irqsave(rwlock_t *lock, unsigned long flags);
int read_seqretry(const seqlock_t *lock, unsigned iv);
unsigned read_retry_irqrestore(rwlock_t *lock, unsigned long flags);

//读单元
do {
    seqnum = read_seqbegin(&seqlock);
    //读操作代码
} while(read_seqretry(&seqlock, seqnum));

读-复制-更新

RCU是一种锁机制，不同于自旋锁，使用RCU的读端没有锁、内存屏障、原子指令的开销；而写端在访问它的共享资源前，先复制一个副本，然后对副本进行修改，最后使用一个回调机制在适当的时机将指向原数据的数据指针重新指向新的被修改的数据，这个时机就是引用该数据的CPU都退出对该共享数据的读操作的时候。

RCU可以看作是读写锁的高性能版本，它不但允许多个读单元的并发，还允许多个写单元的并发。但RCU不能替代读写锁，因为在写操作比较多时，RCU的同步开销将会大大增加，于读写锁相比，反而得不偿失。

如下为RCU的接口:

//读锁定
rcu_read_lock()
rcu_read_lock_bh()

//读解锁
rcu_read_unlock()
rcu_read_unlock_bh()

//同步RCU
synchronize_rcu()

//挂接回调
void call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu));

//RCU写端示例
struct foo {
    struct list_head list;
    int a;
    int b;
    int c;
};

LIST_HEAD(head);

struct foo *gp = NULL;
...
p = kmalloc(sizeof(*p), GFP_KERNEL);
p->a = 1;
p->b = 2;
p->c = 3;
list_add_rcu(&p->list, &head);
//rcu_assign_pointer(gp, p);

//RCU写端示例
rcu_read_lock();
/* p = rcu_dereference(gp) */
list_for_each_entry_rcu(p, head, list) {
    do_something(p->a, p->b, p->c);
}
rcu_read_unlock();

信号量

信号量是操作系统中最典型的同步互斥手段，与操作系统中的经典概念PV操作对应，在linux中，信号量的定义及接口如下：

struct semaphore sem;

void sema_init(struct semaphore *sem, int val);

void down(struct semaphore *sem);//获得信号量
int down_interruptible(struct semaphore *sem);//获得信号量
int down_trylock(struct semaphore *sem);//获得信号量

void up(struct semaphore *sem);//释放信号量

互斥体

互斥体可以说是信号量的简化版本。其在linux中的定义实现为：

struct mutex m;
void mutex_init(struct mutex *m);
void mutex_lock(struct mutex *m);
int mutex_lock_interruptible(struct mutex *m);
int mutex_trylock(struct mutex *m);

void mutex_unlock(struct mutex *m);

完成量(Completion)

Linux定义了一个完成量的概念，其作用是用于一个执行单元等待另一个执行单元完成某事，类似于条件变量。其定义和接口为：

struct completion comp;

void init_completion(struct completion *comp);
void reinit_completion(struct completion *comp);

void wait_for_completion(struct completion *comp);
void complete(struct completion *comp);
void complete_all(struct completion *comp);

转载请注明来源，欢迎对文章中的引用来源进行考证，欢迎指出任何有错误或不够清晰的表达。可以在下面评论区评论，也可以邮件至 yxhlfx@163.com