内核同步措施

内核同步措施

1.内核同步的引入

假设一种情况，只有一个cpu并且cpu处理任务是串行的，也就是严格地执行完一个再执行下一个。这种情况下内核中的共享数据不需要特殊的规则也能达到我们预期。然而这种情况肯定是不能满足我们对计算机的要求，因为它太慢而且如果有一个任务出问题了那么后面任务将无法执行。现实的情况是多CPU并行处理任务，即使是单核cpu，任务也是可调度的。那么对于共享数据的操作就必须遵循一定的访问规则。否则能会出现意想不到的错误。避免并发带来的竞争称为同步。下面介绍几种内核同步措施。

2.原子操作

原子操作可以保证指令以原子的方式被执行，也就是执行过程不能被打断。linux内核提供一种原子类型atomic_t。

typedef struct {

volatile int counter;

} atomic_t;

并为这种类型提供了一些专门的操作，常见操作如下： 

atomic_set(v,i)：把*v置为i 

atomic_add(i,v)：把*v增加i 

atomic_inc(v)：把*v加1 

atomic_add(i,v)操作：

static inline void atomic_add(int i, atomic_t *v)

{

asm volatile(LOCK_PREFIX "addl %1,%0"

: "=m" (v->counter)

: "ir" (i), "m" (v->counter));

}

其中LOCK_PREFIX保证了操作的原子性。

3.自旋锁

自旋锁的实现

typedef struct raw_spinlock {

unsigned int slock;

} raw_spinlock_t;

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;

其实最简单的自旋锁的实现只需要一个字段，就是volatile unsigned int lock 

自旋锁的使用 

定义一个自旋锁：DEFINE_SPINLOCK(mr_lock); 

获取自旋锁：spin_lock(&mr_lock); 

释放自旋锁：spin_unlock(&mr_lock); 

自旋锁是专门为防止多处理器并行而引入的一种锁，多用于中断处理，自旋锁最多只能被一个内核任务持有。由于当自旋锁被持有时，等待它的任务进行自旋，所以自旋锁被持有的时间不能太久，如果需要长时间锁定可以用信号量。 

如果是开中断的情况下使用自旋锁要特别小心，因为中断可能随时会中断正在运行的程序，如果中断打断的是一段自旋锁保护的代码，那么其他申请同一自旋锁的cpu停顿。 

所以一般申请自旋锁都得关中断，内核提供spin_lock_irqsave和spin_unlock_irqrestore两个函数用于关中断并使用自旋锁

4.信号量

信号量也是一种锁，与自旋锁不同的地方是：当资源不可用时，信号量会将进程挂起，而自旋锁则让等待着忙等（特别浪费处理器时间），一般自旋锁用于中断上下文，信号量用于进程上下文。

4.1信号量的定义：

struct semaphore {

spinlock_t  lock;

unsigned int    count;

struct list_head   wait_list;

};

lock：自旋锁，为了防止多处理器并行造成错误。 

count：如果count大于0，那么表示资源空闲可用，如果count等于0，表示信号量正被使用，但没有其他进程等待，如果count小于0则表示有其他进程等待信号量。 

wait_list：等待信号量的进程链表。

4.2信号量的使用：

获取信号量函数：down()和down_interruptible()，down函数获取信号量时，如果信号量不可用则进程进入睡眠等待并且不可被信号中断。而使用down_interruptible进程在等待过程中可以被信号打断

void down(struct semaphore *sem)

{

unsigned long flags;

spin_lock_irqsave(&sem->lock, flags);//加锁，使信号量的操作在关闭中断下

if (likely(sem->count > 0))  //信号量是否空闲可用，如果可用则count减1

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

}

TASK_UNINTERRUPTIBLE表示进程处于不可中断状态，MAX_SCHEDULE_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)) /*如果是TASK_INTERRUPTIBLE或TASK_WAKEKILL状态，并且有信号将进程唤醒，则将当前进程从等待队列中取出并退出*/

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) /*进程被唤醒了，可能是睡眠时间已经到达，或者获得了信号量，或者被信号唤醒了,所以要进行判断若是被其他进程释放信号量而唤醒的，则返回。*/

return 0;

}

timed_out://进程获取信号量等待超时时返回

list_del(&waiter.list);

return -ETIME;

interrupted://进程等待获取信号量时被信号中断

list_del(&waiter.list);

return -EINTR;

}

semaphore_waiter结构体：

struct semaphore_waiter {

struct list_head list;

struct task_struct *task;

int up;

};

获取信号量函数还有 

down_killable：获取信号量，但可被致命信号唤醒 

down_trylock：获取信号量，如果信号量不可用，则返回失败而不等待 

down_timeout：获取信号量，如果在指定的时间内没有被信号量唤醒，则返回失败 

释放信号量函数up：

void up(struct semaphore *sem)

{

unsigned long flags;

spin_lock_irqsave(&sem->lock, flags); // 获得保护信号量的自旋锁

if (likely(list_empty(&sem->wait_list))) //没有进程等待此信号量

sem->count++;

else

__up(sem); //唤醒等待链表中的进程

spin_unlock_irqrestore(&sem->lock, flags);//释放自旋锁

}

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; //设置up为1，表示进程是被信号量唤醒的，而不是信号

wake_up_process(waiter->task); //唤醒进程

}

4.3读写信号量

读写信号量类似于信号量，但它可以允许有多个读者，在不能获得信号量时，当前进程会调度出去。

