常见嵌入式操作系统中的并发与竞争保护机制

覆盖系统(10 个):Linux、FreeRTOS、μC/OS-II、μC/OS-III、RT-Thread、Zephyr、ThreadX(Eclipse ThreadX)、NuttX(Apache)、LiteOS、AliOS Things(Rhino 内核)。
结构说明:3.1~3.10 按机制展开,每个系统均有完整接口详解,每节末尾对照表覆盖全部 10 个系统。


目录

  1. 为什么需要并发保护
  2. 机制总览:一张全景图
  3. 各机制详解与跨 OS 接口对比
    • 3.1 关中断(Interrupt Masking)
    • 3.2 关调度(Scheduler Lock)
    • 3.3 自旋锁(Spinlock)
    • 3.4 互斥锁(Mutex)
    • 3.5 信号量(Semaphore)
    • 3.6 读写锁(RWLock)
    • 3.7 原子操作(Atomic Operations)
    • 3.8 无锁数据结构(Lock-Free / Per-CPU / RCU)
    • 3.9 临界区(Critical Section)通用抽象
    • 3.10 内存屏障(Memory Barrier)
  4. 优先级反转与解决方案
  5. 选型决策指南
  6. 接口速查总表

1. 为什么需要并发保护

并发执行的实体(线程、ISR、SMP 上的另一个核)交错访问共享资源时会产生竞态条件(Race Condition)。典型场景:

场景 竞争双方 典型后果
线程 vs 线程(单核抢占) 被抢占的临界区 数据撕裂、状态不一致
线程 vs ISR 中断打断访问序列 读改丢更新(lost update)
SMP 核 vs 核 真并行访问 数据损坏、缓存一致性问题
编译器/CPU 乱序 指令重排 逻辑上"不可能"的错误

i++(read-modify-write)是最经典的反例:在 Cortex-M 上是 LDR/ADD/STR 三条指令,任意时刻被抢占都可能丢更新。

保护的本质只有三条路:

  1. 串行化——让临界区独占执行(锁、关中断、关调度);
  2. 原子化——让操作不可分割(原子指令、LDREX/STREX、CAS);
  3. 消除共享——每个执行体持有私有副本(Per-CPU、RCU 读侧、消息传递替代共享状态)。

2. 机制总览:一张全景图

在这里插入图片描述

十大机制一句话特征(与上图对应):

  1. 关中断(Interrupt Masking)——最底层、最粗粒度;单核上"万能",SMP 上必须配合自旋锁
  2. 关调度(Scheduler Lock)——只挡任务切换、不挡中断;对中断延迟影响最小
  3. 自旋锁(Spinlock)——忙等不睡眠;SMP 与 ISR 上下文专用;持有者禁止睡眠
  4. 互斥锁(Mutex)——睡眠等待 + 所有权 + 优先级继承(PI);任务上下文专用
  5. 信号量(Semaphore)——计数资源/同步信号;无所有权;ISR 可 give、不可阻塞 take
  6. 读写锁(RWLock)——读共享、写独占;Linux 有自旋与睡眠(rwsem)两族
  7. 原子操作(Atomic Ops)——atomic_inc / CAS 硬件指令级无锁;M0/M0+ 退化为关中断
  8. 无锁结构(Lock-Free)——SPSC 队列、Per-CPU 变量、RCU(Linux 特色)、seqlock
  9. 内存屏障(Memory Barrier)——不解决互斥,解决"可见性与顺序";SMP 上锁的隐含配件
  10. 优先级反转对策——优先级继承(PI)、优先级天花板(PCP)

一个粗略的"该用谁"判断:

  • ISR 与线程共享数据 → 关中断(单核)/ spinlock(SMP/Linux)
  • 线程间共享数据,临界区短 → mutex(RTOS)/ spinlock(不可睡眠时)
  • 事件通知、资源计数 → semaphore
  • 一两个变量的计数/标志 → atomic
  • 高频读、极少写 → rwlock / RCU / seqlock

3. 各机制详解与跨 OS 接口对比

3.1 关中断(Interrupt Masking)

原理深入

关中断是所有保护机制中最底层的一种:通过设置 CPU 的中断屏蔽寄存器,让可屏蔽中断(IRQ)不再被响应。它同时产生两个效果:

  1. ISR 不会执行——与 ISR 共享的数据得到保护;
  2. 在单核 RTOS 上,调度也不会发生——因为 RTOS 的任务切换由 tick 中断(SysTick)或 PendSV 触发,中断被关 = 调度器失去运行机会。

所以它本质上是"以停止时间为代价换取原子性"。两个硬约束由此而来:

  • 临界区必须极短(微秒级):它直接拉长最坏中断延迟,是实时系统抖动的主要来源;
  • 必须支持嵌套:函数 A 关了中断调用函数 B,B 也"关-开"中断,B 返回时若直接开中断,A 的保护就被破坏。因此正确的模式永远是 save → disable → … → restore,而不是 disable → enable。

SMP 上的失效:关中断只作用于本地核,对隔壁核上的线程毫无约束。因此 Linux SMP 内核里关中断永远和自旋锁成对出现(spin_lock_irqsave),从不单独承担互斥职责。

硬件层面
架构 机制 说明
Cortex-M PRIMASKCPSID i/CPSIE i 全开关,一刀切
Cortex-M BASEPRI 只屏蔽优先级数值 ≥ 阈值的中断——FreeRTOS 的关键:它只关"会调用 OS API 的那部分中断"(≤ configMAX_SYSCALL_INTERRUPT_PRIORITY),更高优先级中断(如电机保护、硬实时采样)照常运行,但这些 ISR 里禁止调用任何 OS API
ARM Linux (ARM64) DAIF 标志 local_irq_save 读 PSTATE.DAIF 并置位 I,restore 写回
x86 IF 标志(cli/sti,内核用 pushf/popf 保存)
RISC-V mstatus.MIE(M 态)/ sstatus.SIE(S 态)CSR csrrci 一条指令完成"读并清零",天然 save-and-disable;基础架构无 BASEPRI 式阈值屏蔽(全局开关 + mie 按中断源使能),但 PLIC 的 priority threshold 寄存器(每 hart/context)可实现等价的分级屏蔽——FreeRTOS 早期 RISC-V port 只能全局关中断,新的 PLIC 方案 port 用 threshold 恢复了"只屏蔽低优先级中断"的能力,是评估 RISC-V RTOS port 质量的检查点
Linux 接口详解
unsigned long flags;

/* 推荐:可嵌套,状态保存在栈变量里 */
local_irq_save(flags);
/* ... 极短临界区 ... */
local_irq_restore(flags);

/* 不可嵌套版本:用于确定外层没关过中断的地方 */
local_irq_disable();
local_irq_enable();

/* 只查询不修改 */
irqs_disabled();        /* 当前中断是否被关 */

/* 软中断(下半部)屏蔽——不碰硬中断 */
local_bh_disable();
local_bh_enable();      /* 注意:可能在此触发 do_softirq 执行 */

实际内核代码里极少裸用 local_irq_save,它几乎都出现在 spin_lock_irqsave 内部。裸用的典型场合:架构相关代码、kgdb、early boot、以及 per-cpu 变量访问前配合 preempt_disable

PREEMPT_RT 的特殊性:RT 补丁把大部分硬中断线程化,local_irq_disable 的语义在 RT 内核中被重新解释(关的是"真正的硬中断上下文"),驱动开发者不需要改代码,但要意识到延迟特性变了。

FreeRTOS 接口详解
/* 任务级:对 configMAX_SYSCALL_INTERRUPT_PRIORITY 以下设 BASEPRI */
taskENTER_CRITICAL();
/* ... */
taskEXIT_CRITICAL();
/* 内部带嵌套计数(uxCriticalNesting),可以成对嵌套 */

/* 不可嵌套的裸开关(速度快,port 相关,一般不推荐直接用) */
taskDISABLE_INTERRUPTS();
taskENABLE_INTERRUPTS();

/* ISR 级:从中断里进入临界区(port 支持时) */
UBaseType_t uxSaved = taskENTER_CRITICAL_FROM_ISR();
/* ... */
taskEXIT_CRITICAL_FROM_ISR(uxSaved);

关键细节:

  • taskENTER_CRITICAL单核上既挡中断又挡调度;FreeRTOS 官方 SMP 内核里它内部实现为"关中断 + 获取 ISR 自旋锁",语义变重,临界区长的代码要重新审视;
  • FromISR 版本的返回值必须保存并回传,它记录了进入前的 BASEPRI;
  • configMAX_SYSCALL_INTERRUPT_PRIORITY 之上的中断(数值更小、优先级更高)不被屏蔽,但其中禁止调用任何 FromISR API——这是用"不可被 OS 保护"换来的零延迟通道。
μC/OS-II 接口详解

μC/OS-II 把临界区做成宏,并提供了三种实现方法,由 OS_CPU.H 中的 OS_CRITICAL_METHOD 选择——这是它教科书式的经典设计:

#if OS_CRITICAL_METHOD == 1
    /* 方法1:最简单的开/关。处理器不支持"读状态"指令时用。
       缺点:破坏嵌套——若进入前中断本来就关着,退出时被错误打开 */
    #define OS_ENTER_CRITICAL()  CPU_INT_DIS()
    #define OS_EXIT_CRITICAL()   CPU_INT_EN()

#elif OS_CRITICAL_METHOD == 2
    /* 方法2:把中断状态压栈,退出时弹栈恢复。支持嵌套。
       缺点:部分编译器对"函数内改栈"的优化有兼容问题 */
    #define OS_ENTER_CRITICAL()  asm("PUSH PSW"); CPU_INT_DIS()
    #define OS_EXIT_CRITICAL()   asm("POP PSW")

#elif OS_CRITICAL_METHOD == 3
    /* 方法3:状态保存到局部变量 cpu_sr。最规范,可移植性最好。
       要求调用处声明 OS_CPU_SR cpu_sr; */
    #define OS_ENTER_CRITICAL()  (cpu_sr = OS_CPU_SR_Save())
    #define OS_EXIT_CRITICAL()   (OS_CPU_SR_Restore(cpu_sr))
#endif

常见坑:用方法 3 时忘记在函数开头声明 OS_CPU_SR cpu_sr;,或者一个函数里多个 OS_ENTER_CRITICAL 共用同一个 cpu_sr 导致状态互相覆盖。

μC/OS-III 接口详解

III 把这件事分成了两层,并引入可配置的中断处置模式

CPU_SR_ALLOC();                 /* 宏:声明 cpu_sr 局部变量 */

CPU_CRITICAL_ENTER();           /* cpu_sr = CPU_SR_Save(),关中断 */
CPU_CRITICAL_EXIT();            /* CPU_SR_Restore(cpu_sr) */

/* 调度器级的临界区(内部根据 OS_CFG_ISR_POST_DEFERRED_EN 二选一) */
OS_CRITICAL_ENTER();
OS_CRITICAL_EXIT();
  • OS_CFG_ISR_POST_DEFERRED_EN = 0直接模式):OS_CRITICAL_ENTER = 关中断。ISR 里 post 信号量直接操作等待列表,中断延迟略长;
  • = 1延迟提交模式):ISR 里 post 只把消息挂到"中断延迟队列",由内核内置的 IntQ 任务在退出中断后统一处理。OS_CRITICAL_ENTER 此时退化为只关调度(锁 IntQ 任务),中断延迟大幅缩短且与系统负载无关——这是 III 相对 II 最大的实时性改进,代价是多一个系统任务和一次额外切换。
RT-Thread 接口详解
rt_base_t level;

level = rt_hw_interrupt_disable();   /* 保存状态 + 关中断,返回状态 */
/* ... */
rt_hw_interrupt_enable(level);       /* 恢复传入的状态——名字具有误导性,
                                        语义是 restore,因此天然支持嵌套 */

这是一个高频面试/踩坑点rt_hw_interrupt_enable(level) 并不是"打开中断",而是"恢复到 level 代表的状态"。配对正确时嵌套完全安全;但若把 rt_hw_interrupt_enable(0) 当"开中断"乱用,会破坏外层保护。

Zephyr 接口详解
unsigned int key;

key = irq_lock();      /* 返回 lock 前的中断使能状态(key) */
/* ... */
irq_unlock(key);       /* 恢复到 key 状态 */
  • 单核上 irq_lock() 内部维护嵌套计数,多重 lock 只在最外层真正关中断;
  • SMP(CONFIG_SMP)上 irq_lock() 语义升级:它获取的是一把全局自旋锁 + 关本地中断——也就是说 Zephyr 应用代码不用改,同一个 API 在 SMP 上自动变成正确的跨核保护,但临界区长度的影响也随之放大;
  • 相关补充:irq_is_locked() 查询状态;k_is_in_isr() / k_is_preempt_thread() 判断上下文(写"既可能被任务也可能被 ISR 调用"的函数时必备)。
