通信/互斥/同步的内部机制

队列操作的内部机制

使用“struct mqueue_inode_s”结构体来描述队列，这个结构体如下：

msglist：里面存储按照优先级排序的“消息”
cmn：里面有2个链表，用来记录“等待队列非空”、“等待队列非满”的任务

队列的操作如下图所示：

这个队列最多有2个消息：maxmsgs=2
它现在有2个消息：nmsgs=2
假设任务A调用“mq_send”函数来写队列，它可以阻塞：挂载cmn.waitfornotful链表

创建队列源码情景分析

注意：mq_open函数里，可以传入属性，比如：

struct mq_attr attr;

attr.mq_maxmsg = 10; // 最大消息数

attr.mq_msgsize = 1024; // 单条消息最大长度（字节）

mqd_t mq = mq_open("/myqueue", O_CREAT \| O_RDWR, 0644, &attr);

如果不传入attr，那么消息队列默认可以容纳16个消息，每个消息大小是32字节。

mq_open函数的调用关系如下：

mq_open // nuttx/fs/mqueue/mq_open.c

nxmq_vopen // nuttx/fs/mqueue/mq_open.c

// nuttx/sched/mqueue/mq_msgqalloc.c

ret = nxmq_alloc_msgq(attr, &msgq);

// 分配struct mqueue_inode_s

msgq = (FAR struct mqueue_inode_s *)

kmm_zalloc(sizeof(struct mqueue_inode_s));

msgq-\>maxmsgs = (int16_t)attr-\>mq_maxmsg;

msgq-\>maxmsgsize = (int16_t)attr-\>mq_msgsize;

dq_init(&msgq-\>cmn.waitfornotempty);

dq_init(&msgq-\>cmn.waitfornotfull);

mq_open只是分配了一个“struct mqueue_inode_s”结构体，并没有分配存储消息的空间。这个结构体如下：

msglist：里面存储按照优先级排序的“消息”
cmn：里面有2个链表，用来记录“等待队列非空”、“等待队列非满”的任务

写队列源码情景分析

mq_open函数并没有分配“存储消息的空间”，写队列时：

分配“消息”，把用户传入的数据复制进消息
如果队列已经满，可以阻塞：把自己记录在队列的“waitfornotfull”链表里
当队列非满时，把“消息”按照优先级插入msglist，并且唤醒队列的“waitfornotempty”链表里的任务

以“mq_send”函数为例，分析它的执行过程：

// int mq_send(mqd_t mqdes, const void \*msg, size_t msglen, int prio)

mq_send // nuttx/sched/mqueue/mq_send.c

nxmq_send // nuttx/sched/mqueue/mq_send.c

file_mq_timedsend_internal // nuttx/sched/mqueue/mq_send.c

// 分配消息

mqmsg = nxmq_alloc_msg(msglen);

memcpy(mqmsg-\>mail, msg, msglen);

mqmsg-\>priority = prio;

mqmsg-\>msglen = msglen;

if (msgq-\>nmsgs \>= msgq-\>maxmsgs) {

// a. 非阻塞则立刻返回

// b. 阻塞

ret = nxmq_wait_send(msgq, abstime, ticks);

// 如果有设置超时时间,启动WD(看门狗定时器)

// 参数都是(...,xmq_sndtimeout, (wdparm_t)rtcb)

wd_start_realtime/wd_start

rtcb-\>waitobj = msgq;

msgq-\>cmn.nwaitnotfull++;

// 从就绪链表移除

nxsched_remove_self(rtcb);

// 放入“这个队列的waitfornotfull链表”

rtcb-\>task_state = TSTATE_WAIT_MQNOTFULL;

nxsched_add_prioritized(rtcb, MQ_WNFLIST(msgq-\>cmn));

// 切换任务

up_switch_context(this_task(), rtcb);

}

// 把消息加入队列的msglist链表, 按照优先级排序

