7.1 并发与竞态

• 并发是指多个执行单元同时、并发的被执行，而并发的执行单元对共享资源（硬件资源、软件上的的全局变量、静态变量等）的访问则很容易导致竟态
• 竟态发生在以下几种情况
  • 对称多处理器（SMP）的多个CPU
  • 单CPU内进程与抢占它的进程
  • 中断与进程
• 解决方法是保证对共享资源的互斥访问
  • 访问共享资源的代码区域称为临界区，临界区需要以某种互斥机制保护
    • 互斥途径：中断屏蔽、原子操作、信号量、自旋锁、互斥体等

7.2 编译乱序和执行乱序

• 编译乱序是编译器的问题，而执行乱序是处理器运行时的行为
• 编译屏障：在代码中设置barrier()屏障来阻挡编译器优化
• 内存屏障指令：解决多核间一个核的内存行为对另一个核可见的问题
  • arm处理器的内存屏障指令有
    • DMB：数据内存屏障
    • DSB：数据同步屏障
    • LSB：指令同步屏障
  • linux的自旋锁、互斥量的等互斥逻辑需要用到上述指令

7.3 中断屏蔽

• 中断屏蔽的使用方法
  • local_irp_disable()与local_irp_enable()只能禁止或是能本CPU内的中断
  • local_irp_save(flags)与local_irp_restore(flags)：处理屏蔽或使能中断还能保存目前CPU的中断信息，对arm就是保存和恢复cpsr
• local_bh_disable()和loacal_bh_diable()：禁止和使能中断的底半部

local_irp_disable();
...
critical section /*临界区*/
...
local_irp_enable();

7.4 原子操作

• 原子操作可以保证对一个整型数据的修改是排他性的
• 原子操作函数分为对整型和位的原子操作

7.4.1 对整型的原子操作

• 设置原子变量的值

//设置原子变量的值为i
void atomic_set(atomic_t *v, int i);
//定义原子变量v并初始化为0
atomic_t v = ATOMIC_INIT(0);

• 获取原子变量的值

//返回原子变量的值
atomic_read(atomic_t *v);

• 原子变量加、减

//原子变量增加1
void atomic_sub(int i, atomic_t *v);

• 原子变量自增、自减

//原子变量增加1
void atomic_inc(atomic_t *v);
//原子变量减少1
void atomic_dec(atomic_t *v);

• 操作并测试，下属操作对原子变量执行自增、自检和减操作后（没有加操作）。测试其是否为0，位0返回true，否则返回false

int atomic_inc_and_test(atomic_t *v);
int atomic_dec_and_test(atomic_t *v);
int atomic_sub_and_test(int i, atomic_t *v);

• 操作并返回，丅述操作是对原子变量的加、减和自增、自减操作，并返回新的值

int atomic_add_return(int i, atomic_t *v);
int atomic_sub_return(int i, atomic_t *v);
int atomic_intc_return(atomic_t *v);
int atomic_dec_return(atomic_t *v);

7.4.2 位原子操作

• 设置位，丅述设置addr地址的底nr位

void set_bit(nr,void *addr);

• 清除位

void change_bit(nr, void *addr);

• 改变位，下述代码对addr的nr位进行反置

void change_bit(nr, void *addr);

• 测试位，返回addr的第nr位

test_bit(nr, void *addr)

• 测试并操作位

int test_and_se_bit(nr, void *addr);
int test_and_clear_bit(nr, void *addr);
int test_and_change_bit(nr, void *addr);

• 例如：使用原子变量使得文字只能被一个进程打开