ThreadX 接口详解
UINT old_posture;

old_posture = tx_interrupt_control(TX_INT_DISABLE);   /* 关中断,返回进入前状态 */
/* ... 极短临界区 ... */
tx_interrupt_control(old_posture);                    /* 恢复——save/restore 语义,
                                                         天然支持嵌套 */
  • TX_INT_ENABLE 也可传入,但正确用法永远是"传回旧状态"而非"显式使能"——与 RT-Thread rt_hw_interrupt_enable(level) 同一设计哲学;
  • 内核与驱动内部更常用宏形式 TX_DISABLE / TX_RESTORE(配合编译单元内声明的状态变量),本质相同,省去函数调用开销;
  • 关中断区间同样直接决定系统最坏中断延迟,ThreadX 官方手册明确建议临界区控制在极短范围。
NuttX 接口详解
irqstate_t flags;

/* 方式一:纯关中断(等价 Linux local_irq_save) */
flags = irqsave();
/* ... */
irqrestore(flags);

/* 方式二:临界区——单核 = irqsave;SMP = irqsave + 获取全局 IRQ 自旋锁 */
flags = enter_critical_section();
/* ... */
leave_critical_section(flags);
  • 与 Zephyr irq_lock() 完全同构的设计enter_critical_sectionCONFIG_SMP 构建下自动升级为跨核保护,同一份驱动代码在单核/多核上都正确——这是 NuttX SMP 支持最关键的应用层约定;
  • 仅关中断用 irqsave/irqrestore;需要"单核 SMP 通吃"的保护一律用 enter_critical_section
  • 另有 sched_lock()/sched_unlock() 提供纯调度锁(见 3.2),与中断锁分工明确。
LiteOS 接口详解
UINT32 intSave;

intSave = LOS_IntLock();       /* 保存状态 + 关中断 */
/* ... */
LOS_IntRestore(intSave);       /* 恢复到 intSave 代表的状态——支持嵌套 */

LOS_IntUnLock();               /* 纯开中断(无状态恢复)——仅在确定外层未关中断
                                  的场合使用,嵌套环境下会破坏外层保护 */
  • LOS_IntLock/LOS_IntRestore 是规范配对,语义与 Linux local_irq_save/restore 一致;
  • 中断开关计数与任务切换联动:持有中断锁期间不会发生任务切换(单核语义);
  • 配套查询:LOS_IntActive()LOS_IntNumGet() 等中断上下文判断接口。
AliOS Things(Rhino 内核)接口详解
RHINO_CRITICAL_ENTER();        /* 宏:内部自动声明并保存 CPSR(cpu_intrpt_save()) */
/* ... */
RHINO_CRITICAL_EXIT();         /* 恢复保存的状态 */

/* 底层原语(一般无需直接用):
   cpu_intrpt_t cpsr = cpu_intrpt_save();
   cpu_intrpt_restore(cpsr);
   cpu_intrpt_disable();  cpu_intrpt_enable();  */
  • RHINO_CRITICAL_ENTER/EXIT 宏把"声明状态变量 + 保存 + 恢复"打包,用法上最接近 μC/OS-II 的 OS_CRITICAL_METHOD == 3——注意该宏内部声明变量,进入/退出必须处于同一作用域
  • Rhino 从设计之初支持 SMP,其临界区在多核构建中同样升级为"关本地中断 + 内核自旋锁"语义;
  • 另有 krhino_intrpt_enter()/exit() 用于中断嵌套计数(与临界区是不同机制,勿混淆)。
对照小结(10 系统)
OS 接口 嵌套安全 ISR 可用 备注
Linux local_irq_save/restore(flags) 本身即在 裸用少,主要包在 spinlock 里
FreeRTOS taskENTER/EXIT_CRITICAL 是(计数) _FROM_ISR BASEPRI 只屏蔽部分优先级
μC/OS-II OS_ENTER/EXIT_CRITICAL 方法2/3 是 三种 OS_CRITICAL_METHOD
μC/OS-III CPU_CRITICAL_ENTER/EXIT 另有延迟提交模式改变语义
RT-Thread rt_hw_interrupt_disable/enable(level) enable 实为 restore
Zephyr irq_lock/irq_unlock(key) SMP 上自动升级为全局自旋锁
ThreadX tx_interrupt_control / TX_DISABLE/TX_RESTORE save/restore 语义
NuttX irqsave/irqrestoreenter_critical_section 后者 SMP 下自动升级(同 Zephyr 思路)
LiteOS LOS_IntLock/IntRestore LOS_IntUnLock 为纯开,慎用
AliOS Things RHINO_CRITICAL_ENTER/EXIT 宏内声明变量,须同一作用域配对

适用场景:几条指令的寄存器/标志位更新;进入临界前状态未知的可重入库函数;boot 早期调度器尚未启动。反模式:在关中断区内做循环等待、调用任何可能触发上下文切换或打印日志的函数。


3.2 关调度(Scheduler Lock)

原理深入

关调度只冻结任务切换:当前任务独占 CPU 运行,但中断照常响应,ISR 里该唤醒谁还唤醒谁(唤醒动作被记录下来,等调度恢复时统一结算——即"pending"机制)。

它与关中断形成明确的权衡:

关中断 关调度
中断延迟 恶化(全系统最坏延迟来源) 不受影响
能否保护与 ISR 共享的数据 不能(ISR 照样跑)
能否保护与任务共享的数据
临界区可承受长度 微秒级 可以稍长(但仍不宜长)

因此关调度的定位很精确:保护"只在任务之间共享"的数据,同时不想伤害中断延迟

通用禁忌(所有 OS 一致):关调度期间禁止调用任何可能阻塞的 API(take 信号量、sleep、malloc 触发等待等)。调度器被锁了还去阻塞自己,轻则断言失败,重则系统挂死。

Linux:preempt_disable
preempt_disable();
/* 当前进程不会被换出(单核视角的原子) */
preempt_enable();
  • 实现:每核/每任务的 preempt_count 计数器,支持嵌套;preempt_enable 减到 0 且 need_resched 置位时触发 preempt_schedule
  • preempt_count 是位段结构:低 8 位抢占计数 + softirq 计数 + hardirq 计数 + NMI 位——in_interrupt() 等上下文判断宏读的就是它;
  • 注意preempt_disable 保护的是"本核不被切走",这正是 per-CPU 变量访问的前提(get_cpu_var() 内部就是 preempt_disable);
  • 用户态没有对应物(用户态进程本来就不能关内核抢占)。
FreeRTOS:vTaskSuspendAll
vTaskSuspendAll();
/* ... 可以执行较长时间的任务级原子操作 ... */
xTaskResumeAll();    /* 返回 pdTRUE 表示期间有待处理的切换,
                        建议:if (xTaskResumeAll() == pdFALSE) taskYIELD(); */
  • 实现极简:uxSchedulerSuspended++ / --;恢复时遍历"悬挂期间被唤醒的任务"清单补做切换,并重算 tick 丢失(xTaskCatchUpTicks,低功耗 tickless 也用同一路径);
  • 挂起期间 tick 中断照常发生,只是不做切换——这与关中断(tick 会丢)是本质区别,系统时间不漂移;
  • 官方限制:挂起期间调用 API 必须是不阻塞的(xTaskResumeAll 自己例外)。
μC/OS-II / III:OSSchedLock/Unlock
OSSchedLock();
/* ... */
OSSchedUnlock();          /* III 原型:OSSchedUnlock(&err) */
  • 嵌套计数 OSSchedLockNestingCtr,锁期间 OSIntNestingCtr(中断嵌套计数)照常工作;
  • 在 III 的延迟提交模式下,OS_CRITICAL_ENTER 内部就用它(见 3.1)——所以 III 的很多"临界区"实际是关调度而非关中断;
  • II 中 OSSchedLock 期间时钟节拍照常、任务延迟(OSTimeDly 到期的)会被置就绪但不切换。
RT-Thread:rt_enter_critical —— 命名陷阱重灾区
rt_enter_critical();
/* ... */
rt_exit_critical();
  • 这是关调度,不关中断! 名字里的 “critical” 让从 FreeRTOS/μC/OS 转过来的工程师大量误用——用它保护与 ISR 共享的队列就会出竞态;
  • 实现:rt_scheduler_lock_nest++;嵌套计数,rt_exit_critical 归零后检查 rt_current_thread 是否该切换;
  • rt_hw_interrupt_disable 的分工在 RT-Thread 设计文档里写得很清楚:任务间共享→critical(调度锁);与 ISR 共享→hw_interrupt
Zephyr:k_sched_lock
k_sched_lock();
/* ... */
k_sched_unlock();
  • 实现:把当前线程的调度锁计数 +1,本质上等同于把当前线程临时提升到"不可被抢占"的协作式地位(优先级高于一切可抢占线程);
  • 锁期间 IRQ 照常;调度锁只防抢占,线程仍可主动睡眠/阻塞(让出 CPU,唤醒后继续持锁)——但主动让出等于自己放弃了锁的保护意义,工程上应避免;
  • Zephyr 另有更细的手段:协作式线程(负优先级,天然不可被抢占)——如果某线程整段生命周期都不能被抢占,直接建成协作式比反复 sched_lock 更干净。
ThreadX:抢占阈值(preemption-threshold)—— 独此一家的"部分关调度"

ThreadX 没有全局关调度 API,取而代之的是每个任务除优先级外的第二个调度参数——抢占阈值

UINT old_threshold;

/* 把本任务抢占阈值提高到 5:此后只有优先级高于 5 的任务能抢占本任务 */
tx_thread_preemption_change(&my_thread, 5, &old_threshold);
/* ... 临界区:低优先级任务进不来,高优先级紧急任务照常运行 ... */
tx_thread_preemption_change(&my_thread, old_threshold, &old_threshold);
  • 与全局锁调度的对比:既不关中断(中断延迟零影响),也不冻结所有任务(高于阈值的紧急任务照常抢占)——粒度是全部 10 个系统中最精细的;
  • 典型用法:某任务处理一段"不希望被同层任务打断、但紧急任务必须能进"的代码(如半双工总线收发窗口);
  • 代价:阈值是调度器全局规划的一部分,设置不当会造成分析范围外的优先级倒挂,需要像规划优先级一样规划阈值。
NuttX:sched_lock/unlock
sched_lock();
/* ... */
sched_unlock();
  • 沿用 POSIX 命名(POSIX 中 sched_lock 本义即禁止本线程被抢占),嵌套计数;
  • 与 Linux preempt_disable 同样充当"per-CPU/本核变量安全访问"的前提(SMP 构建中配合 up_cpu_index() 使用);
  • 锁期间中断照常,语义与其他 RTOS 的调度锁完全一致。
LiteOS:LOS_TaskLock/Unlock
LOS_TaskLock();
/* ... */
LOS_TaskUnlock();
  • 嵌套计数,锁期间 tick 中断照常、到期任务置就绪但不切换;
  • LOS_IntLock 的分工同其他系统:任务间共享→TaskLock;与 ISR 共享→IntLock
  • 锁期间调用阻塞 API 会触发内核断言(debug 构建可早期暴露错误)。
AliOS Things(Rhino):krhino_sched_disable/enable
krhino_sched_disable();
/* ... */
krhino_sched_enable();
  • 嵌套计数;Rhino SMP 构建下该接口锁定的是本地核的调度,跨核保护仍需自旋锁(见 3.3);
  • 恢复时若有更高优先级任务就绪则立即触发切换。
对照小结(10 系统)
OS 接口 实现核心 特殊点
Linux preempt_disable/enable preempt_count 位段 per-CPU 变量访问的前提
FreeRTOS vTaskSuspendAll/xTaskResumeAll uxSchedulerSuspended tick 照常、不丢时间;恢复时补切换
μC/OS-II/III OSSchedLock/Unlock LockNestingCtr III 延迟提交模式的基础
RT-Thread rt_enter/exit_critical scheduler_lock_nest 名字误导,实为关调度
Zephyr k_sched_lock/unlock 线程锁计数 可用协作式线程整体替代
ThreadX tx_thread_preemption_change 每任务抢占阈值 部分关调度:高优先级任务仍可抢占,粒度最细
NuttX sched_lock/unlock 嵌套计数 POSIX 命名;SMP 下配 per-CPU 访问
LiteOS LOS_TaskLock/Unlock 嵌套计数 debug 构建对锁内阻塞有断言
AliOS Things krhino_sched_disable/enable 嵌套计数 SMP 下仅锁本地核调度

3.3 自旋锁(Spinlock)

原理深入

拿不到锁时不睡眠,而是在循环里反复尝试(“自旋”),直到持有者释放。它的成立依赖一个前提:等待的代价 < 一次上下文切换的代价,因此自旋锁只适用于极短临界区(几条到几十条指令)。