nxmq_add_queue(msgq, mqmsg, prio);

msgq->nmsgs++

// 通知等待消息的任务

nxmq_notify_send(msgq);

if (msgq-\>cmn.nwaitnotempty \> 0)

{

// 从队列的waitfornotempty链表取出第1个任务

btcb = (FAR struct tcb_s \*)dq_remfirst(MQ_WNELIST(msgq-\>cmn));

msgq-\>cmn.nwaitnotempty--;

// 唤醒它

if (nxsched_add_readytorun(btcb))

{

up_switch_context(btcb, rtcb);

}

}

读队列源码情景分析

调用mq_receive函数读队列时：

如果发现队列中为空，可以阻塞：把自己记录在队列的“waitfornotempty”链表里
当队列有消息时，从msglist里取出第1个“消息”，并且唤醒队列的“waitfornotfull”链表里的任务
把“消息”的数据复制给APP，释放“消息”

以“mq_receive”函数为例，分析它的执行过程：

// ssize_t mq_receive(mqd_t mqdes, void \*msg, size_t msglen, int \*prio);

mq_receive // nuttx/sched/mqueue/mq_receive.c

nxmq_receive // nuttx/sched/mqueue/mq_receive.c

file_mq_receive // nuttx/sched/mqueue/mq_receive.c

file_mq_timedreceive_internal // nuttx/sched/mqueue/mq_receive.c

// msglist里的第1个消息是否为空

mqmsg = (FAR struct mqueue_msg_s \*)list_remove_head(&msgq-\>msglist);

if (mqmsg == NULL)

{

// a. 非阻塞则立刻返回

// b. 阻塞

nxmq_wait_receive

// 如果有设置超时时间,启动WD(看门狗定时器)

// 参数都是(...,nxmq_rcvtimeout, (wdparm_t)rtcb)

wd_start_realtime/wd_start

msgq-\>cmn.nwaitnotempty++;

// 从就绪链表移除

nxsched_remove_self(rtcb);

// 放入“这个队列的waitfornotempty链表”

rtcb-\>task_state = TSTATE_WAIT_MQNOTEMPTY;

nxsched_add_prioritized(rtcb, MQ_WNELIST(msgq-\>cmn));

// 切换任务

up_switch_context(this_task(), rtcb);

// 读取消息

*rcvmsg = newmsg;

}

msgq->nmsgs--

// 通知队列有空间的任务

nxmq_notify_receive(msgq);

if (msgq-\>cmn.nwaitnotfull \> 0)

{

// 从队列的waitfornotfull链表取出第1个任务

btcb = (FAR struct tcb_s \*)dq_remfirst(MQ_WNFLIST(msgq-\>cmn));

msgq-\>cmn.nwaitnotfull--;

// 唤醒它

if (nxsched_add_readytorun(btcb))

{

up_switch_context(btcb, rtcb);

}

}

// 记录读到的消息的优先级

*prio = mqmsg-\>priority;

// 复制给APP

memcpy(msg, mqmsg-\>mail, mqmsg-\>msglen);

ret = mqmsg-\>msglen;

// 释放消息

nxmq_free_msg(mqmsg);

信号量和互斥量的内部机制

openvel里信号量、互斥量在内核部分使用的是同一套代码：信号量里根据“sem->flags”执行额外的操作——比如“优先级继承”、“优先级保护”。

信号量互斥量机制与代码层次

（1）信号量本质上是一个受保护的计数器，用于控制对有限资源的访问。

互斥 (Mutex)：当信号量初始值为 1 时，它可作为互斥锁，确保在任意时刻只有一个任务能够访问被其保护的临界区代码。
同步 (Synchronization)：任务可以等待（wait）一个信号量，直到另一个任务或中断服务程序（ISR）释放（post）该信号量，从而实现任务间的执行顺序协调。

（2）优先级反转 (Priority Inversion)

