This is an automated email from the ASF dual-hosted git repository. xiaoxiang781216 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/nuttx.git
commit 9bf99434864104a5c64dc0222227a336e42e5cd9 Author: Marco Casaroli <[email protected]> AuthorDate: Tue Nov 3 15:40:21 2026 +0100 sched/mqueue: Wait for a free message in msgsnd(). All System V message queues share one pool of CONFIG_PREALLOC_MQ_MSGS messages. When the pool was empty, msgsnd() gave ENOMEM. POSIX says: when the system-wide limit of messages is reached, msgsnd() blocks, or gives EAGAIN with IPC_NOWAIT. Add a wait list for the pool, g_msgpoolwait. msgsnd() waits there when its queue has room but the pool is empty, and gives EAGAIN with IPC_NOWAIT or in an interrupt handler. The waiters use the state TSTATE_WAIT_MQNOTFULL with the wait object g_msgpoolwait (a struct mqueue_cmn_s), so a signal, a cancellation or a task restart removes them as for a queue. msgrcv() and IPC_RMID return messages to the pool and wake up all pool waiters. After the wake-up, msgsnd() checks again that its queue exists (EIDRM) and has room for the message. Assisted-by: Claude Code:claude-opus-5-5 Signed-off-by: Marco Casaroli <[email protected]> --- include/sys/msg.h | 3 ++- sched/mqueue/msg.h | 14 ++++++++++++++ sched/mqueue/msginternal.c | 15 +++++++++++++++ sched/mqueue/msgrcv.c | 1 + sched/mqueue/msgsnd.c | 46 ++++++++++++++++++++++++++++------------------ 5 files changed, 60 insertions(+), 19 deletions(-) diff --git a/include/sys/msg.h b/include/sys/msg.h index 0bf0ba048f2..b20c4ebbab4 100644 --- a/include/sys/msg.h +++ b/include/sys/msg.h @@ -165,7 +165,8 @@ int msgget(key_t key, int msgflg); * On success, msgsnd() returns 0 (OK); on error, -1 (ERROR) * is returned, with errno set to indicate the error: * - * EAGAIN The queue is full and IPC_NOWAIT is set. + * EAGAIN The queue is full or no free message is left, and + * IPC_NOWAIT is set or the caller is an interrupt handler. * EINVAL msqid is not a valid message queue identifier, mtype is * less than 1, or msgsz is greater than the maximum message * size of the queue. diff --git a/sched/mqueue/msg.h b/sched/mqueue/msg.h index 23b3c0ab804..29af821079d 100644 --- a/sched/mqueue/msg.h +++ b/sched/mqueue/msg.h @@ -85,6 +85,10 @@ extern "C" extern struct list_node g_msgfreelist; +/* The senders that wait for a free message in g_msgfreelist */ + +extern struct mqueue_cmn_s g_msgpoolwait; + /**************************************************************************** * Public Function Prototypes ****************************************************************************/ @@ -129,6 +133,16 @@ void nxmsg_wakeup_senders(FAR struct msgq_s *msgq); void nxmsg_wakeup_receivers(FAR struct msgq_s *msgq); +/**************************************************************************** + * Name: nxmsg_wakeup_pool + * + * Description: + * Wake up all tasks that wait in msgsnd() for a free message + * + ****************************************************************************/ + +void nxmsg_wakeup_pool(void); + /**************************************************************************** * Name: nxmsg_lookup * diff --git a/sched/mqueue/msginternal.c b/sched/mqueue/msginternal.c index 4df6b39e087..d54d1477666 100644 --- a/sched/mqueue/msginternal.c +++ b/sched/mqueue/msginternal.c @@ -49,6 +49,7 @@ static FAR struct msgq_s **g_msgqs; /* The queues; the index is the id - 1 */ ****************************************************************************/ struct list_node g_msgfreelist; +struct mqueue_cmn_s g_msgpoolwait; /**************************************************************************** * Private Functions @@ -243,6 +244,11 @@ void nxmsg_free(FAR struct msgq_s *msgq) list_add_tail(&g_msgfreelist, &entry->node); } + /* Also the senders of this queue that wait for a free message: they + * get EIDRM. + */ + + nxmsg_wakeup_pool(); kmm_free(msgq); } @@ -264,6 +270,15 @@ void nxmsg_wakeup_receivers(FAR struct msgq_s *msgq) nxmsg_wakeup_list(MQ_WNELIST(msgq->cmn), &msgq->cmn.nwaitnotempty); } +/**************************************************************************** + * Name: nxmsg_wakeup_pool + ****************************************************************************/ + +void nxmsg_wakeup_pool(void) +{ + nxmsg_wakeup_list(MQ_WNFLIST(g_msgpoolwait), &g_msgpoolwait.nwaitnotfull); +} + /**************************************************************************** * Name: nxmsg_lookup ****************************************************************************/ diff --git a/sched/mqueue/msgrcv.c b/sched/mqueue/msgrcv.c index 4134540fd35..d170e2e4266 100644 --- a/sched/mqueue/msgrcv.c +++ b/sched/mqueue/msgrcv.c @@ -256,6 +256,7 @@ ssize_t msgrcv(int msqid, FAR void *msgp, size_t msgsz, long msgtyp, /* The room in the queue can be enough for more than one sender */ nxmsg_wakeup_senders(msgq); + nxmsg_wakeup_pool(); errout_with_critical: leave_critical_section(flags); diff --git a/sched/mqueue/msgsnd.c b/sched/mqueue/msgsnd.c index e7315950481..fe4689fe3ae 100644 --- a/sched/mqueue/msgsnd.c +++ b/sched/mqueue/msgsnd.c @@ -41,6 +41,7 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) { + FAR struct mqueue_cmn_s *cmn; FAR struct tcb_s *rtcb; int msqid = msgq->id; @@ -59,13 +60,27 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) } #endif - /* Loop until the queue has room for msgsz more bytes */ + /* Loop until the queue has room for msgsz more bytes and the pool has + * a free message. Wait in the list of the queue or in the list of the + * pool. + */ - while (msgq->cbytes + msgsz > msgq->qbytes) + for (; ; ) { - /* Should we block until there is sufficient space in the - * message queue? - */ + if (msgq->cbytes + msgsz > msgq->qbytes) + { + cmn = &msgq->cmn; + } + else if (list_is_empty(&g_msgfreelist)) + { + cmn = &g_msgpoolwait; + } + else + { + return OK; + } + + /* Should we block until there is sufficient space? */ if ((msgflg & IPC_NOWAIT) != 0) { @@ -77,8 +92,8 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) */ rtcb = this_task(); - rtcb->waitobj = msgq; - msgq->cmn.nwaitnotfull++; + rtcb->waitobj = cmn; + cmn->nwaitnotfull++; /* Initialize the errcode used to communication wake-up error * conditions. @@ -99,7 +114,7 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) /* Add the task to the specified blocked task list */ rtcb->task_state = TSTATE_WAIT_MQNOTFULL; - nxsched_add_prioritized(rtcb, MQ_WNFLIST(msgq->cmn)); + nxsched_add_prioritized(rtcb, MQ_WNFLIST(*cmn)); /* Now, perform the context switch */ @@ -123,8 +138,6 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) return -EIDRM; } } - - return OK; } /**************************************************************************** @@ -152,7 +165,8 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) * On success, msgsnd() returns 0 (OK); on error, -1 (ERROR) * is returned, with errno set to indicate the error: * - * EAGAIN The queue is full and IPC_NOWAIT is set. + * EAGAIN The queue is full or no free message is left, and + * IPC_NOWAIT is set or the caller is an interrupt handler. * EINVAL msqid is not a valid message queue identifier, mtype is * less than 1, or msgsz is greater than the maximum message * size of the queue. @@ -190,9 +204,9 @@ int msgsnd(int msqid, FAR const void *msgp, size_t msgsz, int msgflg) goto errout_with_critical; } - /* Is the message queue FULL? */ + /* Is the message queue FULL, or the pool of free messages empty? */ - if (msgq->cbytes + msgsz > msgq->qbytes) + if (msgq->cbytes + msgsz > msgq->qbytes || list_is_empty(&g_msgfreelist)) { if (!up_interrupt_context()) /* In an interrupt handler? */ { @@ -215,11 +229,7 @@ int msgsnd(int msqid, FAR const void *msgp, size_t msgsz, int msgflg) /* Now allocate the message. */ msg = (FAR struct msgbuf_s *)list_remove_head(&g_msgfreelist); - if (msg == NULL) - { - ret = -ENOMEM; - goto errout_with_critical; - } + DEBUGASSERT(msg != NULL); /* Check if the message was successfully allocated */
