Linux驱动设计——并发与竞态控制

并发的概念：多个执行单元同时、并行被执行。

共享资源：硬件资源（IO/外设等），软件上的全局变量、静态变量等。

四种并发控制机制(对共享资源互斥的访问)：原子操作、自旋锁（spinlock）、信号量(semaphore)和完成量(completion)。中断屏蔽也可以作为一种并发控制机制。

发生竞态情况：

1. 对称多处理器（SMP）的多个CPU之间的竞态
2. 单CPU内进程间的竞态
3. 中断（硬中断、软中断、Tasklet、底半部）与进程之间的竞态

---

 中断屏蔽

可以解决中断与进程之间、内核抢占进程之间的并发。

主要函数：

local_irq_disable();
local_irq_enable();
local_irq_save(flags);        //禁止中断并保存中断寄存器信息到flags
local_irq_restore(flags);    //打开中断，并回复flags中的值到中断寄存器    
local_bh_disable();           //仅禁止中断底半部中断
local_bh_enable();           //打开中断底半部中断

注意：不要长时间地屏蔽中断，因为Linux系统的异步IO、进程调度等很多重要操作都依赖于中断，在屏蔽中断期间，所有的中断都无法得到处理，长时间屏蔽中断可能造成数据丢失甚至系统崩溃。

//中断使用的模板
local_irq_disable();      //屏蔽中断
...
critical section             //临界区（保护对共享资源的操作）
...
local_irq_enable();      //打开中断
//中断屏蔽实例

void s3c2410_gpio_setpin(unsigned int pin, unsigned int to)
{
　　void __iomem *base = S3C24XX_GPIO_BASE(pin);
　　unsigned long offs = S3C2410_GPIO_OFFSET(pin);
　　unsigned long flags;
　　unsigned long dat;

　　local_irq_save(flags);   //禁止中断并保存中断寄存器信息到flags

　　dat = __raw_readl(base + 0x04);  
　　dat &= ~(1 << offs);　　　　　　  //这四行为临界区
　　dat |= to << offs;
　　__raw_writel(dat, base + 0x04);  

　　local_irq_restore(flags);  //打开中断，并回复flags中的值到中断寄存器
}

　　

---

 原子变量操作

类型：atomic_t

设置原子变量的值

void atomic_set(atomic_t *v,int i);  //动态初始化
atomic_t v=ATOMIC_INIT(0);　　　　　　 //竞态初始化

获取原子变量的值

int atomic_read(atomic_t *v); 　　//返回v当前值

原子变量加减

void atomic_add(int i, atomic_t *v);  //*v 累加i;无返回值
void atomic_sub(int i, atomic_t *v);  //*v递减i,无返回值

void atomic_inc(atomic_t *v);
void atomic_dec(atomic_t *v);

操作并测试

//操作结束后，原子值为0，返回true，否则返回false
int atomic_inc_and_test(atomic_t *v);　　　　　//原子值加1
int atomic_dec_and_test(atomic_t *v);　　　　　//原子值减1
int atomic_sub_and_test(int i, atomic_t *v); //原子值减i，再与0比较

操作并返回

//操作结束后返回新值
int atomic_add_return(int i, atomic_t *v);
int atomic_sub_return(int i, atomic_t *v);
int atomic_inc_return(atomic_t *v);
int atomic_dec_return(atomic_t *v);

位原子操作

void set_bit(nr,void *addr);　　　　　　　　//设置第nr个bit的值
void clear_bit(nr, void *addr);　　　　　　//清除第nr个bit的值
void change_bit(nr, void *addr);         //改变第nr个bit的值，即将值翻转
int test_bit(nr, void *addr);            //检测第nrbit是否被设置
int test_and_set_bit(nr, void *addr); 　 //检测，并返回先前的值再进行对应的操作
int test_and_clear_bit(nr, void *addr);  //同上
int test_and_change_bit(nr, void *addr); //同上

　　

原子变量操作绝对不会在执行完毕前被任何其他任务或事件打断。原子操作需要硬件的支持，因此是架构相关的，其API和原子类型的定义都定义在内核源码树中的include/asm/atomic.h文件中，它们都是使用汇编语言实现的。

常用于多个应用程序对同一个共享的值进行操作的情况。

---

自旋锁

自旋锁是实现信号量和完成量的基础。对资源有很好的保护作用。

通常实现为一个整数值

Linux系统中提供了一些锁机制来避免竞争条件，最简单的一种就是自旋锁。引入锁的机制是因为单独的原子操作已经不能满足复杂的内核设计需要。

Linux中一般可以认为有两种锁：自旋锁和信号量。这两种锁是为了解决内核中不同的问题开发的。