static atomic_t xxx_available = ATOMIC_INIT(1);static int xxx_open(struct inode *inode, struct file *filp)
{...if(!atomic_dev_and_test(&xxx_available)){atomic_inc(&xxx_available);return -EBUSY;//已打开}...return 0;//成功
}static int xxx_release(struct inode* inode, struct file *filp){atomic_inc(&xxx_available);//释放设备return 0;}

7.5 自旋锁

7.5.1 自旋锁的使用

• 自旋锁是一种典型的对临界资源进行互斥访问的手段，其名称来源于他的工作方式
• 为了获取自旋锁，在某CPU上的运行需要先执行一个原子操作，该操作测试并设置某个内存变量。
  • 如果测试结果表明锁已经空闲，则程序成功获取自旋锁并继续执行
  • 如果测试结果表明锁仍在被使用，则程序将在一个小的循环内重复这个“测试并设置”操作，即所谓的自旋
• linux系统中对自旋锁的相关操作
  • 定义自旋锁：spinlock_t lock;
  • 初始化自旋锁：spin_lock_init(lock);
  • 获得自旋锁：spin_lock(lock);或者spin_locktry(lock);
  • 释放自旋锁：spin_unlock(lock);
• 自旋锁主要针对SMP或者单CPU单内核可抢占的情况
• 对于单CPU但是内核不支持抢占的情况，自旋锁自动转化为空操作
  • 单CPU系统和可抢占的系统中，自旋锁持有期间中内核的抢占将被禁止
  • 多核SMP的情况下，任何一个核拿到了自旋锁，该核上的抢占调度也暂时禁止，但是没有禁止另外一个核的抢占调度
• 自旋锁的临界区可以被中断或者底半部影响，因而需要自旋锁和其他操作的配合
  • spin_lock_irp() = spin_lock() + local_irq_disable()
  • spin_unlock_irp = spin_unlock() +local_irq_enable();
  • spin_lock_irqsave() = spin_lock() + local_irq_save();
  • spin_unlock_irqrestore() = spin_unlock() + local_irq_restore()
  • spin_lock_bh() = spin_lock() + local_bh_disable()
  • spin_unlock_bh() = spin_unlock() + local_bh_enable()
• 使用自旋锁应该谨慎，同时需要注意几个问题
  • 自旋锁是忙等待，在自旋的时候CPU不做其他的操作，因此只有在占用锁很短的情况下才使用自旋锁
  • 可能因为递归的使用而使系统死锁
  • 在自旋期间，不能调用调用可能引起进程调度的函数，可能导致内核崩溃
  • 在单核情况下，也该认为自己的CPU是多核的，以突出驱动的跨平台性

7.5.2 读写自旋锁

• 自旋锁不关心临界区在进行什么操作，一视同仁
• 自旋锁的衍生锁读写自旋锁允许读的并发，它保留了自选的概念，在写方面只能有一个进程，读方面可以有多个执行单元，但是读和写不能同时进行
• 定义并初始化读写自旋锁:

rwlock_t my_rwlock1 = RW_LOCK_UNLOCKED; //静态初始化
rwlock_t my_rwlock2;
rwlock_init(&my_rwlock2); //动态初始化

• 读锁定:

void read_lock(rwlock_t* lock);
void read_lock_irqsave(rwlock_t* lock, unsigned long flags);
void read_lock_irq(rwlock_t* lock);
void read_lock_bh(rwlock_t* lock);

• 读解锁:

void read_unlock(rwlock_t* lock);
void read_unlock_irqrestore(rwlock_t* lock, unsigned long flags);
void read_unlock_irq(rwlock_t* lock);
void read_unlock_bh(rwlock_t* lock);

在对共享资源进行读取之前,应该先调用读锁定函数锁定共享资源,完成之后再调用读解锁函数释放共享资源;

• 写锁定:

void write_lock(rwlock_t* lock);
void write_lock_irqsave(rwlock_t* lock, unsigned long flags);
void write_lock_irq(rwlock_t* lock);
void write_lock_bh(rwlock_t* lock);
void write_trylock(rwlock_t* lock);

• 写解锁:

void write_unlock(rwlock_t* lock);
void write_unlock_irqrestore(rwlock_t* lock);
void write_unlock_irq(rwlock_t* lock);
void write_unlock_bh(rwlock_t* lock);

 - 在对共享资源进行写操作之前,应该先调用写锁定函数锁定共享资源,完成之后再调用写解锁函数释放共享资源;与spin_trylock()一样,write_trylock()也只是尝试获得写自旋锁,不管是否成功,都会立即返回;

• 读写自旋锁使用套路:

rwlock_t lock;      //定义读写自旋锁
rwlock_init(&lock); //初始化读写自旋锁read_lock(&lock);   //读时加锁
......
//临界区操作
......
read_unlock(&lock); //读后解锁;write_lock_irqsave(&lock, flags);    //写时加锁
......
//临界区操作
......
write_lock_irqrestore(&lock, flags); //写后解锁;

7.5.3 顺序锁

• 顺序锁相关操作：

void write_seqlock(seqlock_t *s1);
void write_sequnlock(seqlock_t *s1);//宏调用，相当于：write_seqlock()+local_irq_save()
write_seqlock_irqsave(lock,flags)
write_sequnlock_irqrestore(lock,flags)//宏调用，相当于：write_seqlock()+local_irq+disable()
write_seqlock_irq(lock)
write_sequnlock_irq(lock)//宏调用，相当于：write_seqlock()+local_bh_disable()
write_seqlock_bh(lock)
write_sequnlock_bh(lock)int write_tryseqlock(seqlock_t *s1)，此函数和上面提到的类似。

• 写执行单元使用如下一种模式的顺序锁：

write_seqlock(&seqlock);
.........//写操作代码块
write_sequnlock(&seqlock);

• 读执行单元涉及如下顺序锁操作：

读开始：unsigned read_seqbegin(const seqlock_t *s1);
//读执行单元在访问共享资源时要，调用该函数，返回锁s1的顺序号
read_seqbegin_irqsave(lock,flags)
//等同于：local_irq_save()+read_seqbegin()
重读：int read_seqretry(const seqlock_t *s1,unsigned iv)
//在读结束后调用此函数来检查，是否有写执行单元对资源进行操作，若有，则重新读。iv 为锁的顺序号。

7.5.4 读-复制-更新（RCU）

• RCU的读端没有锁、内存屏蔽、原子指令类的开销，机会可以认为是直接读
• RCU的写端访问它的共享资源前首先要复制一个副本，然后对副本进行修改，然后使用一个回调机制在适当的时机把指向原来数据的指针重新指向新的修改数据
  • 这个时机就是所有引用该数据的CPU都退出对共享数据读操作的时候
  • 等待适当时机的这一时期称为宽限期
• RCU的有点
  • 允许多个读执行单元同时访问保护区
  • 允许多个读写执行单元同时访问保护区数据
• RCU不能替代读写锁
  • 因为RCU的写执行单元的同步开销较大，当写较多的时候，对读执行单元的性能提高不能弥补写执行单元同步导致的损失
• 参考:
  • RCU机制
  • TREE RCU实现之一 —— 数据结构
  • TREE RCU实现之二 —— 主干函数
  • TREE RCU实现之三 —— 定期调用
    http://blog.csdn.net/junguo/article/details/8244530/
    http://blog.csdn.net/junguo/article/details/8258231
    http://blog.csdn.net/junguo/article/details/8258261
    http://blog.csdn.net/junguo/article/details/8268277

7.6 信号量

• 定义信号量

struct semaphore sem;

• 初始化：

void sema_init (struct semaphore *sem, int val);
void init_MUTEX (struct semaphore *sem); //将sem的值置为1，表示资源空闲
void init_MUTEX_LOCKED (struct semaphore *sem); //将sem的值置为0，表示资源忙

• 申请内核信号量所保护的资源：

//可引起睡眠
void down(struct semaphore * sem);
//down_interruptible能被信号打断
int down_interruptible(struct semaphore * sem);
// 非阻塞函数，不会睡眠。无法锁定资源则马上返回
int down_trylock(struct semaphore * sem); 

• 释放内核信号量所保护的资源：

void up(struct semaphore * sem);

7.7 互斥体

• 互斥体提供了两种机制：经典互斥体和实时互斥体
• 经典互斥体结构体： （会导致无限制优先级反转问题）

struct mutex {/* 1: unlocked, 0: locked, negative: locked, possible waiters */atomic_t  count;spinlock_twait_lock;struct list_headwait_list;#if defined(CONFIG_DEBUG_MUTEXES) || defined(CONFIG_SMP)struct task_struct*owner;#endif#ifdef CONFIG_DEBUG_MUTEXESconst char *name;void  *magic;#endif#ifdef CONFIG_DEBUG_LOCK_ALLOCstruct lockdep_mapdep_map;#endif
};

• 实时互斥体结构体：

struct rt_mutex {raw_spinlock_twait_lock;struct plist_headwait_list;struct task_struct*owner;#ifdef CONFIG_DEBUG_RT_MUTEXESint  save_state;const char *name, *file;int  line;void  *magic;#endif
};

• 操作：

struct mutex my_mutex;
mutex_init(&my_mutex);
void mutex_lock(struct mutex* lock);                       //获取互斥体,不可被信号中断
void mutex_lock_interruptible(struct mutex* lock);     //获取互斥体,可被信号打断
int mutex_trylock(struct mutex* lock);                     //尝试获取互斥体
void mutex_unlock(struct mutex* lock);                     //释放互斥体int mutex_is_locked(struct mutex* lock):
//该函数检查互斥锁lock是否处于锁定状态。返回1,表示已锁定;返回0,表示未锁定;
int mutex_lock_interruptible(struct mutex* lock);  //该函数可被信号打断
int mutex_lock_killable(struct mutex* lock);       //该函数可被kill信号打断

• 用例：

struct mutex my_mutex;
mutex_init(&my_mutex);   mutex_lock(&my_mutex);
...临界区...
mutex_unlock(&my_mutex); 

7.8 完成量

• 完成量：表示一个执行单元需要等待另一个执行单元完成某事后方可执行。
  • 它是一种轻量级机制，为了完成进程间的同步而设计
  • 使用完成量等待时，调用进程是以独占睡眠方式进行等待的
  • 不是忙等待
• 结构体
  • done变量是完成量要保护的对象
  • wait则是申请完成量的进程等待队列

struct completion {unsigned int done;wait_queue_head_t wait;
};

• 初始化函数
  • done变量被初始化为0.
  • 内核代码中wait_for_common函数其实就是对done变量作判断，若done变量没有大于0，则它一直处于while循环中。
  • complete函数就是对done变量加1。wait_for_common函数便会退出while循环，同时将done减1，表示申请完成量成功。

static inline void init_completion(struct completion *x)
{
x->done = 0;
init_waitqueue_head(&x->wait);
}

• 操作：

struct completion my_completion;    //定义完成量my_completioninit_completion(&my_completion);    //初始化完成量my_completionvoid wait_for_completion(struct completion* comp)
//该函数等待一个完成量被唤醒。该函数会阻塞调用进程,如果所等待的完成量没有被唤醒,那就一直阻塞下去,而且不会被信号打断;int wait_for_completion_interruptible(struct completion* comp)
//该函数等待一个完成量被唤醒。但是它可以被外部信号打断;int wait_for_completion_killable(struct completion* comp)
//该函数等待一个完成量被唤醒。但是它可以被kill信号打断;unsigned long wait_for_completion_timeout(struct completion* comp, unsigned long timeout)
//该函数等待一个完成量被唤醒。该函数会阻塞调用进程,如果所等待的完成量没有被唤醒,调用进程也不会一直阻塞下去,而是等待一个指定的超时时间timeout,当超时时间到达时,如果所等待的完成量仍然没有被唤醒,那就返回;超时时间timeout以系统的时钟滴答次数jiffies计算bool try_wait_for_completion(struct completion* comp)
//该函数尝试等待一个完成量被唤醒。不管所等待的完成量是否被唤醒,该函数都会立即返回bool completion_done(struct completion* comp)
//该函数用于检查是否有执行单元阻塞在完成量comp上(是否已经完成),返回0,表示有执行单元被完成量comp阻塞;相当于wait_for_completion_timeout()中的timeout=0void complete(struct completion* comp)
//该函数只唤醒一个正在等待完成量comp的执行单元void complete_all(struct completion* comp)
//该函数唤醒所有正在等待同一个完成量comp的执行单元NORET_TYPE void complete_and_exit(struct completion* comp, long code)
//该函数唤醒一个正在等待完成量comp的执行单元,并退出,code为退出码

• 注意:在内核处理完请求之后,必须调用这三个函数中的一个,来唤醒其它正在等待的进程