RTOS系统 -- ARM Cortex-M4 RPMSG之通道初始化函数

RPMsg Lite 在 ARM Cortex-M4 RTOS 中的使用

简介

在ARM Cortex-M4处理器上使用的RTOS（实时操作系统）中，rpmsg_lite是一个轻量级的远程处理消息传递框架，通常用于多核处理器或多核系统中不同处理器之间的通信。本文档将介绍 rpmsg_lite_remote_init 和 rpmsg_lite_master_init 两个函数的作用及使用场景。

rpmsg_lite_remote_init

rpmsg_lite_remote_init函数用于初始化RPMsg Lite的远程端。远程端通常指从处理器或从核，在多核系统中，这个核通常接收来自主核的指令或数据，并进行相应的处理。

作用

• 初始化RPMsg Lite远程端的结构体和资源。
• 设置远程端与主端之间的通信通道。
• 使能远程端接收主端发送的消息。

使用场景

• 从处理器（或从核）启动时，作为通信的接收端或响应端。
• 在多核系统中，远程核需要与主核进行消息传递时。

示例代码

struct rpmsg_lite_instance *rpmsg_instance;
rpmsg_instance = rpmsg_lite_remote_init(shmem_base, RL_PLATFORM_HIGHEST_LINK_ID, RL_NO_FLAGS);

rpmsg_lite_master_init

rpmsg_lite_master_init函数用于初始化RPMsg Lite的主端。主端通常指主处理器或主核，在多核系统中，主核负责发送指令或数据到远程核，并处理返回的结果。

作用

• 初始化RPMsg Lite主端的结构体和资源。
• 设置主端与远程端之间的通信通道。
• 使能主端发送消息到远程端。

使用场景

• 主处理器（或主核）启动时，作为通信的发送端或控制端。
• 在多核系统中，主核需要与远程核进行消息传递时。

示例代码

struct rpmsg_lite_instance *rpmsg_instance;
rpmsg_instance = rpmsg_lite_master_init(shmem_base, RL_PLATFORM_HIGHEST_LINK_ID, RL_NO_FLAGS);

小结

• rpmsg_lite_remote_init：用于从核初始化，主要用于接收和响应主核的消息。
• rpmsg_lite_master_init：用于主核初始化，主要用于发送消息到从核并处理响应。

在实际应用中，当多核系统中需要进行通信时，主核会调用rpmsg_lite_master_init初始化通信，而从核则会调用rpmsg_lite_remote_init进行相应的初始化，以建立起主从核之间的消息传递通道。

高性能接收队列调度示例

代码

