并发和竞态
谨以此文纪念过往的岁月。
在linux中处理并发和竞态有如下机制：
1.中断屏蔽
2.原子量
3.信号量(semaphore)
4.自旋锁(spin lock)
5.完成量(completion)
一.中断屏蔽
采用以下组合来实现
保存irq flags
local_save_flags(flags);
关闭中断
local_irq_disable();
使能中断
local_irq_enable();
恢复中断flags
local_irq_restore(flags);
对于短时间内避免并发，这个是个好办法，但是不能长时间的关闭中断。并且只能关闭本CPU，对于SMP多CPU引发的竞态无法解决。
关于其实现
#define local_save_flags(flags) \
do { \
typecheck(unsigned long, flags); \ --检测类型是否正确。
raw_local_save_flags(flags); \
} while (0)
将寄存器cpsr值保存到x中
#define raw_local_save_flags(x) \ --内嵌asm语句 将寄存器cpsr中值保存到x中。
({ \
__asm__ __volatile__( \ -- __volatile__ 禁止编译器优化
"mrs %0, cpsr @ local_save_flags" \ --汇编代码 其中%0代表第一个参数条目
: "=r" (x) \ --input 条目
: \ --output 条目
: "memory", "cc"); \ --篡改条目 "memory" 代表内存修改 "cc" 代表修改了condition registor标志位，如果修改r2寄存器则可以使用
--"r2"来标示r2寄存器的值被修改
})
如何去理解上段代码参见
http://blogold.chinaunix.net/u3/93290/showart_1933440.html，
该文中解释了如何在C语言中内嵌asm语句。在理解上面的内嵌asm后理解下面的就会简单很多。
设置cpsr寄存器的bit7位为1
#define raw_local_irq_disable() \
({ \
unsigned long temp; \
__asm__ __volatile__( \
"mrs %0, cpsr @ local_irq_disable\n" \
" orr %0, %0, #128\n" \
" msr cpsr_c, %0" \
: "=r" (temp) \
: \
: "memory", "cc"); \
})
设置cpsr寄存器的bit7位为0
#define raw_local_irq_enable() \
({ \
unsigned long temp; \
__asm__ __volatile__( \
"mrs %0, cpsr @ local_irq_enable\n" \
" bic %0, %0, #128\n" \
" msr cpsr_c, %0" \
: "=r" (temp) \
: \
: "memory", "cc"); \
})
将x值写入cpsr中
#define raw_local_irq_restore(x) \
__asm__ __volatile__( \
"msr cpsr_c, %0 @ local_irq_restore\n" \
: \
: "r" (x) \
: "memory", "cc")
二.原子量
对于原子量有两种操作，一个是基于整型，一个是基于原子。
对于原子操作的比较简单，这里就不描述了。这个也是类似于屏蔽中断加一个counter而组合成的。
以atomic_add_return为例：
static inline int atomic_add_return(int i, atomic_t *v)
{
unsigned long flags;
int val;
raw_local_irq_save(flags); --屏蔽中断
val = v->counter;
v->counter = val += i; --计数器值的增减
raw_local_irq_restore(flags);
return val;
}
三.自旋锁
spin lock为一个互斥设备，其状态只有"锁定"和"解锁"两种状态。在使用时如果锁可用，则"锁定"同时进入临界区，如果不可用则一直自旋，等待锁可用。
自旋锁是不睡眠的，可以在中断处理函数中使用。但是要合理使用自旋锁避免发生死锁的情况。
一般会在以下几种情况中发生死锁：
1.当程序运行时，已经获得一个锁，但是在临界区运行时发生中断，而在中断处理程序中仍需要获得锁，这种情况会发生死锁。由于在中断程序中自旋，而非中断程序无法释放锁，故发生死锁。
2.当程序运行时，已经获得一个锁，在临界区运行时却失去了CPU的控制权，被其他进程抢占了CPU，在其他进程中却需要获得锁，由于锁在前面程序中已经被获取抢占CPU的进程无法获得锁，而此时锁又无法被释放，造成当前进程死锁。在现在的spin lock中都会先禁止CPU被抢占，然后再获得锁。
3.进程获得自旋锁之后再阻塞，也是有可能导致死锁的发生。copy_from_user,copy_to_user以及kmalloc都有可能造成阻塞。
定义一个spin lock
spinlock_t lock
初始化自旋锁
spin_lock_init(lock);
获得自旋锁
spin_lock(lock); --能够获得自旋锁立即返回，获取不到则进程中自旋
spin_trylock(lock); --能够获得自旋锁则锁定并返回真，获取不到则立即返回假，不会自旋
释放自旋锁
spin_unlock(lock);
来分析自旋锁的实现：
初始化自旋锁：
spin_lock_init
这个函数其实是通过一系列的宏定义实现lock->lock = 0;
spin_lock则是通过嵌入asm实现lock->lock = 1;为了确保这个值被改变，使用了嵌入asm，而不是简单的赋值。赋值前主要是禁止抢占，这对于避免死锁的发生很重要。
void __lockfunc _spin_lock(spinlock_t *lock)
{
preempt_disable(); --禁止抢占
spin_acquire(&lock->dep_map, 0, 0, _RET_IP_); --这个具体不清楚干什么
LOCK_CONTENDED(lock, _raw_spin_trylock, _raw_spin_lock); --真正的lock，其实是一段内嵌的asm实现lock->lock = 1;即代表锁被获取。
}
spin_unlock与spin_lock的作用恰恰相反，主要是使能内核抢占，以及释放锁。在嵌入式中内核抢占一般是使能的，不会退化为空。
http://blogold.chinaunix.net/u/15071/showart_92863.html 是一篇讲述spin_lock不错的博客。
而关于其他自旋锁的变种，以后再看。
四.信号量
semaphore也是一种互斥的手段，其使用方式与spin lock类似，但是其使用的场合以及其实现方法与spin lock不同。spin lock 在获取不到资源时会一直自旋等待，
不会发生睡眠，除非是CPU被抢占切换，而semaphore则不会自旋等待而是睡眠，直到获得信号量。
下面来详细的看信号量的实现：
1.从信号量的成员来看，他其实是包括自旋锁，计数器和等待链表
struct semaphore {
spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};
2.信号量初始化
static inline void sema_init(struct semaphore *sem, int val)
{
static struct lock_class_key __key;
*sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val); --这个是实际的信号量成员初始化
lockdep_init_map(&sem->lock.dep_map, "semaphore->lock", &__key, 0); --这个不管
}
#define __SEMAPHORE_INITIALIZER(name, n) \
{ \
.lock = __SPIN_LOCK_UNLOCKED((name).lock), \ --初始化自旋锁为解锁状态
.count = n, \ --设置计数器
.wait_list = LIST_HEAD_INIT((name).wait_list), \ --初始化等待链表
}
#define init_MUTEX(sem) sema_init(sem, 1)
#define init_MUTEX_LOCKED(sem) sema_init(sem, 0)
3.获得信号量
获得信号量有几种不同的获得方式，只是退出的条件不同。
down
down的功能其实在该函数说明中就说明该函数过时了，请使用down_interruptible和down_killable代替。
down -> _down -> _down_common (TASK_UNINTERRUPTIBLE ，MAX_SCHEDULE_TIMEOUT)
down_interruptible -> _down_interruptible -> _down_common (TASK_INTERRUPTIBLE，MAX_SCHEDULE_TIMEOUT)
down_killable -> __down_killable -> _down_common (TASK_KILLABLE，MAX_SCHEDULE_TIMEOUT)
down_timeout -> __down_timeout -> _down_common (TASK_UNINTERRUPTIBLE，jiffies)
上述几个函数最终都会调用_down_common 只是在传替的state和timeout参数不同。
关于上面几种state的参数说明一下。
TASK_UNINTERRUPTIBLE 表明该进程不可以在睡眠时被中断，这样极有可能导致僵尸进程
TASK_INTERRUPTIBLE 表明该进程可以在睡眠中被中断，即被某些信号中断
TASK_KILLABLE 其为宏定义 #define TASK_KILLABLE (TASK_WAKEKILL | TASK_UNINTERRUPTIBLE) 表明不可以被普通信号中断，但是可以被kill
以down为例：
void down(struct semaphore *sem)
{
unsigned long flags;
spin_lock_irqsave(&sem->lock, flags); --锁定信号量自旋锁
if (likely(sem->count > 0)) --关于likely以前说过，有和没有对于条件没有影响，只是通知编译器条件满足的情况居多而已。
sem->count--;
else
__down(sem);
spin_unlock_irqrestore(&sem->lock, flags); --解锁
}
static noinline void __sched __down(struct semaphore *sem)
{
__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT); --传替的state与timeout值与其他几种不同。
}
下面是几种获得信号量公用的部分。
static inline int __sched __down_common(struct semaphore *sem, long state,
long timeout)
{
struct task_struct *task = current;
struct semaphore_waiter waiter;
list_add_tail(&waiter.list, &sem->wait_list);
waiter.task = task;
waiter.up = 0;
for (;;) {
if (signal_pending_state(state, task)) --判断是否有中断信号
goto interrupted;
if (timeout <= 0)
goto timed_out;
__set_task_state(task, state); --设置进程状态
spin_unlock_irq(&sem->lock); --解锁，这个十分重要，如果不解锁的话，极有可能在进程被切换后造成死锁。
timeout = schedule_timeout(timeout); --调度切换进程，睡眠等待该进程再次得到运行。
spin_lock_irq(&sem->lock); --重新锁定
if (waiter.up) --这个需要等up释放信号量
return 0;
}
timed_out:
list_del(&waiter.list);
return -ETIME;
interrupted:
list_del(&waiter.list);
return -EINTR;
}
来看up释放信号量
在另一个进程中调用up释放信号量
static noinline void __sched __up(struct semaphore *sem)
{
struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,struct semaphore_waiter, list); --这个很常见
list_del(&waiter->list);
waiter->up = 1; --释放标志
wake_up_process(waiter->task); --唤醒因为无法得到该信号量的进程。
}
可以将信号量与自旋锁进行比较：
很明显的可以看出信号量其实是进程级的，主要是用于多个进程之间对资源的互斥。其消耗的资源较多，在无法顺利得到信号量的情况下，会发生上下文的切换。信号量保护的临界区中可以包含可能引起阻塞的代码。而自旋锁则需要避免在临界区中包括引起阻塞的代码。
五.completion
completion其作用于semaphore类似。
1.completion的结构：
其包括完成标志符以及等待队列链表
struct __wait_queue_head {
spinlock_t lock;
struct list_head task_list;
};
struct completion {
unsigned int done;
wait_queue_head_t wait;
};
completion也是基于spin lock和等待队列的一种互斥。
2.初始化completion
static inline void init_completion(struct completion *x)
{
x->done = 0; --设置完成标志符为0
init_waitqueue_head(&x->wait); --初始化等待队列
}
3.等待completion
wait_for_completion -> wait_for_common (MAX_SCHEDULE_TIMEOUT, TASK_UNINTERRUPTIBLE)
wait_for_completion_interruptible -> wait_for_common (MAX_SCHEDULE_TIMEOUT, TASK_INTERRUPTIBLE)
wait_for_completion_killable -> wait_for_common ( MAX_SCHEDULE_TIMEOUT, TASK_KILLABLE)
wait_for_completion_timeout -> wait_for_common (timeout, TASK_UNINTERRUPTIBLE)
wait_for_completion_interruptible_timeout -> wait_for_common (timeout, TASK_INTERRUPTIBLE)
关于上面的几种等待完成的函数都会同semaphore一样最终调用同一个函数wait_for_common。只是传入参数不同
static long __sched wait_for_common(struct completion *x, long timeout, int state)
{
might_sleep(); --退化为空
spin_lock_irq(&x->wait.lock); --锁定
timeout = do_wait_for_common(x, timeout, state);
spin_unlock_irq(&x->wait.lock); --解锁
return timeout;
}
static inline long __sched do_wait_for_common(struct completion *x, long timeout, int state)
{
if (!x->done) {
DECLARE_WAITQUEUE(wait, current); --为当前进程创建一个等待队列
wait.flags |= WQ_FLAG_EXCLUSIVE; --将等待标识符设置为单独唤醒
__add_wait_queue_tail(&x->wait, &wait); --将competion中的等待队列添加入进程等待队列。
do {
if (signal_pending_state(state, current)) { --如果是信号唤醒
timeout = -ERESTARTSYS;
break;
}
__set_current_state(state); --设置当前进程的状态
spin_unlock_irq(&x->wait.lock); --自旋锁解锁很重要的。
timeout = schedule_timeout(timeout); --进程切换(在一定时间内)
spin_lock_irq(&x->wait.lock); --自旋锁锁定
} while (!x->done && timeout);
__remove_wait_queue(&x->wait, &wait); --从进程的等待队列中移除competion的等待队列
if (!x->done)
return timeout;
}
x->done--;
return timeout ?: 1;
}
对于try类型的获得互斥体，则是简单的查询互斥体中的计数器是否满足条件，无论满足与否都会返回值，不会在其中涉及任何关于进程的代码。
4.完成completion
完成completion
void complete(struct completion *x)
{
unsigned long flags;
spin_lock_irqsave(&x->wait.lock, flags);
x->done++;
__wake_up_common(&x->wait, TASK_NORMAL, 1, 0, NULL);
spin_unlock_irqrestore(&x->wait.lock, flags);
}
void __wake_up_common(wait_queue_head_t *q, unsigned int mode,int nr_exclusive, int sync, void *key)
{
wait_queue_t *curr, *next;
list_for_each_entry_safe(curr, next, &q->task_list, task_list) { --这个比较难理解。
unsigned flags = curr->flags;
if (curr->func(curr, mode, sync, key) &&(flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
break;
}
}
六.总结
在linux中主要是以上五种实现互斥的手段。
1.中断屏蔽实现上最简单，其主要是关闭CPU中断，使CPU屏蔽进程并发以及中断。如果临界区的代码能够短时间内执行完毕，这是一种可取的手段。但是长时间屏蔽中断，易造成数据丢失甚至系统崩溃。
2.原子量也是将临界区的代码屏蔽。原子量主要是用于临界区的代码简单，在使用锁机制不适合的情况下。比如仅仅是维护一个多进程公用的公共计数器count，在这种情况下，使用锁机制是不适合的，这种情况下，就可以使用原子量。
3.自旋锁，自旋锁其最主要的是不可以在临界区发生阻塞。其屏蔽了内核的抢占，同时维护一个锁。自旋锁可以使用在中断中。自旋锁是轻量级的。
4.信号量，信号量是在临界区时可以发生阻塞的，其在获得不了信号量的情况下会发生进程睡眠，同时让出CPU。其采用维护一个链表来实现的。信号量是不可以使用在中断中的。信号量是进程级的，由于需要发生进程切换，其耗费的资源较多。应该合理的使用自旋锁和信号量。
5.完成量，competion其同样会发生进程的切换，其实现基于等待队列完成。
在使用某种互斥的时候，应该考虑几点：1.临界区的大小 2.临界区的性质 3.竞争临界区的执行路径的数量。
5.1临界区的大小：
这个比较好理解，即临界区需要执行的时间。以多进程维护同一个计数器时，就可以使用原子量或者直接屏蔽中断。而使用锁机制则开销太大。
5.2临界区的性质
即临界区内的代码执行时是否会发生阻塞，中断等。
5.3竞争临界区的执行路径的数量
即在执行临界区代码时，会发生哪些可能会竞争的进程。
