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 bea53e76c80fcee75e739478ad5b108bf8de0289 Author: Marco Casaroli <[email protected]> AuthorDate: Tue Nov 3 14:50:54 2026 +0100 sched/mqueue: Make msg_qbytes a byte limit of the System V message queue. The queue limit was a number of messages. IPC_SET divided msg_qbytes by CONFIG_MQ_MAXMSGSIZE (33 became 32 with the default size), without a range check, and stored the result in an int16_t. POSIX says that msg_qbytes is the maximum number of bytes in the queue. Keep msg_qbytes as a byte limit and the number of bytes in the queue. msgsnd() blocks, or gives EAGAIN with IPC_NOWAIT, when the message would make the queue larger than msg_qbytes. IPC_STAT reports msg_qbytes as IPC_SET set it. IPC_SET gives EINVAL for 0: no message with data could ever enter the queue. The byte limit replaces the per-queue message limit of 16. The default msg_qbytes stays 16 * CONFIG_MQ_MAXMSGSIZE, so a queue still takes 16 messages of the maximum size. The system-wide pool of CONFIG_PREALLOC_MQ_MSGS messages stays the limit of the number of messages; its default (4 or 8) is smaller than 16, so in the default configuration the old per-queue limit never applied. With a larger pool, a queue of small messages can now hold more than 16 messages; when the pool is empty, msgsnd() gives ENOMEM, as before. msgrcv() and IPC_SET now wake up all senders that wait for room: the room that one message frees can be enough for more than one smaller message, and each sender checks again if its own message fits. Assisted-by: Claude Code:claude-opus-5-5 Signed-off-by: Marco Casaroli <[email protected]> --- sched/mqueue/msg.h | 15 ++++++++++++--- sched/mqueue/msgctl.c | 14 ++++++++++++-- sched/mqueue/msginternal.c | 37 ++++++++++++++++++++++++++++++++++++- sched/mqueue/msgrcv.c | 41 ++--------------------------------------- sched/mqueue/msgsnd.c | 14 +++++--------- 5 files changed, 67 insertions(+), 54 deletions(-) diff --git a/sched/mqueue/msg.h b/sched/mqueue/msg.h index f8726e77283..6497f1d7ccc 100644 --- a/sched/mqueue/msg.h +++ b/sched/mqueue/msg.h @@ -43,7 +43,7 @@ ****************************************************************************/ #define MSG_MAX_BYTES CONFIG_MQ_MAXMSGSIZE -#define MSG_MAX_MSGS 16 +#define MSG_DEF_QBYTES (16 * MSG_MAX_BYTES) /* Default msg_qbytes */ /**************************************************************************** * Public Type Definitions @@ -61,9 +61,8 @@ struct msgq_s pid_t lspid; /* PID of the last msgsnd() */ pid_t lrpid; /* PID of the last msgrcv() */ msglen_t cbytes; /* Number of bytes in the queue */ + msglen_t qbytes; /* Maximum bytes in the queue */ int16_t id; /* The message queue identifier */ - int16_t maxmsgs; /* Maximum number of messages in the queue */ - int16_t nmsgs; /* Number of message in the queue */ uint16_t maxmsgsize; /* Max size of message in message queue */ }; @@ -110,6 +109,16 @@ int nxmsg_alloc(FAR struct msgq_s **pmsgq); void nxmsg_free(FAR struct msgq_s *msgq); +/**************************************************************************** + * Name: nxmsg_wakeup_senders + * + * Description: + * Wake up all tasks that wait in msgsnd() for room in the queue + * + ****************************************************************************/ + +void nxmsg_wakeup_senders(FAR struct msgq_s *msgq); + /**************************************************************************** * Name: nxmsg_lookup * diff --git a/sched/mqueue/msgctl.c b/sched/mqueue/msgctl.c index e0bcfb9f65f..c8e4d8fdd11 100644 --- a/sched/mqueue/msgctl.c +++ b/sched/mqueue/msgctl.c @@ -88,9 +88,19 @@ int msgctl(int msqid, int cmd, FAR struct msqid_ds *buf) break; } - msgq->maxmsgs = buf->msg_qbytes / CONFIG_MQ_MAXMSGSIZE; + if (buf->msg_qbytes == 0) + { + ret = -EINVAL; + break; + } + + msgq->qbytes = buf->msg_qbytes; msgq->perm.mode = buf->msg_perm.mode & IPC_MODE; msgq->ctime = time(NULL); + + /* A larger msg_qbytes can make room for the waiting senders */ + + nxmsg_wakeup_senders(msgq); break; } @@ -108,7 +118,7 @@ int msgctl(int msqid, int cmd, FAR struct msqid_ds *buf) buf->msg_ctime = msgq->ctime; buf->msg_cbytes = msgq->cbytes; buf->msg_qnum = list_length(&msgq->msglist); - buf->msg_qbytes = msgq->maxmsgs * CONFIG_MQ_MAXMSGSIZE; + buf->msg_qbytes = msgq->qbytes; buf->msg_lspid = msgq->lspid; buf->msg_lrpid = msgq->lrpid; break; diff --git a/sched/mqueue/msginternal.c b/sched/mqueue/msginternal.c index 8d4f72c9f40..9a23e566045 100644 --- a/sched/mqueue/msginternal.c +++ b/sched/mqueue/msginternal.c @@ -24,6 +24,8 @@ * Included Files ****************************************************************************/ +#include <assert.h> + #include <nuttx/kmalloc.h> #include "mqueue/msg.h" #include "sched/sched.h" @@ -155,7 +157,7 @@ int nxmsg_alloc(FAR struct msgq_s **pmsgq) list_initialize(&msgq->msglist); - msgq->maxmsgs = MSG_MAX_MSGS; + msgq->qbytes = MSG_DEF_QBYTES; msgq->maxmsgsize = MSG_MAX_BYTES; *pmsgq = msgq; @@ -196,6 +198,39 @@ void nxmsg_free(FAR struct msgq_s *msgq) kmm_free(msgq); } +/**************************************************************************** + * Name: nxmsg_wakeup_senders + ****************************************************************************/ + +void nxmsg_wakeup_senders(FAR struct msgq_s *msgq) +{ + FAR struct tcb_s *rtcb = this_task(); + FAR struct tcb_s *btcb; + bool need_switch = false; + + /* Each sender checks again if its message fits */ + + while (msgq->cmn.nwaitnotfull > 0) + { + btcb = (FAR struct tcb_s *)dq_remfirst(MQ_WNFLIST(msgq->cmn)); + DEBUGASSERT(btcb != NULL); + + wd_cancel(&btcb->waitdog); + msgq->cmn.nwaitnotfull--; + btcb->waitobj = NULL; + + if (nxsched_add_readytorun(btcb)) + { + need_switch = true; + } + } + + if (need_switch) + { + up_switch_context(this_task(), rtcb); + } +} + /**************************************************************************** * Name: nxmsg_lookup ****************************************************************************/ diff --git a/sched/mqueue/msgrcv.c b/sched/mqueue/msgrcv.c index 8f4742b2708..4134540fd35 100644 --- a/sched/mqueue/msgrcv.c +++ b/sched/mqueue/msgrcv.c @@ -174,7 +174,6 @@ static int msgrcv_wait(FAR struct msgq_s *msgq, FAR struct msgbuf_s **rcvmsg, } found: - msgq->nmsgs--; msgq->cbytes -= newmsg->msize; msgq->lrpid = nxsched_getpid(); msgq->rtime = time(NULL); @@ -224,7 +223,6 @@ ssize_t msgrcv(int msqid, FAR void *msgp, size_t msgsz, long msgtyp, FAR struct msgbuf_s *msg = NULL; FAR struct mymsg *buf = msgp; FAR struct msgq_s *msgq; - FAR struct tcb_s *btcb; irqstate_t flags; int ret; @@ -255,44 +253,9 @@ ssize_t msgrcv(int msqid, FAR void *msgp, size_t msgsz, long msgtyp, list_add_tail(&g_msgfreelist, &msg->node); - /* Check if any tasks are waiting for the MQ not full event. */ + /* The room in the queue can be enough for more than one sender */ - if (msgq->cmn.nwaitnotfull > 0) - { - FAR struct tcb_s *rtcb = this_task(); - - /* Find the highest priority task that is waiting for - * this queue to be not-full, in the waitfornotfull list of the queue. - * This must be performed in a critical section because - * messages can be sent from interrupt handlers. - */ - - btcb = (FAR struct tcb_s *)dq_remfirst(MQ_WNFLIST(msgq->cmn)); - - /* If one was found, unblock it. NOTE: There is a race - * condition here: the queue might be full again by the - * time the task is unblocked - */ - - DEBUGASSERT(btcb != NULL); - - wd_cancel(&btcb->waitdog); - - msgq->cmn.nwaitnotfull--; - - /* Indicate that the wait is over. */ - - btcb->waitobj = NULL; - - /* Add the task to ready-to-run task list and - * perform the context switch if one is needed - */ - - if (nxsched_add_readytorun(btcb)) - { - up_switch_context(this_task(), rtcb); - } - } + nxmsg_wakeup_senders(msgq); errout_with_critical: leave_critical_section(flags); diff --git a/sched/mqueue/msgsnd.c b/sched/mqueue/msgsnd.c index 7a370ba8c1b..a216b5afa56 100644 --- a/sched/mqueue/msgsnd.c +++ b/sched/mqueue/msgsnd.c @@ -39,7 +39,7 @@ * Name: msgsnd_wait ****************************************************************************/ -static int msgsnd_wait(FAR struct msgq_s *msgq, int msgflg) +static int msgsnd_wait(FAR struct msgq_s *msgq, size_t msgsz, int msgflg) { FAR struct tcb_s *rtcb; int msqid = msgq->id; @@ -59,12 +59,9 @@ static int msgsnd_wait(FAR struct msgq_s *msgq, int msgflg) } #endif - /* Verify that the queue is indeed full as the caller thinks - * Loop until there are fewer than max allowable messages in the - * receiving message queue - */ + /* Loop until the queue has room for msgsz more bytes */ - while (msgq->nmsgs >= msgq->maxmsgs) + while (msgq->cbytes + msgsz > msgq->qbytes) { /* Should we block until there is sufficient space in the * message queue? @@ -196,7 +193,7 @@ int msgsnd(int msqid, FAR const void *msgp, size_t msgsz, int msgflg) /* Is the message queue FULL? */ - if (msgq->nmsgs >= msgq->maxmsgs) + if (msgq->cbytes + msgsz > msgq->qbytes) { if (!up_interrupt_context()) /* In an interrupt handler? */ { @@ -204,7 +201,7 @@ int msgsnd(int msqid, FAR const void *msgp, size_t msgsz, int msgflg) * available in the message queue. */ - ret = msgsnd_wait(msgq, msgflg); + ret = msgsnd_wait(msgq, msgsz, msgflg); } else if ((msgflg & IPC_NOWAIT) != 0) { @@ -233,7 +230,6 @@ int msgsnd(int msqid, FAR const void *msgp, size_t msgsz, int msgflg) list_add_tail(&msgq->msglist, &msg->node); - msgq->nmsgs++; msgq->cbytes += msgsz; msgq->lspid = nxsched_getpid(); msgq->stime = time(NULL);
