并发控制

beyes

并发 ( concurrency )指的是多个执行单元同时、并行执行，而并发的执行单元对共享资源( 硬件资源和软件上的全局变量、静态变量等 )的访问则很容易导致竞态( race conditions )。

1、对称多处理器 (SMP)
SMP 是一种紧耦合、共享存储的系统模型，其体系结构如下图所示，它的特点是多个 CPU 使用共同的系统总线，因此可访问共同的外设和存储器。


Linux 2.6 内核支持抢占调度，一个进程在内核执行的时候可能被另一个高优先级进程打断，进程与抢占它的进程访问共享资源的情况类似于 SMP 的多个 CPU 。

中断可以打断正在执行的进程，如果中断处理程序访问进程正在访问的资源，则竞态也会发生。此外，中断也有可能被更高优先级的中断打断，因此，多个中断之间本身也可能引起并发而导致竞态。

除了 SMP 是真正的并行以外，其他的都是“宏观并行、微观串行”，但其引发的实质问题和 SMP 相似。

beyes

原子操作是指在执行过程中不会被别的代码路径所中断的操作。

Linux 内核提供了一系列函数来实现内核中的原子操作，这些函数分为两类：位原子操作和整形变量原子操作。这两种操作在任何情况下都是原子性的，内核代码可以安全地调用它们而不会被打断，它们都依赖于底层 CPU 的原子操作实现，因此所有这些函数与 CPU 架构密切想关。

为了处理原子操作，Linux 内核提供了一种特殊类型 : atomic_t  ，它的定义是( 一个以原子性访问的计数器 )：

typedef struct {
       volatile int counter;
} atomic_t;

整型原子操作：

1、设置原子变量的值

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

2、获取原子变量的值

atomic_read (atomic_t *v);       /*返回 *v 的值*/

3、原子变量加/减

void atomic_add (atomic_t *v);
void atomic_sub (atomic_t *v);

4、原子变量自增/自减

void atomic_inc (atomic_t *v);         /* *v 加 1*/
void atomic_dec (atomic_t *v)        /* *v 减 1*/

5、操作并测试

int atomic_inc_and_test (atomic_t *v);
int atomic_dec_and_test (atomic_t *v);
int atomic_sub_and_test (atomic_t *v);
上述操作对原子变量执行自增、自减和减操作后(注意没有加)，测试其是否为 0 ，为 0 则返回 true ，否则返回 false 。

6、操作并返回

int atomic_add_returen (int i, atomic_t *v);
int atomic_sub_returen (int i, atomic_t *v);
int atomic_inc_returen (int i, atomic_t *v);
int atomic_dec_returen (int i, atomic_t *v);
上述操作对原子变量进行加/减 和 自增/自减操作，并返回新值。

位原子操作

1、设置位

void set_bit (nr, void *addr);
addr 地址的第 nr 位，置位即将位写为1。

2、清除位

void clear_bit (nr, void *addr);
清除地址 addr 上的第 nr 位，即对其位写0。

3、改变位

void change_bit (nr, void *addr);
对 addr 上的第 nr 进行反置。

4、测试位

test_bit (nr, void *addr);
返回 addr 地址上的第 nr 位。

5、测试并操作位

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

beyes

自旋锁 (spin lock) 是一种对临界资源进行互斥访问的典型手段，其名称来源于它的工作方式。为了获得一个自旋锁，在某 CPU 上运行的代码需要先执行一个原子操作，该操作测试并设置( test and set ) 某个内存变量。由于是原子操作，所以在该操作完成之前其它执行单元不可能访问这个内存变量。

如果测试结果标名锁已经空闲，则程序获得这个自旋锁并继续执行；如果测试结果表明锁仍然被占用，那么程序将在一个小的循环内重复这个“测试并设置”操作，即进行所谓的“自旋“，通俗地说，就是在”原地打转“。当自旋锁的持有者通过重置该变量并释放这个自旋锁后，某个等待的"测试并设置“操作向其调用者报告锁已经释放。

