Linux 下的同步机制

Posted 2020-09-08 太初有道，道与神同在，道就是神……

2017-03-10

---

 回想下最初的计算机设计，在单个CPU的情况下，同一时刻只能由一个线程（在LInux下为进程）占用CPU，且2.6之前的Linux内核并不支持内核抢占，当进程在系统地址运行时，能打断当前操作的只有中断，而中断处理完成后发现之前的状态是在内核，就不触发地调度，只有在返回用户空间时，才会触发调度。所以内核中的共享资源在单个CPU的情况下其实不需要考虑同步机制，尽管表面上看起来是多个进程在同时运行，其实那只是调度器以很小的时间粒度，调度各个进程运行的结果，事实上是一个伪并行。但是随着时代的发展，单个处理器根本满足不了人们对性能的需求，多处理器架构才应运而生。这种情况下，多个处理器之间的工作互不干扰，可实现真正的并行。

　　但是操作系统只有一个，其中不乏很多全局共享的变量，即使是多CPU也不能同时对其进程操作。然而在多处理器情况下，如果我们不加以防护措施，极有可能两个进程同时对同一变量进行访问，这样就容易造成数据的不同步。这种情况是开发者和用户都无法忍受的。况且，在2.6之后的内核启用了内核抢占，即使进程运行在系统地址空间也有可能被抢占，基于此，内核同步机制便被提出来。

内核中的同步机制又很多，具体由原子操作、信号量、自旋锁、读写者锁，RCU机制等。每种方案都有其优缺点，且适用于不同的应用场景。

原子操作

原子操作在内核中主要保护某个共享变量，防止该变量被同时访问造成数据不同步问题。为此，内核中定义了一系列的API，在内核中定义了atomic_t数据类型，其定义的数据操作都像是一条汇编指令执行，中间不会被中断。atomic_t定义的数据类型和标准数据类型int/short等不兼容，数据的加减不能通过标准运算符，必须通过其本身的API，下面是一些该类型操作的API

static __inline__ void atomic_add(int i, atomic_t * v)
static __inline__ void atomic_sub(int i, atomic_t * v)
static inline int atomic_add_return(int i, atomic_t *v)
static __inline__ long atomic_sub_return(int i, atomic_t * v)

 

基于上面的基础API，还实现了其他的API，这里就不在列举。

信号量

信号量一般实现互斥操作，但是可以指定处于临界区的进程数目，当规定数目为1时，表示此为互斥信号量。信号量在内核中的结构如下

struct semaphore {
    raw_spinlock_t        lock;
    unsigned int        count;
    struct list_head    wait_list;
};

 

开头是一个自旋锁，用以保护该数据结构的操作，count指定了信号量关联的资源允许同时访问的进程数目，wait_list是等待访问资源的进程链表。和自旋锁相比，信号量的一个好处允许等待的进程睡眠，而不是一直在轮询请求。所以信号量比较适合于较长的临界区。信号量操作很简单，初始初始化一个信号量，在临界资源前需要down操作以请求获得信号量，执行完毕执行up操作释放资源。

相关代码如下

void down(struct semaphore *sem)
{
    unsigned long flags;

    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(sem->count > 0))
        sem->count--;
    else
        __down(sem);
    raw_spin_unlock_irqrestore(&sem->lock, flags);
}

 

 

void up(struct semaphore *sem)
{
    unsigned long flags;

    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(list_empty(&sem->wait_list)))
        sem->count++;
    else
        __up(sem);
    raw_spin_unlock_irqrestore(&sem->lock, flags);
}

 

