一文搞懂 | Linux 同步管理（上）

秒速五厘米 2022-09-15 12:35 187阅读 0赞

因为现代操作系统是多处理器计算的架构，必然更容易遇到多个进程，多个线程访问共享数据的情况，如下图所示：

![3eddc62b2f0c5a4851a9df73753628e3.png][]

图中每一种颜色代表一种竞态情况，主要归结为三类：

1.  进程与进程之间：单核上的抢占，多核上的SMP；
2.  进程与中断之间：中断又包含了上半部与下半部，中断总是能打断进程的执行流；
3.  中断与中断之间：外设的中断可以路由到不同的CPU上，它们之间也可能带来竞态；

这时候就需要一种同步机制来保护并发访问的内存数据。本系列文章分为两部分，这一章主要讨论原子操作，自旋锁，信号量和互斥锁

## 原子操作 ##

原子操作是在执行结束前不可打断的操作，也是最小的执行单位。以 arm 平台为例，原子操作的 API 包括如下:

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>int atomic_read(atomic_t *v)</td> 
   <td>读操作</td> 
  </tr> 
  <tr> 
   <td>void atomic_set(atomic_t *v, int i)</td> 
   <td>设置变量</td> 
  </tr> 
  <tr> 
   <td>void atomic_add(int i, atomic_t *v)</td> 
   <td>增加 i</td> 
  </tr> 
  <tr> 
   <td>void atomic_sub(int i, atomic_t *v)</td> 
   <td>减少 i</td> 
  </tr> 
  <tr> 
   <td>void atomic_inc(atomic_t *v)</td> 
   <td>增加 1</td> 
  </tr> 
  <tr> 
   <td>void atomic_dec(atomic_t *v)</td> 
   <td>减少 1</td> 
  </tr> 
  <tr> 
   <td>void atomic_inc_and_test(atomic_t *v)</td> 
   <td>加 1 是否为 0</td> 
  </tr> 
  <tr> 
   <td>void atomic_dec_and_test(atomic_t *v)</td> 
   <td>减 1 是否为 0</td> 
  </tr> 
  <tr> 
   <td>void atomic_add_negative(int i, atomic_t *v)</td> 
   <td>加 i 是否为负</td> 
  </tr> 
  <tr> 
   <td>void atomic_add_return(int i, atomic_t *v)</td> 
   <td>增加 i 返回结果</td> 
  </tr> 
  <tr> 
   <td>void atomic_sub_return(int i, atomic_t *v)</td> 
   <td>减少 i 返回结果</td> 
  </tr> 
  <tr> 
   <td>void atomic_inc_return(int i, atomic_t *v)</td> 
   <td>加 1 返回</td> 
  </tr> 
  <tr> 
   <td>void atomic_dec_return(int i, atomic_t *v)</td> 
   <td>减 1 返回</td> 
  </tr> 
 </tbody> 
</table>

原子操作通常是内联函数，往往是通过内嵌汇编指令来实现的，如果某个函数本身就是原子的，它往往被定义成一个宏，以下为例。

#define ATOMIC_OP(op, c_op, asm_op)     \
    static inline void atomic_##op(int i, atomic_t *v)   \
    {         \
     unsigned long tmp;      \
     int result;       \
             \
     prefetchw(&v->counter);      \
     __asm__ __volatile__("@ atomic_" #op "\n"   \
    "1: ldrex %0, [%3]\n"      \
    " " #asm_op " %0, %0, %4\n"     \
    " strex %1, %0, [%3]\n"      \
    " teq %1, #0\n"      \
    " bne 1b"       \
     : "=&r" (result), "=&r" (tmp), "+Qo" (v->counter)  \
     : "r" (&v->counter), "Ir" (i)     \
     : "cc");       \
    }

可见原子操作的原子性依赖于 ldrex 与 strex 实现，ldrex 读取数据时会进行独占标记，防止其他内核路径访问，直至调用 strex 完成写入后清除标记。

ldrex 和 strex 指令，是将单纯的更新内存的原子操作分成了两个独立的步骤:

1.  ldrex 用来读取内存中的值，并标记对该段内存的独占访问：

> ldrex Rx, \[Ry\]

