IPC 的 mb 和 mq 无法确保 FIFO #4940

Activity

1. BernardXiong commented on Aug 13, 2021

   

   BernardXiong

   on Aug 13, 2021

   Member

   More actions

   是的，此处没有做锁定，而semaphore相当于做了锁定。后面可以看看怎么来fix下

   😄React with 😄1Reacted by dongwencai
2. BernardXiong

   added

   bugThis PR/issue is a bug in the current code.This PR/issue is a bug in the current code.

   on Aug 13, 2021
3. dongwencai commented on Nov 21, 2021

   

   dongwencai

   on Nov 21, 2021

   More actions

   uclinux里面也存在这样的问题，但是它的做法并没有把thread从mutex的suspend队列移除，只是wakeup线程，线程可以被调度，当线程获取锁的时候，才会从mutex的suspend队列将自己移除！
4. Rbb666

   added this to 单元测试（UTEST）Frameworkon Nov 7, 2025
5. github-project-automation

   moved this to Todo in 单元测试（UTEST）Frameworkon Nov 7, 2025
6. Rbb666

   removed this from 单元测试（UTEST）Frameworkon Nov 7, 2025
7. Yunlk commented on Oct 22, 2026

   

   Yunlk

   on Oct 22, 2026

   Contributor

   More actions

   按 #BernardXiong 的建议写了一个最小复现，在当前 master（1c801efa）上可以稳定重现。没有提交到任何分支，代码在下面。

   场景

   消息队列 rt_mq_create(..., msg_size=4, max_msgs=1, RT_IPC_FLAG_FIFO)：

   1. waiter_low（priority 20）先 rt_mq_recv(RT_WAITING_FOREVER)，队列空 → 挂起，成为等待队列第一个
   2. waiter_high（priority 10）不在队列上等待，而是先阻塞在一个信号量上
   3. 主线程（priority 10）rt_mq_send() 一条消息 → 唤醒 waiter_low（此时它被从等待队列摘下、进入就绪队列）
   4. 主线程紧接着 rt_sem_release() → waiter_high 就绪
   5. 主线程延时让出 CPU：waiter_high（10）先于 waiter_low（20）运行，调用非阻塞 rt_mq_recv() 把那条消息取走

   实测输出（QEMU vexpress-a9 / ARM，未开 SMP）

   [LOW ] recv, blocking on empty queue
   [HIGH] waiting for its turn to receive
   [SEND] one message -> wakes LOW
   [SEND] rt_mq_send -> 0, entry now = 1
   [SEND] releasing semaphore -> makes HIGH ready
   [HIGH] scheduled before LOW, recv now (mq entry=1)
   [HIGH] got message 42
   [REPRO] acquire order = "H"
   [REPRO] VERDICT: first waiter got nothing, later arrival took the data (FIFO inverted)
   

   注意最后 waiter_low 始终没有打印 "done"——它被唤醒后重新进入 rt_mq_recv 又阻塞了，也就是按 issue 描述回到了等待队列，而此时它在队列里的位置已经排到后来者之后。

   机理

   • rt_mq_send() 只是把消息挂到队列并把队首等待线程 rt_ipc_list_resume()，消息仍留在队列里（日志里 entry now = 1 可以确认）
   • 被唤醒的线程在同一时刻失去了一条不变式：它"被承诺有数据"，但数据没有被预留
   • 任何优先级更高的线程（无论它是否在等待队列里）都可以先把它取走
   • 被唤醒线程再取时队列已空，只能重新挂到等待队列尾部

   这正是 @BernardXiong 说的"此处没有做锁定"，以及 @dongwencai 提到 uClinux 的做法——只唤醒、不把它从 suspend 队列摘除，等真正拿到锁/数据时才出队。

   想请内核维护者决定的方向

   1. 改成"唤醒但不摘链"（uClinux 方式）：线程被唤醒后仍留在等待队列原位，真正取到数据时才移除。语义最符合 RT_IPC_FLAG_FIFO 的字面意思，但需要处理"被唤醒后又超时/被信号打断"的边界，改动面在 ipc.c 的 suspend/resume 路径。
   2. 数据预留：rt_mq_send 直接把消息交给队首等待者（类似 mailbox 的 handoff），队列里不再有可被抢的副本。改动更大，且要处理接收方超时的情况。
   3. 确认为设计取舍，只补文档：明确写出"FIFO 只保证等待队列的排队顺序，不保证数据最终获取顺序"，让使用者知道 RT_IPC_FLAG_FIFO 的边界。

   我个人无法判断哪种符合内核长期意图，但从复现看，行为与 RT_IPC_FLAG_FIFO 的名字存在落差，至少第 3 项（文档）是需要的。

   复现代码（临时 app，未提交）

   #include <rtthread.h>
   
   #define TEST_PRIO_SENDER  5
   #define TEST_PRIO_HIGH    10
   #define TEST_PRIO_LOW     20
   
   static rt_mq_t  _mq;
   static rt_sem_t _go;
   static char     _order[16];
   static int      _order_len;
   
   static void _record(char c)
   {
       if (_order_len < (int)sizeof(_order) - 1) { _order[_order_len++] = c; _order[_order_len] = '\0'; }
   }
   
   static void _waiter_low(void *p)
   {
       rt_int32_t msg = 0;
       rt_kprintf("[LOW ] recv, blocking on empty queue\n");
       if (rt_mq_recv(_mq, &msg, sizeof(msg), RT_WAITING_FOREVER) > 0)
       {
           rt_kprintf("[LOW ] got message %d\n", (int)msg);
           _record('L');
       }
       rt_kprintf("[LOW ] done at tick %d\n", (int)rt_tick_get());
   }
   
   static void _waiter_high(void *p)
   {
       rt_int32_t msg = 0;
       rt_kprintf("[HIGH] waiting for its turn to receive\n");
       rt_sem_take(_go, RT_WAITING_FOREVER);
       rt_kprintf("[HIGH] scheduled before LOW, recv now (mq entry=%d)\n", (int)_mq->entry);
       if (rt_mq_recv(_mq, &msg, sizeof(msg), 0) > 0)
       {
           rt_kprintf("[HIGH] got message %d\n", (int)msg);
           _record('H');
       }
       else
       {
           rt_kprintf("[HIGH] recv failed\n");
       }
   }
   
   static int _repro(void)
   {
       rt_thread_t low, high;
       rt_int32_t  msg = 42;
   
       _mq = rt_mq_create("rf_mq", sizeof(rt_int32_t), 1, RT_IPC_FLAG_FIFO);
       _go = rt_sem_create("rf_go", 0, RT_IPC_FLAG_FIFO);
   
       low  = rt_thread_create("waiter_low",  _waiter_low,  RT_NULL, 1024, TEST_PRIO_LOW,  5);
       high = rt_thread_create("waiter_high", _waiter_high, RT_NULL, 1024, TEST_PRIO_HIGH, 5);
   
       rt_thread_startup(low);
       rt_thread_mdelay(50);              /* LOW blocks first */
       rt_thread_startup(high);
       rt_thread_mdelay(50);              /* HIGH sleeps on the semaphore */
   
       rt_kprintf("[SEND] one message -> wakes LOW\n");
       rt_kprintf("[SEND] rt_mq_send -> %d, entry now = %d\n",
                  (int)rt_mq_send(_mq, &msg, sizeof(msg)), (int)_mq->entry);
       rt_kprintf("[SEND] releasing semaphore -> makes HIGH ready\n");
       rt_sem_release(_go);
   
       rt_thread_mdelay(200);
   
       rt_kprintf("[REPRO] acquire order = \"%s\"\n", _order_len ? _order : "(none)");
       if (rt_strcmp(_order, "H") == 0)
           rt_kprintf("[REPRO] VERDICT: first waiter got nothing, later arrival took the data (FIFO inverted)\n");
       else if (rt_strcmp(_order, "L") == 0)
           rt_kprintf("[REPRO] VERDICT: first waiter got the data (FIFO kept)\n");
       return 0;
   }
   INIT_APP_EXPORT(_repro);

   环境：bsp/qemu-vexpress-a9，QEMU 11.0.0，xpack arm-none-eabi-gcc 15.2.1，BSP 默认配置（未开 SMP）。RT_IPC_FLAG_FIFO 与 RT_IPC_FLAG_PRIO 两种队列 flag 我都试过，现象相同。

   如果维护者认为应该修，我可以按上面任一方向出补丁并补 utest；如果确认是取舍，我可以顺手把文档那条补上。