LDD3第五章并发和竞态（2）

首先看一下scull的write函数中的一段代码，是申请内存的一段代码

if (!dptr->data)
    {
        dptr->data = kmalloc(qset * sizeof(char *), GFP_KERNEL);
        if (!dptr->data)
            goto out;
        memset(dptr->data, 0, qset * sizeof(char *));
    }

    假定有两个进程A和B， 在同时执行上面这段代码，如果指针为NULL，则两个进程都会执行分配内存的操作，因为两个进程对同一个指针赋值，显然只有一个赋值会
成功，而且是第二个完成赋值的进程会成功，这样第二个进程的赋值操作就覆盖了第一个进程的赋值操作，第一个进程分配的内存将会丢失，永远不会返回到系统中

上述事件的过程就是一种竞态，竞态导致对共享数据的非法访问。

一、并发及其管理
   竞态通常座位对资源的共享访问结果而产生，在设计驱动程序时，只要可能就应该避免资源的共享，如果没有并发的访问也就不会有竞态的产生，这种思想的最明显
应用就是尽量避免使用全局变量，在实际工作中也吃过不少全局变量的亏，背了好几个低级失误。
  硬件资源的本质就是共享，资源共享的规则为：
  1、在单个执行线程之外共享硬件或软件资源的任何时候，因为另外一个线程可能产生对该资源的不一致观察，因此必须显式的管理对该资源的访问。
  2、在内核代码创建了一个可能和其他内核部分共享的对象时，该对象必须在有其他组件引用自己时保持存在并正确工作。在对象尚不能正确工作时，不能对内核可用。

二、信号量和互斥体
   信号量是一个比较熟悉的概念，一个信号量本身就是一个整数值，它和一对函数联合使用，P和V，希望进入临界区的进程将在相关信号量调用P，如果信号量值大于0
则值减一，进程可以继续，相反进程必须等待，直到其他人释放该信号量，对信号量的解锁通常调用V完成，该函数增加信号量的值，并在必要时唤醒等待的进程。
    当信号量用于互斥时，信号量的值应初始化为1，这种信号量任何时刻只能有单个进程或线程所有，在这种使用模式下，一个信号量有时也称为一个“互斥体(mutex)”
它是互斥（mutual exclusion）的简称。Linux内核中几乎所有的信号量均用于互斥。

三、linux信号量的实现

要使用信号量，内核代码必须包含，相关的类型是struct semaphore，可以通过下面几种途径来声明和初始化信号量：
1、直接创建信号量，通过sema_init实现
    

/*初始化函数*/
void sema_init(struct semaphore *sem, int val)

其中value是赋予信号量的初始值。

2、信号量通常被用于互斥模式，内核提供了一组辅助函数和宏：
    

/*方法一、声明+初始化宏*/
DECLARE_MUTEX(name);
DECLARE_MUTEX_LOCKED(name);

/*方法二、初始化函数*/
void init_MUTEX(struct semaphore *sem);
void init_MUTEX_LOCKED(struct semaphore *sem);

/*带有“_LOCKED”的是将信号量初始化为0，即锁定，允许任何线程访问时必须先解锁。没带的为1。*/

在linux中，P函数称为down，这里的down指的是该函数减小了信号量的指，它也许会将调用者置于睡眠状态，等待信号量变的可用，down的三个版本如下：

void down(struct semaphore *sem); /*不推荐使用，会建立不可杀进程*/
int down_interruptible(struct semaphore *sem);/*推荐使用，使用down_interruptible需要格外小心，若操作被中断，该函数会返回非零值，而调用这不会拥有该信号量。对down_interruptible的正确使用需要始终检查返回值，并做出相应的响应。*/
int down_trylock(struct semaphore *sem);/*带有“_trylock”的永不休眠，若信号量在调用是不可获得，会立刻返回非零值。*/

V函数为：

