实践Linux的理论
全部博文(61)
分类: LINUX
2014-04-28 15:18:16
原文地址:(转)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 );