2015-06-03
敬叶 初稿

并发场景

中断上半部

硬件事件在不知道何时发生的情况下发生，因此要知道有没有事件发生，一种方式是CPU不断的去询问硬件，一种方式是发生之后硬件通知CPU。大多数情况下，由硬件去通知CPU是很好的处理方法，但是硬件在某些情况下密集发生事件时，采用询问更快，因为只要保证每次询问都能拿到消息，就可以省掉中断时间了。

设备通过中断线关联到特定中断，中断线IRQ就是一个整数值。中断产生时由中断控制器将信号发送给CPU，CPU调用中断处理函数。中断处理函数是一个普通C函数，但是要求不能睡眠，并且要快速处理返回。因为中断会打断别的进程，所以必须要快，但是有时候又有很多事情要做，所以内核中引入了两半处理法，中断上半部做必要的标记，把真正麻烦的工作交给下半部，这样中断处理函数就可以快速返回了。

系统中断信息可以通过/proc/interrupts查看。

有些代码需要避免被ISR中断，这时可以禁用本地中断，禁用中断也意味着禁用抢占。 

local_irq_disable();
/* do something */
local_irq_enable();

但是上面这样的写法是错误的，因为如果在禁用前已经禁用了，那么启用的时候就会意外开启已经禁用的中断，正确的做法是如下写法。 

local_irq_save(flags);                  /* flags is ulong, not ptr */
/* do something */
local_irq_restore(flags);

中断下半部

前面提到中断必须要快速返回，一般的做法就是做一些标记，然后开启下半部，中断就可以返回了。

tasklet

小任务是比较常用的一种下半部，其重要约束条件是不允许休眠。使用tasklet包括如下几点：处理函数，初始化，调度等 

void tasklet_handler(unsigned long data); /* write handler */

tasklet_init(tasklet, tasklet_handler, dev);

tasklet_schedule(&tasklet);

workqueue

工作队列可以休眠，效率上相对于tasklet可能会低一点。使用workqueue包括如下几点：处理函数，初始化，调度等 

void work_handler(void *data);

schedule_work(&work);

void flush_scheduled_work(void);
int cancel_delayed_work(struct work_struct *work);

条件等待

延迟指定时间

Linux提供了一个滴答计时器，HZ表示一秒钟的滴答数，这个值在不同平台上是不一样的，大部分平台是100。全局变量jiffies记录了开机到当前时间点所经历的滴答数，这个变量的长度和平台的字长相同，因此Linux提供了一个jiffies_64变量。

改变量会溢出，所以要进行时间比较，需要使用如下函数： 

#define time_after(unknown, known) ((long)(known) - (long)(unknown) < 0)
#define time_before(unknown, known) ((long)(unknown) - (long)(known) < 0)
#define time_after_eq(unknown, known) ((long)(unknown) - (long)(known) >= 0)
#define time_before_eq(unknown, known) ((long)(known) - (long)(unknown) >= 0)

短时间的延迟可以用delay函数，delay其实就是忙等，所以毫秒级别的忙等是需要避免的。 

void udelay(unsigned long usecs);
void ndelay(unsigned long nsecs);
void mdelay(unsigned long msecs);

定时器

定时器用于在指定时间之后调用指定函数。 

void handler(unsigned long data);
setup_timer(timer, func, data);

add_timer(&timer);
mod_timer(&timer, jiffies + new_delay);
del_timer(&timer);

完成变量

完成变量是一种异步等待机制，一个线程初始化之后，等待另一个线程唤醒。 

init_completion(x);
wait_for_completion(x);

另一个线程调用如下函数来通知条件成熟： 

complete(x);

内核同步方法

所谓同步方法，目的是为了让临界资源得到保护，防止竞态出现。如果不去保护临界资源，多个线程同时访问和修改会造成状态混乱。为了避免这样的混乱，内核提供了锁机制，当一个线程访问临界资源时，禁止其它线程并发访问。

内核中出现并发的情况是非常多的，如中断上半部，延迟下半部，内核抢占，SMP等等。所以通常情况下，全局资源和共享资源都是要保护的对象。当然用锁也经常会遇到设计不当产生死锁的情况，一个重要的原则是如果要获取多个锁，那么确保在所有代码中以相同的顺序获取。

同步方法

原子操作

原子类型本质上就是一个整数，只不过对它访问不会产生竞态，其具体实现是和架构相关的。比较常见的接口如下： 