void up(struct semaphore *sem);/*任何拿到信号量的线程都必须通过一次（只有一次）对up的调用而释放该信号量。在出错时，要特别小心；若在拥有一个信号量时发生错误，必须在将错误状态返回前释放信号量。*/

一般，我们很容易犯忘记释放信号量的的错误。

四、在scull中使用信号量

scull结构定义如下：

struct scull_dev
{
    struct scull_qset *data; /* Pointer to first quantum set */
    int quantum; /* the current quantum size */
    int qset; /* the current array size */
    unsigned long size; /* amount of data stored here */
    unsigned int access_key; /* used by sculluid and scullpriv */
    struct semaphore sem; /* mutual exclusion semaphore */
    struct cdev cdev;      /* Char device structure        */
}

其中，sem成员就是我们的信号量，信号量在使用前必须初始化，scull对信号量的初始化如下代码：

for (i = 0; i < scull_nr_devs; i++)
    {
        scull_devices[i].quantum = scull_quantum;
        scull_devices[i].qset = scull_qset;
        init_MUTEX(&scull_devices[i].sem);
        scull_setup_cdev(&scull_devices[i], i);
    }

这里的顺序需要注意，是先初始化的信号量，之后再注册CDEV，即驱动在对内核可用之前已经初始化好信号量。

五、读取者/写入者信号量

信号量对所有的调用者执行互斥，而不管每个线程想做什么。但是许多的任务我们可以划分为两种类型：一些是读取，一些是写入，允许
多个并发的读取是允许的，而不用等待其他读取者推出临界区，linux内核提供了读取/写入者信号量“rwsem”，使用rwsem的代码必须包含


初始化

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);
void downgrade_write(struct rw_semaphore *sem);/*该函数用于把写者降级为读者，这有时是必要的。因为写者是排他性的，因此在写者保持读写信号量期间，任何读者或写者都将无法访问该读写信号量保护的共享资源，对于那些当前条件下不需要写访问的写者，                                                  降级为读者将，使得等待访问的读者能够立刻访问，从而增加了并发性，提高了效率。*/

一个rwsem允许一个写入者或者无限多个读取者拥有该信号量，写入者拥有更高的优先级，当某个写入者试图进入临界区，在所有写入者完成其工作之前，不允许读取者获得访问。

六、completion

内核中常见的一种模式，在当前线程之外初始化某个活动，然后等待该活动的结束，因此内核中出现了completion接口，completion是一种轻量级的机制，它允许一个线程告诉另
一个线程某个工作已经完成，为了使用completion，代码必须包含，

DECLARE_COMPLETION(my_completion);    /* 创建completion（声明+初始化） */

struct completion my_completion;      /* 动态声明completion 结构体*/
init_completion(&my_completion);      /*动态初始化completion*/

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

/*如果未使用completion_all，completion可重复使用；否则必须使用以下函数重新初始化completion*/
INIT_COMPLETION(struct completion c);          /*快速重新初始化completion*/

completion机制的典型使用是模块退出时的内核线程终止，在这种原型中，某些驱动程序的内部工作由一个内核线程在while(1)循环中完成
当内核准备清除该模块时，exit函数会告诉该线程退出并等待completion，为了实现这个目的，内核包含了可用于这种线程的一个特殊函数

void complete_and_exit(struct completion *c, long retval)

七、自旋锁

信号量对互斥来讲是非常有用的工具，但是它不是内核提供的唯一的这类工具，自旋锁是一个互斥设备，它只有两个值：“锁定”和“解锁”
它通常实现为某个整数值中的单个位，希望获得某特定锁的代码测试相关的位。如果锁可用，则“锁定”位被设置，代码进入临界区，相反
如果锁不可用，则代码进入忙循环并重复检查这个锁，直到锁可用为止，自旋锁和信号量的最大区别为：信号量是一种睡眠锁，信号量不
可用时，则会导致进程睡眠，而自旋锁不会，代码一直会处于忙等状态。