理解自旋锁最简单的方法是把它作为一个变量看待，该变量把一个临界区或者标记为“我当前正在运行，请稍等一会”或者标记为“我当前不在运行，可以被使用“。

如果 A 执行单元首先进入例程，它将持有自旋锁；当 B 执行单元试图进入同一个例程时，将获知自旋锁已被持有，需等待 A 执行单元释放后才能进入。

一般说来，在由自旋锁保护的每个临界区里内核抢占不起作用。但对于单 CPU 系统，把自己给锁起来没什么意义，自旋锁原语只是禁用或者启用内核抢占；需要注意的是，内核抢占在忙等待状态仍然有效，因此一个进程等待一个自旋锁释放也可被一个更高优先级的进程所取代。

Linux 系统中与自旋锁想关的操作主要有 4 种：

1、定义自旋锁
spinlock_t spin;

spinlock_t 的定义在include/linux/spinlock_types.h 中有：

typedef struct {
       raw_spinlock_t raw_lock;

#ifdef CONFIG_GENERIC_LOCKBREAK
       unsigned int break_lock;
#endif

#ifdef CONFIG_DEBUG_SPINLOCK
        unsigned int magic, owner_cpu;
        void *owner;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
        struct lockdep_map dep_map;
#endif
} spinlock_t;

其中，raw_spinlock_t 的定义在 include/linux/spinlock_types_up.h 中：

typedef struct {
        volatile unsigned int slock;
} raw_spinlock_t;

上面，slock 变量为 1 表示非锁状态，如果为 0 或是任一负数则表示锁状态。

break_lock 表示进程是否在忙等待自旋锁。

2、初始化自旋锁
spin_lock_init (lock);

3、获得自旋锁
spin_lock (lock);
该宏用于获得自旋锁 lock ，如果能够立即获得锁，它就马上返回，否则，它将自旋在那里，直到该自旋锁的保持者释放。

spin_trylock (lock);
该宏尝试获得自旋锁 lock，如果能立即获得，它将获得锁并返回真，否则立即返回假，实际上不再“在原地打转” 。

4、释放自旋锁
spin_unlock (lock);
该宏释放自旋锁 lock ，它与 spin_trylock 或 spin_lock 配对使用。

自旋锁一般如下使用：

spinlock_t lock;
spin_lock_init (&lock);

spin_lock (&lock);      /*获得自旋锁,保护临界区*/
...//临界区
spin_unlock (&lock);     /*解锁*/

自旋锁主要针对 SMP 或单 CPU 但内核可抢占的情况，对于单 CPU 且内核不支持抢占的系统，自旋锁退化为空操作。
在但 CPU 和内核可抢占的系统中，自旋作持有期间，内核的抢占将被禁止。由于内核可抢占的单 CPU 系统的行为实际类似于 SMP 系统，因此，在这样的单 CPU 系统中使用自旋锁仍十分必要。

尽管用了自旋锁可以保证临界区不受别的 CPU 和本 CPU 内的抢占进程打扰，但是得到锁的代码路径在执行临界区的时候仍然可能受到中断和底半部 (BH) 的影响。为了防止这种影响，就需要用到自旋锁的衍生。 spin_lock() / spin_unlock() 是自旋锁机制的基础，它们和关中断 lock_iqr_disable() / 开中断 lock_irq_enable() 、关底半部 lock_bh_disable() / 开底半部 local_bh_enable() 、关中断保存状态字 local_iqr_save() / 开中断并恢复状态 local_irq_restore() 结合，就形成了整套自旋锁机制，关系如下：

spin_lock_irq() = spin_lock() + local_irq_disable()
spin_unlock_irq() = 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 在等待自旋锁时不做任何有用的工作，仅仅是等待。因此，只有在占用锁时间极短的情况下，使用自旋锁才是合理的。当临界区很大或有共享设备的时候，需要较长时间占用锁，使用自旋锁会降低系统的性能。
自旋锁可能导致系统死锁。引发这个问题最常见的情况是递归使用一个自旋锁，即如果一个已经拥有某个自旋锁的 CPU 想第 2 次获得这个锁，则该 CPU 死锁。此外，如果进程获得自旋锁后再阻塞，也有可能导致死锁的发生。copy_from_user()、copy_to_user() 和 kmalloc() 等函数都有可能引起阻塞，因此在自旋锁的占用期间不能调用这些函数。