读取寄存器 Ry 指向的4字节内存值，将其保存到 Rx 寄存器中，同时标记对 Ry 指向内存区域的独占访问。如果执行 ldrex 指令的时候发现已经被标记为独占访问了，并不会对指令的执行产生影响。

1.  strex 在更新内存数值时，会检查该段内存是否已经被标记为独占访问，并以此来决定是否更新内存中的值：

> strex Rx, Ry, \[Rz\]

如果执行这条指令的时候发现已经被标记为独占访问了，则将寄存器 Ry 中的值更新到寄存器 Rz 指向的内存，并将寄存器 Rx 设置成 0。指令执行成功后，会将独占访问标记位清除。如果执行这条指令的时候发现没有设置独占标记，则不会更新内存，且将寄存器 Rx 的值设置成 1。

ARM 内部的实现如下所示，这里不再赘述。

![a14920ce64580abd213b74d43ec6f3d0.png][]

## 自旋锁 spin\_lock ##

Linux内核中最常见的锁是自旋锁，自旋锁最多只能被一个可执行线程持有。如果一个线程试图获取一个已被持有的自旋锁，这个线程会进行忙循环——旋转等待(会浪费处理器时间)锁重新可用。自旋锁持有期间不可被抢占。

另一种处理锁争用的方式：让等待线程睡眠，直到锁重新可用时再唤醒它，这样处理器不必循环等待，可以去执行其他代码，但是这会有两次明显的上下文切换的开销，**信号量**便提供了这种锁机制。

自旋锁的使用接口如下：

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>spin_lock()</td> 
   <td>获取指定的自旋锁</td> 
  </tr> 
  <tr> 
   <td>spin_lock_irq()</td> 
   <td>禁止本地中断并获取指定的锁</td> 
  </tr> 
  <tr> 
   <td>spin_lock_irqsave()</td> 
   <td>保存本地中断当前状态，禁止本地中断，获取指定的锁</td> 
  </tr> 
  <tr> 
   <td>spin_unlock()</td> 
   <td>释放指定的锁</td> 
  </tr> 
  <tr> 
   <td>spin_unlock_irq()</td> 
   <td>释放指定的锁，并激活本地中断</td> 
  </tr> 
  <tr> 
   <td>spin_unlock_irqrestore()</td> 
   <td>释放指定的锁，并让本地中断恢复以前状态</td> 
  </tr> 
  <tr> 
   <td>spin_lock_init()</td> 
   <td>动态初始化指定的锁</td> 
  </tr> 
  <tr> 
   <td>spin_trylock()</td> 
   <td>试图获取指定的锁，成功返回0，否则返回非0</td> 
  </tr> 
  <tr> 
   <td>spin_is_locked()</td> 
   <td>测试指定的锁是否已被占用，已被占用返回非0，否则返回0</td> 
  </tr> 
 </tbody> 
</table>

以 spin\_lock 为例看下它的用法：

DEFINE_SPINLOCK(mr_lock);
    spin_lock(&mr_lock);
    /* 临界区 */
    spin_unlock(&mr_lock);

![a788d97f535c95d824c817a0cfcd0d9b.png][]

static inline void arch_spin_lock(arch_spinlock_t *lock)
    {
     unsigned int tmp;
     arch_spinlock_t lockval, newval;
    
     asm volatile(
     /* Atomically increment the next ticket. */
     ARM64_LSE_ATOMIC_INSN(
     /* LL/SC */
    " prfm pstl1strm, %3\n"
    "1: ldaxr %w0, %3\n"
    " add %w1, %w0, %w5\n"
    " stxr %w2, %w1, %3\n"
    " cbnz %w2, 1b\n",
     /* LSE atomics */
    " mov %w2, %w5\n"
    " ldadda %w2, %w0, %3\n"
     __nops(3)
     )
    
     /* Did we get the lock? */
    " eor %w1, %w0, %w0, ror #16\n"
    " cbz %w1, 3f\n"
     /*
      * No: spin on the owner. Send a local event to avoid missing an
      * unlock before the exclusive load.
      */
    " sevl\n"
    "2: wfe\n"
    " ldaxrh %w2, %4\n"
    " eor %w1, %w2, %w0, lsr #16\n"
    " cbnz %w1, 2b\n"
     /* We got the lock. Critical section starts here. */
    "3:"
     : "=&r" (lockval), "=&r" (newval), "=&r" (tmp), "+Q" (*lock)
     : "Q" (lock->owner), "I" (1 << TICKET_SHIFT)
     : "memory");
    }