struct rw_semaphore {

__s32   activity;

spinlock_t  wait_lock;

struct list_head   wait_list;

#ifdef CONFIG_DEBUG_LOCK_ALLOC

struct lockdep_map dep_map;

#endif

};

activity：如果activity为0，那么没有读者或写者， 如果activity>0，表示当前的并发读者数。 如果activity为-1，表示有一个写者。 

wait_lock：自旋锁 

wait_list：等待信号量的等待队列链表。 

获取读信号量函数down_read：

void __sched down_read(struct rw_semaphore *sem)

{

might_sleep();

rwsem_acquire_read(&sem->dep_map, 0, 0, _RET_IP_);

LOCK_CONTENDED(sem, __down_read_trylock, __down_read);

}

这个函数没看懂，不过它是调用 __down_read完成加读者的具体操作

void __sched __down_read(struct rw_semaphore *sem)

{

struct rwsem_waiter waiter;

struct task_struct *tsk;

spin_lock_irq(&sem->wait_lock);

//加锁并关中断

if (sem->activity >= 0 && list_empty(&sem->wait_list)) {

//如果有0或者多个读者且等待队列为空获得读写信号量

/* granted */

sem->activity++;//读者数加1

spin_unlock_irq(&sem->wait_lock);//解锁

goto out;

}

//若不能获得sem,将当前进程加入等待队列

tsk = current;

set_task_state(tsk, TASK_UNINTERRUPTIBLE);

/* set up my own style of waitqueue */

waiter.task = tsk;

waiter.flags = RWSEM_WAITING_FOR_READ;//表示读写等待操作

get_task_struct(tsk);//进程使用计数加1

list_add_tail(&waiter.list, &sem->wait_list);

/* we don't need to touch the semaphore struct anymore */

spin_unlock_irq(&sem->wait_lock);

/* wait to be given the lock */

for (;;) {//读着等待sem

if (!waiter.task)

break;

schedule();//调度出去，并等待被唤醒

set_task_state(tsk, TASK_UNINTERRUPTIBLE);

}

tsk->state = TASK_RUNNING; //获得sem

out:

;

}

释放读信号量函数up_read：

void up_read(struct rw_semaphore *sem)

{

rwsem_release(&sem->dep_map, 1, _RET_IP_);

__up_read(sem);

}

__up_read：

void __up_read(struct rw_semaphore *sem)

{

unsigned long flags;

spin_lock_irqsave(&sem->wait_lock, flags);

if (--sem->activity == 0 && !list_empty(&sem->wait_list))

//如果没有读者持有信号且等待队列有写者则唤醒等待的写者

sem = __rwsem_wake_one_writer(sem);//只唤醒第1个写者

spin_unlock_irqrestore(&sem->wait_lock, flags);

}

static inline struct rw_semaphore *

__rwsem_wake_one_writer(struct rw_semaphore *sem)

{

struct rwsem_waiter *waiter;

struct task_struct *tsk;

sem->activity = -1;//将activity置-1，标志写者持有sem

waiter = list_entry(sem->wait_list.next, struct rwsem_waiter, list);//获得第一个写者

list_del(&waiter->list);//将其从队列中删除

tsk = waiter->task;

smp_mb();

waiter->task = NULL;

wake_up_process(tsk);//唤醒进程

put_task_struct(tsk);

return sem;

}

释放与获得写信号量操作与读信号量相似 

获得写信号量：

void __sched down_write(struct rw_semaphore *sem)

{

might_sleep();

rwsem_acquire(&sem->dep_map, 0, 0, _RET_IP_);

LOCK_CONTENDED(sem, __down_write_trylock, __down_write);

}

static inline void __down_write(struct rw_semaphore *sem)

{

__down_write_nested(sem, 0);

}

void __sched __down_write_nested(struct rw_semaphore *sem, int subclass)

{

struct rwsem_waiter waiter;

struct task_struct *tsk;

spin_lock_irq(&sem->wait_lock);

if (sem->activity == 0 && list_empty(&sem->wait_list)) {   //没有读者和写者，并且没有其他处于等待状态的读者和写者

/* granted */

 sem->activity = -1;//将activity置-1，标志写者持有sem

spin_unlock_irq(&sem->wait_lock);

goto out;

}

//运行到这里，说明有读者或者写者，将当前任务挂起

tsk = current;

set_task_state(tsk, TASK_UNINTERRUPTIBLE);

/* set up my own style of waitqueue */

waiter.task = tsk;

waiter.flags = RWSEM_WAITING_FOR_WRITE;/*这个标志表示等待者是一个写者*/

get_task_struct(tsk);

list_add_tail(&waiter.list, &sem->wait_list);

/* we don't need to touch the semaphore struct anymore */

spin_unlock_irq(&sem->wait_lock);

/* wait to be given the lock */

for (;;) {

if (!waiter.task)

break;

schedule();

set_task_state(tsk, TASK_UNINTERRUPTIBLE);

}

tsk->state = TASK_RUNNING;

out:

;

}

void up_write(struct rw_semaphore *sem)

{

rwsem_release(&sem->dep_map, 1, _RET_IP_);

__up_write(sem);

}

void __up_write(struct rw_semaphore *sem)

{

unsigned long flags;

spin_lock_irqsave(&sem->wait_lock, flags);

sem->activity = 0;

/*因为仅仅只能有一个写者，也不能有读者能够与写者同在，因此写者释放信号量时，可以将activity设置为0，表示没有写者和读者了*/

if (!list_empty(&sem->wait_list))

sem = __rwsem_do_wake(sem, 1);

spin_unlock_irqrestore(&sem->wait_lock, flags);

}