int atomic_read(atomic_t *v);
void atomic_set(atomic_t *v, int i);
void atomic_add(int i, atomic_t *v);
void atomic_sub(int i, atomic_t *v);
void atomic_inc(atomic_t *v);
void atomic_dec(atomic_t *v);
int atomic_sub_and_test(int i, atomic_t *v);
int atomic_add_negative(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_inc_return(int i, atomic_t *v);
int atomic_dec_return(int i, atomic_t *v);
int atomic_dec_and_test(atomic_t *v);
int atomic_inc_and_test(atomic_t *v);

要注意原子变量的长度是32位，即便在64位机上也是如此。内核也提供了64位原子变量，只需要将atomic改为atomic64即可。

内核针对位操作也提供了一组原子操作，不过针对指针操作。既然是指针，那么就是平台依赖的，32位机上位的范围为0-31， 64位机上范围为0-63，常见接口如下： 

void set_bit(int nr, void *addr);
void clear_bit(int nr, void *addr);
void change_bit(int nr, void *addr);
int test_and_set_bit(int nr, void *addr);
int test_and_clear_bit(int nr, void *addr);
int test_and_change_bit(int nr, void *addr);
int test_bit(int nr, void *addr);

内存屏障

屏障的作用是确保对变量的操作是按顺序完成的，主要可以防止编译器优化。

PERCPU变量

PERCPU数据存储在数组中，将index关联到一个对应的CPU，因为只能有一个CPU访问，所以不需要关注并发问题。 

unsigned long my_percpu[NR_CPUS];
int cpu;

cpu = get_cpu();                        /* disable kernel preemption */
my_percpu[cpu]++;
put_cpu();                              /* enable kernel preemption */

上面的代码禁用了抢占，所以不要长期占用。之所以要禁止抢占，是因为如果拿到cpu号又调度出去，会导致拿到的cpu号就是错误的。 

DEFINE_PER_CPU(type, name);
get_cpu_var(name)++;
put_cpu_var(name);

void *alloc_percpu(type);
get_cpu_var(ptr);
/* do something on ptr */
put_cpu_var(ptr);
void free_percpu(const void *);

自旋锁

自旋锁调用方法很简单： 

spin_lock(&mr_lock);
/* critical region ... */
spin_unlock(&mr_lock);

自旋锁可以在中断上下文使用，因为自旋锁在获取过程中CPU一直忙等，所以在持有自旋锁期间不能休眠。

很多驱动在获取自旋锁的时候，也要禁止中断，所以最常用的其实是如下一对： 

spin_lock_irqsave(flags);
/* critical region ... */
spin_lock_irqrestore(flags);

自旋锁还有一种变体，就是读写自旋锁，读锁可以多次加锁，但是获取写入锁必须等到所有的读取锁释放。读写锁也有禁止IRQ形式，使用上和自旋锁完全一样。 

read_lock(&mr_rwlock);
/* critical section (read only) ... */
read_unlock(&mr_rwlock);

write_lock(&mr_rwlock);
/* critical section (read and write) ... */
write_unlock(&mr_lock);

信号量

信号量的好处是获取锁的过程可能会休眠，信号量不为0就可以获取。 

sema_init(sem, count);
down(sem)
/* critical region ... */
up(sem);

信号量也有读写形式，名字很奇怪，和自旋锁不统一： 

down_write(sem);
/* critical region ... */
up_write(sem);

互斥锁

互斥锁不允许多次加锁，因此相比于信号量更简单，并且互斥锁在使用上还有限制：谁加锁谁释放。不允许进程退出后不解锁。 

mutex_lock(mutex);
/* critical region ... */
mutex_unlock(mutex);

起始互斥锁强调的是互斥的概念，而不仅仅是同步，因为只能有一个进程可以持有互斥锁。

顺序锁

顺序锁的使用方法和别的锁有所不同，在Linux文件系统中有很多地方用到了顺序锁，顺序锁严格意义来讲是一个重试机制： 

do {
        seq = read_seqbegin(&lock);
        /* critical region ... */
} while (read_seqretry(&lock, seq));

首先读取顺序锁的值，然后进入临界区操作，操作完了之后再检查顺序值，如果顺序值被修改了，说明在操作期间被人动过，就需要重试，否则不需要重试。

RCU-Read-Copy-Update

这是一种高级互斥机制，如果用在正确的场景下，效率会非常好。要注意这种锁的应用场景：

  • 读取频率很高，写入频率很低！
  • 受保护的资源只能通过指针访问
  • 引用保护资源的地方必须是原子上下文

其工作原理是，如果发生写入，那么写入将复制资源，在复制的资源上修改，修改完毕之后更新指针。这种锁是严重偏向读取优先的，读取几乎不需要等待，当然要求就是读取期间需要禁止抢占。相比较于顺序锁，读取不需要重试，顺序锁也是用在读取多写入少的情况，但是人家写入是不会被读取阻塞的。 

rcu_read_lock();
ptr = rcu_dereference_raw(rcu_ptr);
do_something_with(ptr);
rcu_read_unlock();

但是更新就比较费劲，因为要等待所有人释放才能更新，更新只需调用如下API： 

ptr = rcu_dereference_raw(rcu_ptr);
new = copy_memory(ptr);
rcu_assign_pointer(rcu_ptr, new);
call_rcu(&ptr->rcu_head, free_function);

首先需要复制内存并修改，假定新内存地址为new，通过rcu_assign_pointer可以修改，但是我们需要释放之前的内存，那么需要先保存事先的地址，通过调用call_rcu来释放。另外要注意的是，每个受RCU保护的资源都应该有一个成员叫rcu_head，这样我们才能够在释放的时候通过container_of由rcu_head 找到真正需要释放的资源。