static inline void arch_spin_unlock(arch_spinlock_t *lock)
    {
     unsigned long tmp;
    
     asm volatile(ARM64_LSE_ATOMIC_INSN(
     /* LL/SC */
     " ldrh %w1, %0\n"
     " add %w1, %w1, #1\n"
     " stlrh %w1, %0",
     /* LSE atomics */
     " mov %w1, #1\n"
     " staddlh %w1, %0\n"
     __nops(1))
     : "=Q" (lock->owner), "=&r" (tmp)
     :
     : "memory");
    }

上边的代码中，核心逻辑在于 asm volatile() 内联汇编中，有很多独占的操作指令，只有基于指令的独占操作，才能保证软件上的互斥。把核心逻辑翻译成 C 语言：

![245a2f4720765588307f668997f196b3.png][]

可以看出，Linux 中针对每一个 spin\_lock 有两个计数。分别是 next 和 owner（初始值为0）。进程 A 申请锁时，会判断 next 和 owner 的值是否相等。如果相等就代表锁可以申请成功，否则原地自旋。直到 owner 和 next 的值相等才会退出自旋。

## 信号量 Semaphore ##

信号量是在多线程环境下使用的一种措施，它负责协调各个进程，以保证他们能够正确、合理的使用公共资源。它和 spin\_lock 最大的不同之处就是：无法获取信号量的进程可以睡眠，因此会导致系统调度。

信号量的定义如下：

struct semaphore {
     raw_spinlock_t  lock;      //利用自旋锁同步
     unsigned int  count;      //用于资源计数
     struct list_head wait_list; //等待队列
    };

信号量在创建时设置一个初始值 count，用于表示当前可用的资源数。一个任务要想访问共享资源，首先必须得到信号量，获取信号量的操作为 count - 1。若当前 count 为负数，表明无法获得信号量，该任务必须挂起在该信号量的等待队列等待；若当前 count 为非负数，表示可获得信号量，因而可立刻访问被该信号量保护的共享资源。

当任务访问完被信号量保护的共享资源后，必须释放信号量，释放信号量是操作 count + 1，如果加一后的 count 为非正数，表明有任务等待，则唤醒所有等待该信号量的任务。

了解了信号量的结构与定义，接下来我们看下常用的信号量接口：

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>DEFINE_SEMAPHORE(name)</td> 
   <td>声明信号量并初始化为 1</td> 
  </tr> 
  <tr> 
   <td>void sema_init(struct semaphore *sem, int val)</td> 
   <td>声明信号量并初始化为 val</td> 
  </tr> 
  <tr> 
   <td>down</td> 
   <td>获得信号量，task 不可被中断，除非是致命信号</td> 
  </tr> 
  <tr> 
   <td>down_interruptible</td> 
   <td>获得信号量，task 可被中断</td> 
  </tr> 
  <tr> 
   <td>down_trylock</td> 
   <td>能够获得信号量时，count --，否则立刻返回，不加入 waitlist</td> 
  </tr> 
  <tr> 
   <td>down_killable</td> 
   <td>获得信号量，task 可被 kill</td> 
  </tr> 
  <tr> 
   <td>up</td> 
   <td>释放信号量</td> 
  </tr> 
 </tbody> 
</table>

这里我们看下最核心的两个实现 **down** 和 **up**。

*  down

down 用于调用者获得信号量，若 count 大于0，说明资源可用，将其减一即可。

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);
    }
    EXPORT_SYMBOL(down);

若 count < 0，调用函数 \_\_down()，将 task 加入等待队列，并进入等待队列，并进入调度循环等待，直至其被 \_\_up 唤醒，或者因超时以被移除等待队列。

static inline int __sched __down_common(struct semaphore *sem, long state,
            long timeout)
    {
     struct semaphore_waiter waiter;
    
     list_add_tail(&waiter.list, &sem->wait_list);
     waiter.task = current;
     waiter.up = false;
    
     for (;;) {
      if (signal_pending_state(state, current))
       goto interrupted;
      if (unlikely(timeout <= 0))
       goto timed_out;
      __set_current_state(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;
    }

*  up

up 用于调用者释放信号量，若 waitlist 为空，说明无等待任务，count + 1，该信号量可用。

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);
    }
    EXPORT_SYMBOL(up);

