2026/7/19大约 12 分钟附录源码附录
os_mq.c
路径: kernel\source\os_mq.c
/**
***********************************************************************************************************************
* Copyright (c) 2020-2021, China Mobile Communications Group Co.,Ltd.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*
* @file os_mq.c
*
* @brief This file implements the message queue functions.
*
* @revision
* Date Author Notes
* 2020-02-14 OneOS Team First Version
* 2021-01-08 OneOS team Refactor message queue implementation.
***********************************************************************************************************************
*/
#include <oneos_config.h>
#include <os_stddef.h>
#include <os_errno.h>
#include <os_clock.h>
#include <arch_interrupt.h>
#include <os_mq.h>
#include <os_spinlock.h>
#include <string.h>
#include "os_kernel_internal.h"
#ifdef OS_USING_MESSAGEQUEUE
#define MQ_TAG "MSGQUEUE"
static os_list_node_t gs_os_mq_resource_list_head = OS_LIST_INIT(gs_os_mq_resource_list_head);
static OS_DEFINE_SPINLOCK(gs_os_mq_resource_list_lock);
#define _k_mq_put_msg_to_queue(msgqueue, message) \
do \
{ \
message->next = OS_NULL; \
\
if (msgqueue->msg_queue_tail != OS_NULL) \
{ \
msgqueue->msg_queue_tail->next = message; \
} \
\
/* Set new tail */ \
msgqueue->msg_queue_tail = message; \
\
if (OS_NULL == msgqueue->msg_queue_head) \
{ \
msgqueue->msg_queue_head = message; \
} \
\
msgqueue->used_msgs++; \
} while (0)
#define _k_mq_put_urgentmsg_to_queue(msgqueue, message) \
do \
{ \
message->next = msgqueue->msg_queue_head; \
msgqueue->msg_queue_head = message; \
\
if (OS_NULL == msgqueue->msg_queue_tail) \
{ \
msgqueue->msg_queue_tail = message; \
} \
\
msgqueue->used_msgs++; \
} while (0)
#define _k_mq_get_msg_from_queue(msgqueue, message) \
do \
{ \
OS_ASSERT((OS_NULL != msgqueue->msg_queue_head)); \
\
message = msgqueue->msg_queue_head; \
msgqueue->msg_queue_head = message->next; \
\
if (msgqueue->msg_queue_tail == message) \
{ \
msgqueue->msg_queue_tail = OS_NULL; \
} \
\
msgqueue->used_msgs--; \
} while (0)
#define _k_mq_release_free_msg(msgqueue, message) \
do \
{ \
message->next = msgqueue->msg_queue_free; \
msgqueue->msg_queue_free = message; \
} while (0)
#define _k_mq_get_free_msg(msgqueue, message) \
do \
{ \
OS_ASSERT((OS_NULL != msgqueue->msg_queue_free)); \
\
message = msgqueue->msg_queue_free; \
msgqueue->msg_queue_free = message->next; \
} while (0)
static os_err_t _k_mq_init(os_msgqueue_t *mq,
const char *name,
void *msg_pool,
os_size_t msg_pool_size,
os_size_t msg_size,
uint8_t object_alloc_type)
{
void *pool_align_begin;
const void *pool_end;
os_size_t pool_align_size;
os_mq_msg_t *msg;
os_size_t msg_align_size;
os_size_t msg_res_size;
uint16_t index;
os_err_t ret;
ret = OS_SUCCESS;
msg_align_size = OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE);
msg_res_size = msg_align_size + sizeof(os_mq_msg_hdr_t);
pool_align_begin = (void *)OS_ALIGN_UP((os_ubase_t)msg_pool, OS_ALIGN_SIZE);
pool_end = (void *)((uint8_t *)msg_pool + msg_pool_size);
pool_align_size = (os_ubase_t)pool_end - (os_ubase_t)pool_align_begin;
mq->queue_depth = (uint16_t)(pool_align_size / msg_res_size);
if (mq->queue_depth > 0)
{
mq->msg_pool = msg_pool;
mq->max_msg_size = msg_align_size;
mq->used_msgs = 0U;
mq->object_alloc_type = object_alloc_type;
mq->wake_type = OS_MQ_WAKE_TYPE_PRIO;
if (OS_NULL != name)
{
(void)strncpy(&mq->name[0], name, OS_NAME_MAX);
mq->name[OS_NAME_MAX] = '\0';
}
else
{
mq->name[0] = '\0';
}
os_list_init(&mq->send_task_list_head);
os_list_init(&mq->recv_task_list_head);
mq->msg_queue_head = OS_NULL;
mq->msg_queue_tail = OS_NULL;
mq->msg_queue_free = OS_NULL;
for (index = 0U; index < mq->queue_depth; index++)
{
msg = (os_mq_msg_t *)((uint8_t *)pool_align_begin + index * msg_res_size);
msg->msg_len = 0U;
msg->next = mq->msg_queue_free;
mq->msg_queue_free = msg;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_init(&mq->lock);
#endif
}
else
{
OS_KERN_LOG(KERN_ERROR, MQ_TAG, "The count of calculated message entry is less than 1.");
ret = OS_NOMEM;
}
return ret;
}
#ifdef OS_USING_HEAP
/**
***********************************************************************************************************************
* @brief This function creates a message queue with dynamic memory allocation.
*
* @details Both control block and message pool of the message queue are allocated in memory heap.
*
* @attention This interface is not allowed in the following cases:
* 1. In interrupt context.
* 2. Interrupt is disabled.
* 3. Scheduler is locked.
*
* @param[in] name Message queue name.
* @param[in] msg_size The size of a message.
* @param[in] max_msgs The maximum number of messages in this message queue.
*
* @return The message queue id.
* @retval OS_NULL Failed to create a message queue with dynamic memory allocation.
* @retval else Return the message queue id.
***********************************************************************************************************************
*/
os_msgqueue_id os_msgqueue_create_dynamic(const char *name, os_size_t msg_size, os_size_t max_msgs)
{
os_msgqueue_t *mq;
void *msg_pool;
os_size_t msg_pool_size;
os_size_t align_msg_size;
os_err_t ret;
OS_ASSERT(msg_size > 0U);
OS_ASSERT(max_msgs > 0U);
OS_ASSERT(OS_FALSE == os_is_irq_active());
align_msg_size = OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE);
msg_pool_size = max_msgs * (align_msg_size + sizeof(os_mq_msg_hdr_t));
mq = (os_msgqueue_t *)OS_KERNEL_MALLOC(sizeof(os_msgqueue_t));
msg_pool = OS_KERNEL_MALLOC_ALIGN(OS_ALIGN_SIZE, msg_pool_size);
if ((OS_NULL == mq) || (OS_NULL == msg_pool))
{
OS_KERN_LOG(KERN_ERROR, MQ_TAG, "Malloc failed, mq(%p), msg_pool(%p)", mq, msg_pool);
if (OS_NULL != mq)
{
OS_KERNEL_FREE(mq);
mq = OS_NULL;
}
if (OS_NULL != msg_pool)
{
OS_KERNEL_FREE(msg_pool);
msg_pool = OS_NULL;
}
ret = OS_NOMEM;
}
else
{
ret = _k_mq_init(mq, name, msg_pool, msg_pool_size, align_msg_size, OS_KOBJ_ALLOC_TYPE_DYNAMIC);
if (OS_SUCCESS == ret)
{
os_spin_lock(&gs_os_mq_resource_list_lock);
os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node);
os_spin_unlock(&gs_os_mq_resource_list_lock);
mq->object_inited = OS_KOBJ_INITED;
}
else
{
OS_KERNEL_FREE(mq);
OS_KERNEL_FREE(msg_pool);
mq = OS_NULL;
msg_pool = OS_NULL;
}
}
return OS_TYPE_CONVERT(os_msgqueue_id, mq);
}
#endif /* OS_USING_HEAP */
/**
***********************************************************************************************************************
* @brief This function initialize a message queue with static memory allocation.
*
* @attention This interface is not allowed in interrupt context.
*
* @param[in] msgqueue Message queue's control block.
* @param[in] name Message queue name.
* @param[in] msg_pool The address of message queue's pool.
* @param[in] msg_pool_size The size of message queue's pool in bytes.
* @param[in] msg_size The size of a message.
*
* @return The message queue id.
***********************************************************************************************************************
*/
os_msgqueue_id os_msgqueue_create_static(os_msgqueue_dummy_t *msgqueue,
const char *name,
void *msg_pool,
os_size_t msg_pool_size,
os_size_t msg_size)
{
os_msgqueue_t *iter_mq;
os_list_node_t *pos;
os_err_t ret;
os_msgqueue_t *mq;
OS_ASSERT(OS_NULL != msgqueue);
OS_ASSERT(OS_NULL != msg_pool);
OS_ASSERT(msg_pool_size > 0U);
OS_ASSERT(msg_size > 0U);
OS_ASSERT(os_is_irq_active() == OS_FALSE);
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue);
os_spin_lock(&gs_os_mq_resource_list_lock);
os_list_for_each(pos, &gs_os_mq_resource_list_head)
{
iter_mq = os_list_entry(pos, os_msgqueue_t, resource_node);
if (iter_mq == mq)
{
OS_KERN_LOG(KERN_ERROR, MQ_TAG, "The mq(addr: %p, name: %s) has been exist", iter_mq, iter_mq->name);
mq = OS_NULL;
break;
}
}
if (OS_NULL != mq)
{
os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node);
os_spin_unlock(&gs_os_mq_resource_list_lock);
ret = _k_mq_init(mq, name, msg_pool, msg_pool_size, msg_size, OS_ALLOC_TYPE_STATIC);
if (OS_SUCCESS == ret)
{
mq->object_inited = OS_KOBJ_INITED;
}
else
{
os_spin_lock(&gs_os_mq_resource_list_lock);
os_list_del(&mq->resource_node);
os_spin_unlock(&gs_os_mq_resource_list_lock);
}
}
else
{
os_spin_unlock(&gs_os_mq_resource_list_lock);
}
return OS_TYPE_CONVERT(os_msgqueue_id, mq);
}
/**
***********************************************************************************************************************
* @brief Destroy a message queue.
*
* @param[in] msgqueue_id the message queue id to destroy.
*
* @return The result of destroy the message queue.
* @retval OS_SUCCESS Destroy the message queue successfully.
* @retval else Destroy the message queue failed.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_destroy(os_msgqueue_id msgqueue_id)
{
os_bool_t need_sched;
os_msgqueue_t *mq;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT(OS_FALSE == os_is_irq_active());
#ifndef OS_USING_HEAP
OS_ASSERT(OS_ALLOC_TYPE_STATIC == mq->object_alloc_type);
#endif
need_sched = OS_FALSE;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
mq->object_inited = OS_KOBJ_DEINITED;
if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))
{
need_sched = OS_TRUE;
}
if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))
{
need_sched = OS_TRUE;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
if (OS_TRUE == need_sched)
{
OS_KERNEL_ENTER();
OS_KERNEL_EXIT_SCHED();
}
#else
if (OS_TRUE == need_sched)
{
OS_KERNEL_EXIT_SCHED();
}
else
{
OS_KERNEL_EXIT();
}
#endif
os_spin_lock(&gs_os_mq_resource_list_lock);
os_list_del(&mq->resource_node);
os_spin_unlock(&gs_os_mq_resource_list_lock);
if (OS_KOBJ_ALLOC_TYPE_DYNAMIC == mq->object_alloc_type)
{
OS_KERNEL_FREE(mq->msg_pool);
mq->msg_pool = OS_NULL;
OS_KERNEL_FREE(mq);
mq = OS_NULL;
}
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function will send a message to message queue.
*
* @attention This interface is not allowed in the following cases:
* 1. In interrupt context and timeout is not OS_NO_WAIT.
* 2. Interrupt is disabled and timeout is not OS_NO_WAIT.
* 3. Scheduler is locked and timeout is not OS_NO_WAIT.
*
* @param[in] msgqueue_id The message queue id to destroy.
* @param[in] buffer The message buffer to be send.
* @param[in] buff_size The size of the message to be send.
* @param[in] timeout The timeout. This parameter can be the following value:
* 1. OS_NO_WAIT, if the message queue is full, return immediately.
* 2. OS_WAIT_FOREVER, if the message queue is full, wait until mesage queue
* is not full.
* 3. Else, if the message queue is full, wait until message queue is not
* full or timeout.
*
* @return The result of sending a message.
* @retval OS_SUCCESS Send a message successfully.
* @retval else Send a message failed.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_send(os_msgqueue_id msgqueue_id, const void *buffer, os_size_t buff_size, os_tick_t timeout)
{
os_err_t ret;
os_task_t *block_task;
os_bool_t need_sched;
register os_msgqueue_t *mq;
register os_mq_msg_t *msg;
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_KERNEL_INIT();
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_NULL != buffer);
OS_ASSERT((buff_size > 0U) && (buff_size <= mq->max_msg_size));
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));
ret = OS_SUCCESS;
need_sched = OS_FALSE;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
if (OS_NULL != mq->msg_queue_free)
{
_k_mq_get_free_msg(mq, msg);
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
#endif
(void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
msg->msg_len = buff_size;
}
else
{
if (OS_NO_WAIT == timeout)
{
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
ret = OS_FULL;
}
else
{
OS_ASSERT((OS_NULL != _k_task_self()));
if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
{
ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_TRUE);
}
else
{
ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_FALSE);
}
if (OS_SUCCESS == ret)
{
/*
* swap_data field filled by the receiver is point to free message buffer,
* so put the sending message into this buffer.
*/
msg = (os_mq_msg_t *)_k_task_self()->swap_data;
(void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
msg->msg_len = buff_size;
#ifdef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
#endif
}
}
}
if (OS_SUCCESS == ret)
{
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
#endif
if (&mq->recv_task_list_head != mq->recv_task_list_head.next)
{
/*
* When the receiver task has been blocked, put address of the message buffer filled data on swap_data,
* and unblock the receiver task. The receiver task will get message from swap_data.
*/
block_task = os_list_first_entry(&mq->recv_task_list_head, os_task_t, task_node);
need_sched = k_unblock_task(block_task);
block_task->swap_data = (os_ubase_t)msg;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
os_spin_unlock_irq(&mq->lock, irq_save);
if (OS_TRUE == need_sched)
{
OS_KERNEL_ENTER();
OS_KERNEL_EXIT_SCHED();
}
#else
if (OS_TRUE == need_sched)
{
OS_KERNEL_EXIT_SCHED();
}
else
{
OS_KERNEL_EXIT();
}
#endif
}
else
{
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
#endif
_k_mq_put_msg_to_queue(mq, msg);
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
}
}
return ret;
}
/**
***********************************************************************************************************************
* @brief This function will send a message to message queue head.
*
* @attention This interface is not allowed in the following cases:
* 1. In interrupt context and timeout is not OS_NO_WAIT.
* 2. Interrupt is disabled and timeout is not OS_NO_WAIT.
* 3. Scheduler is locked and timeout is not OS_NO_WAIT.
*
* @param[in] msgqueue_id The message queue id to destroy.
* @param[in] buffer The message buffer to be send.
* @param[in] buff_size The size of the message to be send.
* @param[in] timeout The timeout. This parameter can be the following value:
* 1. OS_NO_WAIT, if the message queue is full, return immediately.
* 2. OS_WAIT_FOREVER, if the message queue is full, wait until mesage queue
* is not full.
* 3. Else, if the message queue is full, wait until message queue is not
* full or timeout.
*
* @return The result of sending a message.
* @retval OS_SUCCESS Send a message successfully.
* @retval else Send a message failed.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_send_urgent(os_msgqueue_id msgqueue_id, const void *buffer, os_size_t buff_size, os_tick_t timeout)
{
os_err_t ret;
os_task_t *block_task;
os_bool_t need_sched;
register os_msgqueue_t *mq;
register os_mq_msg_t *msg;
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_KERNEL_INIT();
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_NULL != buffer);
OS_ASSERT((buff_size > 0U) && (buff_size <= mq->max_msg_size));
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));
ret = OS_SUCCESS;
need_sched = OS_FALSE;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
if (OS_NULL != mq->msg_queue_free)
{
_k_mq_get_free_msg(mq, msg);
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
#endif
(void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
msg->msg_len = buff_size;
}
else
{
if (OS_NO_WAIT == timeout)
{
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
ret = OS_FULL;
}
else
{
OS_ASSERT((OS_NULL != _k_task_self()));
if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
{
ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_TRUE);
}
else
{
ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_FALSE);
}
if (OS_SUCCESS == ret)
{
/*
* swap_data field filled by the receiver is point to free message buffer,
* so put the sending message into this buffer.
*/
msg = (os_mq_msg_t *)_k_task_self()->swap_data;
(void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
msg->msg_len = buff_size;
#ifdef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
#endif
}
}
}
if (OS_SUCCESS == ret)
{
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
#endif
if (&mq->recv_task_list_head != mq->recv_task_list_head.next)
{
/*
* When the receiver task has been blocked, put address of the message buffer filled data on swap_data,
* and unblock the receiver task. The receiver task will get message from swap_data.
*/
block_task = os_list_first_entry(&mq->recv_task_list_head, os_task_t, task_node);
need_sched = k_unblock_task(block_task);
block_task->swap_data = (os_ubase_t)msg;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
os_spin_unlock_irq(&mq->lock, irq_save);
if (OS_TRUE == need_sched)
{
OS_KERNEL_ENTER();
OS_KERNEL_EXIT_SCHED();
}
#else
if (OS_TRUE == need_sched)
{
OS_KERNEL_EXIT_SCHED();
}
else
{
OS_KERNEL_EXIT();
}
#endif
}
else
{
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
#endif
_k_mq_put_urgentmsg_to_queue(mq, msg);
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
}
}
return ret;
}
/**
***********************************************************************************************************************
* @brief This function will receive a message from message queue.
*
* @attention This interface is not allowed in the following cases:
* 1. In interrupt context and timeout is not OS_NO_WAIT.
* 2. Interrupt is disabled and timeout is not OS_NO_WAIT.
* 3. Scheduler is locked and timeout is not OS_NO_WAIT.
*
* @param[in] msgqueue_id The message queue id to destroy.
* @param[in] buffer The receive buffer is provided by the caller.
* @param[in] buff_size The size of the receive buffer.
* @param[in] timeout The timeout. This parameter can be the following value:
* 1. OS_NO_WAIT, if the message queue is empty, return immediately.
* 2. OS_WAIT_FOREVER, if the message queue is empty, wait until message queue is
* not empty.
* 3. Else, if the message queue is empty, wait until message queue is not empty
* or timeout.
* @param[in] recv_msg_size The size of the received message.
*
* @return The result of receiving a message.
* @retval OS_SUCCESS Receive a message successfully.
* @retval else Receive a message failed.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_recv(os_msgqueue_id msgqueue_id,
void *buffer,
os_size_t buff_size,
os_tick_t timeout,
os_size_t *recv_msg_size)
{
os_task_t *block_task;
os_err_t ret;
os_bool_t need_sched;
register os_msgqueue_t *mq;
register os_mq_msg_t *msg;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_NULL != buffer);
OS_ASSERT(buff_size > 0U);
OS_ASSERT(OS_NULL != recv_msg_size);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));
ret = OS_SUCCESS;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
if (OS_NULL == mq->msg_queue_head)
{
if (OS_NO_WAIT == timeout)
{
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
ret = OS_EMPTY;
}
else
{
OS_ASSERT((OS_NULL != _k_task_self()));
if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
{
ret = k_block_task(&mq->lock, irq_save, &mq->recv_task_list_head, timeout, OS_TRUE);
}
else
{
ret = k_block_task(&mq->lock, irq_save, &mq->recv_task_list_head, timeout, OS_FALSE);
}
if (OS_SUCCESS == ret)
{
#ifdef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
#endif
/*
* When the sender wakes up this task, it puts address of the message buffer on the swap_data,
* so there's no need to get message from the queue.
*/
msg = (os_mq_msg_t *)_k_task_self()->swap_data;
}
}
}
else
{
_k_mq_get_msg_from_queue(mq, msg);
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
#endif
}
if (OS_SUCCESS == ret)
{
if (buff_size < msg->msg_len)
{
/* When receive buffer size is less than message size, discard this message. */
OS_KERN_LOG(KERN_ERROR,
MQ_TAG,
"Recv buff size(%lu) of task(%s) is less than msg size(%lu)",
buff_size,
_k_task_self()->name,
msg->msg_len);
ret = OS_NOMEM;
}
else
{
(void)memcpy(buffer, (void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), msg->msg_len);
*recv_msg_size = msg->msg_len;
}
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
#endif
if (&mq->send_task_list_head != mq->send_task_list_head.next)
{
/*
* When the sender task was blocked, the message buffer that has just been processed will
* not be released into free pool. Instead, this message buffer is assigned to the sender task.
*/
block_task = os_list_first_entry(&mq->send_task_list_head, os_task_t, task_node);
need_sched = k_unblock_task(block_task);
block_task->swap_data = (os_ubase_t)msg;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
os_spin_unlock_irq(&mq->lock, irq_save);
if (OS_TRUE == need_sched)
{
OS_KERNEL_ENTER();
OS_KERNEL_EXIT_SCHED();
}
#else
if (OS_TRUE == need_sched)
{
OS_KERNEL_EXIT_SCHED();
}
else
{
OS_KERNEL_EXIT();
}
#endif
}
else
{
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
#endif
_k_mq_release_free_msg(mq, msg);
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
}
}
return ret;
}
/**
***********************************************************************************************************************
* @brief This function to set the mq wake type(by prio or fifo)
*
* @param[in] msgqueue_id The message queue id to set.
* @param[in] wake_type The message queue wake type to set.
*
* @return The operation result.
* @retval OS_SUCCESS If the operation successful.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_set_wake_type(os_msgqueue_id msgqueue_id, uint8_t wake_type)
{
os_err_t ret;
os_msgqueue_t *mq;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT((OS_MQ_WAKE_TYPE_PRIO == wake_type) || (OS_MQ_WAKE_TYPE_FIFO == wake_type));
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
if (os_list_empty(&mq->recv_task_list_head) && os_list_empty(&mq->send_task_list_head))
{
mq->wake_type = wake_type;
ret = OS_SUCCESS;
}
else
{
ret = OS_BUSY;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
return ret;
}
/**
***********************************************************************************************************************
* @brief This function to reset message queue
*
* @param[in] msgqueue_id The message queue id to reset.
*
* @return The operation result.
* @retval OS_SUCCESS Reset success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_reset(os_msgqueue_id msgqueue_id)
{
os_mq_msg_t *msg;
os_bool_t need_sched;
os_msgqueue_t *mq;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
need_sched = OS_FALSE;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
while (1)
{
if (OS_NULL == mq->msg_queue_head)
{
break;
}
_k_mq_get_msg_from_queue(mq, msg);
_k_mq_release_free_msg(mq, msg);
}
if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))
{
need_sched = OS_TRUE;
}
if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))
{
need_sched = OS_TRUE;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
if (OS_TRUE == need_sched)
{
OS_KERNEL_ENTER();
OS_KERNEL_EXIT_SCHED();
}
#else
if (OS_TRUE == need_sched)
{
OS_KERNEL_EXIT_SCHED();
}
else
{
OS_KERNEL_EXIT();
}
#endif
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function to test the message queue is empty
*
* @param[in] msgqueue_id The message queue id to test.
* @param[out] empty_flag A pointer to return the empty flag of message queue.
* OS_TRUE indicates that the message queue is empty.
*
* @return The operation result.
* @retval OS_SUCCESS Operation success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_is_empty(os_msgqueue_id msgqueue_id, os_bool_t *empty_flag)
{
os_msgqueue_t *mq;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT(OS_NULL != empty_flag);
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
if (OS_NULL == mq->msg_queue_head)
{
*empty_flag = OS_TRUE;
}
else
{
*empty_flag = OS_FALSE;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function to test the message queue is full
*
* @param[in] msgqueue_id The message queue id to test.
* @param[out] full_flag A pointer to return the full flag of message queue.
* OS_TRUE indicates that the message queue is full.
*
* @return The operation result.
* @retval OS_SUCCESS Operation success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_is_full(os_msgqueue_id msgqueue_id, os_bool_t *full_flag)
{
os_msgqueue_t *mq;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT(OS_NULL != full_flag);
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
if (OS_NULL == mq->msg_queue_free)
{
*full_flag = OS_TRUE;
}
else
{
*full_flag = OS_FALSE;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function to get the message queue's capacity
*
* @param[in] msgqueue_id The message queue id to get.
* @param[out] msgqueue_depth A pointer to return the capacity of message queue.
*
* @return The operation result.
* @retval OS_SUCCESS Get success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_get_queue_depth(os_msgqueue_id msgqueue_id, uint16_t *msgqueue_depth)
{
os_msgqueue_t *mq;
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT(OS_NULL != msgqueue_depth);
*msgqueue_depth = mq->queue_depth;
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function to get the message queue's msgs size
*
* @param[in] msgqueue_id The message queue id to get.
* @param[out] max_msg_size A pointer to return the max msg size of message queue.
*
* @return The operation result.
* @retval OS_SUCCESS Get success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_get_max_msg_size(os_msgqueue_id msgqueue_id, os_size_t *max_msg_size)
{
os_msgqueue_t *mq;
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT((OS_NULL != mq));
OS_ASSERT((OS_KOBJ_INITED == mq->object_inited));
OS_ASSERT((OS_NULL != max_msg_size));
*max_msg_size = mq->max_msg_size;
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function to get the message queue's used entry
*
* @param[in] msgqueue_id The message queue id to get.
* @param[out] used_msgs A pointer to return the used entry count of message queue.
*
* @return The operation result.
* @retval OS_SUCCESS Get success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_get_used_msgs(os_msgqueue_id msgqueue_id, uint16_t *used_msgs)
{
os_msgqueue_t *mq;
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT(OS_NULL != used_msgs);
*used_msgs = mq->used_msgs;
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief This function to get the message queue's unused entry
*
* @param[in] mq_id The message queue id to get.
* @param[out] unused_msgs A pointer to return the unused entry count of message queue.
*
* @return The operation result.
* @retval OS_SUCCESS Get success.
* @retval else Error code.
***********************************************************************************************************************
*/
os_err_t os_msgqueue_get_unused_msgs(os_msgqueue_id msgqueue_id, uint16_t *unused_msgs)
{
os_msgqueue_t *mq;
OS_KERNEL_INIT();
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
OS_ASSERT(OS_NULL != unused_msgs);
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
#else
OS_KERNEL_ENTER();
#endif
*unused_msgs = mq->queue_depth - mq->used_msgs;
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
return OS_SUCCESS;
}
/**
***********************************************************************************************************************
* @brief Get the name of the specified msgqueue.
*
* @param[in] msgqueue_id The ID of msgqueue.
*
* @return The name of the specified msgqueue.
***********************************************************************************************************************
*/
const char *os_msgqueue_get_name(os_msgqueue_id msgqueue_id)
{
os_msgqueue_t *mq;
mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);
OS_ASSERT(OS_NULL != mq);
OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
return mq->name;
}
#if defined(OS_USING_SHELL)
#include <shell.h>
#define SH_SHOW_TASK_CNT_MAX 10
typedef struct
{
os_msgqueue_t *mq;
uint16_t used_msgs;
uint16_t recv_block_task_count;
uint16_t send_block_task_count;
block_task_info_t send_block_task[SH_SHOW_TASK_CNT_MAX];
block_task_info_t recv_block_task[SH_SHOW_TASK_CNT_MAX];
} sh_mq_info_t;
static os_err_t os_mq_show(os_msgqueue_t *mq)
{
sh_mq_info_t mq_info;
os_task_t *iter_task;
uint16_t task_send_index;
uint16_t task_recv_index;
OS_KERNEL_INIT();
if ((OS_NULL == mq) || (OS_KOBJ_INITED != mq->object_inited))
{
os_kprintf("The input parameter is an illegal message queue object.\r\n");
return OS_FAILURE;
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
os_spin_lock_irq(&mq->lock, &irq_save);
k_kernel_spin_lock();
#else
OS_KERNEL_ENTER();
#endif
mq_info.mq = mq;
mq_info.used_msgs = mq->used_msgs;
mq_info.recv_block_task_count = os_list_len(&mq->recv_task_list_head);
mq_info.send_block_task_count = os_list_len(&mq->send_task_list_head);
task_send_index = 0;
os_list_for_each_entry(iter_task, &mq->send_task_list_head, os_task_t, task_node)
{
mq_info.send_block_task[task_send_index].current_priority = iter_task->current_priority;
mq_info.send_block_task[task_send_index].name = iter_task->name;
task_send_index++;
if (task_send_index >= SH_SHOW_TASK_CNT_MAX)
{
break;
}
}
task_recv_index = 0;
os_list_for_each_entry(iter_task, &mq->recv_task_list_head, os_task_t, task_node)
{
mq_info.recv_block_task[task_recv_index].current_priority = iter_task->current_priority;
mq_info.recv_block_task[task_recv_index].name = iter_task->name;
task_recv_index++;
if (task_recv_index >= SH_SHOW_TASK_CNT_MAX)
{
break;
}
}
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
k_kernel_spin_unlock();
os_spin_unlock_irq(&mq->lock, irq_save);
#else
OS_KERNEL_EXIT();
#endif
os_kprintf("%-*s %-12u %-10u",
OS_NAME_MAX,
(mq_info.mq->name[0] != '\0') ? mq_info.mq->name : "-",
mq_info.used_msgs,
mq_info.mq->queue_depth);
if (mq_info.send_block_task_count > 0)
{
os_kprintf(" %-6u(send):", mq_info.send_block_task_count);
k_show_blocked_task(mq_info.send_block_task, task_send_index);
if (mq_info.send_block_task_count > task_send_index)
{
os_kprintf("/...\r\n");
}
else
{
os_kprintf("\r\n");
}
}
else if (mq_info.recv_block_task_count > 0)
{
os_kprintf(" %-6u(recv):", mq_info.recv_block_task_count);
k_show_blocked_task(mq_info.recv_block_task, task_recv_index);
if (mq_info.recv_block_task_count > task_recv_index)
{
os_kprintf("/...\r\n");
}
else
{
os_kprintf("\r\n");
}
}
else
{
os_kprintf(" %-6u\r\n", 0);
}
return OS_SUCCESS;
}
static os_err_t sh_show_mq_info(int32_t argc, char **argv)
{
os_msgqueue_t *iter_mq;
os_msgqueue_t *mq_tmp;
uint16_t len;
OS_UNREFERENCE(argc);
OS_UNREFERENCE(argv);
os_kprintf("%-*s %-13s %-11s %-10s\r\n",
OS_NAME_MAX,
"Message queue",
"Message Count",
"Queue depth",
"Block Task");
len = OS_NAME_MAX;
while (len--)
{
os_kprintf("-");
}
os_kprintf(" ------------- ----------- ----------\r\n");
if (argc >= 2)
{
if (argv[1])
{
mq_tmp = (os_msgqueue_t *)strtoul(argv[1], 0, 0);
os_spin_lock(&gs_os_mq_resource_list_lock);
os_list_for_each_entry(iter_mq, &gs_os_mq_resource_list_head, os_msgqueue_t, resource_node)
{
if (mq_tmp == iter_mq)
{
os_mq_show(mq_tmp);
os_spin_unlock(&gs_os_mq_resource_list_lock);
return OS_SUCCESS;
}
}
os_spin_unlock(&gs_os_mq_resource_list_lock);
}
os_kprintf("Invalid Mq Object.\r\n");
return OS_FAILURE;
}
os_spin_lock(&gs_os_mq_resource_list_lock);
os_list_for_each_entry(iter_mq, &gs_os_mq_resource_list_head, os_msgqueue_t, resource_node)
{
os_mq_show(iter_mq);
}
os_spin_unlock(&gs_os_mq_resource_list_lock);
return OS_SUCCESS;
}
SH_CMD_EXPORT(show_mq, sh_show_mq_info, "Show message queue information");
#endif /* defined(OS_USING_SHELL) */
#endif /* end of OS_USING_MESSAGEQUEUE */