对于down操作，首先获取信号量结构的自旋锁，并会关闭当前CPU的中断，然后如果count还大于0，则直接分配资源，count--，否则调用down函数阻塞当前进程，down函数中直接调用了down_common函数。

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 = false;

    for (;;) {
        if (signal_pending_state(state, task))
            goto interrupted;
        if (unlikely(timeout <= 0))
            goto timed_out;
        __set_task_state(task, state);
        raw_spin_unlock_irq(&sem->lock);
        timeout = schedule_timeout(timeout);
        raw_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结构，插入到信号量结构的等待进程链表中。timeout是一个超时时间，当设置为小于等于0时表示不在此等待资源。通过这些检查后，设置当前进程为TASK_INTERRUPTIBLE状态，表示可被中断唤醒的阻塞。然后开启本地中断表示当前任务告一段落，下面要调用schedule_timeout进程调度。在具体切换进程后，下半部分的代码就是下次被调度的时候执行了。

而对于up操作，首先获取自旋锁，如果当前等待队列为空，则单纯的增加count表示可用资源增加，否则执行_up操作，该函数实现比较简单。首先从等待链表中移除对应节点，设置结构的up信号为true，然后调用wake_up_process函数唤醒执行进程。这样唤醒是吧进程加入就绪链表中，可以被调度器正常调度。

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 = true;
    wake_up_process(waiter->task);
}

 

自旋锁

自旋锁恐怕是内核中应用最为广泛的同步机制了，在内核中表现为两个功用：

1、对于数据结构或者变量的保护

2、对于临界区代码的保护

 对于自旋锁的操作很简单，其结构spinlock_t，对于自旋锁的操作，根据对临界区的不会要求级别，有多种API可以选择

static inline void spin_lock(spinlock_t *lock)
static inline void spin_unlock(spinlock_t *lock)
static inline void spin_lock_bh(spinlock_t *lock)
static inline void spin_unlock_bh(spinlock_t *lock)
static inline void spin_lock_irq(spinlock_t *lock)
static inline void spin_unlock_irq(spinlock_t *lock)

 

前面最基础的还是spin_lock，用以获取自旋锁，在具体获取之前会调用preempt_disable禁止内核抢占，所以自旋锁保护的临界代码执行期间会不会被调度。本局临界代码的性质，可以调用spin_lock_bh禁止软中断或者通过调用spin_lock_irq禁止本地CPU的中断。有自旋锁保护的代码不能进入睡眠状态，因为等待获取锁的CPU会一直轮询，不做其他事情，如果在临界区内睡眠，则对CPU性能耗能较大。

通过上面函数获取锁和释放锁主要用于对临界代码的保护，操作本身是一个原子操作。

对于数据结构的保护，自旋锁往往作为一个字段嵌入到数据结构中，在操作具体的结构之前，需要获取锁，操作完毕释放锁。

 

读写者锁

 读写者问题其实就是针对读写操作分别做的处理，可以看到其他的同步机制没有区分读写操作，只要是线程访问，就需要加锁，但是很多资源在不是写操作的情况下，是可以允许多进程访问的。因此为了提高效率，读写者锁就应运而生。读写者锁在执行写操作时，需要加writelock，此时只有一个线程可以进入临界区，而在执行读操作时，加readlock，此时可以允许多个线程进入临界区。适用于读操作明显多于写操作的临界区。

RCU机制

RCU机制是一种较新的内核同步机制，可以提供两种类型的保护：对数据结构和对链表。在内核中应用的相当频繁。

RCU机制使用条件：

• 对共享资源的访问大部分时间是只读的，写操作相对较少。
• 在RCU保护的代码范围内，不能进入睡眠。
• 受保护资源必须通过指针访问。

RCU保护的数据结构，不能反引用其指针，即不能*ptr获取其内容，必须使用其对应的API。同时反引用指针并使用其结果的代码，必须使用rcu_read_lock()和rcu_read_unlock()保护起来。

如果要修改ptr指向的对象，需要先创建一个副本，然后调用rcu_assign_pointer(ptr,new_ptr)进行修改。所以这种情况，受保护的数据结构允许读写并发执行，因为实质上是操作两个结构，只有在对旧的数据结构访问完成后，才会修改指针指向。

 内存和优化屏障

在看内核源码的时候经常看见有barrier()的出现，相当于一堵墙，让编译器在处理完屏障之前的代码之前，不会处理屏障后面的代码。原来为了提高代码的执行效率，编译器都会适当的对代码进行指令重排，一般情况下这种重排不会影响程序功能，但是编译器毕竟不是人，某些对顺序有严格要求的代码，很可能无法被编译器准确识别，比如关闭和启用抢占的代码，这样，如果编译器把核心代码移出关闭抢占区间，那么很可能影响最终结果，因此，这种时候在关闭抢占后应该加上内存屏障，保障不会把后面的代码排到前面来。