三条铁律:

  1. 持有者绝不睡眠/阻塞——单核上:睡眠意味着永远没人来放锁(死锁);SMP 上:持有者睡几秒,其他核就烧几秒 CPU;
  2. 临界区必须短——自旋是纯浪费,等待时间越长浪费越大;
  3. 防止"同锁不同上下文"死锁——线程持锁时被 ISR 打断,ISR 又来拿同一把锁:单核立即死锁。所以"线程侧拿锁"必须同时关本地中断(_irqsave 版本存在的意义)。

单核上的退化:非抢占单核内核里 spinlock 可编译为空操作;抢占单核内核里只需"关抢占"。真正的互斥语义只在 SMP 上有意义——所以自旋锁是 SMP 的原生公民

Linux:spinlock 家族全解
DEFINE_SPINLOCK(lock);            /* 静态定义+初始化 */
spinlock_t lock;
spin_lock_init(&lock);            /* 动态初始化 */

/* ① 基础版:关本地抢占 + 拿锁 */
spin_lock(&lock);
spin_unlock(&lock);
spin_trylock(&lock);              /* 拿不到立即返回 0,不等待 */

/* ② _irq 版:额外关本地硬中断(不保存状态,不可嵌套进已关中断区) */
spin_lock_irq(&lock);
spin_unlock_irq(&lock);

/* ③ _irqsave 版:保存并恢复中断状态——最常用、最安全 */
unsigned long flags;
spin_lock_irqsave(&lock, flags);
spin_unlock_irqrestore(&lock, flags);

/* ④ _bh 版:额外禁本地下半部(softirq/tasklet) */
spin_lock_bh(&lock);
spin_unlock_bh(&lock);

版本选择的决策逻辑——问自己"这把锁保护的数据,最高会被哪个上下文访问":

数据被谁访问 进程上下文侧用 ISR 侧用
只有进程/线程上下文 spin_lock
softirq/tasklet 也会碰 spin_lock_bh softirq 里 spin_lock
硬中断 ISR 也会碰 spin_lock_irqsave ISR 里 spin_lock(ISR 在自己上下文已原子)

选错的经典后果:进程侧用了 spin_lock 而 ISR 也拿同一把锁 → 进程持锁被 ISR 抢占 → ISR 自旋 → 单核死锁 / SMP 上本地核死锁。

实现演化(了解有助于理解性能行为):

  • 早期:TAS 自旋锁(test-and-set 单变量),多核抢同一缓存行,核数一多总线爆炸;
  • Ticket lock:取号排队,公平但所有核仍盯着同一行刷;
  • 现代:qspinlock(queued spinlock,MCS 锁的紧凑版)——每个等待者在自己的 per-CPU 节点上自旋,缓存行不再乒乓,是 x86/ARM64 的默认实现。

配套还有 raw_spinlock_t(PREEMPT_RT 下不退化为 rtmutex 的"真"自旋锁,调度器、中断子系统内部用)。

RT-Thread(SMP)
rt_spinlock_t lock;
rt_spin_lock_init(&lock);

rt_base_t level = rt_spin_lock_irqsave(&lock);
rt_spin_unlock_irqrestore(&lock, level);

rt_spin_lock(&lock) / rt_spin_unlock(&lock);   /* 不操作中断 */

RT-Thread 单核默认构建里自旋锁基本退化为关调度/关中断;SMP 构建(RT_USING_SMP)才是完整语义,内核的调度器、IPC 对象在 SMP 下全部改用自旋锁保护。

Zephyr:k_spinlock
K_SPINLOCK_DEFINE(lock);          /* 静态定义 */
struct k_spinlock l;
/* 无需显式 init(K_SPINLOCK_DEFINE 已含) */

k_spinlock_key_t key = k_spin_lock(&l);   /* = 拿锁 + 关本地中断,返回 key */
k_spin_unlock(&l, key);                    /* 放锁 + 恢复中断 */

k_spin_trylock(&l, &key);         /* 非阻塞尝试 */
  • 语义精确对应 Linux 的 spin_lock_irqsave,key 里同时打包了中断状态与锁状态;
  • Zephyr 的 irq_lock() 在 SMP 下就是用一把内部 k_spinlock 实现的(见 3.1);
  • 持锁期间的规则与 Linux 一致:不许 sleep、不许 take 任何会阻塞的内核对象。
NuttX:命名直接对齐 Linux
spinlock_t lock;
spin_lock_init(&lock);            /* 或 SPIN_LOCK_INITIALIZER 静态初始化 */

irqstate_t flags;
spin_lock_irqsave(&lock, flags);  /* 拿锁 + 关本地中断 + 保存状态 */
spin_unlock_irqrestore(&lock, flags);

spin_lock(&lock) / spin_unlock(&lock);    /* 不操作中断 */
spin_trylock(&lock);
  • CONFIG_SPINLOCK(SMP 构建默认开启);单核构建下退化为关调度/空操作,语义保持兼容;
  • NuttX SMP 的调度器、等待队列内部均以自旋锁保护;enter_critical_section() 内部就是"全局 IRQ 自旋锁 + irqsave"的组合。
LiteOS(SMP 版本)
SPIN_LOCK_S lock;
LOS_SpinInit(&lock);

UINT32 intSave;
LOS_SpinLockSave(&lock, &intSave);   /* 拿锁 + 关中断,状态存入 intSave */
LOS_SpinUnlockRestore(&lock, intSave);

LOS_SpinLock(&lock) / LOS_SpinUnlock(&lock);   /* 不操作中断 */
LOS_SpinTrylock(&lock);                        /* 非阻塞 */
  • 命名体系同样对齐 Linux 习惯(LockSave/UnlockRestore ≈ irqsave/irqrestore);
  • LiteOS-A(带 MMU 的多核版本)内核调度器使用自旋锁;LiteOS-M(Cortex-M 轻量版)单核无自旋锁需求。
AliOS Things(Rhino):SMP 原生自旋锁
kspinlock_t lock;
krhino_spin_lock_init(&lock);

krhino_spin_lock(&lock);
krhino_spin_unlock(&lock);

cpu_intrpt_t flags;
krhino_spin_lock_irq_save(&lock, &flags);
krhino_spin_unlock_irq_restore(&lock, &flags);
  • Rhino 是 10 个系统中从设计之初就按 SMP 设计内核的 RTOS:调度器的全局就绪表、任务状态迁移均以自旋锁保护,应用层 API 同步开放;
  • 单核构建时同样退化为关中断语义,代码无需改动。
FreeRTOS / μC/OS / ThreadX:没有应用层自旋锁
  • FreeRTOS / μC/OS-II/III:单核设计哲学——"自旋"在单核上没有意义(等待者占着 CPU,持有者反而跑不了),用关中断/关调度完全替代。FreeRTOS 官方 SMP 内核(v11+)引入了内核内部自旋锁(task lock / ISR lock),并把 taskENTER_CRITICAL 重定义在其之上,但对应用层不暴露独立 spinlock API;
  • ThreadX:应用层无自旋锁 API;ThreadX SMP 版内核内部使用锁保护调度数据结构,应用接口与单核完全一致——对应用开发者透明。
对照小结(10 系统)
OS API 上下文 备注
Linux spin_lock/_irq/_irqsave/_bh + trylock 线程/softirq/ISR qspinlock 实现;PREEMPT_RT 下普通 spinlock 退化为可睡眠锁,真自旋用 raw_spinlock_t
RT-Thread rt_spin_lock(_irqsave) 线程/ISR SMP 构建下才有完整语义
Zephyr k_spin_lock(unlock) 带 key 线程/ISR ≈ spin_lock_irqsave
NuttX spin_lock_irqsave/lock/unlock/trylock 线程/ISR 命名语义直接对齐 Linux
LiteOS LOS_SpinLock(Save)/Unlock(Restore)/Trylock 线程/ISR LiteOS-A(SMP);LiteOS-M 单核无
AliOS Things krhino_spin_lock(_irq_save)/unlock(_irq_restore) 线程/ISR SMP 原生设计,内核调度器即用
FreeRTOS —(SMP 内核内部用) 单核用临界区替代
μC/OS-II/III 单核哲学,关中断替代
ThreadX —(SMP 版内核内部) 对应用透明

3.4 互斥锁(Mutex)

原理深入

互斥锁 = 自旋锁的"可睡眠版" + 所有权 + (通常)优先级继承

  • 可睡眠:拿不到就挂到等待队列、让出 CPU——临界区可以长,不浪费算力;
  • 所有权(ownership):锁记录持有者,只能由持有者释放。这是它与信号量的分水岭,也是能做优先级继承的前提(系统知道"该提升谁");
  • 优先级继承(PI):高优先级任务等待时,持有者临时继承该优先级,避免被中间优先级任务间接阻塞(详见第 4 章)。

上下文硬约束:只能在任务/线程上下文使用。ISR 里 take 会阻塞(非法),give 无意义(ISR 不持有锁)——所有 OS 都禁止 ISR 操作 mutex。

递归锁:同一持有者重复加锁不阻塞,配对的释放次数清零才真正解锁。方便"函数 A 持锁调用同样持锁的函数 B",但会掩盖设计问题,且递归释放不完整会造成假象泄漏,慎用。

Linux(内核态)
struct mutex m;
mutex_init(&m);                  /* 或 DEFINE_MUTEX(m) 静态定义 */

mutex_lock(&m);                  /* 不可中断睡眠(TASK_UNINTERRUPTIBLE) */
mutex_lock_interruptible(&m);    /* 可被信号打断,返回 -EINTR——驱动首选 */
mutex_lock_killable(&m);         /* 只响应致命信号 */
mutex_trylock(&m);               /* 非阻塞,返回是否拿到 */
mutex_unlock(&m);

mutex_is_locked(&m);             /* 仅查询,结果立即过时,别用它做决策 */

实现要点:

  • struct mutex 内含 atomic_long owner(持有者 task 指针 + 标志位,单字段即锁状态)+ 自旋锁保护的等待队列;
  • 乐观自旋(optimistic spinning / MCS):拿锁失败时先看持有者是否在别的核上运行中——是,则原地自旋等它放锁(省去睡眠+唤醒两次切换);持有者睡眠或换出才真的挂队列。这是 Linux mutex 在长临界区下仍有高吞吐的关键;
  • DEBUG 配置(CONFIG_DEBUG_MUTEXESlockdep)能查:非持有者释放、ISR 中使用、持锁睡眠、锁顺序反转(AB-BA 死锁静态检测)。

用户态对应:pthread mutex(glibc 基于 futex 实现——无竞争时纯用户态原子操作,零系统调用;有竞争才 futex(FUTEX_WAIT) 陷入内核):

pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_mutexattr_t attr;
pthread_mutexattr_init(&attr);
pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT);  /* 开 PI */
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);   /* 递归 */
pthread_mutex_init(&m, &attr);
pthread_mutex_lock(&m) / pthread_mutex_unlock(&m);

PI 用户态锁走 futex(FUTEX_LOCK_PI),内核用 rtmutex(与内核 mutex PI 同一套红黑树基础设施)实现。

FreeRTOS
SemaphoreHandle_t m;

m = xSemaphoreCreateMutex();                    /* 动态分配(heap) */
StaticSemaphore_t buf;
m = xSemaphoreCreateMutexStatic(&buf);          /* 静态分配(免堆,推荐用于
                                                   认证/确定性场景) */

xSemaphoreTake(m, portMAX_DELAY);               /* 超时可用 tick 数或 0 */
xSemaphoreGive(m);

关键特性与坑:

  1. 内建 PIconfigUSE_MUTEXES=1):持锁的低优先级任务在等待者等待时被临时提升,give 时恢复;
  2. PI 的局限:FreeRTOS 的继承不做链式传递,且一个任务持有多个 mutex 时优先级恢复逻辑简单——复杂嵌套持锁场景下继承可能不彻底,这是与 μC/OS-III、Linux rtmutex 的实质差距;
  3. 创建时就是"已释放"状态(take 立即成功),而二值信号量创建时是"空"——互换使用时的第一个坑;
  4. 无 FromISR 版本:ISR 中 take/give 互斥锁行为未定义——要"ISR 通知任务"请用二值信号量或任务通知;
  5. 递归锁单独一族:
SemaphoreHandle_t rm = xSemaphoreCreateRecursiveMutex();
xSemaphoreTakeRecursive(rm, timeout);   /* 同一任务可重复 take */
xSemaphoreGiveRecursive(rm);            /* 次数配对归零才真正释放 */
μC/OS-II:优先级天花板式 mutex
INT8U  err;
OS_EVENT *m;

/* prio = 优先级天花板(PIP 值):一个比所有可能使用者都高的空闲优先级 */
m = OSMutexCreate(prio, &err);