优先级反转是抢占式多任务系统中一个常见的问题。当一个高优先级任务需要等待一个被低优先级任务持有的资源（如信号量）时，它将被阻塞。如果此时出现一个中等优先级的任务，它会抢占低优先级任务的执行，导致持有资源的任务迟迟无法释放资源，从而间接阻塞了高优先级任务。

典型场景：

低优先级任务 C 获取信号量，进入临界区。
高优先级任务 A 启动，尝试获取同一信号量，因资源被占用而阻塞。
中优先级任务 B 启动，由于其优先级高于任务 C，它抢占了 CPU。
结果：任务 B（以及其他中等优先级任务）持续运行，导致任务 C 无法执行并释放信号量。这使得高优先级任务 A 被中等优先级任务 B 无限期地阻塞，仿佛任务 A 的优先级低于任务 B。

为解决此问题，openvela引入了优先级保护和优先级继承机制。

（3）信号量使用模式

在 openvela中，信号量主要有两种应用模式：互斥锁模式和信号通知模式。正确区分并使用它们对于系统稳定至关重要。

互斥锁模式 (Locking Semaphore)

此模式用于保护共享资源，实现临界区的互斥访问。

用途：确保同一时间只有一个任务可以访问某个共享数据结构或硬件资源。

特点：获取（wait）和释放（post）信号量的操作由同一个任务完成。

强烈建议启用优先级继承或优先级保护，以避免优先级反转。

信号通知模式 (Signaling Semaphore)

此模式用于任务间的同步或通信。

用途：一个任务（或 ISR）通知另一个任务某个事件已发生。例如，任务 A 等待一个信号量，直到任务 B 完成数据准备后释放该信号量以唤醒任务 A。

特点：获取（wait）和释放（post）信号量的操作由不同的任务（或任务与 ISR）完成。

必须禁用优先级继承。否则，释放信号量的任务 B 可能会错误地继承等待任务 A 的优先级，引发意外的调度行为。

最佳实践：在初始化信号量时，根据其用途通过 sem_setprotocol() 设置正确的优先级策略。

（4）解决优先级反转

openvela提供了两种策略来解决优先级反转问题：优先级保护和优先级继承。

优先级保护 (Priority Protection)

优先级保护基于优先级天花板协议 (Priority Ceiling Protocol) 的思想。

当一个任务获取信号量时，系统会立即将其优先级提升到为该信号量预设的“天花板”优先级。该优先级通常设置为可能访问此信号量的所有任务中的最高优先级。任务释放信号量后，其优先级恢复为原始值。

这种机制实现简单，能有效防止持有锁的任务被其他任务抢占。

配置：通过 Kconfig 选项 CONFIG_PRIORITY_PROTECT 启用此功能。

实现：优先级保护通过两个核心函数 nxsem_protect_wait 和 nxsem_protect_post 实现。

优先级继承 (Priority Inheritance)

优先级继承是一种动态调整优先级的策略。

当一个高优先级任务因等待信号量而被阻塞时，系统会自动将当前持有该信号量的低优先级任务的优先级临时提升至与被阻塞的高优先级任务相同。这使得持有锁的任务能尽快完成其临界区操作并释放锁，从而最大限度地减少高优先级任务的阻塞时间。当锁被释放后，持有任务的优先级将恢复。

由于信号量是 openvela中最基础的同步原语，为它实现优先级继承意味着互斥锁（Mutex）等上层机制也自动具备了此能力。

优先级继承的实现依赖于以下核心数据结构：

① struct sem_s：描述信号量本身，包含计数值、等待任务列表以及持有者信息。

② struct semholder_s：描述信号量的持有关系，将任务控制块 (Task Control Block, TCB) 与其持有的信号量关联起来。

struct sem_s 关键成员：

/* This is the generic semaphore structure. */

struct sem_s