上面描述中“测试”和“设置”的操作必须以原子的方式进行，
适用于自旋锁的核心规则：
（1）任何拥有自旋锁的代码都必须使原子的，不能因为任何原因放弃处理器，除服务中断外（某些情况下也不能放弃CPU,如中断服务也要
    获得自旋锁。为了避免这种锁陷阱需要在拥有自旋锁时禁止中断），不能放弃CPU（如休眠，休眠可发生在许多无法预期的地方）。否
    则CPU将有可能永远自旋下去（死机）。
（2）拥有自旋锁的时间越短越好。
自旋锁原语所需包含的文件是 ,以下是自旋锁的内核API: 

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

/* 所有spinlock等待本质上是不可中断的，一旦调用spin_lock，在获得锁之前一直处于自旋状态*/
void spin_lock(spinlock_t *lock);           /* 获得spinlock*/
void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);/* 获得spinlock，禁止本地cpu中断，保存中断标志于flags*/
void spin_lock_irq(spinlock_t *lock);       /* 获得spinlock,禁止本地cpu中断*/
void spin_lock_bh(spinlock_t *lock)         /* 获得spinlock,禁止软件中断，保持硬件中断打开*/

/* 以下是对应的锁释放函数*/
void spin_unlock(spinlock_t *lock);
void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags);
void spin_unlock_irq(spinlock_t *lock);
void spin_unlock_bh(spinlock_t *lock);

/* 以下非阻塞自旋锁函数（立即返回），成功获得，返回非零值；否则返回零*/
int spin_trylock(spinlock_t *lock);
int spin_trylock_bh(spinlock_t *lock)

读取者/写入者自旋锁

这种锁和前面介绍的读取/写入者信号量很类似，允许任意数量的读取者进入临界区，但是写入者必须互斥访问，读取/写入者锁具有
rwlock_t类型，API如下：

rwlock_t my_rwlock = RW_LOCK_UNLOCKED;/* 编译时初始化*/

rwlock_t my_rwlock;
rwlock_init(&my_rwlock); /* 运行时初始化*/

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);

/* 新内核已经有了read_trylock*/

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);
int write_trylock(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)

八、锁陷阱

锁定模式的规则必须要清晰和明确，而且必须在一开始就需要明确，否则其后的改进会比较困难
不明确规则：如果某个获得锁的函数要调用其他同样试图获取这个锁的函数，代码就会锁死。（无论是信号量还是自旋锁，不允许锁的拥有者第二次获得同个锁）
           为了锁的正确工作，不得不编写一些函数，这些函数假定调用者已经获得了相关的锁。
锁的顺序规则：在必须获取多个锁时，应始终以相同顺序获取。
若必须获得一个局部锁和一个属于内核更中心位置的锁，应先获得局部锁。
若我们拥有信号量和自旋锁的组合，必须先获得信号量。
不得再拥有自旋锁时调用down。（可导致休眠）
尽量避免需要多个锁的情况。
细颗粒度和粗颗粒度的对比：应该在最初使用粗颗粒度的锁，除非有真正的原因相信竞争会导致问题。

九、锁之外的方法

    (1)免锁方法
   经常用于免锁的生产者/消费者任务的数据结构之一是循环缓冲区，在这个算法中，一个生产者将数据结构放入数组的结尾，而消费者从数组的另一端
   移走数据。
   (2)原子变量
    内核提供了一种原子的证书类型，成为atomic_t类型，定义在中，这种操作的速度非常快，因为只要可能，它们都会被编译成单个机器
  指令，接口函数如下：
  

void atomic_set(atomic_t *v, int i); /*设置原子变量 v 为整数值 i.*/
atomic_t v = ATOMIC_INIT(0); /*编译时使用宏定义 ATOMIC_INIT 初始化原子值.*/

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

void atomic_add(int i, atomic_t *v);/*由 v 指向的原子变量加 i. 返回值是 void*/
void atomic_sub(int i, atomic_t *v); /*从 *v 减去 i.*/

