分类： LINUX
2015-04-03 07:56:22
原文地址：(转)Linux内核的同步机制 作者：liebo
Linux内核的同步机制（一）：原子操作
原子操作：UP和SMP的异同
原子操作是不可分割的，在执行完毕不会被任何其它任务或事件中断。在单处理器系统（UniProcessor）中，能够在单条指令中完成的操作都可以认为是"原子操作"，因为中断只能发生于指令之间。这也是某些CPU指令系统中引入了test_and_set、test_and_clear等指令用于临界资源互斥的原因。但是，在对称多处理器（Symetric Multi-Processor）结构中就不同了，由于系统中有多个处理器在独立地运行，即使能在单条指令中完成的操作也有可能受到干扰。我们以decl（递减指令）为例，这是一个典型的"读－改－写"过程，涉及两次内存访问。设想在不同CPU运行的两个进程都在递减某个计数值，可能发生的情况是：
1． CPU A（上的进程，以下同）从内存单元把当前计数值（2）装载进它的寄存器中；
2． CPU B从内存单元把当前计数值（2）装载进它的寄存器中。
3． CPU A在它的寄存器中将计数值递减为1；
4． CPU B在它的寄存器中将计数值递减为1；
5． CPU A把修改后的计数值（1）写回内存单元。
6． CPU B把修改后的计数值（1）写回内存单元。
我们看到，内存里的计数值应该是0，然而它却是1。如果该计数值是一个共享资源的引用计数，每个进程都在递减后把该值与0进行比较，从而确定是否需要释放该共享资源。这时，两个进程都去掉了对该共享资源的引用，但没有一个进程能够释放它--两个进程都推断出：计数值是1，共享资源仍然在被使用。
原子性不可能由软件单独保证--必须需要硬件的支持，因此是和架构相关的。在x86平台上，CPU提供了在指令执行期间对总线加锁的手段。CPU芯片上有一条引线#HLOCK pin，如果汇编语言的程序中在一条指令前面加上前缀"LOCK"，经过汇编以后的机器代码就使CPU在执行这条指令的时候把#HLOCK pin的电位拉低，持续到这条指令结束时放开，从而把总线锁住，这样同一总线上别的CPU就暂时不能通过总线访问内存了，保证了这条指令在多处理器环境中的原子性。
Linux内核中的原子操作
Linux 2.4.21中，原子类型的定义和原子操作API都放在内核源码树的include/asm/atomic.h文件中，大部分使用汇编语言实现，因为c语言并不能实现这样的操作。
在x86的原子操作实现代码中，定义了LOCK宏，这个宏可以放在随后的内联汇编指令之前。如果是SMP，LOCK宏被扩展为lock指令；否则被定义为空--单CPU无需防止其它CPU的干扰，锁内存总线完全是在浪费时间。
#ifdef CONFIG_SMP
#define LOCK "lock ; "
#else
#define LOCK ""
#endif
typedef struct { volatile int counter; } atomic_t;
在所有支持的体系结构上原子类型atomic_t都保存一个int值。在x86的某些处理器上，由于工作方式的原因，原子类型能够保证的可用范围只有24位。volatile是一个类型描述符，要求编译器不要对其描述的对象作优化处理，对它的读写都需要从内存中访问。
#define ATOMIC_INIT(i) { (i) }
用于在定义原子变量时，初始化为指定的值。如：
static atomic_t count = ATOMIC_INIT(1);
#define atomic_read(v) ((v)->counter)
读取v指向的原子变量的值。由于该操作本身就是原子的，只需要一次内存访问就能完成，因此定义为一个宏，并用C代码实现。
#define atomic_set(v,i) (((v)->counter) = (i))
设置v指向的原子变量为i。由于该操作本身就是原子的，只需要一次内存访问就能完成，因此定义为一个宏，并用C代码实现。
static __inline__ void atomic_add(int i, atomic_t *v)
将v指向的原子变量加上i。该函数不关心原子变量的新值，返回void类型。在下面的实现中，使用了带有C/C++表达式的内联汇编代码，格式如下（参考《AT&T ASM Syntax》）：
__asm__ __volatile__("Instruction List" : Output : Input : Clobber/Modify);
__asm__ __volatile__指示编译器原封不动保留表达式中的汇编指令系列，不要考虑优化处理。涉及的约束还包括：
1. 等号约束（=）：只能用于输出操作表达式约束，说明括号内的左值表达式v->counter是write-only的。
2. 内存约束（m）：表示使用不需要借助寄存器，直接使用内存方式进行输入或输出。
3. 立即数约束（i）：表示输入表达式是一个立即数（整数），不需要借助任何寄存器。
4. 寄存器约束（r）：表示使用一个通用寄存器，由GCC在%eax/%ax/%al、%ebx/%bx/%bl、%ecx/%cx/%cl和%edx/%dx/%dl中选取一个合适的。
{
__asm__ __volatile__(
LOCK "addl %1,%0"
:"=m" (v->counter)
:"ir" (i), "m" (v->counter));
}
static __inline__ void atomic_sub(int i, atomic_t *v)
从v指向的原子变量减去i。
static __inline__ int atomic_sub_and_test(int i, atomic_t *v)
从v指向的原子变量减去i，并测试是否为0。若为0，返回真，否则返回假。由于x86的subl指令会在结果为0时设置CPU的zero标志位，而且这个标志位是CPU私有的，不会被其它CPU影响。因此，可以执行一次加锁的减操作，再根据CPU的zero标志位来设置本地变量c，并相应返回。
{
unsigned char c;
__asm__ __volatile__(
LOCK "subl %2,%0; sete %1"
:"=m" (v->counter), "=qm" (c)
:"ir" (i), "m" (v->counter) : "memory");
return c;
}
static __inline__ void atomic_inc(atomic_t *v)
递增v指向的原子变量。
static __inline__ void atomic_dec(atomic_t *v)
递减v指向的原子变量。
static __inline__ int atomic_dec_and_test(atomic_t *v)
递减v指向的原子变量，并测试是否为0。若为0，返回真，否则返回假。
static __inline__ int atomic_inc_and_test(atomic_t *v)
递增v指向的原子变量，并测试是否为0。若为0，返回真，否则返回假。
static __inline__ int atomic_add_negative(int i, atomic_t *v)
将v指向的原子变量加上i，并测试结果是否为负。若为负，返回真，否则返回假。这个操作用于实现semaphore。
原子操作的应用
在Linux内核中，原子操作被大量用于实现资源的引用计数。待补充...
参考文献：
1. Linux内核注释
2. LINUX内核源代码情景分析
label1:
flag = lock--;
if (flag >= 0) {
return;
} else {
do {
nop;
} while(lock <= 0);
goto label1;
}
}
当然，我们需要保证lock--为原子操作，就像在汇编中在decb指令前加上lock前缀一样。释放锁时使用spin_unlock_string宏，定义如下：
static inline void spin_unlock(spinlock_t *lock)
{
__asm__ __volatile__(
spin_unlock_string
);
}
#define spin_unlock_string \
"movb $1,%0" \
:"=m" (lock->lock) : : "memory"
需要强调的是，为了和spin_lock相配合，我们在释放锁是必须将lock设置为1，而不能使用lock++的方式。
#define SPIN_LOCK_UNLOCKED (spinlock_t) { 1 SPINLOCK_MAGIC_INIT }
#define spin_lock_init(x) do { *(x) = SPIN_LOCK_UNLOCKED; } while(0)
#define spin_is_locked(x) (*(volatile signed char *)(&(x)->lock) <= 0)
#define spin_unlock_wait(x) do { barrier(); } while(spin_is_locked(x))
由于spin_trylock不需要自旋，实现中采用xchgb指令（该指令将自动锁总线，而不需要再使用lock前缀）。若lock原来的值为1，说明未上锁，因此返回为真，表明成功获得锁；否则返回为假。
static inline int spin_trylock(spinlock_t *lock)
{
char oldval;
__asm__ __volatile__(
"xchgb %b0,%1"
:"=q" (oldval), "=m" (lock->lock)
:"0" (0) : "memory");
return oldval > 0;
}
自旋锁的应用
在讨论自旋锁的应用时，我们一般区分两种平台：单处理器非抢占式内核和对称多处理器或抢占式内核。在前面我们看到，在单处理器非抢占式内核下，自旋锁根本不存在。这体现了一种出色的设计策略，既然没有别人能够同时刻执行，就没有理由加锁。对于抢占式内核，我们将它等同于对称多处理器来考虑。
1． 用户上下文之间
如果数据结构只可能被用户上下文访问，最高效的办法就是使用信号量。（我们在后面将讨论信号量机制）。
2． 用户上下文与softirq之间
这种情况下，使用spin_lock_bh()/spin_unlock_bh()可以满足要求。如果是单处理器非抢占式内核，自旋锁消失了，spin_lock_bh等同于local_bh_disable，会进制在用户上下文时进制softirq，从而避免用户上下文和softirq同时进入临界区。如果是对称多处理器或者抢占式内核，即使是在不同CPU上的用户上下文和softirq同时运行，自旋锁机制保证了只有一个持有者，只有在它释放锁之后，另一个才能进入临界区。
3． 用户上下文和Tasklet/Timer之间
同上。同加锁观点来看，Tasklet和Timer的地位是同样的。
4． Tasklet或Timer之间
这里有两点需要说明：（1）同一时刻，一个Tasklet或Timer不会同时在两个CPU上执行；（2）如果CPU已经处在Tasklet或Timer中，它不会同时再执行其它的Tasklet或Timer。因此我们只需要考虑在不同CPU上运行两个不同Tasklet或Timer的情况，而这种情况只需要使用自旋锁机制，即spin_lock和spin_unlock函数。
5． Softirq之间或和Tasklet/Timer之间
同一个softirq可能在不同的CPU上执行，同上道理，使用spin_lock和spin_unlock可以在不同CPU的同一个或不同softirq，或者Softirq与Tasklet或Timer之间保护共享数据。
6． 硬件中断之间
如果被保护的共享资源在软中断（包括tasklet和timer）或进程上下文和硬中断上下文访问，那么在软中断或进程上下文访问期间，可能被硬中断打断，从而进入硬中断上下文对共享资源进行访问，因此，在进程或软中断上下文需要使用spin_lock_irq和spin_unlock_irq来保护对共享资源的访问。
而在中断处理句柄中使用什么版本，需依情况而定，如果只有一个中断处理句柄访问该共享资源，那么在中断处理句柄中仅需要spin_lock和spin_unlock来保护对共享资源的访问就可以了。
因为在执行中断处理句柄期间，不可能被同一CPU上的软中断或进程打断。但是如果有不同的中断处理句柄访问该共享资源，那么需要在中断处理句柄中使用spin_lock_irq和spin_unlock_irq来保护对共享资源的访问。
在使用spin_lock_irq和spin_unlock_irq的情况下，完全可以用spin_lock_irqsave和spin_unlock_irqrestore取代，具体应该使用哪一个也需要依情况而定，如果可以确信在对共享资源访问前中断是打开的，那么使用spin_lock_irq更好一些，因为它比spin_lock_irqsave要快一些。
但是如果不能确定是否中断使能，那么使用spin_lock_irqsave和spin_unlock_irqrestore更好，因为它将恢复访问共享资源前的中断标志而不是直接打开中断。 当然，有些情况下需要在访问共享资源时必须禁止中断，而访问完后必须打开中断，这样的情形使用spin_lock_irq和spin_unlock_irq最好。
在这个链表中，有两种数据结构：等待队列头（wait_queue_head_t）和等待队列项（wait_queue_t）。等待队列头和等待队列项中都包含一个list_head类型的域作为"连接件"。由于我们只需要对队列进行添加和删除操作，并不会修改其中的对象（等待队列项），因此，我们只需要提供一把保护整个基础设施和所有对象的锁，这把锁保存在等待队列头中，为wq_lock_t类型。在实现中，可以支持读写锁（rwlock）或自旋锁（spinlock）两种类型，通过一个宏定义来切换。如果使用读写锁，将wq_lock_t定义为rwlock_t类型；如果是自旋锁，将wq_lock_t定义为spinlock_t类型。无论哪种情况，分别相应设置wq_read_lock、wq_read_unlock、wq_read_lock_irqsave、wq_read_unlock_irqrestore、wq_write_lock_irq、wq_write_unlock、wq_write_lock_irqsave和wq_write_unlock_irqrestore等宏。
等待队列头
struct __wait_queue_head {
wq_lock_t lock;
struct list_head task_list;
};
typedef struct __wait_queue_head wait_queue_head_t;
前面已经说过，等待队列的主体是进程，这反映在每个等待队列项中，是一个任务结构指针（struct task_struct * task）。flags为该进程的等待标志，当前只支持互斥。
等待队列项
struct __wait_queue {
unsigned int flags;
#define WQ_FLAG_EXCLUSIVE 0x01
struct task_struct * task;
struct list_head task_list;
};
typedef struct __wait_queue wait_queue_t;
声明和初始化
#define DECLARE_WAITQUEUE(name, tsk) \
wait_queue_t name = __WAITQUEUE_INITIALIZER(name, tsk)
#define __WAITQUEUE_INITIALIZER(name, tsk) { \
task: tsk, \
task_list: { NULL, NULL }, \
__WAITQUEUE_DEBUG_INIT(name)}
通过DECLARE_WAITQUEUE宏将等待队列项初始化成对应的任务结构，并且用于连接的相关指针均设置为空。其中加入了调试相关代码。
#define DECLARE_WAIT_QUEUE_HEAD(name) \
wait_queue_head_t name = __WAIT_QUEUE_HEAD_INITIALIZER(name)
#define __WAIT_QUEUE_HEAD_INITIALIZER(name) { \
lock: WAITQUEUE_RW_LOCK_UNLOCKED, \
task_list: { &(name).task_list, &(name).task_list }, \
__WAITQUEUE_HEAD_DEBUG_INIT(name)}
通过DECLARE_WAIT_QUEUE_HEAD宏初始化一个等待队列头，使得其所在链表为空，并设置链表为"未上锁"状态。其中加入了调试相关代码。
static inline void init_waitqueue_head(wait_queue_head_t *q)
该函数初始化一个已经存在的等待队列头，它将整个队列设置为"未上锁"状态，并将链表指针prev和next指向它自身。
{
q->lock = WAITQUEUE_RW_LOCK_UNLOCKED;
INIT_LIST_HEAD(&q->task_list);
}
static inline void init_waitqueue_entry(wait_queue_t *q, struct task_struct *p)
该函数初始化一个已经存在的等待队列项，它设置对应的任务结构，同时将标志位清0。
{
q->flags = 0;
q->task = p;
}
static inline int waitqueue_active(wait_queue_head_t *q)
该函数检查等待队列是否为空。
{
return !list_empty(&q->task_list);
}
static inline void __add_wait_queue(wait_queue_head_t *head, wait_queue_t *new)
将指定的等待队列项new添加到等待队列头head所在的链表头部，该函数假设已经获得锁。
{
list_add(&new->task_list, &head->task_list);
}
static inline void __add_wait_queue_tail(wait_queue_head_t *head, wait_queue_t *new)
将指定的等待队列项new添加到等待队列头head所在的链表尾部，该函数假设已经获得锁。
{
list_add_tail(&new->task_list, &head->task_list);
}
static inline void __remove_wait_queue(wait_queue_head_t *head, wait_queue_t *old)
将函数从等待队列头head所在的链表中删除指定等待队列项old，该函数假设已经获得锁，并且old在head所在链表中。
{
list_del(&old->task_list);
}
睡眠和唤醒操作
对等待队列的操作包括睡眠和唤醒（相关函数保存在源代码树的/kernel/sched.c和include/linux/sched.h中）。思想是更改当前进程（CURRENT）的任务状态，并要求重新调度，因为这时这个进程的状态已经改变，不再在调度表的就绪队列中，因此无法再获得执行机会，进入"睡眠"状态，直至被"唤醒"，即其任务状态重新被修改回就绪态。
常用的睡眠操作有interruptible_sleep_on和sleep_on。两个函数类似，只不过前者将进程的状态从就绪态（TASK_RUNNING）设置为TASK_INTERRUPTIBLE，允许通过发送signal唤醒它（即可中断的睡眠状态）；而后者将进程的状态设置为TASK_UNINTERRUPTIBLE，在这种状态下，不接收任何singal。以interruptible_sleep_on为例，其展开后的代码是：
void interruptible_sleep_on(wait_queue_head_t *q)
{
unsigned long flags;
wait_queue_t wait;
/* 构造当前进程对应的等待队列项 */
init_waitqueue_entry(&wait, current);
/* 将当前进程的状态从TASK_RUNNING改为TASK_INTERRUPTIBLE */
current->state = TASK_INTERRUPTIBLE;
/* 将等待队列项添加到指定链表中 */
wq_write_lock_irqsave(&q->lock,flags);
__add_wait_queue(q, &wait);
wq_write_unlock(&q->lock);
/* 进程重新调度，放弃执行权 */
schedule();
/* 本进程被唤醒，重新获得执行权，首要之事是将等待队列项从链表中删除 */
wq_write_lock_irq(&q->lock);
__remove_wait_queue(q, &wait);
wq_write_unlock_irqrestore(&q->lock,flags);
/* 至此，等待过程结束，本进程可以正常执行下面的逻辑 */
}
对应的唤醒操作包括wake_up_interruptible和wake_up。wake_up函数不仅可以唤醒状态为TASK_UNINTERRUPTIBLE的进程，而且可以唤醒状态为TASK_INTERRUPTIBLE的进程。wake_up_interruptible只负责唤醒状态为TASK_INTERRUPTIBLE的进程。这两个宏的定义如下：
#define wake_up(x) __wake_up((x),TASK_UNINTERRUPTIBLE | TASK_INTERRUPTIBLE, 1)
#define wake_up_interruptible(x) __wake_up((x),TASK_INTERRUPTIBLE, 1)
__wake_up函数主要是获取队列操作的锁，具体工作是调用__wake_up_common完成的。
void __wake_up(wait_queue_head_t *q, unsigned int mode, int nr)
{
if (q) {
unsigned long flags;
wq_read_lock_irqsave(&q->lock, flags);
__wake_up_common(q, mode, nr, 0);
wq_read_unlock_irqrestore(&q->lock, flags);
}
}
/* The core wakeup function. Non-exclusive wakeups (nr_exclusive == 0) just wake everything up. If it's an exclusive wakeup (nr_exclusive == small +ve number) then we wake all the non-exclusive tasks and one exclusive task.
There are circumstances in which we can try to wake a task which has already started to run but is not in state TASK_RUNNING. try_to_wake_up() returns zero in this (rare) case, and we handle it by contonuing to scan the queue. */
static inline void __wake_up_common (wait_queue_head_t *q, unsigned int mode, int nr_exclusive, const int sync)
参数q表示要操作的等待队列，mode表示要唤醒任务的状态，如TASK_UNINTERRUPTIBLE或TASK_INTERRUPTIBLE等。nr_exclusive是要唤醒的互斥进程数目，在这之前遇到的非互斥进程将被无条件唤醒。sync表示？？？
{
struct list_head *tmp;
struct task_struct *p;
CHECK_MAGIC_WQHEAD(q);
WQ_CHECK_LIST_HEAD(&q->task_list);
/* 遍历等待队列 */
list_for_each(tmp,&q->task_list) {
unsigned int state;
/* 获得当前等待队列项 */
wait_queue_t *curr = list_entry(tmp, wait_queue_t, task_list);
CHECK_MAGIC(curr->__magic);
/* 获得对应的进程 */
p = curr->task;
state = p->state;
/* 如果我们需要处理这种状态的进程 */
if (state & mode) {
WQ_NOTE_WAKER(curr);
if (try_to_wake_up(p, sync) && (curr->flags&WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
break;
}
}
}
/* 唤醒一个进程，将它放到运行队列中，如果它还不在运行队列的话。"当前"进程总是在运行队列中的（except when the actual re-schedule is in progress)，and as such you're allowed to do the simpler "current->state = TASK_RUNNING" to mark yourself runnable without the overhead of this. */
static inline int try_to_wake_up(struct task_struct * p, int synchronous)
{
unsigned long flags;
int success = 0;
/* 由于我们需要操作运行队列，必须获得对应的锁 */
spin_lock_irqsave(&runqueue_lock, flags);
/* 将进程状态设置为TASK_RUNNING */
p->state = TASK_RUNNING;
/* 如果进程已经在运行队列中，释放锁退出 */
if (task_on_runqueue(p))
goto out;
/* 否则将进程添加到运行队列中 */
add_to_runqueue(p);
/* 如果设置了同步标志 */
if (!synchronous || !(p->cpus_allowed & (1UL << smp_processor_id())))
reschedule_idle(p);
/* 唤醒成功，释放锁退出 */
success = 1;
out:
spin_unlock_irqrestore(&runqueue_lock, flags);
return success;
}
等待队列应用模式
等待队列的的应用涉及两个进程，假设为A和B。A是资源的消费者，B是资源的生产者。A在消费的时候必须确保资源已经生产出来，为此定义一个资源等待队列。这个队列同时要被进程A和进程B使用，我们可以将它定义为一个全局变量。
DECLARE_WAIT_QUEUE_HEAD(rsc_queue); /* 全局变量 */
在进程A中，执行逻辑如下：
while (resource is unavaiable) {
interruptible_sleep_on( &wq );
}
consume_resource();
在进程B中，执行逻辑如下：
produce_resource();
wake_up_interruptible( &wq );