{

volatile int32_t semcount; // >0: 可用资源数; <0: 等待任务数

uint8_t flags; // 协议标志 (SEM_PRIO_INHERIT, etc.)

dq_queue_t waitlist; // 等待任务的优先级队列

#ifdef CONFIG_PRIORITY_INHERITANCE

# if CONFIG_SEM_PREALLOCHOLDERS > 0

FAR struct semholder_s *hhead; // 持有者链表头指针

# else

struct semholder_s holder; // 单一持有者槽位

# endif

#endif

#ifdef CONFIG_PRIORITY_PROTECT

uint8_t ceiling; // 优先级天花板

uint8_t saved; // 任务原始优先级

#endif

};

struct semholder_s 关键成员：

/* This structure contains information about the holder of a semaphore */

struct semholder_s

{

FAR struct semholder_s *flink; // 指向同一信号量的下一个持有者

FAR struct semholder_s *tlink; // 指向同一任务持有的下一个信号量

FAR struct sem_s *sem; // 指向对应的信号量

FAR struct tcb_s *htcb; // 指向持有者任务的TCB

int32_t counts; // 持有计数

};

实现过程：

① 获取与继承：当任务尝试获取一个已被占用的信号量时，系统会将持有该信号量的低优先级任务的优先级提升到当前任务的优先级。

② 恢复：当任务释放信号量时，系统会重新计算其优先级。它会检查该任务是否还持有其他信号量，并根据等待这些信号量的最高优先级任务来设置其新优先级。如果它不再持有任何导致优先级提升的信号量，则其优先级恢复为基础值。

配置：

① CONFIG_PRIORITY_INHERITANCE：启用优先级继承功能。

② CONFIG_SEM_PREALLOCHOLDERS：定义预分配的 semholder_s 结构数量，决定了一个信号量能被多少个不同任务同时持有（适用于计数值大于1的信号量）。

在openvela里，互斥量是“特殊的信号量”，在内核里它们使用同一套代码:

创建信号量/互斥量源码情景分析

在创建互斥量时，比创建信号量多了一些设置：

sem_open：用作“信号量”，“sem->flags = 0”
pthread_mutex_init：用作“互斥量”

相比于sem_open，它把sem的初始值设置为1，并且设置了sem->flags。

如果pthread_mutex_init函数里没有提供attr，则根据内核的配置设置默认的flags：当内核配置了“CONFIG_PRIORITY_INHERITANCE”，则“sem->flags = SEM_TYPE_MUTEX | SEM_PRIO_INHERIT”——使用优先级继承；否则“sem->flags = SEM_TYPE_MUTEX”——不使用优先级继承。

如果pthread_mutex_init函数里提供了attr，则使用attr来设置是使用“优先级继承”还是使用“优先级保护”。如果使用“优先级保护”，还要设置互斥量的“天花板优先级”。

总结：

信号量：sem->flags = 0
没有使用“优先级继承”、“优先级保护”的普通互斥量：sem->flags = SEM_TYPE_MUTEX
使用“优先级继承”的互斥量：sem->flags = SEM_TYPE_MUTEX | SEM_PRIO_INHERIT
使用“优先级保护”的互斥量：sem->flags = SEM_TYPE_MUTEX | SEM_PRIO_PROTECT

这2个函数的调用差异如下图所示：

函数调用关系如下：

sem_open // nuttx/libs/libc/semaphore/sem_open.c

nxsem_open // nuttx/fs/semaphore/sem_open.c

nsem = group_malloc(NULL, sizeof(struct nsem_inode_s));

// nuttx/libs/libc/semaphore/sem_init.c

nxsem_init(&nsem-\>ns_sem, 0, value);

// 参数: (FAR pthread_mutex_t \*mutex, FAR const pthread_mutexattr_t \*attr)

pthread_mutex_init // nuttx/sched/pthread/pthread_mutexinit.c

mutex_init(&mutex-\>mutex); // nuttx/sched/pthread/pthread.h

nxmutex_init // nuttx/libs/libc/misc/lib_mutex.c