若 waitlist 非空，将 task 从等待队列移除，并唤醒该 task，对应 \_\_down 条件。

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

## 互斥锁 mutex ##

Linux 内核中，还有一种类似信号量的同步机制叫做互斥锁。互斥锁类似于 count 等于 1 的信号量。所以说信号量是在多个进程/线程访问某个公共资源的时候，进行保护的一种机制。而互斥锁是单个进程/线程访问某个公共资源的一种保护，于互斥操作。

互斥锁有一个特殊的地方：只有持锁者才能解锁。如下图所示：

![b5d923d6a6519c82a807470d64c1db9e.png][]

用一句话来讲信号量和互斥锁的区别，就是**信号量用于线程的同步，互斥锁用于线程的互斥。**

互斥锁的结构体定义：

struct mutex {
     atomic_long_t  owner; //互斥锁的持有者
     spinlock_t  wait_lock; //利用自旋锁同步
    #ifdef CONFIG_MUTEX_SPIN_ON_OWNER
     struct optimistic_spin_queue osq; /* Spinner MCS lock */
    #endif
     struct list_head wait_list; //等待队列
    ......
    };

其常用的接口如下所示：

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>DEFINE_MUTEX(name)</td> 
   <td>静态声明互斥锁并初始化解锁状态</td> 
  </tr> 
  <tr> 
   <td>mutex_init(mutex)</td> 
   <td>动态声明互斥锁并初始化解锁状态</td> 
  </tr> 
  <tr> 
   <td>void mutex_destroy(struct mutex *lock)</td> 
   <td>销毁该互斥锁</td> 
  </tr> 
  <tr> 
   <td>bool mutex_is_locked(struct mutex *lock)</td> 
   <td>判断互斥锁是否被锁住</td> 
  </tr> 
  <tr> 
   <td>mutex_lock</td> 
   <td>获得锁，task 不可被中断</td> 
  </tr> 
  <tr> 
   <td>mutex_unlock</td> 
   <td>解锁</td> 
  </tr> 
  <tr> 
   <td>mutex_trylock</td> 
   <td>尝试获得锁，不能加锁则立刻返回</td> 
  </tr> 
  <tr> 
   <td>mutex_lock_interruptible</td> 
   <td>获得锁，task 可以被中断</td> 
  </tr> 
  <tr> 
   <td>mutex_lock_killable</td> 
   <td>获得锁，task 可以被中断</td> 
  </tr> 
  <tr> 
   <td>mutex_lock_io</td> 
   <td>获得锁，在该 task 等待琐时，它会被调度器标记为 io 等待状态</td> 
  </tr> 
 </tbody> 
</table>

![1346f60ed3a31f405dcf79f42e5f0f53.png][]

5T技术资源大放送！包括但不限于：C/C++，Arm, Linux，Android，人工智能，单片机，树莓派，等等。在上面的【人人都是极客】公众号内回复「peter」，即可免费获取！！

![0d4aa9d14cef91767298ee831bd7eeb5.gif][] **记得点击****分享****、****赞****和****在看****，给我充点儿电吧**

[3eddc62b2f0c5a4851a9df73753628e3.png]: /images/20220828/adcebbf11d724ec79ca60e27a28c3809.png
[a14920ce64580abd213b74d43ec6f3d0.png]: /images/20220828/5a892da182974a8fa8ad0966af745bd2.png
[a788d97f535c95d824c817a0cfcd0d9b.png]: /images/20220828/ef5ff1b4127f46cf994bb86b92a9140e.png
[245a2f4720765588307f668997f196b3.png]: /images/20220828/b06eaab624974fdc8b395ad5958d77ba.png
[b5d923d6a6519c82a807470d64c1db9e.png]: /images/20220828/2a967762924a437f83b9f27514161e38.png
[1346f60ed3a31f405dcf79f42e5f0f53.png]: /images/20220828/ce35e809eb964bc0b6e1f54d05fee167.png
[0d4aa9d14cef91767298ee831bd7eeb5.gif]: /images/20220828/b165212f67414b718f11411d11493e7a.png