OSMutexPend(m, timeout, &err);     /* timeout 单位 = tick,0 = 永远等 */
OSMutexPost(m);
INT8U v = OSMutexAccept(m, &err);  /* 非阻塞 */
OSMutexQuery(m, &OS_MUTEX_DATA);   /* 查询持有者/等待者 */
  • II 的实现是**优先级天花板(Priority Ceiling)**而非动态继承:一旦低优先级任务拿到锁,立即提升到创建时指定的 prio,放锁后恢复。优点是实现简单、无链式问题、无死锁;缺点是提升过激(没人等也提升),且天花板值需要人工规划;
  • 不支持递归;ISR 禁止 pend/post。
μC/OS-III:真 PI + 嵌套计数
OS_MUTEX m;
OS_ERR  err;
CPU_TS  ts;

OSMutexCreate(&m, "UartLock", &err);
OSMutexPend(&m, 0, OS_OPT_PEND_BLOCKING, &ts, &err);
OSMutexPost(&m, OS_OPT_POST_NONE, &err);
OSMutexDel(&m, OS_OPT_DEL_ALWAYS, &err);   /* 删除(可唤醒所有等待者) */
  • 真正的动态优先级继承:等待者入队时,持有者继承等待者中最高优先级;支持多级链式继承(A 等 B、B 等 C 时 C 能被一路提升);放锁时按"剩余持有锁中最需要的优先级"恢复;
  • 内建嵌套/递归计数:同一任务对已持有 mutex 重复 pend 成功(计数 +1),post 次数配对才释放——II 不支持,III 内建;
  • 等待队列默认按优先级排序(不是 FIFO),同优先级内 FIFO。
RT-Thread
rt_mutex_t m = rt_mutex_create("m", RT_IPC_FLAG_PRIO);  /* 动态 */
struct rt_mutex sm;
rt_mutex_init(&sm, "m", RT_IPC_FLAG_PRIO);              /* 静态(免堆) */

rt_mutex_take(m, RT_WAITING_FOREVER);   /* 也可用 tick 超时 / RT_WAITING_NO */
rt_mutex_release(m);
rt_mutex_delete(m);                     /* 动态对象删除 */
rt_mutex_detach(&sm);                   /* 静态对象脱离 */
  • RT_IPC_FLAG_PRIO:等待者按优先级排队(实时系统推荐);RT_IPC_FLAG_FIFO:先来后到(公平但可能被插队);
  • 内建 PI:等待时持有者优先级被提升,释放后恢复——与 μC/OS-III 同路线;
  • 内建递归计数:持有计数 >0 时同线程 take 直接成功,release 配对归零才解锁;
  • take 在中断上下文会直接返回 -RT_EFULL 错误(不会真的阻塞,防御式设计)。
Zephyr
K_MUTEX_DEFINE(m);                    /* 静态定义,零初始化成本 */
struct k_mutex m2;  k_mutex_init(&m2);

k_mutex_lock(&m, K_FOREVER);          /* K_NO_WAIT / K_MSEC(n) 超时 */
k_mutex_unlock(&m);
  • 内建 PI:等待时提升持有者,解锁时恢复;
  • 递归内建:同一线程重复 lock 成功(lock_count 递增),unlock 配对;
  • ISR 中 lock 立即返回 -EBUSY
  • 配套原语 k_condvar(Zephyr 在 RTOS 里较少见地提供了条件变量):
K_CONDVAR_DEFINE(cv);
k_mutex_lock(&m);
while (!condition)
    k_condvar_wait(&cv, &m, K_FOREVER);  /* 原子地放锁+睡眠,唤醒时重新持锁 */
k_mutex_unlock(&m);
/* 另一方:k_condvar_signal(&cv) / k_condvar_broadcast(&cv) */
ThreadX
TX_MUTEX m;

/* TX_INHERIT 开启优先级继承;TX_NO_INHERIT 关闭(性能略高,自担反转风险) */
tx_mutex_create(&m, "UartLock", TX_INHERIT);

tx_mutex_get(&m, TX_WAIT_FOREVER);   /* 超时可用 tick 数 / TX_NO_WAIT */
tx_mutex_put(&m);
tx_mutex_delete(&m);
tx_mutex_prioritize(&m);             /* 特色:把等待者中最高优先级者提到队首 */
  • PI 是创建时显式选择的属性而非默认开启——安全关键项目开 TX_INHERIT,确定性/性能敏感的场合可关;
  • 支持递归持有:同一任务对已持有 mutex 重复 get 成功,put 配对计数;
  • tx_mutex_prioritize 是 ThreadX 特有的"等待队列人工提级"接口,用于运行时动态修正等待顺序;
  • ISR 禁止 get/put(get 在 ISR 连 TX_NO_WAIT 都不允许,与信号量不同)。
NuttX:POSIX 层 + 内核原生层双接口
/* ① 应用层(POSIX,可移植) */
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_lock(&m);
pthread_mutex_unlock(&m);
/* PI 由内核配置 CONFIG_PRIORITY_INHERITANCE 全局使能(默认开) */

/* ② 内核/驱动层(更轻,无 POSIX 开销) */
nxmutex_t m;   nxmutex_init(&m);
nxmutex_lock(&m);  nxmutex_unlock(&m);

/* ③ 递归版 */
nxrmutex_t rm;  nxrmutex_init(&rm);
nxrmutex_lock(&rm);    /* 同任务可重复加锁 */
nxrmutex_unlock(&rm);  /* 配对归零才释放 */
  • nxmutex/nxrmutex 底层基于 nxsem 实现但加了所有权检查,语义对齐 Linux 内核 mutex;
  • PI 是全局配置而非逐锁属性——与 ThreadX 的 per-mutex TX_INHERIT 是不同的粒度选择;
  • ISR 同样禁止操作。
LiteOS
UINT32 muxHandle;

LOS_MuxCreate(&muxHandle);                       /* 句柄式创建,内建 PI */
LOS_MuxPend(muxHandle, LOS_WAIT_FOREVER);        /* 超时 tick / LOS_NO_WAIT */
LOS_MuxPost(muxHandle);
LOS_MuxDelete(muxHandle);
LOS_MuxIsValid(muxHandle);                       /* 查询类接口 */
  • PI 内建默认开,持锁任务在等待者出现时被提升,post 后恢复;
  • 递归内建:同任务重复 pend 计数,post 配对;
  • 句柄(UINT32)而非指针的句柄式管理是 LiteOS 对象管理的统一风格(sem/event/queue 同);
  • ISR 中 pend 直接返回错误码,不会真的阻塞。
AliOS Things(Rhino)
/* ① 原生层 */
kmutex_t m;
krhino_mutex_create(&m, "m");
krhino_mutex_lock(&m, RHINO_WAIT_FOREVER);   /* tick 超时 / RHINO_NO_WAIT */
krhino_mutex_unlock(&m);
krhino_mutex_del(&m);

/* ② AOS 抽象层(业务代码推荐,屏蔽内核差异) */
aos_mutex_t am;
aos_mutex_new(&am);
aos_mutex_lock(&am, AOS_WAIT_FOREVER);       /* 注意抽象层超时单位是 ms */
aos_mutex_unlock(&am);
aos_mutex_free(&am);
  • PI 内建:等待者出现时持有者被提优,解锁恢复;
  • 支持递归持有计数;
  • 双层 API 是 AliOS 的体系特点krhino_ 原生层参数用 tick,aos_ 抽象层用 ms 且语义向 POSIX 靠拢——混用时务必确认超时单位;
  • SMP 构建下 mutex 内部由自旋锁保护等待队列。
对照小结(10 系统)
OS 创建/定义 take/give PI 递归 ISR 可用
Linux 内核 DEFINE_MUTEX/mutex_init mutex_lock(_interruptible)/unlock 有(rtmutex 基础设施) 无(需自封装) 禁止
Linux 用户态 pthread_mutex_init pthread_mutex_lock/unlock 可选(PTHREAD_PRIO_INHERIT 可选 attr
FreeRTOS xSemaphoreCreateMutex(Static) xSemaphoreTake/Give 有(简化版,无链式) 单独 API(RecursiveMutex) 禁止(无 FromISR)
μC/OS-II OSMutexCreate(prio) OSMutexPend/Post 优先级天花板 不支持 禁止
μC/OS-III OSMutexCreate OSMutexPend/Post 真 PI,链式继承 内建计数 禁止
RT-Thread rt_mutex_create/init rt_mutex_take/release 内建计数 禁止(返回错误)
Zephyr K_MUTEX_DEFINE/k_mutex_init k_mutex_lock/unlock 内建计数 禁止(-EBUSY)
ThreadX tx_mutex_create(TX_INHERIT) tx_mutex_get/put 可选(创建时指定) 支持 禁止(连 NO_WAIT 都不允许)
NuttX pthread_mutex_* / nxmutex / nxrmutex lock/unlock 有(CONFIG_PRIORITY_INHERITANCE 全局) nxrmutex 即递归锁 禁止
LiteOS LOS_MuxCreate(句柄式) LOS_MuxPend/Post 内建 内建计数 禁止(返回错误码)
AliOS Things krhino_mutex_create / aos_mutex_new lock/unlock 内建 支持 禁止

3.5 信号量(Semaphore)

原理深入

信号量 = 计数器 + 等待队列,Dijkstra 1965 年的原语,是 IPC 机制里用途最广也最被滥用的一种。两种形态:

  • 二值信号量(计数 ∈ {0,1}):本质是"事件旗标"——ISR 完成 DMA、按键按下、数据到达;
  • 计数信号量(计数 ∈ [0, N]):本质是"资源池余量"——N 个 buffer、N 个连接槽。

三种经典用法:

① 事件同步(初值 0):ISR give → 任务 take 被唤醒
② 资源计数(初值 N):take 拿一个资源,give 还一个
③ 互斥(初值 1):能用但不该用——见下

与 mutex 的四个本质区别(面试高频):

Mutex 信号量
所有权 有(谁拿谁放) 无(A 可放 B 拿的)
优先级继承 没有(用作互斥有反转风险)
计数 0/1 0~N
ISR 释放 禁止 允许(give 不阻塞)

"用二值信号量当互斥锁"是 RTOS 工程中最常见的隐患:功能上能跑,但高优先级任务可能被无期限间接阻塞(无 PI),且没有所有权检查,任何任务都能 give——调试困难。

Linux(内核态)
struct semaphore s;
sema_init(&s, 4);

down(&s);                     /* 不可中断——几乎总是错的 */
down_interruptible(&s);       /* 可中断,返回 -EINTR,首选 */
down_killable(&s);
down_trylock(&s);             /* 非阻塞 */
down_timeout(&s, jiffies);    /* 带超时 */
up(&s);
  • 内核演化史:2.4 时代 semaphore 是唯一的睡眠锁,兼任互斥;2.6 引入 struct mutex 后,semaphore 退居"真正需要计数"的少数场合(设备同时打开数限制等);
  • 它真正的"存活形态"是 读写信号量 struct rw_semaphore(见 3.6);
  • 用户态:POSIX 信号量 sem_init/sem_wait/sem_trywait/sem_post/sem_getvalue(可 pshared 放共享内存跨进程,或 sem_open 具名信号量);System V semget/semop/semctl(支持信号量组与 UNDO,历史包袱重,新代码勿用)。
FreeRTOS
/* 二值:初值 0——创建后先 give 一次才 take 得到 */
SemaphoreHandle_t b = xSemaphoreCreateBinary();

/* 计数:uxMaxCount=上限,uxInitialCount=初值 */
SemaphoreHandle_t c = xSemaphoreCreateCounting(10, 0);

xSemaphoreTake(b, portMAX_DELAY);
xSemaphoreGive(b);

/* ISR 中:唤醒动作可能要求上下文切换 */
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
xSemaphoreGiveFromISR(b, &xHigherPriorityTaskWoken);
portYIELD_FROM_ISR(xHigherPriorityTaskWoken);  /* 有此标志才切换 */

机制细节:

  • 所有 FreeRTOS 信号量/互斥锁底层是同一个队列结构xQueue),take = 收消息,give = 发消息——所以开销相同;
  • GiveFromISR 并不在 ISR 里直接切换:它把"要唤醒谁"记入 xPendingReadyList(调度挂起时)或置 xHigherPriorityTaskWoken,由 portYIELD_FROM_ISR 触发 PendSV 完成——这是 FreeRTOS 所有 FromISR API 的统一模式
  • 二值信号量的"中断-任务同步"场景,官方更推荐任务通知
/* 任务通知:每个任务 TCB 内建 32 位通知值,无需创建对象 */
/* ISR 侧 */   vTaskNotifyGiveFromISR(xTaskHandle, &xWoken);
/* 任务侧 */   ulTaskNotifyTake(pdTRUE, portMAX_DELAY);  /* 当二值信号量用 */
/* 或 */       xTaskNotifyWait(0, UINT32_MAX, &val, timeout); /* 当事件组用 */

官方数据:任务通知比二值信号量快约 45%、省 RAM、少一次对象创建——一对一的 ISR→任务通知应首选它;信号量保留给"多对一 / 事件需要排队计数"的场景。

