RIOT OS Mutex Ping-Pong 基准测试:用互斥锁握手测量上下文切换开销
【免费下载链接】RIOTRIOT - The friendly OS for IoT项目地址: https://gitcode.com/GitHub_Trending/riot/RIOT
导读
本文以 RIOT 仓库中 tests/bench/mutex_pingpong 基准测试应用为核心,深入讲解 RIOT 操作系统如何利用两个线程之间反复的 mutex lock/unlock 握手,在 1 秒内统计出 unlock 次数,并据此量化每次上下文切换的时钟开销。读完本文,你将掌握该基准测试的完整实现原理、编译运行方法、输出解析方式,以及它与thread_yield_pingpong、msg_pingpong等姊妹基准在代码结构与测量语义上的差异,并了解core/mutex.c底层同步实现如何影响测量结果。
测试设计:为什么是 "Ping-Pong" 而不是自旋计数
在 RIOT 的tests/bench/目录下,mutex_pingpong属于一组专门用于测量内核调度与同步原语开销的基准测试家族。其设计思想在 tests/bench/mutex_pingpong/README.md 中有明确描述:
- 一个线程反复地对同一个互斥锁执行
mutex_lock; - 另一个线程反复地对它执行
mutex_unlock; - 最终结果是1 秒时间窗内完成的 unlock 次数;
- 由于每一次 unlock 都会唤醒并切换到等待中的锁持有线程,该次数恰好等于期间发生的上下文切换次数的一半。
之所以采用"两个线程来回交接锁"的方式,而不是在一个线程内自旋计数,是因为要测量的恰恰是 mutex 阻塞与唤醒所牵涉的跨线程调度开销:锁的交接意味着当前运行线程被挂起、等待线程被置为可运行并切换出去,这是 RIOT 调度器(core/sched.c)完整工作流的一次体现。
该 README 还特意说明:这个测试应用有意与其他相似基准测试重复代码,目的是为了能够公平地比较各基准应用的代码体积(code sizes)。这解释了为什么mutex_pingpong与thread_yield_pingpong、msg_pingpong的框架代码几乎同构——它们在结构上刻意保持一致,差异只在每次循环中调用的同步原语本身。
核心实现逐行解析
完整的测试逻辑位于 tests/bench/mutex_pingpong/main.c,约 85 行,结构非常紧凑。下面按功能模块拆解。
全局状态与可配置参数
#ifndef TEST_DURATION #define TEST_DURATION (1000000U) #endif volatile unsigned _flag = 0; static char _stack[THREAD_STACKSIZE_MAIN]; static mutex_t _mutex = MUTEX_INIT;TEST_DURATION以微秒为单位,默认1000000U(1 秒)。它可以在编译期通过CFLAGS覆盖,例如传入-DTEST_DURATION=(500000U)即可把测量窗口缩短为 0.5 秒。_flag是测量终止标志,由 xtimer 定时器回调置位(volatile防止编译器优化掉轮询循环)。_stack为第二线程提供栈空间,直接复用主线程的栈大小常量THREAD_STACKSIZE_MAIN。_mutex使用 RIOT 提供的静态初始化宏MUTEX_INIT。从 core/include/mutex.h 可以看到,该宏展开为{ .queue = { .next = NULL } },表示互斥锁处于未锁定状态。
定时回调与第二线程
static void _timer_callback(void*arg) { (void)arg; _flag = 1; } static void *_second_thread(void *arg) { (void)arg; while (1) { mutex_lock(&_mutex); } return NULL; }第二线程的工作极其简单:无限循环地mutex_lock。由于主线程在循环中会反复占有并释放锁,第二线程的mutex_lock绝大多数情况下都会立即成功,不会产生阻塞——它扮演的是"接住锁"的角色,让主线程的下一次mutex_unlock总有等待者可以唤醒。
线程创建与初始握手
thread_create(_stack, sizeof(_stack), THREAD_PRIORITY_MAIN - 1, THREAD_CREATE_WOUT_YIELD, _second_thread, NULL, "second_thread"); /* lock the mutex, then yield to second_thread */ mutex_lock(&_mutex); thread_yield_higher();这里有两个关键细节:
- 第二线程的优先级被设置为
THREAD_PRIORITY_MAIN - 1,即高于主线程。在 RIOT 中数字越小优先级越高,因此第二线程一旦可运行,就会抢占主线程。 - 创建线程时使用
THREAD_CREATE_WOUT_YIELD标志,不立即让出 CPU,保证后续流程在主线程中按顺序执行。 - 主线程先
mutex_lock占住锁,再调用thread_yield_higher()主动让位给更高优先级的第二线程。于是第二线程进入mutex_lock时发现锁已被占用,随即进入阻塞队列(STATUS_MUTEX_BLOCKED,见 core/mutex.c),等待主线程解锁。初始握手完成,测量循环的"乒乓"状态就绪。
测量主循环
xtimer_t timer; timer.callback = _timer_callback; uint32_t n = 0; xtimer_set(&timer, TEST_DURATION); while (!_flag) { mutex_unlock(&_mutex); n++; }主循环在 1 秒内无限执行mutex_unlock并累加计数。其执行节奏是:
- 主线程
mutex_unlock唤醒阻塞中的第二线程(将其置为STATUS_PENDING),并调用thread_yield_higher()让出 CPU(见 core/mutex.c); - 第二线程被调度运行,
mutex_lock成功取得锁; - 第二线程随即再次
mutex_lock,发现锁被自己持有,于是阻塞等待; - 调度器切回主线程,主线程再次
mutex_unlock……
如此往返,每一次循环对应两次上下文切换(主线程→第二线程、第二线程→主线程),因此最终计数n除以 2 就是上下文切换次数。
当xtimer在TEST_DURATION之后触发_timer_callback将_flag置 1 时,循环退出,测量结束。
结果输出
printf("{ \"result\" : %"PRIu32, n); printf(", \"ticks\" : %"PRIu32, (uint32_t)((TEST_DURATION/US_PER_MS) * (coreclk()/KHZ(1)))/n); puts(" }");输出为一行 JSON 风格的文本,包含两个字段:
result:1 秒内的 unlock 次数,即每秒钟完成的锁交接次数;ticks:平均每次 unlock 消耗的 CPU 时钟周期数。
ticks的计算方式为(TEST_DURATION / US_PER_MS) * (coreclk() / KHZ(1)) / n:TEST_DURATION/US_PER_MS把测量窗口换算为毫秒,coreclk()/KHZ(1)得到以 kHz 为单位的 CPU 核心时钟频率(coreclk()声明于 core/include/clk.h),两者相乘即测量窗口内的总时钟周期数,再除以n即为单次 lock/unlock 握手的平均时钟开销。由于两次上下文切换对应一次 unlock,单次上下文切换的时钟成本约为ticks / 2。
编译、烧录与运行
该基准测试是一个标准的 RIOT 测试应用,编译方式与普通 RIOT 应用一致:
# 在 tests/bench/mutex_pingpong 目录下,针对目标板编译 make BOARD=native # 在 native 平台上直接运行 make BOARD=native term # 在真实硬件上烧录并运行 make BOARD=your-board flash termnative平台(cpu/native)是快速验证的首选:它把 RIOT 编译为宿主机程序,无需真实硬件即可观察输出。以 native 为例,运行后终端会打印类似如下的结果:
main starting { "result" : 1234567, "ticks" : 89 }其中result和ticks的具体数值会随宿主平台、编译器优化级别以及 RIOT 配置(如是否启用优先级继承、mutex 调试等)而变化。需要说明的是,native平台测量的是宿主机上的仿真调度开销,与真实 MCU 的绝对数值差异很大;要评估真实硬件上的上下文切换成本,应针对具体开发板烧录运行。
Makefile 结构非常精简(tests/bench/mutex_pingpong/Makefile):
include ../Makefile.bench_common USEMODULE += xtimer include $(RIOTBASE)/Makefile.includeMakefile.bench_common(tests/bench/Makefile.bench_common)负责引入公共的测试构建基础设施Makefile.tests_common;USEMODULE += xtimer声明依赖 xtimer 定时器模块(测量窗口由xtimer_set驱动)。
测试脚本与自动化验证
仓库还提供了自动化测试脚本 tests/bench/mutex_pingpong/tests/01-run.py,它通过 RIOT 的 testrunner 框架断言输出格式:
def testfunc(child): child.expect(r"{ \"result\" : \d+(, \"ticks\" : \d+)? }")正则表达式要求输出匹配{ "result" : <数字> }或带ticks字段的完整形式,任何格式异常都会导致测试失败。这意味着该基准不仅可手动运行,也可接入 CI 或make test流程自动校验。
内存受限板卡的豁免清单
tests/bench/mutex_pingpong/Makefile.ci 列出了BOARD_INSUFFICIENT_MEMORY,即因 Flash/RAM 过小而不参与 CI 构建的板卡:
atmega8 nucleo-l011k4 stm32f030f4-demo从源码结构看,该清单的存在与 README 中"比较代码体积"的意图相呼应——这类基准虽小,但仍需容纳一个完整内核与 xtimer 模块,极小内存的 8 位/入门级 MCU 无法满足。
底层原理:mutex 阻塞与唤醒路径
要真正理解测量结果,需要知道一次mutex_unlock→mutex_lock交接在 core/mutex.c 内部经历了什么。
当第二线程在主线程持有锁时调用mutex_lock,会进入mutex_lock_internal(core/mutex.c):它先关闭中断(irq_disable),检查到mutex->queue.next != NULL(锁已被占用)且需要阻塞时,调用内部函数_block。_block将当前线程状态设为STATUS_MUTEX_BLOCKED、把该线程的rq_entry挂入 mutex 的等待队列,恢复中断后调用thread_yield_higher()主动让出 CPU(core/mutex.c)。
当主线程调用mutex_unlock时(core/mutex.c):
- 关闭中断,检查锁状态;
- 若等待队列非空,用
list_remove_head取出队首等待线程; - 通过
sched_set_status(process, STATUS_PENDING)把等待线程重新置为可运行; - 恢复中断并调用
thread_yield_higher(),让调度器决定是否切换。
从这段实现可以看到,一次锁交接至少包含两次thread_yield_higher()调用和两次调度器调度,这正是基准测试希望计量的核心开销。此外,如果启用了MODULE_CORE_MUTEX_PRIORITY_INHERITANCE(优先级继承)或MODULE_CORE_MUTEX_DEBUG(mutex 调试)模块,_block与mutex_unlock中会额外执行优先级调整或调试信息打印(core/mutex.c、core/mutex.c),这些都会增大单次交接的ticks值——因此做横向对比时,应保持内核模块配置一致。
与姊妹基准测试的对比
tests/bench/目录下还有若干结构高度相似的基准,可通过对照理解各自的测量语义:
| 基准目录 | 循环内核心操作 | 测量内容 |
|---|---|---|
| tests/bench/mutex_pingpong | mutex_unlock | 每秒锁交接次数 ≈ 上下文切换次数的一半 |
| tests/bench/thread_yield_pingpong | thread_yield | 每秒主动让出 CPU 的次数(协作式切换吞吐) |
| tests/bench/msg_pingpong | msg_send/msg_receive | 每秒消息发送次数,即 IPC 通道吞吐 |
以 thread_yield_pingpong/main.c 为例,它同样使用xtimer_set+_flag的 1 秒窗口、同样的result/ticks输出格式,但循环体只是thread_yield(),不涉及 mutex 队列操作。而 msg_pingpong/main.c 使用atomic_flag作为计时终止信号,并通过msg_send/msg_receive在两个线程间传递消息,测量的是内核消息队列(core/msg.c)的吞吐。
三个基准共享的输出格式与代码骨架(README 中"有意重复代码以便比较代码大小"的表述正是针对这一设计),使开发者可以:
- 横向比较同一平台下 mutex 交接、主动让出、消息传递三种同步路径的开销差异;
- 纵向比较不同板卡或不同内核配置下的绝对开销。
总结与使用建议
mutex_pingpong是 RIOT 中一个"小而精"的内核性能探针:
- 测量方法:两个线程通过一个 mutex 反复交接,1 秒内统计 unlock 次数(即上下文切换次数的一半),并输出平均每次交接的 CPU 时钟数;
- 可配置性:通过
-DTEST_DURATION=...调整测量窗口;通过切换目标板卡评估不同硬件平台的调度开销; - 自动化:
tests/01-run.py可接入 testrunner 自动校验输出格式; - 可比性:与
thread_yield_pingpong、msg_pingpong保持同构代码骨架,便于比较不同同步原语的代价与代码体积。
如果你正在为 RIOT 应用做实时性调优(例如评估某临界区频繁 lock/unlock 对系统响应的影响),或想对比不同 CPU 移植(cpu/目录下各架构)的调度器实现质量,这个基准都是最直接的测量起点。
【免费下载链接】RIOTRIOT - The friendly OS for IoT项目地址: https://gitcode.com/GitHub_Trending/riot/RIOT
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考