菜单
Cobalt/Message queue services
详细描述
用于发送任意大小消息的实时 IPC 机制
消息队列是一种实时任务可以通过 xkernel 管理的消息队列交换或传递数据的方法。消息可以有不同的长度，并且可以分配不同的类型或用途。一个任务可以创建消息队列，并由多个任务使用，以发送和/或接收队列中的消息。
函数文档
rt_queue_alloc
void * rt_queue_alloc (RT_QUEUE *queue, size_t size);
分配消息缓冲区。
此服务从队列的内部池中分配一个消息缓冲区。此缓冲区可以在调用 rt_queue_send() 之前填充有效载荷信息。当成对使用时，这些服务提供了一个零拷贝接口用于发送消息。
参数
- queue 队列描述符。
- size 请求的缓冲区大小（以字节为单位）。零是一个可接受的值，这意味着消息不包含有效载荷；在这种情况下，接收方将收到一个零大小的消息。
返回值
成功时返回分配的缓冲区的地址，失败时返回 NULL。
标签
unrestricted, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_bind
int rt_queue_bind (RT_QUEUE *queue, const char *name, RTIME timeout);
绑定到消息队列。
此例程创建一个新的描述符，以引用由其符号名称标识的现有消息队列。如果在进入时对象不存在，调用者可能会阻塞，直到创建具有给定名称的队列。
参数
- queue 操作填充的队列描述符的地址。失败时，此内存的内容未定义。
- name 一个有效的以 NULL 结尾的名称，用于标识要绑定的队列。此字符串应与传递给 rt_queue_create() 的对象名称参数匹配。
- timeout 等待注册发生的时钟滴答数（参见注释）。传递 TM_INFINITE 会导致调用者无限期阻塞，直到对象注册。传递 TM_NONBLOCK 会导致服务立即返回，如果在进入时对象未注册，则不会等待。
返回值
成功时返回零。否则：
- 如果在检索完成之前为当前任务调用了 rt_task_unblock()，则返回 -EINTR。
- 如果 timeout 等于 TM_NONBLOCK 且在进入时未注册搜索的对象，则返回 -EWOULDBLOCK。
- 如果在指定时间内无法检索到对象，则返回 -ETIMEDOUT。
- 如果此服务应阻塞，但不是从 xkernel 线程调用的，则返回 -EPERM。
标签
xthread-nowait, switch-primary
注释
timeout 值被解释为 Alchemy 时钟分辨率的倍数（参见 -alchemy-clock-resolution 选项，默认为 1 纳秒）。
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
RT_QUEUE queue_send;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 绑定到消息队列
ret = rt_queue_bind(&queue_send, "MyQueue", TM_INFINITE);
if (ret < 0) {
fprintf(stderr, "Failed to bind queue\n");
return;
}
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue_send, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue_send, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
// 解除绑定
rt_queue_unbind(&queue_send);
}
void receiver_task_func(void *arg)
{
int ret;
void *msg_buf;
ssize_t msg_size;
RT_QUEUE queue_recv;
printf("Receiver: Waiting for message\n");
// 绑定到消息队列
ret = rt_queue_bind(&queue_recv, "MyQueue", TM_INFINITE);
if (ret < 0) {
fprintf(stderr, "Failed to bind queue\n");
return;
}
msg_size = rt_queue_receive(&queue_recv, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue_recv, msg_buf); // 读取完成要及时free掉内存
}
// 解除绑定
rt_queue_unbind(&queue_recv);
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_create
int rt_queue_create (RT_QUEUE *queue, const char *name, size_t poolsize, size_t qlimit, int mode);
创建消息队列。
创建一个消息队列对象，允许多个任务通过使用可变大小的消息来交换数据。消息队列在创建时为空。
参数
- queue 队列描述符的地址，成功调用后可用于唯一标识创建的对象。
- name 表示队列符号名称的 ASCII 字符串。当非 NULL 且非空时，此字符串的副本将用于将创建的队列索引到对象注册表中。
- poolsize 为保存消息预分配的消息缓冲池的大小（以字节为单位）。消息缓冲区将被声明并释放到此池中。缓冲池内存不能扩展。参见注释。
- qlimit 此参数允许限制在任何时间点可以排队的最大消息数，发送到已满的队列会产生错误。可以传递特殊值 Q_UNLIMITED 以禁用限制检查。
- mode 队列创建模式。以下标志可以 OR 进此位掩码中，每个标志都会影响新队列：
- Q_FIFO 使任务以 FIFO 顺序挂起在队列中以消费消息。
- Q_PRIO 使任务以优先级顺序挂起在队列中。
返回值
成功时返回零。否则：
- 如果 mode 无效或 poolsize 为零，则返回 -EINVAL。
- 如果系统无法从主堆中获取内存以创建队列，则返回 -ENOMEM。
- 如果名称与已注册的队列冲突，则返回 -EEXIST。
- 如果此服务从无效上下文（例如中断或非 xkernel 线程）调用，则返回 -EPERM。
标签
xthread-only, mode-unrestricted, switch-secondary
注释
队列可以由属于同一 xkernel 会话的多个进程共享。
每个挂起在队列中的消息消耗四个长字以及实际有效载荷大小，向下对齐到下一个长字边界。例如，在 32 位平台上，一个 6 字节的消息需要 24 字节的存储空间。
当 qlimit 被指定（即不同于 Q_UNLIMITED）时，此开销会自动计算，因此 poolsize / qlimit 字节的 qlimit 消息可以同时存储到池中。否则，poolsize 会在内部增加 5% 以应对这种开销。
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_delete
int rt_queue_delete (RT_QUEUE *queue);
删除消息队列。
此例程删除先前通过调用 rt_queue_create() 创建的队列对象。所有附加到该队列的资源将自动释放，包括所有挂起的消息。
参数
- queue 队列描述符。
返回值
成功时返回零。否则：
- 如果 queue 不是有效的队列描述符，则返回 -EINVAL。
- 如果此服务从异步上下文调用，则返回 -EPERM。
标签
mode-unrestricted, switch-secondary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_flush
int rt_queue_flush (RT_QUEUE *queue);
刷新队列中的挂起消息。
此例程刷新当前队列中所有挂起的消息，适当地释放所有消息缓冲区。
参数
- queue 队列描述符。
返回值
成功时返回零。否则：
- 如果 queue 不是有效的队列描述符，则返回 -EINVAL。
标签
unrestricted, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
// 刷新队列并释放缓存消息，receiver会触发超时故障
rt_queue_flush(&queue);
printf("Sender: Flush message\n");
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
struct timespec abs_timeout;
rt_task_sleep(500000000); // 睡眠500ms,等待缓存消息被刷新掉
printf("Receiver: Waiting for message\n");
// 设置绝对超时时间为当前时间加上2秒
clock_gettime(CLOCK_MONOTONIC, &abs_timeout);
abs_timeout.tv_sec += 2;
msg_size = rt_queue_receive_timed(&queue, &msg_buf, &abs_timeout);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_free
int rt_queue_free (RT_QUEUE *queue, void *buf);
释放消息缓冲区。
此服务将消息缓冲区释放到队列的内部池中。
参数
- queue 队列描述符。
- buf 要释放的消息缓冲区的地址。即使是不携带有效载荷数据的零大小消息也必须释放，因为它们被分配了有效的内存空间来存储内部信息。
返回值
成功时返回零；如果 buf 不是先前由 rt_queue_alloc() 服务分配的有效消息缓冲区，或者调用者未通过 rt_queue_receive() 成功返回获得消息的所有权，则返回 -EINVAL。
标签
unrestricted, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_inquire
int rt_queue_inquire (RT_QUEUE *queue, RT_QUEUE_INFO *info);
查询队列状态。
此例程返回指定队列的状态信息。
参数
- queue 队列描述符。
- info 指向返回缓冲区的指针，用于复制信息。
返回值
成功时返回零，并将状态信息写入 info 指向的结构体。否则：
- 如果 queue 不是有效的队列描述符，则返回 -EINVAL。
标签
unrestricted, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
RT_QUEUE_INFO info;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
ret = rt_queue_inquire(&queue, &info);
if (ret < 0) {
fprintf(stderr, "Failed to inquire infomation from queue\n");
return;
}
printf("Get Current Queue Infomation:\n");
printf(" name:%s, usedmem:%d, poolsize:%d, qlimit:%d\n",info.name,info.usedmem,info.poolsize,info.qlimit);
printf(" mode:%d, nmessages:%d, nwaiters:%d\n",info.mode,info.nmessages,info.nwaiters);
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_read
ssize_t rt_queue_read (RT_QUEUE *q, void *buf, size_t size, RTIME timeout);
从队列中读取（带相对标量超时）。
此例程是 rt_queue_read_timed() 的变体，接受以标量值表示的相对超时规范。
参数
- q: 队列描述符。
- buf: 一个指向内存区域的指针，成功时将写入接收到的消息有效载荷。
- size: buf 指向的内存区域的长度（以字节为单位）。
- timeout: 以时钟滴答表示的延迟。传递 TM_INFINITE 会导致调用者无限期阻塞，直到有消息可用。传递 TM_NONBLOCK 会导致服务立即返回而不阻塞，如果没有消息可用。
标签
xthread-nowait, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
char msg_buf[MAX_MSG_SIZE];
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_read(&queue, msg_buf, sizeof(msg_buf), TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_read_timed
ssize_t rt_queue_read_timed (RT_QUEUE *q, void *buf, size_t size, const struct timespec *abs_timeout);
从队列中读取消息。
此服务从指定队列中读取下一个可用消息。
参数
- q: 队列描述符。
- buf: 一个指向内存区域的指针，成功时将写入接收到的消息有效载荷。传递数据的内部消息缓冲区由此调用自动释放。如果配置中启用了 --enable-pshared，buf 必须通过 xkernel 内存分配器（如 xnmalloc() 或基于它的任何服务，如 rt_heap_alloc()）获取。
- size: buf 指向的内存区域的长度（以字节为单位）。大于 size 的消息将被适当截断。
- abs_timeout: 以秒/纳秒表示的绝对时间，基于 Alchemy 时钟，指定等待消息可用的时间限制。传递 NULL 会导致调用者无限期阻塞，直到有消息可用。传递 { .tv_sec = 0, .tv_nsec = 0 } 会导致服务立即返回而不阻塞，如果没有消息可用。
返回值
成功时返回复制到 buf 的字节数。零是一个可能的值，对应于传递给 rt_queue_send() 或 rt_queue_write() 的零大小消息。否则：
- 如果在消息到达之前达到 abs_timeout，则返回 -ETIMEDOUT。
- 如果 abs_timeout 为 { .tv_sec = 0, .tv_nsec = 0 } 且在进入调用时没有消息可用，则返回 -EWOULDBLOCK。
- 如果在消息可用之前为当前任务调用了 rt_task_unblock()，则返回 -EINTR。
- 如果 q 不是有效的队列描述符，则返回 -EINVAL。
- 如果在调用者等待消息时 q 被删除，则返回 -EIDRM。在这种情况下，返回此服务时 q 不再有效。
- 如果此服务应阻塞，但不是从 xkernel 线程调用的，则返回 -EPERM。
标签
xthread-nowait, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
char msg_buf[MAX_MSG_SIZE];
ssize_t msg_size;
struct timespec abs_timeout;
printf("Receiver: Waiting for message\n");
// 设置绝对超时时间为当前时间加上2秒
clock_gettime(CLOCK_MONOTONIC, &abs_timeout);
abs_timeout.tv_sec += 2;
msg_size = rt_queue_read_timed(&queue, msg_buf, sizeof(msg_buf), &abs_timeout);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_read_until
ssize_t rt_queue_read_until (RT_QUEUE *q, void *buf, size_t size, RTIME abs_timeout);
从队列中读取（带绝对标量超时）。
此例程是 rt_queue_read_timed() 的变体，接受以标量值表示的绝对超时规范。
参数
- q: 队列描述符。
- buf: 一个指向内存区域的指针，成功时将写入接收到的消息有效载荷。
- size: buf 指向的内存区域的长度（以字节为单位）。
- abs_timeout: 以时钟滴答表示的绝对时间。传递 TM_INFINITE 会导致调用者无限期阻塞，直到有消息可用。传递 TM_NONBLOCK 会导致服务立即返回而不阻塞，如果没有消息可用。
标签
xthread-nowait, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
char msg_buf[MAX_MSG_SIZE];
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
RTIME timeout = rt_timer_read() + 2000000000; // 2 seconds
msg_size = rt_queue_read_until(&queue, msg_buf, sizeof(msg_buf), timeout);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_receive
ssize_t rt_queue_receive (RT_QUEUE *q, void **bufp, RTIME timeout);
从队列中接收消息（带相对标量超时）。
此例程是 rt_queue_receive_timed() 的变体，接受以标量值表示的相对超时规范。
参数
- q: 队列描述符。
- bufp: 一个指向内存位置的指针，成功时将写入接收到的消息地址。
- timeout: 以时钟滴答表示的延迟。传递 TM_INFINITE 会导致调用者无限期阻塞，直到有消息可用。传递 TM_NONBLOCK 会导致服务立即返回而不阻塞，如果没有消息可用。
标签
xthread-nowait, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_receive_timed
ssize_t rt_queue_receive_timed (RT_QUEUE *q, void **bufp, const struct timespec *abs_timeout);
从队列中接收消息（带绝对超时日期）。
此服务从指定队列中接收下一个可用消息。
参数
- q: 队列描述符。
- bufp: 一个指向内存位置的指针，成功时将写入接收到的消息地址。一旦消息被消费，消息空间应使用 rt_queue_free() 释放。
- abs_timeout: 以秒/纳秒表示的绝对时间，基于 Alchemy 时钟，指定等待消息可用的时间限制。传递 NULL 会导致调用者无限期阻塞，直到有消息可用。传递 { .tv_sec = 0, .tv_nsec = 0 } 会导致服务立即返回而不阻塞，如果没有消息可用。
返回值
成功时返回接收到的消息的字节数。零是一个可能的值，对应于传递给 rt_queue_send() 或 rt_queue_write() 的零大小消息。否则：
- 如果在消息到达之前达到 abs_timeout，则返回 -ETIMEDOUT。
- 如果 abs_timeout 为 { .tv_sec = 0, .tv_nsec = 0 } 且在进入调用时没有消息可用，则返回 -EWOULDBLOCK。
- 如果在消息可用之前为当前任务调用了 rt_task_unblock()，则返回 -EINTR。
- 如果 q 不是有效的队列描述符，则返回 -EINVAL。
- 如果在调用者等待消息时 q 被删除，则返回 -EIDRM。在这种情况下，返回此服务时 q 不再有效。
- 如果此服务应阻塞，但不是从 xkernel 线程调用的，则返回 -EPERM。
标签
xthread-nowait, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
struct timespec abs_timeout;
printf("Receiver: Waiting for message\n");
// 设置绝对超时时间为当前时间加上2秒
clock_gettime(CLOCK_MONOTONIC, &abs_timeout);
abs_timeout.tv_sec += 2;
msg_size = rt_queue_receive_timed(&queue, &msg_buf, &abs_timeout);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_receive_until
ssize_t rt_queue_receive_until (RT_QUEUE *q, void **bufp, RTIME abs_timeout);
从队列中接收消息（带绝对标量超时）。
此例程是 rt_queue_receive_timed() 的变体，接受以标量值表示的绝对超时规范。
参数
- q: 队列描述符。
- bufp: 一个指向内存位置的指针，成功时将写入接收到的消息地址。
- abs_timeout: 以时钟滴答表示的绝对时间。传递 TM_INFINITE 会导致调用者无限期阻塞，直到有消息可用。传递 TM_NONBLOCK 会导致服务立即返回而不阻塞，如果没有消息可用。
标签
xthread-nowait, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
RTIME timeout = rt_timer_read() + 2000000000; // 2 seconds
msg_size = rt_queue_receive_until(&queue, &msg_buf, timeout);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_send
int rt_queue_send (RT_QUEUE *q, const void *buf, size_t size, int mode);
发送消息到队列。
此服务将完整消息发送到指定队列。消息必须由之前调用 rt_queue_alloc() 分配。
参数
- q: 队列描述符。
- buf: 要发送的消息缓冲区地址。消息缓冲区必须使用 rt_queue_alloc() 服务分配。一旦传递给 rt_queue_send()，buf 指向的内存不再由发送者控制，因此不应再被引用；此内存的释放必须由接收方处理。
- size: 消息的实际大小（以字节为单位），可能小于从 rt_queue_alloc() 获得的缓冲区的分配大小。零是一个有效值，在这种情况下将发送一个空消息。
- mode: 一组影响操作的标志：
- Q_URGENT 将消息添加到消息队列的前面，确保 LIFO 顺序。
- Q_NORMAL 将消息添加到消息队列的末尾，确保 FIFO 顺序。
- Q_BROADCAST 将消息发送给所有当前等待消息的任务。消息不会被复制；而是维护一个引用计数，以确保消息在最后一个接收者使用 rt_queue_free() 释放其引用之前保持有效，之后消息空间将返回到队列的内部池。
返回值
成功时，此服务返回因操作而被唤醒的接收者数量。如果返回零，则表示队列接收端没有任务在等待，消息已入队。错误时，返回以下错误代码之一：
- 如果 q 不是消息队列描述符，mode 无效，或 buf 为 NULL，则返回 -EINVAL。
- 如果排队消息会超过创建队列时定义的限制，则返回 -ENOMEM。
标签
unrestricted, switch-primary
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
}
void receiver_task_func(void *arg)
{
void *msg_buf;
ssize_t msg_size;
printf("Receiver: Waiting for message\n");
msg_size = rt_queue_receive(&queue, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue, msg_buf); // 读取完成要及时free掉内存
}
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}rt_queue_unbind
int rt_queue_unbind (RT_QUEUE *q);
解除消息队列的绑定。
参数
- q: 队列描述符。
此例程释放之前对消息队列的绑定。此调用返回后，描述符将不再有效，无法再引用该对象。
标签
thread-unrestricted
示例代码
c{filename="app.c"}
复制代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <alchemy/task.h>
#include <alchemy/queue.h>
#include <alchemy/timer.h>
#define TASK_PRIO 50
#define TASK_MODE T_JOINABLE
#define TASK_STKSZ 0
#define QUEUE_SIZE 1024
#define MAX_MSG_SIZE 100
RT_QUEUE queue;
RT_TASK sender_task, receiver_task;
void sender_task_func(void *arg)
{
int ret;
const char *message = "Hello from xkernel sender task!";
void *msg_buf;
RT_QUEUE queue_send;
rt_task_sleep(500000000); // 睡眠500ms,确保接收者任务已经开始等待
// 绑定到消息队列
ret = rt_queue_bind(&queue_send, "MyQueue", TM_INFINITE);
if (ret < 0) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 分配消息缓冲区
msg_buf = rt_queue_alloc(&queue_send, strlen(message) + 1);
if (msg_buf == NULL) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
// 复制消息到缓冲区
strcpy(msg_buf, message);
printf("Sender: Sending message\n");
ret = rt_queue_send(&queue_send, msg_buf, strlen(message) + 1, Q_NORMAL);
if (ret < 0) {
fprintf(stderr, "Sender: Failed to send message, error code: %d\n", ret);
} else {
printf("Sender: Message sent successfully\n");
}
// 解除绑定
rt_queue_unbind(&queue_send);
}
void receiver_task_func(void *arg)
{
int ret;
void *msg_buf;
ssize_t msg_size;
RT_QUEUE queue_recv;
printf("Receiver: Waiting for message\n");
// 绑定到消息队列
ret = rt_queue_bind(&queue_recv, "MyQueue", TM_INFINITE);
if (ret < 0) {
fprintf(stderr, "Failed to allocate message buffer\n");
return;
}
msg_size = rt_queue_receive(&queue_recv, &msg_buf, TM_INFINITE);
if (msg_size < 0) {
if (msg_size == -ETIMEDOUT) {
printf("Receiver: Timeout while waiting for message\n");
} else {
fprintf(stderr, "Receiver: Failed to receive message, error code: %zd\n", msg_size);
}
} else {
printf("Receiver: Received message of size %zd: %s\n", msg_size, (char *)msg_buf);
rt_queue_free(&queue_recv, msg_buf); // 读取完成要及时free掉内存
}
// 解除绑定
rt_queue_unbind(&queue_recv);
}
int main(int argc, char *argv[])
{
int ret;
// 创建消息队列
ret = rt_queue_create(&queue, "MyQueue", QUEUE_SIZE, MAX_MSG_SIZE, Q_FIFO);
if (ret) {
fprintf(stderr, "Failed to create queue, error code: %d\n", ret);
return EXIT_FAILURE;
}
// 创建发送者任务
ret = rt_task_create(&sender_task, "SenderTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create sender task, error code: %d\n", ret);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 创建接收者任务
ret = rt_task_create(&receiver_task, "ReceiverTask", TASK_STKSZ, TASK_PRIO, TASK_MODE);
if (ret) {
fprintf(stderr, "Failed to create receiver task, error code: %d\n", ret);
rt_task_delete(&sender_task);
rt_queue_delete(&queue);
return EXIT_FAILURE;
}
// 启动任务
rt_task_start(&receiver_task, &receiver_task_func, NULL);
rt_task_start(&sender_task, &sender_task_func, NULL);
// 等待任务完成
rt_task_join(&sender_task);
rt_task_join(&receiver_task);
// 清理资源
rt_task_delete(&sender_task);
rt_task_delete(&receiver_task);
rt_queue_delete(&queue);
printf("Program completed\n");
return EXIT_SUCCESS;
}最近修改: 2025-09-30
- SP7010工智机如何使用本地I/O？
- SP7010PING不通且工智机扫描不到怎么解决
- 如何查询SX51系列工智机的串口波特率？
- SX21系列工智机如何更新Bios
- 工程任务执行超时后，如何安全删除该工程？
- 工智机跨设备工程报错"设备版本不匹配"如何解决？
- 工智机通过指令安装deb组件方法
- PC如何连接到工智机？
- 工智机远程文件传输
- 工智机系统时间不正确
- 不使用RTE如何检查硬件串口？
- Modbus运行样例
- MetaOS一级虚拟化系统如何安装软件？
- 工智机如何远程打开GUI程序
- MetaOS一级虚拟化系统默认开机启动浏览器
- MetaFacture扫描不到工智机如何处理？
- 如何添加RTE外部库?
- MetaOS一层虚拟化下如何安装Firefox浏览器？
- socat串口网口透传工具介绍
- SX5/58系列工智机启动共享文件夹
- 外界触摸屏下，工智机如何关闭三指触控？
- 在Windows主机上使用Socat
- SX5系列工智机无法访问网口配置问题
- SX2系列工智机常见网络问题处理方法
- MetaOS一级虚拟化，文件系统损坏如何修复？
- Onlinehelp相关使用问题
- 问题收集反馈