#include "rpmsg_lite.h"
#include "rpmsg_queue.h"
#include "rpmsg_ns.h"
#include "FreeRTOS.h"
#include "task.h"
#include "semphr.h"#define APP_EPT_ADDR (30U) // 应用定义的端点地址
#define BUFFER_SIZE 8 // 缓存大小// 消息缓存结构体
typedef struct {void *data;uint32_t len;
} message_t;// RPMsg-Lite实例、端点和队列句柄
struct rpmsg_lite_instance *my_rpmsg;
struct rpmsg_lite_endpoint *my_ept;
rpmsg_queue_handle my_queue;
SemaphoreHandle_t send_mutex;// 消息缓存和相关变量
message_t message_buffer[BUFFER_SIZE];
uint8_t buffer_head = 0;
uint8_t buffer_tail = 0;
uint8_t buffer_count = 0;
SemaphoreHandle_t buffer_mutex;
SemaphoreHandle_t buffer_sem;// 封装发送消息的函数
int32_t send_message(void *data, uint32_t len)
{int32_t result;// 锁定发送操作xSemaphoreTake(send_mutex, portMAX_DELAY);// 发送消息result = rpmsg_lite_send(my_rpmsg, my_ept, APP_EPT_ADDR, data, len, RL_DONT_BLOCK);// 解锁发送操作xSemaphoreGive(send_mutex);return result;
}// 接收消息的任务
void receive_task(void *pvParameters)
{void *rx_buffer;uint32_t len;int32_t result;while (1){// 等待接收消息result = rpmsg_queue_recv(my_rpmsg, my_queue, &APP_EPT_ADDR, &rx_buffer, &len, RL_BLOCK);if (result == RL_SUCCESS){// 锁定缓冲区xSemaphoreTake(buffer_mutex, portMAX_DELAY);// 检查缓冲区是否已满if (buffer_count < BUFFER_SIZE){// 将消息存入缓冲区message_buffer[buffer_head].data = rx_buffer;message_buffer[buffer_head].len = len;buffer_head = (buffer_head + 1) % BUFFER_SIZE;buffer_count++;// 通知处理任务xSemaphoreGive(buffer_sem);}else{// 如果缓冲区已满，释放接收缓冲区rpmsg_lite_release_rx_buffer(my_rpmsg, rx_buffer);}// 解锁缓冲区xSemaphoreGive(buffer_mutex);}}
}// 处理消息的任务
void process_task(void *pvParameters)
{message_t msg;while (1){// 等待消息缓存有数据xSemaphoreTake(buffer_sem, portMAX_DELAY);// 锁定缓冲区xSemaphoreTake(buffer_mutex, portMAX_DELAY);// 取出消息进行处理if (buffer_count > 0){msg = message_buffer[buffer_tail];buffer_tail = (buffer_tail + 1) % BUFFER_SIZE;buffer_count--;// 解锁缓冲区xSemaphoreGive(buffer_mutex);// 处理消息// ...// 例如：发送回主核int32_t send_result = send_message(msg.data, msg.len);if (send_result != RL_SUCCESS){// 处理发送失败// ...}// 释放消息缓冲区rpmsg_lite_release_rx_buffer(my_rpmsg, msg.data);}else{// 解锁缓冲区xSemaphoreGive(buffer_mutex);}}
}int main(void)
{// 初始化RPMsg-Lite实例my_rpmsg = rpmsg_lite_remote_init(/* RPMsg Lite Transport Initialization Parameters */);// 创建消息队列my_queue = rpmsg_queue_create(my_rpmsg);// 创建RPMsg端点my_ept = rpmsg_lite_create_ept(my_rpmsg, APP_EPT_ADDR, rpmsg_queue_rx_cb, my_queue);// 创建缓冲区互斥量和信号量buffer_mutex = xSemaphoreCreateMutex();buffer_sem = xSemaphoreCreateBinary();send_mutex = xSemaphoreCreateMutex();// 启动接收任务和处理任务xTaskCreate(receive_task, "Receive Task", configMINIMAL_STACK_SIZE, NULL, tskIDLE_PRIORITY + 1, NULL);xTaskCreate(process_task, "Process Task", configMINIMAL_STACK_SIZE, NULL, tskIDLE_PRIORITY + 1, NULL);// 启动调度器vTaskStartScheduler();// 不会到达这里for (;;){}
}

这个代码实现了消息的缓存和逐个处理，具体说明如下：

1. 消息缓存结构体：定义了一个 message_t 结构体用于存储消息数据和长度。

2. 消息缓冲区和相关变量：定义了一个固定大小的消息缓冲区 message_buffer，以及缓冲区的头尾指针和计数器。

3. 信号量和互斥量：使用FreeRTOS的信号量和互斥量来同步和保护对缓冲区的访问。

4. 接收消息任务：receive_task 函数接收消息后，将消息存入缓冲区。如果缓冲区已满，则释放接收缓冲区。

5. 处理消息任务：process_task 函数等待信号量，取出缓冲区中的消息进行处理，并释放消息缓冲区。

6. 创建任务和启动调度器：在 main 函数中创建接收任务和处理任务，并启动FreeRTOS调度器。

任务调度机制

在FreeRTOS中，任务调度是基于任务的优先级和任务的状态（就绪、运行、阻塞、挂起）进行的，而不是按顺序调度。具体调度机制如下：

1. 优先级调度：优先级高的任务优先运行。receive_task 和 process_task 的优先级在这个示例中是一样的（默认最低优先级），所以它们将轮流运行。

2. 时间片轮转：当多个同优先级的任务都处于就绪状态时，调度器会在它们之间进行时间片轮转。也就是说，这些任务会轮流获得CPU时间。

3. 任务阻塞：当一个任务调用阻塞API（例如 xSemaphoreTake 等待信号量时），这个任务会进入阻塞状态，调度器会切换到其他就绪状态的任务。

在上面的代码中：

• receive_task 在没有消息接收时会一直阻塞在 rpmsg_queue_recv 上。
• process_task 在没有消息处理时会一直阻塞在 xSemaphoreTake(buffer_sem) 上。
  当 receive_task 接收到消息后，会通过信号量 buffer_sem 通知 process_task 开始处理消息。这种机制确保了两个任务不会同时访问消息缓冲区，避免了数据竞争。

安全的消息发送

1. 封装发送逻辑：将发送逻辑封装到一个函数send_message(...)中，便于维护和复用。
2. 错误处理：添加错误处理机制，确保在发送失败时能够适当处理。
3. 线程安全：使用互斥量确保发送操作的线程安全。

互斥量与信号量

xSemaphoreCreateBinary() 和 xSemaphoreCreateMutex() 都是FreeRTOS提供的用于任务同步的API函数，但它们在功能和使用场景上有所不同。

xSemaphoreCreateBinary()

• 作用：创建一个二进制信号量（Binary Semaphore）。
• 特点：
  • 初始状态为"空"（不可用）。
  • 可以用于任务之间的同步，也可以用于任务和中断之间的同步。
  • 典型的使用场景包括事件通知（例如一个任务等待另一个任务或中断通知它某个事件发生）。
• 用法：适用于需要简单信号传递的情况，例如从中断服务程序（ISR）通知任务某个事件已经发生。

xSemaphoreCreateMutex()

• 作用：创建一个互斥量（Mutex）。

• 特点：

  • 初始状态为"满"（可用）。
  • 设计用于保护共享资源，确保一次只有一个任务可以访问该资源。
  • 具备优先级继承机制，以防止优先级反转。

• 用法：适用于保护共享资源，防止数据竞争。
  在中断中使用
  如果需要在中断中使用信号量进行同步，FreeRTOS提供了专门的API函数以确保线程安全和中断安全。通常使用二进制信号量或计数信号量。不能直接使用 xSemaphoreGive 等标准API，而是使用对应的 ISR 安全版本。

• 适合中断的API：

  1. 创建信号量：使用 xSemaphoreCreateBinary() 或 xSemaphoreCreateCounting()。
  2. 在中断服务程序中给信号量：
     • xSemaphoreGiveFromISR()
     • xSemaphoreGiveFromISR() 适用于二进制信号量和计数信号量。
  3. 在任务中等待信号量：使用 xSemaphoreTake()。
     示例代码

#include "FreeRTOS.h"
#include "task.h"
#include "semphr.h"// 全局信号量句柄
SemaphoreHandle_t xBinarySemaphore;// 中断服务程序（ISR）
void vAnISR(void)
{BaseType_t xHigherPriorityTaskWoken = pdFALSE;// 在中断中给信号量xSemaphoreGiveFromISR(xBinarySemaphore, &xHigherPriorityTaskWoken);// 如果xHigherPriorityTaskWoken为pdTRUE，说明有更高优先级的任务被唤醒，进行任务切换portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
}// 一个任务，用于等待信号量
void vTaskFunction(void *pvParameters)
{while (1){// 等待信号量if (xSemaphoreTake(xBinarySemaphore, portMAX_DELAY) == pdTRUE){// 处理从中断通知的事件}}
}int main(void)
{// 创建二进制信号量xBinarySemaphore = xSemaphoreCreateBinary();// 检查信号量创建是否成功if (xBinarySemaphore != NULL){// 创建任务xTaskCreate(vTaskFunction, "Task", configMINIMAL_STACK_SIZE, NULL, tskIDLE_PRIORITY + 1, NULL);// 启动调度器vTaskStartScheduler();}// 如果信号量创建失败，处理失败情况for (;;);
}

关键点

1. 创建二进制信号量：使用 xSemaphoreCreateBinary()。
2. 中断服务程序（ISR）：
   • 使用 xSemaphoreGiveFromISR() 给信号量。
   • 检查 xHigherPriorityTaskWoken 并调用 portYIELD_FROM_ISR() 进行任务切换。
3. 任务中等待信号量：使用 xSemaphoreTake()。

发送接口

rpmsg_lite_send 接口

rpmsg_lite_send 是一个标准的发送接口，它会将用户提供的消息拷贝到内部缓冲区，然后发送出去。这个接口的优点是简单易用，不需要用户管理发送缓冲区的内存，但可能会有额外的内存拷贝开销。
用法示例

#include "rpmsg_lite.h"#define SHMEM_BASE         (void*)0x3F800000
#define RPMSG_LITE_LINK_ID RL_PLATFORM_HIGHEST_LINK_ID
#define RL_NO_FLAGS        0#define LOCAL_EPT_ADDR     30
#define REMOTE_EPT_ADDR    40
#define MSG_SIZE           128int main(void)
{struct rpmsg_lite_instance *rpmsg_instance;struct rpmsg_lite_ept_static_context ept_context;struct rpmsg_lite_endpoint *my_ept;char msg[MSG_SIZE];// 初始化RPMsg Lite实例rpmsg_instance = rpmsg_lite_master_init(SHMEM_BASE, RPMSG_LITE_LINK_ID, RL_NO_FLAGS);// 创建本地端点my_ept = rpmsg_lite_create_ept(rpmsg_instance, LOCAL_EPT_ADDR, NULL, NULL, &ept_context);// 初始化消息内容memset(msg, 0, MSG_SIZE);strcpy(msg, "Hello, remote!");// 发送消息到远程端int status = rpmsg_lite_send(rpmsg_instance, my_ept, REMOTE_EPT_ADDR, msg, strlen(msg) + 1, RL_BLOCK);// 检查发送状态if (status != RL_SUCCESS){// 处理发送失败return -1;}// 清理资源rpmsg_lite_destroy_ept(rpmsg_instance, my_ept);rpmsg_lite_deinit(rpmsg_instance);return 0;
}

rpmsg_lite_send_nocopy 接口

rpmsg_lite_send_nocopy 是一种无拷贝的发送接口。用户需要首先申请发送缓冲区的内存，填充消息内容，然后通过该接口发送消息。发送完成后，用户需要手动释放发送缓冲区的内存。这个接口的优点是避免了内存拷贝，提高了效率，但需要用户管理发送缓冲区的内存。

用法示例

#include "rpmsg_lite.h"#define SHMEM_BASE         (void*)0x3F800000
#define RPMSG_LITE_LINK_ID RL_PLATFORM_HIGHEST_LINK_ID
#define RL_NO_FLAGS        0#define LOCAL_EPT_ADDR     30
#define REMOTE_EPT_ADDR    40
#define MSG_SIZE           128int main(void)
{struct rpmsg_lite_instance *rpmsg_instance;struct rpmsg_lite_ept_static_context ept_context;struct rpmsg_lite_endpoint *my_ept;void *msg;// 初始化RPMsg Lite实例rpmsg_instance = rpmsg_lite_master_init(SHMEM_BASE, RPMSG_LITE_LINK_ID, RL_NO_FLAGS);// 创建本地端点my_ept = rpmsg_lite_create_ept(rpmsg_instance, LOCAL_EPT_ADDR, NULL, NULL, &ept_context);// 申请发送内存msg = rpmsg_lite_alloc_tx_buffer(rpmsg_instance, &msg, RL_BLOCK);// 检查内存分配是否成功if (msg == NULL){// 处理内存分配失败return -1;}// 初始化发送内存（例如，将消息内容填充为"Hello, remote!"）memset(msg, 0, MSG_SIZE);strcpy((char*)msg, "Hello, remote!");// 发送消息到远程端int status = rpmsg_lite_send_nocopy(rpmsg_instance, my_ept, REMOTE_EPT_ADDR, msg, strlen((char*)msg) + 1);// 检查发送状态if (status != RL_SUCCESS){// 处理发送失败return -1;}// 手动释放发送内存rpmsg_lite_release_rx_buffer(rpmsg_instance, msg);// 清理资源rpmsg_lite_destroy_ept(rpmsg_instance, my_ept);rpmsg_lite_deinit(rpmsg_instance);return 0;
}

rpmsg_lite_send与rpmsg_lite_send_nocopy关键区别

1. 内存管理：

   • rpmsg_lite_send：消息内存由用户管理，函数内部会进行一次内存拷贝。
   • rpmsg_lite_send_nocopy：消息内存需要用户申请和释放，函数内部不会进行内存拷贝。

2. 效率：

   • rpmsg_lite_send：由于存在内存拷贝，可能略低于rpmsg_lite_send_nocopy。
   • rpmsg_lite_send_nocopy：无内存拷贝，效率较高，但需要用户手动管理内存。

3. 用法：

   • rpmsg_lite_send：适用于简单的消息发送，不需要特别管理发送内存。
   • rpmsg_lite_send_nocopy：适用于性能要求较高的场景，需要用户管理发送内存。
     通过这些示例，可以更好地理解这两个接口的使用场景和方法，选择最适合自己应用需求的接口。

判断远端的RPMsg是否已经读取

为了判断远端是否已经读取消息，可以使用回调函数或同步机制。下面是一个示例，展示了如何申请发送内存、使用rpmsg_lite_send_nocopy发送消息，并在消息被远端处理后释放内存。

示例代码

#include "rpmsg_lite.h"
#include "rpmsg_ns.h"#define SHMEM_BASE         (void*)0x3F800000
#define RPMSG_LITE_LINK_ID RL_PLATFORM_HIGHEST_LINK_ID
#define RL_NO_FLAGS        0#define LOCAL_EPT_ADDR     40
#define REMOTE_EPT_ADDR    30
#define MSG_SIZE           128static struct rpmsg_lite_instance *rpmsg_instance;
static struct rpmsg_lite_endpoint *my_ept;
static struct rpmsg_lite_ept_static_context ept_context;// 回调函数，当消息从远端接收到时被调用
void rpmsg_read_callback(void *payload, uint32_t payload_len, uint32_t src, void *priv)
{// 处理接收到的消息// 这里我们假设只打印消息内容printf("Received message from %d: %s\n", src, (char *)payload);// 根据需求处理消息，例如发送响应消息// 例如，回传一个确认消息void *tx_msg;tx_msg = rpmsg_lite_alloc_tx_buffer(rpmsg_instance, &tx_msg, RL_BLOCK);if (tx_msg != NULL){strcpy(tx_msg, "Ack: Message received");rpmsg_lite_send_nocopy(rpmsg_instance, my_ept, REMOTE_EPT_ADDR, tx_msg, strlen(tx_msg) + 1);}// 释放接收缓冲区rpmsg_lite_release_rx_buffer(rpmsg_instance, payload);
}int main(void)
{// 初始化RPMsg Lite实例rpmsg_instance = rpmsg_lite_remote_init(SHMEM_BASE, RPMSG_LITE_LINK_ID, RL_NO_FLAGS);// 创建本地端点my_ept = rpmsg_lite_create_ept(rpmsg_instance, LOCAL_EPT_ADDR, rpmsg_read_callback, NULL, &ept_context);// 远程端主要任务是等待并处理接收到的消息，因此可以进入一个循环while (1){// 其他处理逻辑或低功耗处理可以放在这里// 例如：// __WFI(); // 等待中断}// 清理资源rpmsg_lite_destroy_ept(rpmsg_instance, my_ept);rpmsg_lite_deinit(rpmsg_instance);return 0;
}

说明

1. 回调函数： 定义一个回调函数rpmsg_read_callback，当从远端接收到消息时会调用该函数。在回调函数中设置一个标志message_received表示消息已被远端处理。
2. 创建端点： 使用rpmsg_lite_create_ept创建端点时，将回调函数作为参数传入。
3. 发送消息： 使用rpmsg_lite_send_nocopy发送消息。
4. 等待消息处理： 使用一个循环等待标志message_received变为true，表示消息已经被远端处理。在实际应用中，可以添加超时处理以防止无限等待。
5. 释放内存和清理资源： 在确认消息已被处理后，手动释放发送内存，并清理其他资源。
   通过上述步骤，可以在RTOS中使用rpmsg_lite_send_nocopy发送消息，并确保在消息被远端处理后正确释放内存。

接收接口

rpmsg_queue_recv 接口

用法
rpmsg_queue_recv 用于从消息队列中接收消息，并将接收到的消息复制到用户提供的缓冲区中。其基本用法如下：

int rpmsg_queue_recv(struct rpmsg_queue *queue,unsigned long *src,void *data,int *len,unsigned long timeout);

• queue：指向消息队列的指针。
• src：接收消息的源端点地址。
• data：用户提供的缓冲区，用于存储接收到的消息。
• len：指向一个整数的指针，用于存储接收到的消息长度。
• timeout：等待消息的超时时间，单位为毫秒。若设置为RL_BLOCK，则表示一直等待直到接收到消息。

高级用法
rpmsg_queue_recv 可以与其他RPMsg接口组合使用，以实现复杂的消息传递机制。例如，可以在一个任务中使用rpmsg_queue_recv不断接收消息，并根据消息的类型或内容进行不同的处理。

注意事项

• 使用rpmsg_queue_recv时需要确保提供的缓冲区足够大，以存储接收到的消息，否则可能导致数据丢失或缓冲区溢出。
• 超时时间的设置需要根据应用需求进行合理配置，以避免任务长时间阻塞。

rpmsg_queue_recv_nocopy 接口

用法
rpmsg_queue_recv_nocopy 用于从消息队列中接收消息，并且不将消息复制到用户提供的缓冲区中，而是直接返回一个指向消息的指针。其基本用法如下：

int rpmsg_queue_recv_nocopy(struct rpmsg_queue *queue,unsigned long *src,void **data,int *len,unsigned long timeout);

• queue：指向消息队列的指针。
• src：接收消息的源端点地址。
• data：指向一个指针的指针，用于返回消息的指针。
• len：指向一个整数的指针，用于存储接收到的消息长度。
• timeout：等待消息的超时时间，单位为毫秒。若设置为RL_BLOCK，则表示一直等待直到接收到消息。

高级用法
rpmsg_queue_recv_nocopy 常用于高性能或低延迟的应用场景，因为它避免了消息复制的开销。可以在接收消息后直接处理消息数据，而不需要额外的内存拷贝操作。

注意事项

• 使用rpmsg_queue_recv_nocopy 接口时，需要确保在处理完消息后调用rpmsg_queue_nocopy_free 释放消息所占用的内存，以避免内存泄漏。
• 由于消息未被复制，处理消息的代码需要确保在消息处理完之前，消息数据不会被其他任务修改或释放。

rpmsg_queue_recv与rpmsg_queue_recv_nocopy关键区别

• 消息复制：rpmsg_queue_recv 将消息复制到用户提供的缓冲区，而rpmsg_queue_recv_nocopy 则直接返回指向消息的指针。
• 性能：rpmsg_queue_recv_nocopy 避免了消息复制的开销，适用于高性能或低延迟的场景。
• 内存管理：使用rpmsg_queue_recv_nocopy 需要额外处理消息内存的释放，而rpmsg_queue_recv 不需要。
• 复杂性：rpmsg_queue_recv_nocopy 的使用复杂度稍高，需要用户更加小心地管理消息内存。

示例代码

以下是一个使用rpmsg_queue_recv 和rpmsg_queue_recv_nocopy 的示例代码：

void rpmsg_recv_example(struct rpmsg_queue *queue)
{unsigned long src;int len;unsigned char buffer[256];unsigned char *msg_ptr;// 使用 rpmsg_queue_recv 接收消息int ret = rpmsg_queue_recv(queue, &src, buffer, &len, RL_BLOCK);if (ret == RL_SUCCESS) {printf("Received message: %s\n", buffer);}// 使用 rpmsg_queue_recv_nocopy 接收消息ret = rpmsg_queue_recv_nocopy(queue, &src, (void **)&msg_ptr, &len, RL_BLOCK);if (ret == RL_SUCCESS) {printf("Received message (no copy): %s\n", msg_ptr);// 处理完消息后释放内存rpmsg_queue_nocopy_free(queue, msg_ptr);}
}