void atomic_inc(atomic_t *v);
void atomic_dec(atomic_t *v); /*递增或递减一个原子变量.*/

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);
/*进行一个特定的操作并且测试结果; 如果在操作结束后, 原子值是 0, 那么返回值是真; 否则, 它是假. 注意没有 atomic_add_and_test.*/

int atomic_add_negative(int i, atomic_t *v);
/*加整数变量 i 到 v. 如果结果是负值返回值是真, 否则为假.*/

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);
/*像 atomic_add 和其类似函数, 除了它们返回原子变量的新值给调用者.*/

    (3)位操作

    内核提供了一组可原子的修改和测试某个位的函数，原子位操作非常快，这种操作可以使用单个机器指令来执行，并不需要禁止中断，函数在
   中声明，nr参数（用来描述要操作的位）通常被定义为int，少数架构上被定义为unsigned long，要修改的地址通常为指向unsigned long的指针，某些架构上为
  void *来代替
  

void set_bit(nr, void *addr); /*设置第 nr 位在 addr 指向的数据项中。*/

void clear_bit(nr, void *addr); /*清除指定位在 addr 处的无符号长型数据.*/

void change_bit(nr, void *addr);/*翻转nr位.*/

test_bit(nr, void *addr); /*这个函数是唯一一个不需要是原子的位操作; 它简单地返回这个位的当前值.*/

int test_and_set_bit(nr, void *addr);
int test_and_clear_bit(nr, void *addr);
int test_and_change_bit(nr, void *addr) /*这三个函数和前面列出的一样具有原子化的行为，不同之处是它同时返回这个位的先前值*/

一个实例，假定当锁在0时空闲，而在非零时忙：

/* 试着锁定 */
while (test_and_set_bit(nr, addr) != 0)
    wait_for_a_while();

/* do your work */

/*释放锁，并检查*/
if (test_and_clear_bit(nr, addr) == 0)
    something_went_wrong(); /* already released: error */

(4)seqlock

当要保护的资源很小、很简单会频繁被访问而且写入访问很少发生切必须快速时，可以使用seqlock，seqlock允许读取者对资源自由访问，但需要
读取者检查是否和写入者发生冲突，当有冲突时，就需要重试对资源的访问，seqlock通常不能用于保护包含有指针的数据结构，因为指针可能是无效
的指针

/*两种初始化方法*/
seqlock_t lock1 = SEQLOCK_UNLOCKED;

seqlock_t lock2;
seqlock_init(&lock2)

这种类型的锁通常用于保护某种类型的简单计算，读取访问在进入临界区时获得一个整数序列，如果动作结束时此序列已经发生并发的修改
则可以简单丢弃结果并重新计算

unsigned int seq;
do {
    seq = read_seqbegin(&the_lock);
    /* Do what you need to do */
} while read_seqretry(&the_lock, seq)

如果在中断处理程序中使用seqlock，则应该受用IRQ安全的版本：

unsigned int read_seqbegin_irqsave(seqlock_t *lock, unsigned long flags);
int read_seqretry_irqrestore(seqlock_t *lock, unsigned int seq, unsigned long flags)

写入者必须在进入seqlock保护的临界资源时获得一个互斥锁，为此，需要调用下面的函数

void write_seqlock(seqlock_t *lock);
void write_sequnlock(seqlock_t *lock)

因为自旋锁用来控制写存取, 所有通常的变体都可用:

void write_seqlock_irqsave(seqlock_t *lock, unsigned long flags);
void write_seqlock_irq(seqlock_t *lock);
void write_seqlock_bh(seqlock_t *lock);

void write_sequnlock_irqrestore(seqlock_t *lock, unsigned long flags);
void write_sequnlock_irq(seqlock_t *lock);
void write_sequnlock_bh(seqlock_t *lock)

(5)读取-复制-更新

读取-拷贝-更新(RCU) 是一个高级的互斥方法, 在合适的情况下能够有高效率. 它在驱动中的使用很少。