自旋锁的类型(结构体)：struct spinlock_t

自旋锁的使用

在驱动程序中，有的设备只允许打开一次，那么就需要一个自旋锁来保护表示设备打开和关闭状态的变量count。

1. 定义和初始化自旋锁

//编译时初始化
spinlock_t xxx_lock=SPIN_LOCK_UNLOCKED;
//运行时初始化
void spin_lock_init(spinlock_t *lock);

2. 锁定自旋锁(四种锁定函数)

void spin_lock(spinlock_t *lock);

//获得自旋锁之前禁止终端（包括软中断和硬中断）
void spin_lock_irq(spinlock_t *lock);

//获得自旋锁之前禁止中断（包括软中断和硬中断），中断状态保存到状态字flags中。
void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);

//获得自旋锁之前禁止软件中断，硬件中断保持打开
void spin_lock_bh(spinlock_t *lock);

3. 释放自旋锁（对应的四种解锁函数）

void spin_unlock(spinlock_t *lock);
void spin_unlock_irq(spinlcok_t *lock);

void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);

void spin_lock_bh(spinlock_t *lock); 

4. 使用自旋锁

自旋锁使用的模板之一

spinlock_t lock;
spin_lock_init(&lock);
spin_lock(&lock);
critical section
spin_unlock(&lock);

自旋锁实例

static void mousedev_hangup(struct mousedev* mousedev)
{
    struct mousedev_client *client;
    spin_lock(&mousedev->client_lock);      //闭锁
    list_for_each_entry(client, &mousedev->client_list, node)
        kill_fasync(&client->fasync, SIGIO, POLL_HUP);
    spin_unlock(&mousedev->client_lock);   //解锁
    wake_up_interruptible(&mousedev->wait);
}

读写自旋锁(使用较少)

读取者获得锁

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_unlock(rwlock_t *lock);
void write_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
void write_unlock_irq(rwlock_t *lock);
void write_unlock_bh(rwlock_t *lock);

读写锁实例

unsigned long flags;
rwlock_t rwlock;
rwlock_init(&rwlock);
read_lock(&rwlock);
...                            //临界资源
read_unlock(&rwlock);
write_lock_irqsave(&rwlock, flags);
...                            //临界资源
write_unlock_irqrestore(&rwlock, flags);

自旋锁使用注意事项

自旋锁是一种忙等待，当自旋锁条件不满足时，会一直不断的循环条件是否满足。所以适合短时间锁定的轻量级加锁机制。

自旋锁不能递归使用。因为自旋锁被设计为在不同线程或函数之间同步。（？）

---

信号量

信号量本质上是一个整数值，是一种外部资源的标识。有的一对操作函数分别被称为P和V。

Linux提供了两种信号量，其中一种用于内核程序中，另一种用于用户程序中。

信号量也是一种保护临界资源的一种有用方法。信号量与自旋锁使用方法基本一样。与自旋锁相比，信号量只有当得到信号量的进程或者线程时才能够进入临界区，执行临界代码。

信号量与自旋锁最大的不同：当一个进程试图获取一个锁定的信号量时，进程不会像自旋锁一样在远处忙等待。而是采用如下方式：当获取的信号量没有被释放时，进程会将自身加入一个等待队列中去睡眠，直到拥有信号量的进程释放掉信号量后，处于等待队列的那个进程才被唤醒。然后继续执行。所以这就要求使用信号量的进程是能够睡眠的进程（像中断处理程序就不能使用信号量）。

进入临界区：

相关信号量上调用P;

如果信号量>0,则该值会减一，而进程可以继续；

如果信号量的值小于或等于0，进程必须阻塞等待直到其他人释放该信号量。

退出临界区：

信号量的解锁通过调用V完成；

该函数增加信号量的值，并在必要时唤醒等待的进程。（信号量的值的增减是原子操作）

Question：

信号量的值为什么可以大于1（即为什么可以有多个进程同时占有临界资源）？

答：当同一资源的数量为N时，意味能够允许N个进程同时使用该资源，此时，可以设置相应信号量的初值为N。
而如果资源只有一个，且互斥使用时，信号量的初值必须为1。

信号量除了用于互斥还能做什么？

信号量的值应该初始化为1。只能由单个线程或进程拥有。一个信号量（二进制信号量）有时也称为一个“互斥体（mutex,即mutual exclusion）”。Linux内核中几乎所有的信号量均用于互斥。

信号量和互斥体的区别（一个用于同步（同步包括互斥），一个用于互斥）：

http://blog.csdn.net/rommi/article/details/6015143

http://www.cnblogs.com/lonelycatcher/archive/2011/12/20/2294161.html