nxsem_init(&mutex-\>sem, 0, 1);

mutex-\>holder = NXMUTEX_NO_HOLDER;

#ifdef CONFIG_PRIORITY_INHERITANCE

// nuttx/sched/semaphore/sem_setprotocol.c

nxsem_set_protocol(&mutex-\>sem, SEM_TYPE_MUTEX \| SEM_PRIO_INHERIT);

#else

// nuttx/sched/semaphore/sem_setprotocol.c

nxsem_set_protocol(&mutex-\>sem, SEM_TYPE_MUTEX);

#endif

if (attr)

{

mutex_set_protocol(&mutex-\>mutex, attr-\>proto);

nxsem_set_protocol

if (attr-\>proto == PTHREAD_PRIO_PROTECT)

// 设置优先级天花板

mutex_setprioceiling(&mutex-\>mutex, attr-\>ceiling, NULL);

}

等待信号量/互斥量源码情景分析

使用“sem_wait”等函数来获取信号量；使用“pthread_mutex_lock”来获取互斥量。它们最终都调用“nxsem_wait”函数。

在“nxsem_wait”函数里，根据“sem->flags”进行不同的操作：

sem->flags == 0：除了等待sem的值非0，没有特殊操作
((sem->flags & SEM_PRIO_MASK) == SEM_PRIO_INHERIT)：

如果互斥量被占用了，就提升holder的优先级

((sem->flags & SEM_PRIO_MASK) == SEM_PRIO_PROTECT)：

自己成功获得互斥量后，提升自己的优先级到天花板，尽快执行以便释放互斥量。

“sem_wait”、“pthread_mutex_lock”代码如下，它们都调用“nxsem_wait”：

sem_wait // nuttx\libs\libc\semaphore\sem_wait.c

nxsem_wait // nuttx\sched\semaphore\sem_wait.c

pthread_mutex_lock // nuttx\libs\libc\pthread\pthread_mutex_lock.c

pthread_mutex_timedlock // nuttx\sched\pthread\pthread_mutextimedlock.c

pthread_mutex_take // nuttx/sched/pthread/pthread_mutex.c

mutex_clocklock // nuttx\sched\pthread\pthread.h

nxmutex_clocklock // nuttx/libs/libc/misc/lib_mutex.c

nxsem_clockwait/nxsem_wait

“nxsem_wait”根据sem->flags决定“额外做什么”，代码如下：

nxsem_wait // nuttx\sched\semaphore\sem_wait.c

nxsem_wait_slow // nuttx\sched\semaphore\sem_wait.c

flags = enter_critical_section(); // 进入临界区

if (atomic_fetch_sub(NXSEM_COUNT(sem), 1) \> 0)

{

// 成功获得且配置了CONFIG_PRIORITY_PROTECT

// 就提升当前任务的优先级让它尽快执行以便是否信号量

nxsem_protect_wait // nuttx\sched\semaphore\sem_wait.c

if ((sem-\>flags & SEM_PRIO_MASK) == SEM_PRIO_PROTECT)

// 提升持有者的优先级

rtcb-\>sched_priority = sem-\>ceiling;

nxsem_add_holder(sem);

}

else

