[kernel][bsp][HUST_CSE]:设计新的消息接收函数，可以返回消息的实际大小

bsp/allwinner/libraries/sunxi-hal/osal/src/hal_queue.c

@@ -182,7 +182,7 @@ int hal_queue_recv(hal_queue_t queue, void *buffer, int timeout)
     }
 
     ret = rt_mq_recv(queue, buffer, queue->msg_size, timeout);
-    if (ret != RT_EOK)
+    if (ret < 0)
     {
         return -2;
     }

bsp/fujitsu/mb9x/mb9bf500r/application.c

@@ -74,7 +74,7 @@ void rt_init_thread_entry(void *parameter)
 
     while(1)
     {
-        if (rt_mq_recv(&mq, &msg, sizeof(msg), RT_WAITING_FOREVER) == RT_EOK)
+        if (rt_mq_recv(&mq, &msg, sizeof(msg), RT_WAITING_FOREVER) >= 0)
         {
             switch(msg.type)
             {

examples/message_queue/messagequeue_tc.c

@@ -0,0 +1,132 @@
+/*
+ * Copyright (c) 2006-2023, RT-Thread Development Team
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Change Logs:
+ * Date           Author       Notes
+ * 2023-04-23     xin-zheqi    the first version
+ */
+/*
+ * 程序清单：消息队列例程
+ *
+ * 发送消息长度为3
+ * 接收函数返回消息实际长度
+ */
+#include <rtthread.h>
+#define THREAD_PRIORITY      25
+#define THREAD_TIMESLICE     5
+/* 消息队列控制块 */
+static struct rt_messagequeue mq;
+/* 消息队列中用到的放置消息的内存池 */
+static rt_uint8_t msg_pool[2048];
+rt_align(RT_ALIGN_SIZE)
+static char thread1_stack[1024];
+static struct rt_thread thread1;
+/* 线程1入口函数 */
+static void thread1_entry(void *parameter)
+{
+    char buf[10] = 0;
+    rt_size_t length = 0;
+    rt_uint8_t cnt = 0;
+    while (1)
+    {
+        /* 从消息队列中接收消息 */
+        if ((length = rt_mq_recv(&mq, buf, sizeof(buf), RT_WAITING_FOREVER)) >= 0)
+        {
+            rt_kprintf("thread1: recv msg from msg queue,the message's length is:%d, the content:%s\n", length, buf);
+            if (cnt == 19)
+            {
+                break;
+            }
+        }
+        /* 延时50ms */
+        cnt++;
+        rt_thread_mdelay(50);
+    }
+    rt_kprintf("thread1: detach mq \n");
+    rt_mq_detach(&mq);
+}
+rt_align(RT_ALIGN_SIZE)
+static char thread2_stack[1024];
+static struct rt_thread thread2;
+/* 线程2入口 */
+static void thread2_entry(void *parameter)
+{
+    int result;
+    /* 实际发送3字节消息 */
+    char buf[10] = "ABC";
+
+    rt_uint8_t cnt = 0;
+    while (1)
+    {
+        if (cnt == 8)
+        {
+            /* 发送紧急消息到消息队列中 */
+            result = rt_mq_urgent(&mq, buf, 3);
+            if (result != RT_EOK)
+            {
+                rt_kprintf("rt_mq_urgent ERR\n");
+            }
+            else
+            {
+                rt_kprintf("thread2: send urgent message - %s\n", buf);
+            }
+        }
+        else if (cnt >= 20)/* 发送20次消息之后退出 */
+        {
+            rt_kprintf("message queue stop send, thread2 quit\n");
+            break;
+        }
+        else
+        {
+            /* 发送消息到消息队列中 */
+            result = rt_mq_send(&mq, buf, 3);
+            if (result != RT_EOK)
+            {
+                rt_kprintf("rt_mq_send ERR\n");
+            }
+            rt_kprintf("thread2: send message - %s\n", buf);
+        }
+        buf++;
+        cnt++;
+        /* 延时5ms */
+        rt_thread_mdelay(5);
+    }
+}
+/* 消息队列示例的初始化 */
+int msgq_sample(void)
+{
+    rt_err_t result;
+    /* 初始化消息队列 */
+    result = rt_mq_init(&mq,
+                        "mqt",
+                        &msg_pool[0],               /* 内存池指向msg_pool */
+                        5,                          /* 每个消息的大小是 5 字节 */
+                        sizeof(msg_pool),           /* 内存池的大小是msg_pool的大小 */
+                        RT_IPC_FLAG_FIFO);          /* 如果有多个线程等待，按照先来先得到的方法分配消息 */
+    if (result != RT_EOK)
+    {
+        rt_kprintf("init message queue failed.\n");
+        return -1;
+    }
+    rt_thread_init(&thread1,
+                   "thread1",
+                   thread1_entry,
+                   RT_NULL,
+                   &thread1_stack[0],
+                   sizeof(thread1_stack),
+                   THREAD_PRIORITY, THREAD_TIMESLICE);
+    rt_thread_startup(&thread1);
+    rt_thread_init(&thread2,
+                   "thread2",
+                   thread2_entry,
+                   RT_NULL,
+                   &thread2_stack[0],
+                   sizeof(thread2_stack),
+                   THREAD_PRIORITY, THREAD_TIMESLICE);
+    rt_thread_startup(&thread2);
+    return 0;
+}
+/* 导出到 msh 命令列表中 */
+MSH_CMD_EXPORT(msgq_sample, msgq sample);

examples/utest/testcases/kernel/messagequeue_tc.c

@@ -111,14 +111,14 @@ static void mq_recv_case(rt_mq_t testmq)
     for (int var = 0; var < MAX_MSGS + 1; ++var)
     {
         ret = rt_mq_recv(testmq, &recv_buf[var], sizeof(recv_buf[0]), RT_WAITING_FOREVER);
-        uassert_true(ret == RT_EOK);
+        uassert_true(ret >= 0);
         uassert_true(recv_buf[var] == (var + 1));
     }
 
     for (int var = 0; var < 3; ++var)
     {
         ret = rt_mq_recv(testmq, &recv_buf[var], sizeof(recv_buf[0]), RT_WAITING_FOREVER);
-        uassert_true(ret == RT_EOK);
+        uassert_true(ret >= 0);
         uassert_true(recv_buf[var] == (var + 1));
     }
 }

include/rtthread.h

@@ -487,15 +487,15 @@ rt_err_t rt_mq_send_wait_killable(rt_mq_t     mq,
                          rt_size_t   size,
                          rt_int32_t  timeout);
 rt_err_t rt_mq_urgent(rt_mq_t mq, const void *buffer, rt_size_t size);
-rt_err_t rt_mq_recv(rt_mq_t    mq,
+rt_ssize_t rt_mq_recv(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout);
-rt_err_t rt_mq_recv_interruptible(rt_mq_t    mq,
+rt_ssize_t rt_mq_recv_interruptible(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout);
-rt_err_t rt_mq_recv_killable(rt_mq_t    mq,
+rt_ssize_t rt_mq_recv_killable(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout);

src/ipc.c

@@ -44,6 +44,7 @@
  * 2022-04-08     Stanley      Correct descriptions
  * 2022-10-15     Bernard      add nested mutex feature
  * 2022-10-16     Bernard      add prioceiling feature in mutex
+ * 2023-04-16     Xin-zheqi    redesigen queue recv and send function return real message size
  */
 
 #include <rtthread.h>
@@ -59,6 +60,8 @@
     #define __on_rt_object_put_hook(parent)         __ON_HOOK_ARGS(rt_object_put_hook, (parent))
 #endif
 
+#define GET_MESSAGEBYTE_ADDR(msg)               (((rt_size_t *)((msg) + 1)) + 1)          /* The first block is message head and the next four bytes are message's length, thus this is the first byte of the real message */
+
 #if defined(RT_USING_HOOK) && defined(RT_HOOK_USING_FUNC_PTR)
 extern void (*rt_object_trytake_hook)(struct rt_object *object);
 extern void (*rt_object_take_hook)(struct rt_object *object);
@@ -2934,8 +2937,8 @@ rt_err_t rt_mq_init(rt_mq_t     mq,
     mq->msg_pool = msgpool;
 
     /* get correct message size */
-    msg_align_size = RT_ALIGN(msg_size, RT_ALIGN_SIZE);
-    mq->msg_size = msg_size;
+    msg_align_size = RT_ALIGN(msg_size + sizeof(rt_size_t), RT_ALIGN_SIZE);
+    mq->msg_size = msg_size + sizeof(rt_size_t);
     mq->max_msgs = pool_size / (msg_align_size + sizeof(struct rt_mq_message));
 
     if (0 == mq->max_msgs)
@@ -3066,8 +3069,8 @@ rt_mq_t rt_mq_create(const char *name,
     /* initialize message queue */
 
     /* get correct message size */
-    msg_align_size = RT_ALIGN(msg_size, RT_ALIGN_SIZE);
-    mq->msg_size = msg_size;
+    msg_align_size = RT_ALIGN(msg_size + sizeof(rt_size_t), RT_ALIGN_SIZE);
+    mq->msg_size = msg_size + sizeof(rt_size_t);
     mq->max_msgs = max_msgs;
 
     /* allocate message pool */
@@ -3160,6 +3163,7 @@ RTM_EXPORT(rt_mq_delete);
  *           thread will be resumed and an error code will be returned. By
  *           contrast, the rt_mq_send() function will return an error code
  *           immediately without waiting when the messagequeue if fully used.
+ *           The front four bytes are the length of the message
  *
  * @see      rt_mq_send()
  *
@@ -3200,7 +3204,7 @@ static rt_err_t _rt_mq_send_wait(rt_mq_t     mq,
     RT_DEBUG_SCHEDULER_AVAILABLE(timeout != 0);
 
     /* greater than one message size */
-    if (size > mq->msg_size)
+    if (size + sizeof(size) > mq->msg_size)
         return -RT_ERROR;
 
     /* initialize delta tick */
@@ -3300,8 +3304,11 @@ static rt_err_t _rt_mq_send_wait(rt_mq_t     mq,
 
     /* the msg is the new tailer of list, the next shall be NULL */
     msg->next = RT_NULL;
+
+    /* add the length */
+    *(rt_size_t *)(msg + 1) = size;
     /* copy buffer */
-    rt_memcpy(msg + 1, buffer, size);
+    rt_memcpy(GET_MESSAGEBYTE_ADDR(msg), buffer, size);
 
     /* disable interrupt */
     level = rt_hw_interrupt_disable();
@@ -3355,7 +3362,7 @@ rt_err_t rt_mq_send_wait(rt_mq_t     mq,
 {
     return _rt_mq_send_wait(mq, buffer, size, timeout, RT_UNINTERRUPTIBLE);
 }
-RTM_EXPORT(rt_mq_send_wait)
+RTM_EXPORT(rt_mq_send_wait);
 
 rt_err_t rt_mq_send_wait_interruptible(rt_mq_t     mq,
                          const void *buffer,
@@ -3364,7 +3371,7 @@ rt_err_t rt_mq_send_wait_interruptible(rt_mq_t     mq,
 {
     return _rt_mq_send_wait(mq, buffer, size, timeout, RT_INTERRUPTIBLE);
 }
-RTM_EXPORT(rt_mq_send_wait_interruptible)
+RTM_EXPORT(rt_mq_send_wait_interruptible);
 
 rt_err_t rt_mq_send_wait_killable(rt_mq_t     mq,
                          const void *buffer,
@@ -3373,7 +3380,7 @@ rt_err_t rt_mq_send_wait_killable(rt_mq_t     mq,
 {
     return _rt_mq_send_wait(mq, buffer, size, timeout, RT_KILLABLE);
 }
-RTM_EXPORT(rt_mq_send_wait_killable)
+RTM_EXPORT(rt_mq_send_wait_killable);
 /**
  * @brief    This function will send a message to the messagequeue object.
  *           If there is a thread suspended on the messagequeue, the thread will be resumed.
@@ -3443,7 +3450,7 @@ rt_err_t rt_mq_urgent(rt_mq_t mq, const void *buffer, rt_size_t size)
     RT_ASSERT(size != 0);
 
     /* greater than one message size */
-    if (size > mq->msg_size)
+    if (size + sizeof(size) > mq->msg_size)
         return -RT_ERROR;
 
     RT_OBJECT_HOOK_CALL(rt_object_put_hook, (&(mq->parent.parent)));
@@ -3467,8 +3474,10 @@ rt_err_t rt_mq_urgent(rt_mq_t mq, const void *buffer, rt_size_t size)
     /* enable interrupt */
     rt_hw_interrupt_enable(level);
 
+    /* add the length */
+    *(rt_size_t *)(msg + 1) = size;
     /* copy buffer */
-    rt_memcpy(msg + 1, buffer, size);
+    rt_memcpy(GET_MESSAGEBYTE_ADDR(msg), buffer, size);
 
     /* disable interrupt */
     level = rt_hw_interrupt_disable();
@@ -3536,10 +3545,10 @@ RTM_EXPORT(rt_mq_urgent);
  *           If use macro RT_WAITING_NO to set this parameter, which means that this
  *           function is non-blocking and will return immediately.
  *
- * @return   Return the operation status. When the return value is RT_EOK, the operation is successful.
+ * @return   Return the real length of the message. When the return value is larger than zero, the operation is successful.
  *           If the return value is any other values, it means that the mailbox release failed.
  */
-static rt_err_t _rt_mq_recv(rt_mq_t    mq,
+static rt_ssize_t _rt_mq_recv(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout,
@@ -3550,6 +3559,7 @@ static rt_err_t _rt_mq_recv(rt_mq_t    mq,
     struct rt_mq_message *msg;
     rt_uint32_t tick_delta;
     rt_err_t ret;
+    rt_size_t len;
 
     /* parameter check */
     RT_ASSERT(mq != RT_NULL);
@@ -3665,8 +3675,10 @@ static rt_err_t _rt_mq_recv(rt_mq_t    mq,
     /* enable interrupt */
     rt_hw_interrupt_enable(level);
 
+    /* get real message length */
+    len = *(rt_size_t *)(msg + 1);
     /* copy message */
-    rt_memcpy(buffer, msg + 1, size > mq->msg_size ? mq->msg_size : size);
+    rt_memcpy(buffer, GET_MESSAGEBYTE_ADDR(msg), len);
 
     /* disable interrupt */
     level = rt_hw_interrupt_disable();
@@ -3686,18 +3698,18 @@ static rt_err_t _rt_mq_recv(rt_mq_t    mq,
 
         rt_schedule();
 
-        return RT_EOK;
+        return len;
     }
 
     /* enable interrupt */
     rt_hw_interrupt_enable(level);
 
     RT_OBJECT_HOOK_CALL(rt_object_take_hook, (&(mq->parent.parent)));
 
-    return RT_EOK;
+    return len;
 }
 
-rt_err_t rt_mq_recv(rt_mq_t    mq,
+rt_ssize_t rt_mq_recv(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout)
@@ -3706,7 +3718,7 @@ rt_err_t rt_mq_recv(rt_mq_t    mq,
 }
 RTM_EXPORT(rt_mq_recv);
 
-rt_err_t rt_mq_recv_interruptible(rt_mq_t    mq,
+rt_ssize_t rt_mq_recv_interruptible(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout)
@@ -3715,7 +3727,7 @@ rt_err_t rt_mq_recv_interruptible(rt_mq_t    mq,
 }
 RTM_EXPORT(rt_mq_recv_interruptible);
 
-rt_err_t rt_mq_recv_killable(rt_mq_t    mq,
+rt_ssize_t rt_mq_recv_killable(rt_mq_t    mq,
                     void      *buffer,
                     rt_size_t  size,
                     rt_int32_t timeout)