使用自旋锁使设备只能被一个进程打开：

int xxx_count = 0;   /*定义文件打开次数*/

static int xxx_open (struct inode *inode, struct file *filp)
{
     ...
     spin_lock (&xxx_lock);  /*获取自旋锁*/
    
     if (xxx_count){     /*获得自旋锁,但已有别的进程打开设备*/
          spin_unlock (&xxx_lock);    /*设备忙,释放自旋锁返回*/
          reurn (-EBUSY);
     }

     xxx_count++;        /*没有别的进程使用设备,增加使用计数*/
     spin_unlock (&xxx_lock);    /*已经成功打开设备,释放自旋锁*/
     ...
     return (0);        /*打开成功*/
}

static int xxx_release (struct inode *inode, struct file *filp)
{
    ...
    spin_lock (&xxx_lock);    /*获得自旋锁*/
    xxx_count--;       /*减少使用计数*/
    spin_unlock (&xxx_lock);

    return (0);
}

上面，使用自旋锁修改了全局变量 xxx_count 。程序段保证了一个设备仅能被一个进程打开。

关于自旋锁的其他介绍：http://www.groad.net/bbs/read.php?tid-1298-page-e.html#a

beyes

自旋锁不关心锁定的临界区究竟进行怎样的操作，不管是读还是写，都一视同仁。即便多个执行单元同时读取临界资源也会被锁住。

实际上，对共享资源并发访问时，多个执行单元同时读取是没有问题的，自旋锁的衍生锁读写自旋锁 (rwlock) 允许读操作并发。

读写自旋锁是一种比自旋锁粒度更小的锁机制，它保留了“自旋”的概念，但是在写操作方面，只能最多有一个写进程，在读操作方面，同时可以有多个读执行单元。当然，读和写不能同时进行！

1、定义和初始化读写自旋锁

rwlock_t  my_rwlock = RW_LOCK_UNLOCKED;    /*静态初始化*/
rwlock_t my_rwlock;
rwlock_init (&my_rwlock);    /*动态初始化*/

2、读锁定

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

3、读解锁

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

在对共享资源进行读取之前，应先调用读锁定函数，完成之后应调用读解锁函数。
read_lock_irqsave() 是 read_lock() 与 local_irq_save() 的组合；
read_lock_irq()  是 read_lock() 和 local_irq_disable() 的组合；
read_lock_bh() 是 read_lock()  和 local_bh_disable() 的组合。

read_unlock_irqrestore() , read_unlock_irq(), read_unlock_bh() 情况类似。

4、写锁定

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

5、写解锁

void write_unlock (rwlock_t *lock);
void write_unlock_iqrrestore (rwlock_t *lock, unsigned long flags);
void write_unlock_iqr (rwlock_t *lock);
void write_unlock_bh (rwlock_t *lock);

write_lock_irqsave() 是 write_lock() 与 local_irq_save() 的组合；
write_lock_irq() 是 write_lock() 与 local_irq_disable() 的组合；
write_lock_bh() 是 write_lock() 与 local_bh_disable() 的组合。

write_unlock_irqrestore() , write_unlock_irq() , write_unlock_bh() 的情况类似。

在对共享资源进行写之前，应该先调用写锁定函数，完成后再调用写锁定函数。

write_trylock() 也只是尝试获取读写自旋锁，不管成功失败，都会立即返回。

读写自旋锁一般如下使用：

rwlock_t lock;   /*定义 rwlock */
rwlock_init (&lock);    /*初始化 rwlock*/

/*读时获取锁*/
read_lock (&lock);
...//临界资源
read_unlock (&lock);

/*写获取锁*/
write_lock_irqsave (&lock, flags);
...//临界资源
write_unlock_irqrestore (&lock, flags);