{

// 无法获得信号量

// 如果配置了CONFIG_PRIORITY_INHERITANCE并且

// sem-\>flags表示使用SEM_PRIO_INHERIT

// 就提升hoder的优先级

// nuttx\sched\semaphore\sem_holder.c

nxsem_boost_priority(sem);

nxsem_boostholderprio(&sem-\>holder, sem, rtcb);

// 把持有者的优先级提升为当前任务的优先级

nxsched_set_priority(htcb, rtcb-\>sched_priority);

// 从就绪链表移除

nxsched_remove_self(rtcb);

// 放入“这个信号量的等待链表”

rtcb-\>task_state = TSTATE_WAIT_SEM;

nxsched_add_prioritized(rtcb, SEM_WAITLIST(sem));

// 切换任务

up_switch_context(this_task(), rtcb);

释放信号量/互斥量源码情景分析

使用“sem_post”等函数来释放信号量；使用“pthread_mutex_unlock”来释放互斥量。它们最终都调用“nxsem_post”函数。

在“nxsem_post”函数里，根据“sem->flags”进行不同的操作：

sem->flags == 0：让sem的值加1、唤醒等待sem的任务，没有特殊操作
((sem->flags & SEM_PRIO_MASK) == SEM_PRIO_INHERIT)：

要恢复自己的优先级

((sem->flags & SEM_PRIO_MASK) == SEM_PRIO_PROTECT)：

要恢复自己的优先级

“sem_post”、“pthread_mutex_unlock”代码如下，它们都调用“nxsem_post”：

sem_post // nuttx/libs/libc/semaphore/sem_post.c

nxsem_post // nuttx/sched/semaphore/sem_post.c

pthread_mutex_unlock // nuttx/sched/pthread/pthread_mutexunlock.c

// 这就是mutex和semaphore的不同点

// mutex: 必须lock后才能unlock

// 而对于semaphore: 任何时候都可以post

if (mutex_is_locked(&mutex-\>mutex))

...

// mutex: 必须由上锁的任务解锁

// 而对于semaphore: 任何任务、中断都可以post

if (errcheck && !mutex_is_hold(&mutex-\>mutex))

...

pthread_mutex_give // nuttx/sched/pthread/pthread_mutex.c

mutex_unlock // nuttx/sched/pthread/pthread.h

nxmutex_unlock // nuttx/libs/libc/misc/lib_mutex.c

nxsem_post

“nxsem_post”根据sem->flags决定“额外做什么”，代码如下：

nxsem_post // nuttx/sched/semaphore/sem_post.c

nxsem_post_slow // nuttx/sched/semaphore/sem_post.c

flags = enter_critical_section(); // 进入临界区

// 取出原数值给sem_count, 累加原数值

sem_count = atomic_fetch_add(NXSEM_COUNT(sem), 1);

nxsem_release_holder(sem);

// 对于使用了优先级继承、优先级保护的任务

// 先关闭调度器

if (proto != SEM_PRIO_NONE)

{

sched_lock();

}

if (sem_count \< 0) // 有任务在等待?

stcb = (FAR struct tcb_s \*)dq_remfirst(SEM_WAITLIST(sem));

nxsem_add_holder_tcb(stcb, sem);

if (nxsched_add_readytorun(stcb))

{

up_switch_context(stcb, rtcb);

}

}

// 恢复优先级

#if defined(CONFIG_PRIORITY_INHERITANCE) || defined(CONFIG_PRIORITY_PROTECT)

if (proto != SEM_PRIO_NONE)

{

if (proto == SEM_PRIO_INHERIT)

{

#ifdef CONFIG_PRIORITY_INHERITANCE

// 对于继承了优先级的任务，恢复它的优先级

nxsem_restore_baseprio(stcb, sem);

#endif

}

else if (proto == SEM_PRIO_PROTECT)

{

// 对于提升优先级到天花板的任务，恢复它的优先级

nxsem_protect_post(sem);

}

// 开启调度器

sched_unlock();

}

#endif

队列操作的内部机制​

创建队列源码情景分析​

写队列源码情景分析​

读队列源码情景分析​

信号量和互斥量的内部机制​

信号量互斥量机制与代码层次​

创建信号量/互斥量源码情景分析​

等待信号量/互斥量源码情景分析​

释放信号量/互斥量源码情景分析​

队列操作的内部机制

创建队列源码情景分析

写队列源码情景分析

读队列源码情景分析

信号量和互斥量的内部机制

信号量互斥量机制与代码层次

创建信号量/互斥量源码情景分析

等待信号量/互斥量源码情景分析

释放信号量/互斥量源码情景分析