μC/OS-II
OS_EVENT *s = OSSemCreate(0);         /* 初值 */
OSSemPend(s, timeout, &err);          /* timeout 单位 tick */
OSSemPost(s);                         /* ISR 中可调用 */
INT16U v = OSSemAccept(s);            /* 非阻塞读当前值 */
OSSemSet(s, cnt, &err);               /* 强制设置(需无等待者) */
OSSemQuery(s, &OS_SEM_DATA);

II 用统一的 OS_EVENT 结构承载信号量/互斥锁/消息队列/消息邮箱,等待表是优先级位图OSEventGrp + OSEventTbl[])——这是 II 能实现 O(1) 调度的基础,也直接决定优先级数上限(位图规模:经典版本 8×8=64 级,V2.8x 起扩展到 16×16=255 级)。

μC/OS-III
OS_SEM s;
OSSemCreate(&s, "RxSem", 0, &err);
OSSemPend(&s, 0, OS_OPT_PEND_BLOCKING, &ts, &err);
OSSemPost(&s, OS_OPT_POST_1, &err);      /* 只唤醒最高优先级一个 */
OSSemPost(&s, OS_OPT_POST_ALL, &err);    /* 广播:唤醒所有等待者 */
OSSemPendAbort(&s, OS_OPT_PEND_ABORT_ALL, &err);  /* 强制中止等待 */
  • OS_OPT_POST_ALL 广播是 III 新增——实现"条件变量式"的群体唤醒;
  • 任务内建信号量(每个 TCB 自带一个 sem,免创建):
OS_SEM_CTR n = OSTaskSemPend(0, OS_OPT_PEND_BLOCKING, &ts, &err);
OSTaskSemPost(&tcb, OS_OPT_POST_NONE, &err);   /* ISR 可用 */
OSTaskSemSet(&tcb, 0, &err);

等价于 FreeRTOS 任务通知,同样更快更省——III 官方推荐优先使用。

RT-Thread
rt_sem_t s = rt_sem_create("s", 0, RT_IPC_FLAG_PRIO);   /* 动态 */
struct rt_semaphore ss;
rt_sem_init(&ss, "s", 0, RT_IPC_FLAG_FIFO);             /* 静态 */

rt_sem_take(s, rt_tick_from_millisecond(100));
rt_sem_release(s);                    /* ISR 中安全 */
rt_sem_control(s, RT_IPC_CMD_RESET, (void*)0);  /* 复位计数 */
  • 计数值溢出保护:release 到上限(RT_UINT16_MAX)时返回 -RT_EFULL
  • 等待队列 PRIO(优先级)或 FIFO 可选——与 mutex 同一套 rt_ipc_object 基础设施。
Zephyr
K_SEM_DEFINE(s, 0, 10);               /* 静态:初值0,上限10 */
struct k_sem s2; k_sem_init(&s2, 4, 4);

k_sem_take(&s, K_MSEC(100));
k_sem_give(&s);                       /* ISR 可用 */
unsigned int n = k_sem_count_get(&s);
k_sem_reset(&s);

Zephyr 还提供一族补充同步原语,可视为信号量的"特化版本":

  • k_event:32 位事件标志组,k_event_wait(&e, mask, reset_all, timeout)——对应 FreeRTOS EventGroup / μC/OS Flag;
  • k_poll:等待多个内核对象(信号量、队列、事件)中任意一个就绪——嵌入式版的 select/poll,多路复用等待;
  • k_futex:用户态(userspace)线程的轻量等待原语,k_futex_wait/wake,语义照搬 Linux futex。
ThreadX
TX_SEMAPHORE s;

tx_semaphore_create(&s, "RxSem", 0);       /* 初值 0 */
tx_semaphore_get(&s, TX_WAIT_FOREVER);     /* tick 超时 / TX_NO_WAIT */
tx_semaphore_put(&s);                      /* ISR 可 put */
tx_semaphore_delete(&s);
tx_semaphore_info_get(&s, ...);            /* 查询计数/等待者 */
tx_semaphore_ceiling_put(&s, 1);           /* 带上限的 put:防计数溢出 */
tx_semaphore_put_notify(&s, notify_cb);    /* 特色:put 时触发注册回调 */
  • ISR 规则精确:put 允许;get 仅允许 TX_NO_WAIT(拿不到立即返回,不阻塞);
  • tx_semaphore_ceiling_put 防止生产者溢出计数上限——对应 RT-Thread 的 -RT_EFULL 保护,但做成了显式 API;
  • tx_semaphore_put_notify 是 ThreadX 特色:post 动作直接触发回调函数,可省掉一个专职等待任务——注意回调运行在调用 put 的一方上下文中(ISR 中 put 则运行在 ISR 上下文),必须短小且不能阻塞。
NuttX
/* ① 应用层(POSIX) */
sem_t s;
sem_init(&s, 0, 0);
sem_wait(&s);                /* 可中断 */
sem_trywait(&s);  sem_timedwait(&s, &ts);
sem_post(&s);                /* ISR 可 post */

/* ② 内核/驱动层 */
nxsem_t sem;
nxsem_init(&sem, 0, 0);
nxsem_wait(&sem);  nxsem_trywait(&sem);  nxsem_tickwait(&sem, delay);
nxsem_post(&sem);            /* 中断安全 */
  • 双层语义一致,内核层少 POSIX 校验开销;
  • NuttX 的信号量是其实现 mutex、cond、mqueue 等高层对象的基础原语(nxmutex 即构建在 nxsem 之上)。
LiteOS
UINT32 semHandle;

LOS_SemCreate(4, &semHandle);          /* 计数信号量,初值 4 */
LOS_BinarySemCreate(1, &semHandle);    /* 二值信号量单独建模 */

LOS_SemPend(semHandle, LOS_WAIT_FOREVER);   /* tick 超时 / LOS_NO_WAIT */
LOS_SemPost(semHandle);                     /* ISR 可 post */
LOS_SemDelete(semHandle);
UINT32 cnt = LOS_SemValueGet(semHandle);    /* 读当前计数 */
  • 二值与计数分开建模是 LiteOS 的语义洁癖:LOS_BinarySemCreate(1) 语义 = 初值可用的二值锁旗标,LOS_SemCreate 用于计数——比"计数信号当初值 1"的隐式用法更不易误用;
  • 但注意:二值信号量依然不是互斥锁(无所有权、无 PI),互斥请用 LOS_Mux*
AliOS Things(Rhino)
/* ① 原生层 */
ksem_t s;
krhino_sem_create(&s, "s", 0);
krhino_sem_take(&s, RHINO_WAIT_FOREVER);
krhino_sem_give(&s);                /* ISR 可 give */
krhino_sem_give_all(&s);            /* 广播:唤醒所有等待者 */
krhino_sem_count_get(&s, &cnt);
krhino_sem_del(&s);

/* ② AOS 抽象层 */
aos_sem_t as;
aos_sem_new(&as, 0);
aos_sem_wait(&as, AOS_WAIT_FOREVER);   /* ms 超时 */
aos_sem_signal(&as);                   /* ISR 可用 */
aos_sem_free(&as);
  • krhino_sem_give_all 广播对应 μC/OS-III 的 OS_OPT_POST_ALL
  • 事件标志(krhino_evflags_*)与定长消息队列(krhino_buf_queue_*)是 Rhino 体系中与信号量配套的主力同步对象(见下表)。
对照小结(10 系统)
OS 创建 take/give ISR give 特色
Linux 内核 sema_init down(_interruptible)/up 可(up 不阻塞) 地位已被 mutex 取代,rwsem 是其主力形态
Linux 用户态 sem_init/sem_open sem_wait/post 可跨进程(共享内存/具名)
FreeRTOS xSemaphoreCreateBinary/Counting xSemaphoreTake/Give(FromISR) 底层 = 队列;一对一通知改用任务通知(快 45%)
μC/OS-II OSSemCreate OSSemPend/Post OS_EVENT + 优先级位图
μC/OS-III OSSemCreate OSSemPend/Post POST_ALL 广播;任务内建 OSTaskSem
RT-Thread rt_sem_create/init rt_sem_take/release 溢出保护;PRIO/FIFO 排队
Zephyr K_SEM_DEFINE/k_sem_init k_sem_take/give 另有 k_event / k_poll / k_futex 补充
ThreadX tx_semaphore_create tx_semaphore_get/put 可(get 仅 NO_WAIT) ceiling_put 防溢出;put_notify 唤醒回调
NuttX sem_init / nxsem_init (nx)sem_wait/post POSIX + 内核双层;是 nxmutex 等对象的基座
LiteOS LOS_SemCreate / LOS_BinarySemCreate LOS_SemPend/Post 二值/计数分开建模;句柄式管理
AliOS Things krhino_sem_create / aos_sem_new take/give;give_all 广播 双层 API(tick vs ms,注意单位)

3.6 读写锁(RWLock)

原理深入

读写锁把"互斥"细化为两种模式:读共享(任意多读者同时持有)+ 写独占(一个写者排斥一切)。它成立的前提是读操作远多于写操作——否则写者等待期间读者不断进入(读者优先实现)会造成写者饥饿,或者反过来(写者优先实现)造成读延迟抖动。

写者饥饿是 rwlock 的原生缺陷,各实现的应对:

  • Linux rwlock(自旋版):后期改为类 ticket 的排队机制,新读者在写者排队后不得插队;
  • Linux rwsem:乐观自旋 + 等待队列混合,写者排队后有抢断限制;
  • seqlock:干脆允许写者不阻塞读者,让读者"读错就重读"(见下)。
Linux:三种形态

① 自旋读写锁 rwlock_t——短临界区,可用于 ISR:

DEFINE_RWLOCK(lock);

read_lock(&lock);      read_unlock(&lock);
write_lock(&lock);     write_unlock(&lock);
/* 全套 _irq/_irqsave/_bh 变体,规则与 spinlock 相同 */
read_lock_irqsave(&lock, flags);
write_unlock_irqrestore(&lock, flags);

② 读写信号量 struct rw_semaphore——长临界区,进程上下文(可睡眠):

DECLARE_RWSEM(rwsem);

down_read(&rwsem);   up_read(&rwsem);
down_write(&rwsem);  up_write(&rwsem);
down_write_trylock(&rwsem);
downgrade_write(&rwsem);   /* 写→读降级,无需先释放再竞争——
                              避免"放写锁后被新写者抢断"的窗口 */

内核最著名的 rwsem 是内存管理的 mmap_lock(旧名 mmap_sem)——页 fault(读)与 mmap/munmap(写)的并发控制,直接决定系统内存路径性能,近年内核大量优化(per-CPU 计数读侧、乐观自旋)都围着它做。

③ seqlock(顺序锁)——"写少到几乎没有"时的极致优化:

DEFINE_SEQLOCK(sl);

/* 写者:互斥进入(写者间仍需互斥),序号 +1 */
write_seqlock(&sl);  ...  write_sequnlock(&sl);

/* 读者:不加锁!读到奇数序号或前后序号不一致就重试 */
unsigned seq;
do {
    seq = read_seqbegin(&sl);
    /* 读数据 */
} while (read_seqretry(&sl, seq));

典型用户:jiffies_64 更新、时间戳、统计快照。限制:被保护数据不能含指针(读者重读时指针可能已释放);读者重试循环在写频繁时会饿死——它只赢"读极多、写极少、数据小"的场景。

NuttX:唯一的标准 RWLock

NuttX 是全部嵌入式 RTOS 中唯一提供 POSIX 标准读写锁的:

pthread_rwlock_t rw = PTHREAD_RWLOCK_INITIALIZER;

pthread_rwlock_rdlock(&rw);    pthread_rwlock_unlock(&rw);
pthread_rwlock_wrlock(&rw);    pthread_rwlock_unlock(&rw);
pthread_rwlock_tryrdlock(&rw); pthread_rwlock_trywrlock(&rw);
pthread_rwlock_timedrdlock(&rw, &ts);   /* 带超时变体 */

实现基于 nxsem 体系,读者计数 + 写者互斥的经典结构;随 POSIX 语义天然只能在任务上下文使用。

RT-Thread
struct rt_rwlock lock;
rt_rwlock_init(&lock, "lock", RT_IPC_FLAG_PRIO);   /* 或 rt_rwlock_create 动态 */

rt_rwlock_read_lock(&lock, RT_WAITING_FOREVER);
rt_rwlock_read_unlock(&lock);
rt_rwlock_write_lock(&lock, RT_WAITING_FOREVER);
rt_rwlock_write_unlock(&lock);

基于信号量实现的组件级 rwlock,POSIX 风格命名,随 RT-Thread 组件配置启用。