信号量的结构类型：struct semaphore  头文件：asm/semaphore.h

struct semaphore{
    spinlock_t     lock;
    unsigned int    count;
    struct list_head    wait_list;
};

信号量的使用

1. 初始化信号量

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

    静态的声明互斥信号量

//静态的声明互斥体通过两个辅助宏实现
DECLARE_MUTEX(name),        //声明信号量name，并初始化为1
DECLARE_MUTEX_LOCKED(name); //声明信号量name,并初始化为0，即阻塞状态

    动态的初始化互斥信号量

void init_MUTEX(strcut semaphore *sem),
void init_MUTEX_LOCKED(strcut semaphore *sem);

2. 获得信号量 (P函数被称为down)

void down(struct semaphore *sem);    //减小信号量的值，如果当前进程不能获得信号量就阻塞
void down_interruptible(struct semaphore *sem); //可中断的down函数
int down_trylock(struct semaphore *sem); //不会阻塞的down函数，信号量不可用的话，立刻返回非零值1，成功获取返回0.

down(struct semaphore *sem);创建了不可杀的进程，应少用。

使用down_interruptiable(struct semaphore *sem);时要额外小心（其操作被中断的话，返回非零值，调用者不会拥有该信号量），要始终查看返回值，并作出相应的响应。

3. 释放信号量 (V操作)

void up(struct semaphore *sem);

4. 使用信号量

　　模板之一，与其他类似

DECLARE_MUTEX(sem);          //定义信号量
if(down_interruptible(&sem))   //获取信号量保护临界区
{
    return -ERESTARTSYS;
}
...
critical section                         //临界区
...
up(&sem);                              //释放信号量

　鼠标设备驱动实例 linux-2.6.32.2driversinputmousedev.c

static void mousedev_remove_chrdev(struct mousedev* mousedev)
{
    mutex_lock(&mousedev_table_mutex);
    mousedev_table[mousedev->minor] = NULL;
    mutex_unlock(&mousedev_table_mutex);
}

5. 信号量用于同步操作（可以用于保证线程的执行先后顺序）

读取者/写入者信号量

读操作允许并发，写操作只能互斥。即rwsem可以有无数个读取者拥有此信号量，而只能有一个写入者拥有此信号量。

写入者优先级别更高，当有大量写入者竞争改信号量时，可能会长时间拒绝读者访问。

数据类型：struct rw_semaphore

初始化：

void init_rwsem(struct rw_semaphore* sem);

主要函数

void down_read(struct rw_semaphore* sem);
int down_read_trylock(struct rw_semaphore* sem);
void up_read(struct rw_semaphore* sem);
void down_write(struct rw_semaphore* sem);
int down_write_trylock(struct rw_semaphore* sem);
void up_write(struct rw_semaphore* sem);

模板

rw_semaphore_t rw_sem;
init_rwsem(&rw_sem);

down_read(&rw_sem); //获取临界资源
...　　　　　　　　　　//临界区
up_read(&rw_sem);   //释放临界资源
down_write(&rw_sem);
...
up_write(&rw_sem);

自旋锁与信号量的选择

当代码在很短的时间内可以执行完成时，那么选择自旋锁；当一个进程对被保护资源占用时间比进程切换的时间长很多时，选择信号量。

---

完成量 completion

针对一个线程要等待另一个线程执行完成某个操作后才能继续执行的情况。虽然信号量也可以完成这项工作，但是效率比完成量要差些。

完成量实现一个线程发送信号通知另一个线程开始完成某个任务。

完成量的实现

完成量的结构：includelinuxcompletion.h

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

完成量的使用

1. 定义和初始化完成量

DECLARE_COMPLETION(xxx_completion);  //静态创建

struct completion xxx_completion;  //动态创建
init_completion(&xxx_completion); //初始化

2. 等待完成量

void wait_for_completion(struct completion *c);   

3. 释放完成量

void complete(struct completion *c); //唤醒一个等待的线程 
void complete_all(struct completion *c); //唤醒所有等待的线程

　　如果使用complete_all并想重复使用completion结构，则必须在重复使用该结构之前重新初始化它。可以使用宏：

　　INIT_COMPLETION(struct completion *c);

4. 使用完成量

 实例

DECLARE_COMPLETION(xxx_comp);
ssize_t complete_read(struct file *filp, char __user *buf, size_t count, loff_t *pos)
{
　　wait_for_completion(&xxx_comp);

　　return 0;
}
ssize_t complete_write(struct file *filp, char __user *buf, size_t count, loff_t *pos)
{
　　complete(&xxx_comp);
　　return count;
} 

实验任务：

1. 不加Semaphore的多进程写试验

2. 加Semaphore的多进程写试验

To be continue...