其余系统:无原生,自构造或换思路
OS 情况
FreeRTOS 无原生 rwlock。读者写者问题一般用:① 直接用 mutex(读也互斥,简单可靠)② 自构造"读者计数 + 二值信号量"经典方案 ③ 读多写少场景改用双缓冲/指针交换(无锁思路)
μC/OS-II/III 无原生。同上;III 用户常自封装
Zephyr 无原生 rwlock,但有 k_condvar 条件变量,配合 k_mutex 十几行即可实现读者优先或写者优先的 rwlock
ThreadX 无原生。mutex 或事件标志组合替代
LiteOS 无原生。LOS_Mux + 计数器自构造
AliOS Things 无原生。mutex 或事件标志组合
自构造参考(嵌入式通用模式)
/* 读者优先 rwlock:一个 mutex 保护读者计数,一个信号量挡写者 */
typedef struct { SemaphoreHandle_t cnt_mtx; SemaphoreHandle_t wr; int readers; } rwlock_t;

void read_lock(rwlock_t *l) {
    xSemaphoreTake(l->cnt_mtx, portMAX_DELAY);
    if (++l->readers == 1) xSemaphoreTake(l->wr, portMAX_DELAY); /* 第一个读者挡写者 */
    xSemaphoreGive(l->cnt_mtx);
}
void read_unlock(rwlock_t *l) {
    xSemaphoreTake(l->cnt_mtx, portMAX_DELAY);
    if (--l->readers == 0) xSemaphoreGive(l->wr);
    xSemaphoreGive(l->cnt_mtx);
}
/* write_lock = take(wr);write_unlock = give(wr)。注意写者饥饿问题 */

工程建议:嵌入式场景里"双缓冲 + 原子指针交换"通常比 rwlock 更好——读者永远读旧缓冲,写者写新缓冲后一次性切换指针,零阻塞零饥饿。


3.7 原子操作(Atomic Operations)

原理深入

原子操作把 read-modify-write 交给硬件指令保证不可分割,从而在"单个变量"粒度上完全绕开锁。硬件实现分两代:

架构 机制 特点
Cortex-M0/M0+/M1(ARMv6-M) 无独占指令 GCC __atomic_* 在此类核心上退化为"关中断实现"__atomic_always_lock_free 返回假)——PY32F071 等 M0+ 平台的"原子操作"实质是隐式临界区,开销和实时性影响要按关中断评估
Cortex-M3/M4/M7(ARMv7-M) LDREX/STREX(独占加载/条件存储) LL/SC 范式:加载时监视地址,存储时若被干扰则失败重试。天然构成 CAS 循环
ARMv8.1+(LSE 扩展) LDADD/LDCLR/CASAL 单指令完成,无重试循环,大核数下性能远好于 LL/SC
x86 LOCK 前缀(lock inclock cmpxchg 锁缓存行/总线,保证 RMW 原子

CAS(Compare-And-Swap)是万能原语cmpxchg(addr, old, new) 语义为"若 *addr == old 则写入 new,返回是否成功"。任何单变量的无锁算法都能用它构造(计数、栈指针、链表头)。同时记住两点:

  1. 原子操作只保证操作本身不可分割,不保证顺序——需要顺序时选带 acquire/release 语义的变体或另加屏障(见 3.10);
  2. ABA 问题:值从 A→B→A 变回去,CAS 检测不到中间变化。指针场景用"指针+计数器"打包(双字 CAS)或改用 RCU 解决。
Linux:最完整的原子类型体系
/* ① 整型原子量 */
atomic_t v = ATOMIC_INIT(0);
atomic_set(&v, 10);
int i = atomic_read(&v);
atomic_inc(&v);  atomic_dec(&v);
atomic_add(5, &v);  atomic_sub(3, &v);

/* ② 操作并返回(返回值参与判断时用) */
atomic_inc_return(&v);
atomic_inc_and_test(&v);        /* 加后为 0 返回真 */
atomic_dec_and_test(&v);        /* 减后为 0 返回真——引用计数"归零即释放"的标准模式 */
atomic_add_unless(&v, 1, 0);    /* 除非为 0 否则加 1 */

/* ③ CAS 原语 */
atomic_cmpxchg(&v, old, new);
atomic_xchg(&v, new);

/* ④ 位操作(天然原子,常用于标志位图) */
set_bit(3, &flags);
clear_bit(3, &flags);
test_and_set_bit(3, &flags);    /* 返回旧值——单标志位"锁"的经典实现 */
test_and_clear_bit(3, &flags);
test_bit(3, &flags);

/* ⑤ 引用计数封装(推荐替代手写 atomic 计数) */
refcount_t r;  refcount_set(&r, 1);
refcount_inc(&r);
if (refcount_dec_and_test(&r))  free(obj);   /* 防溢出/防 UAF,越界时 WARN */
/* 带释放回调的高层封装 */
struct kref k;  kref_init(&k);  kref_get(&k);  kref_put(&k, release_fn);

/* ⑥ 64 位与长整型 */
atomic64_t / atomic_long_t   /* 32 位平台上 atomic64 会自动改用锁实现,注意性能差异 */

使用军规atomic_* 不带顺序语义时编译器仍可乱序,内核惯例是搭配 smp_mb() 或使用 *_relaxed/_acquire/_release 后缀变体(新代码推荐显式语义版本)。

C11 标准层:跨平台正解

裸机/RTOS 工程里,stdatomic.h 才是可移植写法——GCC/Clang/IAR 都支持,在 M3/M4/M7 上编译为 LDREX/STREX 序列,在 M0/M0+ 上自动退化为关中断实现(对调用方透明,但开销不同):

#include <stdatomic.h>

_Atomic uint32_t g_counter;
atomic_fetch_add(&g_counter, 1);                    /* seq_cst 默认 */
atomic_fetch_add_explicit(&g_counter, 1, memory_order_relaxed);  /* 显式宽松序 */

uint32_t expected = 0;
atomic_compare_exchange_strong(&g_counter, &expected, 1);   /* CAS */

atomic_flag f = ATOMIC_FLAG_INIT;
while (atomic_flag_test_and_set(&f));   /* 自旋锁的最小实现 */
atomic_flag_clear(&f);
GCC 内建(不支持 C11 的老工程)
__sync_fetch_and_add(&x, 1);        /* 旧接口(全屏障语义) */
__sync_bool_compare_and_swap(&x, old, new);
__atomic_fetch_add(&x, 1, __ATOMIC_RELAXED);   /* 新接口,可指定内存序 */
__atomic_compare_exchange_n(&x, &old, new, 0, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
RT-Thread:ATOMIC 组件
/* menuconfig: RT_USING_ATOMIC */
rt_atomic_t a;

rt_atomic_add(&a, 1);      rt_atomic_sub(&a, 1);
rt_atomic_inc(&a);         rt_atomic_dec(&a);
rt_atomic_or(&a, mask);    rt_atomic_and(&a, mask);  rt_atomic_xor(&a, mask);
rt_atomic_compare_exchange_strong(&a, &old, new);    /* CAS */
rt_atomic_flag_test_and_set(&a);   rt_atomic_flag_clear(&a);
rt_atomic_load(&a) / rt_atomic_store(&a, v);

实现:硬件支持时走指令,否则自动退化为关中断——上层代码无感,这是 RT-Thread 给移植性做的兜底。

Zephyr:sys/atomic.h
atomic_t v = ATOMIC_INIT(0);

atomic_inc(&v);  atomic_dec(&v);
atomic_add(&v, 5);  atomic_sub(&v, 3);
atomic_set(&v, 0);  atomic_get(&v);
atomic_cas(&v, old, new);           /* 返回布尔 */
atomic_set_bit(&v, n);  atomic_clear_bit(&v, n);
atomic_test_bit(&v, n);
atomic_test_and_set_bit(&v, n);  atomic_test_and_clear_bit(&v, n);

三种实现由 Kconfig 选择:CONFIG_ATOMIC_OPERATIONS_BUILTIN(编译器内建,默认推荐)、CONFIG_ATOMIC_OPERATIONS_C(通用 C + 单全局锁兜底,无原子指令的目标用)、CONFIG_ATOMIC_OPERATIONS_ARCH(架构汇编写死)。

LiteOS:RTOS 中少见的官方原子 API 全套

LiteOS 是 Linux/RT-Thread/Zephyr 之外唯一提供官方原子操作 API 的 RTOS(los_atomic.h):

UINT32 v = 0;

LOS_AtomicAdd(&v, 1);       LOS_AtomicSub(&v, 1);
LOS_AtomicInc(&v);          LOS_AtomicDec(&v);
LOS_AtomicSet(&v, 10);      UINT32 x = LOS_AtomicRead(&v);

LOS_AtomicXchg32bits(&v, new);          /* 原子交换,返回旧值 */
LOS_CmpXchg32bits(&v, new, old);        /* CAS:*v==old 则写 new */
/* 64 位版本:LOS_Atomic64Add / LOS_AtomicXchg64bits / LOS_CmpXchg64bits */

注意:LOS_Atomic* 保证不可分割但不含内存顺序语义,跨核传递数据时仍需自行配屏障(见 3.10)。

其余系统:无封装,三条路
  1. C11 stdatomic.h / __atomic_* 内建——首选(FreeRTOS、μC/OS、ThreadX、NuttX、AliOS Things 均走此路径);
  2. 进临界区包住 RMW——万能但重;
  3. 对齐单字读写天然原子——仅"读"或"写"本身原子,flag = 1 这类可以裸写;但 flag++ 不行。ARM 文档明确:对齐的字/半字/字节 Load/Store 单指令不可分割(中断只能发生在指令之间)。
对照小结(10 系统)
OS 接口 实现 备注
Linux atomic_t/atomic64_t、bitops、refcount/kref LDREX/STREX 或 LSE、LOCK 前缀 _relaxed/_acquire/_release 语义变体
RT-Thread rt_atomic_* 指令优先,无指令退化关中断 需开 RT_USING_ATOMIC
Zephyr atomic_* builtin/C/arch 三种实现可选 cas 返回布尔
LiteOS LOS_Atomic*LOS_CmpXchg32bits(+64 位版) 硬件指令 不含顺序语义,需自配屏障
NuttX C11 stdatomic.h 标准路径 编译器生成
ThreadX C11 / __atomic_* 自行解决 编译器生成
AliOS Things C11 / __atomic_* 自行解决 编译器生成
FreeRTOS 无(用 C11/__atomic_* 编译器生成 SMP 内核尤其需要
μC/OS-II/III 无(用 C11/临界区) 同上
通用 C11 stdatomic.h 跨平台 裸机/RTOS 首选写法

3.8 无锁数据结构(Lock-Free / Per-CPU / RCU)

思路总览

锁的代价:上下文切换、优先级反转、死锁风险、ISR 不可用。无锁路线用三种策略绕开:

  1. 私有化:每核/每任务一份副本,物理上消除共享(per-CPU);
  2. 版本化:写者制作新副本再原子切换指针,读者永远读一个自洽版本(RCU、双缓冲、seqlock);
  3. CAS 循环:用原子指令竞争更新,冲突者重试(无锁栈/队列;SPSC 队列甚至不需要 CAS,靠内存屏障即可)。

嵌入式最实用的结论先行:单生产者单消费者(SPSC)环形缓冲 + 正确的内存屏障,可以免锁解决 80% 的"ISR 与任务共享数据"问题——包括日志、串口收发、采样数据上传,既快又绝无优先级反转。

Linux:四大无锁武器

① Per-CPU 变量——从源头消灭共享:

DEFINE_PER_CPU(int, pkt_cnt);            /* 每个核一份 */
int *p = get_cpu_var(pkt_cnt);           /* = preempt_disable + 取本核副本 */
(*p)++;
put_cpu_var(pkt_cnt);                    /* preempt_enable */

this_cpu_inc(pkt_cnt);                   /* 单指令 RMW 版,无需关抢占 */
int x = per_cpu(pkt_cnt, 3);             /* 访问 3 号核的副本(统计汇总用) */
for_each_possible_cpu(cpu) sum += per_cpu(pkt_cnt, cpu);

典型应用:网络栈计数器、slub 分配器每核缓存、调度器运行队列。注意 get_cpu_var 的关抢占只是保证"访问期间不换核",并不挡别的核访问它自己的副本——本来就不需要。

② RCU(Read-Copy-Update)——Linux 独有的精密武器:

/* 读者:几乎零开销(不阻塞、不原子操作,RT 内核外就是关抢占/计数) */
rcu_read_lock();
p = rcu_dereference(g_ptr);        /* 带依赖屏障的指针读 */
if (p) use(p->data);
rcu_read_unlock();

/* 写者:制作新副本 → 原子发布指针 → 等待"宽限期" → 释放旧副本 */
new = kmalloc(...);  *new = *old;  new->field = x;
rcu_assign_pointer(g_ptr, new);    /* 带 release 语义的发布 */
synchronize_rcu();                 /* 等所有在临界区里的读者离开 */
kfree(old);
/* 或异步回收,不阻塞写者: */
call_rcu(&old->rcu, free_fn);

核心思想:读者从不等待写者;写者靠"宽限期"(所有 CPU 都经历过一次上下文切换/静止状态,意味着老读者必然已退出)保证旧副本释放时无人引用。典型用户:路由表、netfilter、fd 表、dentry 缓存——读 : 写 > 1000:1 的场景。链表/哈希还有专用 RCU 版本 API(list_add_rcuhlist_for_each_entry_rcu)。

③ seqlock:见 3.6——本质也是"版本化"思想的特化。

④ kfifo / lockless ring

DECLARE_KFIFO(fifo, char, 256);
kfifo_in(&fifo, buf, len);
kfifo_out(&fifo, buf, len);
/* 单生产单消费场景下,kfifo 在配对使用内存屏障后是免锁的;
   多生产/多消费仍需外加锁(内核提供 __kfifo + 自管锁的原始版) */

perf、ftrace 的 ring buffer 是手工内存屏障版无锁环的教科书实现(reader 页不可写回绕、writer 保留提交协议)。

RTOS 侧的免锁武器

FreeRTOS:StreamBuffer / MessageBuffer——官方承诺单写任务(或 ISR)+ 单读任务(或 ISR)场景免锁

StreamBufferHandle_t sb = xStreamBufferCreate(1024, 1);  /* 容量, 触发水位 */
xStreamBufferSend(sb, data, len, timeout);      /* 任务或 FromISR 版 */
xStreamBufferReceive(sb, buf, len, timeout);
/* MessageBuffer 版带消息边界(长度前缀),Stream 版是裸字节流 */

实现上仍是"无显式锁 + 临界区极短"的设计,而非真 lockless;多写者必须外加互斥。

RT-Thread:ringbuffer + pipe

struct rt_ringbuffer rb;
rt_ringbuffer_init(&rb, pool, size);
rt_ringbuffer_put(&rb, data, len);    /* SPSC 下免锁 */
rt_ringbuffer_get(&rb, buf, len);
rt_ringbuffer_putchar/getchar;        /* 字节级版 */
/* rt_pipe:带阻塞等待和回调的设备级封装,串口驱动大量使用 */

Zephyr:ring_buf + MPSC 队列

RING_BUF_DECLARE(rb, 256);
ring_buf_put(&rb, data, len);    /* SPSC 下免锁(ISR↔线程常用) */
ring_buf_get(&rb, buf, len);
/* ring_buf_item_put/get:带类型+长度头的消息版 */

μC/OS-III:无免锁承诺对象,OSQ/OSTaskQ 内部走临界区保护——SPSC 场景建议自带环形缓冲。

其他主线系统

  • ThreadX:无免锁对象;SPSC 场景自实现环形缓冲(索引更新用 TX_DISABLE/TX_RESTORE 保护或 C11 原子);
  • NuttX:无 RCU;链表工具宏 sq_addlast/dq_addfirst 等(queue.h)需自管锁;SPSC 自实现;
  • LiteOSLOS_QueueWrite/Read 内部走临界区保护,无免锁承诺;SPSC 自实现;
  • AliOS Thingskrhino_buf_queue 内部有锁;SPSC 自实现。
手写 SPSC 环形缓冲(跨平台模板)
/* 关键:head 只由写者更新,tail 只由读者更新;发布用 release,观察用 acquire */
typedef struct { uint8_t buf[N]; _Atomic uint32_t head, tail; } spsc_t;

bool spsc_put(spsc_t *q, uint8_t d) {
    uint32_t h = atomic_load_explicit(&q->head, memory_order_relaxed);
    uint32_t t = atomic_load_explicit(&q->tail, memory_order_acquire);
    if (h - t >= N) return false;                 /* 满 */
    q->buf[h % N] = d;
    atomic_store_explicit(&q->head, h + 1, memory_order_release);  /* 数据先就位再发布 */
    return true;
}
bool spsc_get(spsc_t *q, uint8_t *d) {
    uint32_t t = atomic_load_explicit(&q->tail, memory_order_relaxed);
    uint32_t h = atomic_load_explicit(&q->head, memory_order_acquire); /* 先看到数据再看 head */
    if (t == h) return false;                     /* 空 */
    *d = q->buf[t % N];
    atomic_store_explicit(&q->tail, t + 1, memory_order_release);
    return true;
}

要点:① 一头一尾各归一方独占写,是免锁成立的根本;② 内存序必须正确——否则弱序架构(ARM)上读者可能先看到 head 更新、后读到旧数据;③ 在单核 ISR↔任务场景,严格说可以用 volatile + 关中断临界点替代,但写成 acquire/release 版本在任何平台都对。

对照小结(10 系统)
系统 per-CPU / RCU 类 SPSC 免锁缓冲
Linux per-CPU 全套 + RCU 全家桶 + seqlock kfifo
FreeRTOS StreamBuffer/MessageBuffer(官方承诺 SPSC 免锁)
μC/OS-III 自实现
RT-Thread rt_ringbuffer(SPSC 免锁)
Zephyr ring_buf(SPSC 免锁)
ThreadX 自实现
NuttX 无 RCU 自实现
LiteOS 自实现
AliOS Things 自实现

3.9 临界区(Critical Section)通用抽象

同名不同义:移植最大的坑

"临界区"这个词在各 OS 里有三种不同实现,跨 OS 移植时不辨语义直接替换 API,是竞态 bug 的高发源头:

OS 名为 “Critical” 的 API 实际实现 挡任务切换 挡中断
FreeRTOS taskENTER_CRITICAL 关中断(BASEPRI 阈值) ✓(部分优先级)
μC/OS-II OS_ENTER_CRITICAL 关中断
μC/OS-III CPU_CRITICAL_ENTER 关中断
μC/OS-III OS_CRITICAL_ENTER 可配置:直接模式=关中断;延迟提交模式=关调度 视配置 视配置
RT-Thread rt_enter_critical 关调度 ✗(中断照进!)
Zephyr irq_lock 关中断(SMP 下+全局自旋锁)
NuttX enter_critical_section 关中断(SMP 下+全局自旋锁)
ThreadX TX_DISABLE/TX_RESTORE 关中断
LiteOS LOS_IntLock 关中断
AliOS Things RHINO_CRITICAL_ENTER 关中断
Linux 无此名 local_irq_save / preempt_disable 分开提供 分开 分开

最危险的一对:FreeRTOS taskENTER_CRITICAL ↔ RT-Thread rt_enter_critical。前者保护的代码(例如与 UART ISR 共享的环形缓冲索引)直接换成后者编译不报错,运行时 ISR 照样进来改索引——偶发数据错乱,极难复现。

统一抽象层设计(推荐做法)

跨 OS 的组件(协议栈、文件系统、中间件)应当定义自己的原语层,语义取最强公约数——“可嵌套、可在 ISR 使用、同时挡中断与调度”:

/* plat_lock.h —— 语义 = save interrupt & scheduler state */
typedef unsigned long plat_lock_key_t;
plat_lock_key_t plat_lock(void);        /* 保存状态 + 关中断(可嵌套,ISR 可用) */
void plat_unlock(plat_lock_key_t key);  /* 恢复状态 */
void plat_sched_lock(void);             /* 仅关调度(中断不受影响) */
void plat_sched_unlock(void);

映射表 A(经典系统):

抽象 Linux FreeRTOS μC/OS-III RT-Thread Zephyr
plat_lock local_irq_save taskENTER_CRITICAL_FROM_ISR 版语义 CPU_CRITICAL_ENTER rt_hw_interrupt_disable irq_lock
plat_unlock local_irq_restore taskEXIT_CRITICAL_FROM_ISR CPU_CRITICAL_EXIT rt_hw_interrupt_enable(key) irq_unlock(key)
plat_sched_lock preempt_disable vTaskSuspendAll OSSchedLock rt_enter_critical k_sched_lock

映射表 B(其余系统):

抽象 ThreadX NuttX LiteOS AliOS Things
plat_lock tx_interrupt_control irqsave LOS_IntLock RHINO_CRITICAL_ENTER
plat_unlock tx_interrupt_control(old) irqrestore LOS_IntRestore RHINO_CRITICAL_EXIT
plat_sched_lock tx_thread_preemption_change(近似) sched_lock LOS_TaskLock krhino_sched_disable

两个设计要点:

  1. 返回值/key 模式(save/restore)必须贯穿抽象层——RT-Thread 与 Zephyr 的 enable/unlock 本来就是 restore 语义,若抽象层设计成 lock()/unlock() 无参对,在这两个 OS 上无法安全实现;
  2. 文档写清"允许 ISR 调用"——这决定了实现必须选关中断而不是关调度。
临界区使用检查清单
  • 临界区内的代码路径是否全部确定?(不能含条件调用到阻塞 API)
  • 最长执行时间是否有上界?(关中断区按微秒计,用逻辑分析仪/SEGGER SystemView 实测)
  • 进入时是否保存了状态?(支持嵌套)
  • 是否存在"拿着临界区等待条件"的逻辑?(永远错误——等待条件必须用睡眠原语)
  • SMP 版本是否还成立?(单核关中断 ≠ 多核互斥,见 Zephyr irq_lock 的语义升级)

3.10 内存屏障(Memory Barrier)

原理深入

互斥锁解决"不要同时访问",内存屏障解决"按我写的顺序被看到"。乱序来自三层:

  1. 编译器优化:重排、合并、消除、把变量提升进寄存器——volatile 只能治这一层(且不保证顺序);
  2. CPU 乱序执行:ARM 的弱内存模型允许 store buffer 里的写延迟可见、独立地址的访问乱序完成;
  3. 缓存一致性协议延迟:SMP 上本核的写不会瞬间被其他核看到。

两个经典翻车现场:

/* 现场1:发布数据(无锁队列、双缓冲切指针同理) */
buf[0] = data;   flag = 1;          /* 写者 */
while (!flag);   x = buf[0];        /* 读者:ARM 上可能先看到 flag=1 再读 buf,
                                       结果读到旧值!需要写者 wmb / 读者 rmb 配对 */

/* 现场2:DMA 描述符(驱动工程师的家常便饭) */
desc->addr = pa;  desc->len = n;
writel(DOORBELL, 1);                /* 必须 wmb()/dma_wmb(),否则门铃先于描述符
                                       到达,DMA 读到垃圾描述符 */

屏障类型速记:DMB(Data Memory Barrier,保顺序不等完成)、DSB(Data Sync Barrier,等到所有访存完成,改 MPU/缓存策略后必用)、ISB(Instruction Sync Barrier,冲刷流水线,改了代码/映射后必用)。

Linux:分层屏障体系
/* ① 编译器屏障:只挡编译器 */
barrier();

/* ② SMP 屏障:只保证核间顺序(UP 编译为空),配对的成对出现 */
smp_wmb();    /* 写屏障:之前的写先于之后的写被看到 */
smp_rmb();    /* 读屏障 */
smp_mb();     /* 全屏障 */

/* ③ 强制屏障:含对外设/DMA 的顺序(映射到 DSB 等)——驱动用 */
mb();  rmb();  wmb();
dma_wmb();    /* DMA 专用变体 */

/* ④ 带语义的访问(推荐新代码使用,把屏障和访问绑定) */
WRITE_ONCE(x, v);        /* 防编译器拆/并/发明写——不是屏障但常一起用 */
READ_ONCE(x);
smp_store_release(&p, v);   /* 此写之前的所有写,先于本写被看到 */
smp_load_acquire(&p);       /* 本读之后的所有读,不得提前 */

/* ⑤ 无锁编程的隐含屏障 */
smp_mb__before_atomic();  smp_mb__after_atomic();   /* 配原子操作 */

Linux 内核文档 memory-barriers.txt 的核心结论:屏障必须成对使用——写者 release 配对读者 acquire;只在一边放屏障等于没放。所有 spin_lock/mutex_lock 内部隐含 acquire,unlock 隐含 release——这就是"用锁就不用管乱序"的原因。

RTOS 层:CMSIS / 编译器内建

Zephyr、FreeRTOS、μC/OS、RT-Thread、ThreadX、NuttX、LiteOS、AliOS Things 均无独立屏障封装,直接使用 CMSIS 或编译器内建:

#include <cmsis_core.h>
__DMB();   /* 顺序:无锁队列发布、描述符更新 */
__DSB();   /* 完成:改 MPU 配置、开关 cache 后 */
__ISB();   /* 流水线:自修改代码、向量表重定位后 */

__sync_synchronize();        /* 编译器全屏障(ARM 上 = DMB) */
C11 标准层
atomic_thread_fence(memory_order_release);   /* 释放屏障 */
atomic_thread_fence(memory_order_acquire);   /* 获取屏障 */
atomic_signal_fence(memory_order_seq_cst);   /* 仅编译器级(信号处理用) */
/* 更推荐:直接给原子变量操作指定内存序(见 3.7),比裸 fence 更精确 */
跨 OS 对照(10 系统)与军规
OS 屏障接口 备注
Linux smp_*mb/rmb/wmbdma_*mbREAD_ONCE/WRITE_ONCE、acquire/release 变体 体系最完整,屏障成对原则
FreeRTOS 无封装,CMSIS __DMB/__DSB/__ISB SMP 版需要格外注意
μC/OS-II/III 同上 单核场景主要防编译器(barrier/volatile)
RT-Thread 同上(部分 BSP 提供 rt_hw_dmb/dsb 驱动 DMA 场景必用
Zephyr CMSIS 宏 SMP 下 k_spin_* 已含正确屏障
ThreadX 无封装,CMSIS/__sync_synchronize()
NuttX C11 fence / 架构宏 SMP 下 spinlock 已含屏障
LiteOS 无封装,CMSIS 宏 LOS_Atomic* 不含顺序语义,需自配屏障
AliOS Things 无封装,CMSIS 宏 SMP 原生内核,手写无锁代码必须配屏障
通用 C11 atomic_thread_fence / 内存序参数 可移植首选

军规

  1. 用了锁,就不用额外管顺序(锁内含 acquire/release);
  2. 绕过锁(无锁队列、标志位通知、双缓冲),就必须成对补屏障;
  3. volatile ≠ 屏障 ≠ 原子——它只承诺"每次都去内存读",三样各管各的;
  4. DMA 场景永远用 wmb()/dma_wmb() 再敲门铃寄存器,先 flush cache 再交描述符。

4. 优先级反转与解决方案

问题建模

经典三任务场景:低优先级 L 持有互斥锁 → 高优先级 H 来拿锁被阻塞 → 中优先级 M(不需要锁)抢占 L → H 实际在等 M。M 们可以无限接力,H 的阻塞时间无界——实时性崩溃。1997 年火星探路者号的总线互斥锁反转导致系统反复复位,最后靠开启优先级继承修复,是教科书案例。

方案一:优先级继承(PI)

等待发生时,持有者临时继承等待者中的最高优先级,放锁后恢复。反转窗口被压缩到"L 以 H 的优先级跑完临界区"的时间。

OS PI 实现 完整度
Linux 内核 mutex 底层走 rtmutex 基础设施(PI 等待链用红黑树管理) 完整,支持链式提升;用户态经 FUTEX_LOCK_PI
Linux 用户态 pthread_mutexattr_setprotocol(PTHREAD_PRIO_INHERIT) 需显式开启,默认关闭
μC/OS-III mutex 内建,支持多级链式继承(A→B→C 一路提升) 完整
RT-Thread mutex 内建 PI 完整
Zephyr mutex 内建 PI 完整
FreeRTOS mutex 内建简化版 PI 有局限:继承不沿等待链传递(A 等 B、B 等 C 时 C 不会被提升);同一任务持多把 mutex 时优先级恢复逻辑简单,可能提前回到基准优先级——复杂嵌套持锁场景下不彻底,是与 μC/OS-III、Linux rtmutex 的公认差距点
ThreadX 创建时 TX_INHERIT 显式开启 完整(注意默认不继承,需显式选择)
NuttX CONFIG_PRIORITY_INHERITANCE 全局使能,pthread mutex 生效 完整
LiteOS mux 内建 完整
AliOS Things (Rhino) mutex 内建 完整

方案二:优先级天花板(PCP)

创建锁时指定一个天花板优先级(≥ 所有可能使用者的最高优先级),任何任务拿到锁立即提升。反转根本不发生(持锁者永远跑得比竞争者快),还顺带防死锁。

  • μC/OS-II 的 mutex 就是 PCPOSMutexCreate(prio, &err) 的 prio 参数即天花板;
  • POSIX:pthread_mutexattr_setprotocol(PTHREAD_PRIO_PROTECT) + pthread_mutexattr_setprioceiling
  • 缺点:优先级规划成本高(天花板值是全局资源)、提升过激(无竞争也提升,影响调度公平性)。

方案三:工程规避(所有 OS 通用兜底)

  1. 极短临界区用关中断/关调度——锁不存在,反转无从谈起;
  2. SPSC 无锁队列替代"锁+共享缓冲"(见 3.8)——嵌入式首选;
  3. ISR 永远不拿 mutex(所有 OS 强制)——斩断一条反转链;
  4. 统一优先级访问:让所有访问某资源的任务跑在同一优先级(牺牲灵活性换可分析性);
  5. 持锁时间不调用可能阻塞的 API——反转窗口 = 持锁时间,缩短它永远有效。

5. 选型决策指南

在这里插入图片描述

决策分支对应的各 OS 接口(导图呈现决策骨架,接口细节见下表):

叶子节点 首选接口 说明
原子操作 C11 stdatomic.hatomic_t(Linux)/ rt_atomic_* / Zephyr atomic_* / LOS_Atomic* 见 3.7;M0/M0+ 上是关中断实现
SPSC 无锁队列 StreamBuffer(FreeRTOS)/ rt_ringbuffer / ring_buf(Zephyr)/ 自实现 见 3.8;免锁、无优先级反转
关中断 local_irq_save 及 3.1 各节 save/restore 接口 微秒级临界区专用
自旋锁 spin_lock_irqsave(线程侧)+ spin_lock(ISR 侧) 版本选择决策表见 3.3
互斥锁(带 PI) 各 OS mutex(3.4 节对照表) Linux 驱动用 mutex_lock_interruptible
读写锁 / RCU Linux:rwlock / rwsem / seqlock / RCU;NuttX:pthread_rwlock 其余 RTOS 用双缓冲 + 指针交换,见 3.6
任务通知 xTaskNotifyGive(FromISR) / OSTaskSemPost 一对一 ISR→任务,比信号量快约 45%
信号量 各 OS sem(3.5 节对照表) 多对一 / 计数排队
k_poll / 事件组 Zephyr k_poll;各 OS 事件标志组 等"多个事件任意一个"
关调度 vTaskSuspendAll / OSSchedLock / rt_enter_critical 允许中断、只挡任务切换,见 3.2
抢占阈值 ThreadX tx_thread_preemption_change 允许中断和高优先级任务,只挡低优先级
内存屏障成对 写者 wmb ↔ 读者 rmb,或 release ↔ acquire 见 3.10 军规

跨 OS 通用军规

  1. 能无锁就无锁(SPSC 队列、数据私有化);
  2. 必须上锁就用带 PI 的 mutex,不用二值信号量做互斥;
  3. ISR 里永远不睡眠:只 give、只 notify、只拿 spinlock;
  4. 持锁时间 = 反转窗口 = 延迟上限,能短则短;
  5. 跨 OS 代码抽象自己的原语层(lock/sched_lock/mutex 三级),不直接散布 OS API——尤其小心 rt_enter_critical 这类同名异义词;
  6. 用静态分析/调试工具兜底:Linux lockdep、FreeRTOS configCHECK_FOR_STACK_OVERFLOW+队列断言、各家的 mutex DEBUG 配置。

6. 接口速查总表

总表:10 系统 × 全机制

机制 Linux (kernel) FreeRTOS μC/OS-II μC/OS-III RT-Thread Zephyr ThreadX NuttX LiteOS AliOS Things
关中断 local_irq_save/restore taskENTER/EXIT_CRITICAL OS_ENTER/EXIT_CRITICAL CPU_CRITICAL_ENTER/EXIT rt_hw_interrupt_disable/enable irq_lock/unlock tx_interrupt_control / TX_DISABLE/TX_RESTORE irqsave/irqrestoreenter_critical_section LOS_IntLock/IntRestore RHINO_CRITICAL_ENTER/EXIT
关调度 preempt_disable/enable vTaskSuspendAll/xTaskResumeAll OSSchedLock/Unlock OSSchedLock/Unlock rt_enter/exit_critical k_sched_lock/unlock tx_thread_preemption_change(抢占阈值) sched_lock/unlock LOS_TaskLock/Unlock krhino_sched_disable/enable
自旋锁 spin_lock(_irq/_irqsave/_bh) —(SMP 内核内部) rt_spin_lock(_irqsave)(SMP) k_spin_lock/unlock —(SMP 版内核内部) spin_lock_irqsave 等(SMP) LOS_SpinLock(Save)(LiteOS-A) krhino_spin_lock(_irq_save)(SMP 原生)
互斥锁 mutex_lock(_interruptible) xSemaphoreCreateMutex+Take/Give OSMutexCreate/Pend/Post(天花板) OSMutexCreate/Pend/Post(真 PI+链式) rt_mutex_take/release(PI+递归) k_mutex_lock/unlock(PI+递归) tx_mutex_create(TX_INHERIT)/get/put(PI 可选,递归) pthread_mutex_*/nxmutex/nxrmutex(递归) LOS_MuxCreate/Pend/Post(PI+递归) krhino_mutex_lock/unlock(PI)/aos_mutex_*
信号量 down/up;rwsem xSemaphoreCreateBinary/Counting+FromISR OSSemCreate/Pend/Post OSSem*+POST_ALL+OSTaskSem* rt_sem_take/release k_sem_take/give tx_semaphore_get/putceiling_putput_notify sem_wait/post/nxsem_* LOS_SemPend/PostLOS_BinarySem* krhino_sem_take/give(_all)/aos_sem_*
事件标志 —(用 completion/fd 通知) EventGroup OSFlag* rt_event_* k_event_* tx_event_flags_set/get nxevent(内核内) LOS_EventWrite/Read krhino_evflags_set/get
消息队列 msg_*/mqueue xQueueSend/Receive OSQ* OSQ*OSTaskQ* rt_mb/rt_mq k_msgq_* tx_queue_send/receive/front_send POSIX mqueue、nxmq LOS_QueueWrite/Read krhino_buf_queue_*/aos_queue_*
读写锁 rwlock_t/rw_semaphore/seqlock —(自构造/双缓冲) rt_rwlock_* —(mutex+condvar 构造) pthread_rwlock_*(RTOS 中唯一)
原子操作 atomic_t、bitops、refcount/kref、cmpxchg C11/__atomic_* C11/临界区 C11/临界区 rt_atomic_*(可退化关中断) atomic_*(builtin/C/arch 三实现) C11 自行解决 C11 stdatomic.h LOS_Atomic*LOS_CmpXchg32bits(官方全套) C11 自行解决
无锁结构 per-CPU/RCU/seqlock/kfifo StreamBuffer(SPSC 免锁) 自实现 ring 自实现 ring rt_ringbuffer(SPSC 免锁) ring_bufk_fifo 自实现 自实现 自实现 自实现
任务通知 completion:complete/wait_for_completion xTaskNotifyGive(FromISR)/ulTaskNotifyTake OSTaskSemPend/PostOSTaskQ* rt_mb/rt_mq k_event_*k_pollk_condvark_futex put_notify 回调 pthread_cond、work queue
内存屏障 smp_mb/rmb/wmbmbREAD_ONCE/WRITE_ONCE、acquire/release CMSIS __DMB/__DSB/__ISB 同左 同左 同左(部分 BSP rt_hw_dmb CMSIS 宏;k_spin_* 内含屏障 CMSIS/__sync_synchronize() C11 fence/架构宏 CMSIS 宏 CMSIS 宏
PI 互斥 内建(rtmutex) 内建简化版 优先级天花板 内建(链式) 内建 内建 创建时 TX_INHERIT 可选 CONFIG_PRIORITY_INHERITANCE 全局 内建 内建
SMP 支持 完整 官方 SMP(v11+) 无(商业多核扩展) RT_USING_SMP CONFIG_SMP 有(SMP 版) LiteOS-A 有 原生支持
安全认证 —(有商业 RT 衍生) 生态有 SafeRTOS 有(商业认证版) 有(μC/OS-III for Safety-Critical) 有(睿赛德认证版) 功能安全项目进行中 TÜV SIL4 / ASIL D / 医疗证据链 消费级量产验证(小米 Vela) OpenHarmony 轻量内核 阿里云 IoT 生态

结语

并发保护机制的演化主线很清晰:RTOS 用"关中断/关调度 + 带 PI 的 mutex"打天下(单核、确定性优先),Linux 在 SMP 压力下发展出 spinlock 家族、RCU、per-CPU、seqlock 的精细分层(吞吐优先)。新生代系统则在收敛中各找差异:Zephyr 兼取两家(spinlock、futex、condvar、ring_buf),ThreadX 用抢占阈值做出最精细的调度控制,NuttX 把 POSIX 体验(含唯一的标准 RWLock)塞进 MCU,LiteOS 补上了官方原子操作,Rhino 从第一天就按 SMP 设计内核。

选型时先问三个问题:谁和谁竞争(线程/ISR/核)、临界区多长、读多还是写多——答案基本唯一确定了该用的机制。最后记住一句话:所有锁机制都是为"不得不共享"准备的,最高明的并发保护,是通过设计(私有化、消息传递、SPSC 缓冲)让共享根本不发生。

Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