samples: tflite-micro: micro_speech: add OpenAMP transport (Linux/Zephyr)

Thong Phan · Thong Phan · commit 74ce0bd79135 · 2025-10-13T21:20:15.000+02:00
This commit adds OpenAMP Remote Processor Messaging (RPMsg) support
to enable communication between Linux userspace and the Zephyr
micro_speech application running on the remote core.

Implementation details:
- Endpoint name: "audio_pcm"
- Buffer flow: hold/release mechanism for efficient memory management
- Queue depth: 16 messages with 4-byte alignment (K_MSGQ_DEFINE)
- Notify path: mailbox/IPM for inter-processor interrupts
- Linux userspace maps to this endpoint via /dev/ttyRPMSG* device

Signed-off-by: Thong Phan &lt;quang.thong2001@gmail.com&gt;
diff --git a/samples/modules/tflite-micro/micro_speech/CMakeLists.txt b/samples/modules/tflite-micro/micro_speech/CMakeLists.txt
@@ -17,6 +17,7 @@ file(
         app_sources
         src/*
         src/inference/*
+        src/transport/*
 )
 
 target_include_directories(
diff --git a/samples/modules/tflite-micro/micro_speech/src/inference/model_runner.cpp b/samples/modules/tflite-micro/micro_speech/src/inference/model_runner.cpp
@@ -29,6 +29,8 @@ LOG_MODULE_REGISTER(model_runner);
 #include "audio_preprocessor_int8_model.hpp"
 #include "micro_speech_quantized_model.hpp"
 
+#include "transport/rpmsg_transport.h"
+
 #include <zephyr/kernel.h>
 #include <algorithm>
 #include <stdio.h>
diff --git a/samples/modules/tflite-micro/micro_speech/src/main_functions.cpp b/samples/modules/tflite-micro/micro_speech/src/main_functions.cpp
@@ -22,6 +22,7 @@
 LOG_MODULE_REGISTER(micro_speech_openamp);
 
 #include "inference/model_runner.hpp"
+#include "transport/rpmsg_transport.h"
 
 #include <zephyr/kernel.h>
 #include <zephyr/device.h>
@@ -40,20 +41,93 @@ LOG_MODULE_REGISTER(micro_speech_openamp);
 #define BUFFER_SIZE_SAMPLES 				(SAMPLES_PER_SECOND)
 #define BUFFER_SIZE_BYTES 				(BUFFER_SIZE_SAMPLES * SAMPLE_SIZE_BYTES)
 /* Double buffering for receiving and processing */
+static int16_t buffer_a[BUFFER_SIZE_SAMPLES];
 static int16_t buffer_b[BUFFER_SIZE_SAMPLES];
+static int16_t *write_buffer = buffer_a;
 static int16_t *processing_buffer = buffer_b;
 /* Threading & Synchronization */
 K_THREAD_STACK_DEFINE(thread_receive_stack, APP_RECEIVE_TASK_STACK_SIZE);
 K_THREAD_STACK_DEFINE(thread_audio_processing_stack, APP_AUDIO_PROCESSING_TASK_STACK_SIZE);
+static struct k_thread thread_receive_data;
 static struct k_thread thread_audio_processing_data;
 static K_SEM_DEFINE(processing_buffer_ready_sem, 0, 1);
 static K_SEM_DEFINE(ml_complete_sem, 1, 1); /* Starts at 1 to allow the first file */
 static K_MUTEX_DEFINE(buffer_access_mutex);
 /* Shared State (protected by buffer_access_mutex) */
+static uint16_t g_samples_in_write_buffer = 0;
 static uint16_t g_samples_in_processing_buffer = 0;
+static bool g_eof_received = false;
 
 extern "C" {
 
+void app_receive_data_thread(void *arg1, void *arg2, void *arg3)
+{
+	ARG_UNUSED(arg1);
+	ARG_UNUSED(arg2);
+	ARG_UNUSED(arg3);
+
+	int ret;
+
+	LOG_INF("Receiving data thread started");
+
+	k_sem_take(&data_tty_ready_sem, K_FOREVER);
+	ret = rpmsg_create_ept(&tty_ept, rpdev, "audio_pcm", RPMSG_ADDR_ANY, RPMSG_ADDR_ANY, rpmsg_recv_tty_callback, NULL);
+	if (ret != 0) {
+		LOG_ERR("Could not create RPMsg endpoint");
+		return;
+	}
+
+	while (1) {
+                /* Wait for previous buffer to be fully processed */
+		k_sem_take(&ml_complete_sem, K_FOREVER);
+
+		while(1) {
+			struct rpmsg_rcv_msg rx_msg;
+                        /* Wait for messages */
+			k_msgq_get(&tty_msgq, &rx_msg, K_FOREVER);
+
+			bool should_break = false;
+
+			if (rx_msg.len == 3 && memcmp(rx_msg.data, "EOF", 3) == 0) { /* EOF */
+				k_mutex_lock(&buffer_access_mutex, K_FOREVER);
+				g_eof_received = true;
+                                /* Swap and process any leftover data */
+				if (g_samples_in_write_buffer > 0) {
+					int16_t *temp = write_buffer;
+					write_buffer = processing_buffer;
+					processing_buffer = temp;
+					g_samples_in_processing_buffer = g_samples_in_write_buffer;
+					g_samples_in_write_buffer = 0;
+					k_sem_give(&processing_buffer_ready_sem);
+				} else { /* Still signal to trigger final processing if needed */
+					k_sem_give(&processing_buffer_ready_sem);
+				}
+				k_mutex_unlock(&buffer_access_mutex);
+				should_break = true;
+			} else { /* Receive data */
+				k_mutex_lock(&buffer_access_mutex, K_FOREVER);
+				size_t bytes_to_copy = MIN(rx_msg.len, (BUFFER_SIZE_SAMPLES - g_samples_in_write_buffer) * SAMPLE_SIZE_BYTES);
+				memcpy((uint8_t*)write_buffer + g_samples_in_write_buffer * SAMPLE_SIZE_BYTES, rx_msg.data, bytes_to_copy);
+				g_samples_in_write_buffer += bytes_to_copy / SAMPLE_SIZE_BYTES;
+                                /* Buffer is full. Swap buffer */
+				if (g_samples_in_write_buffer >= BUFFER_SIZE_SAMPLES) {
+					int16_t *temp = write_buffer;
+					write_buffer = processing_buffer;
+					processing_buffer = temp;
+					g_samples_in_processing_buffer = g_samples_in_write_buffer;
+					g_samples_in_write_buffer = 0;
+					k_sem_give(&processing_buffer_ready_sem);
+				}
+				k_mutex_unlock(&buffer_access_mutex);
+			}
+			rpmsg_release_rx_buffer(&tty_ept, rx_msg.data);
+			if (should_break) {
+				break; /* Wait for ml_complete_sem */
+			}
+		}
+	}
+}
+
 void app_audio_processing_thread(void *arg1, void *arg2, void *arg3)
 {
 	ARG_UNUSED(arg1);
@@ -68,6 +142,7 @@ void app_audio_processing_thread(void *arg1, void *arg2, void *arg3)
 
 		k_mutex_lock(&buffer_access_mutex, K_FOREVER);
 		size_t samples_to_process = g_samples_in_processing_buffer;
+		bool is_eof = g_eof_received;
 		g_samples_in_processing_buffer = 0; /* Mark buffer as claimed */
 		k_mutex_unlock(&buffer_access_mutex);
 
@@ -81,6 +156,15 @@ void app_audio_processing_thread(void *arg1, void *arg2, void *arg3)
 			/* Clear the buffer after processing */
 			memset(processing_buffer, 0, BUFFER_SIZE_BYTES);
 		}
+
+		if (is_eof) { /* EOF processing */
+			k_mutex_lock(&buffer_access_mutex, K_FOREVER);
+			g_eof_received = false;
+			k_mutex_unlock(&buffer_access_mutex);
+
+			LOG_DBG("Buffer processing complete. Ready for next buffer.");
+			k_sem_give(&ml_complete_sem);
+		}
 		k_sem_give(&ml_complete_sem);
 	}
 }
@@ -92,6 +176,13 @@ void setup() {
         /* Initialize model runner */
         model_runner_init();
 
+        /* Initialize RPMsg transport */
+        rpmsg_transport_start();
+
+        /* Create data receiving thread (higher priority) */
+        k_thread_create(&thread_receive_data, thread_receive_stack, APP_RECEIVE_TASK_STACK_SIZE,
+                        app_receive_data_thread, NULL, NULL, NULL, 5, 0, K_NO_WAIT);
+
         /* Create audio processing thread (lower priority) */
         k_thread_create(&thread_audio_processing_data, thread_audio_processing_stack, APP_AUDIO_PROCESSING_TASK_STACK_SIZE,
                         app_audio_processing_thread, NULL, NULL, NULL, 6, 0, K_NO_WAIT);
diff --git a/samples/modules/tflite-micro/micro_speech/src/transport/rpmsg_transport.c b/samples/modules/tflite-micro/micro_speech/src/transport/rpmsg_transport.c
@@ -0,0 +1,229 @@
+/*
+ * Copyright (c) 2020, STMICROELECTRONICS
+ * Author: Thong Phan <quang.thong2001@gmail.com>
+ *
+ * Heavily based on pwm_mcux_ftm.c, which is:
+ * Copyright (c) 2017, NXP
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "rpmsg_transport.h"
+
+#include <zephyr/logging/log.h>
+LOG_MODULE_REGISTER(rpmsg_transport);
+#include <zephyr/device.h>
+#include <zephyr/drivers/ipm.h>
+
+#include <metal/sys.h>
+#include <metal/io.h>
+#include <resource_table.h>
+#include <addr_translation.h>
+
+#if !DT_HAS_CHOSEN(zephyr_ipc_shm)
+#error "Sample requires definition of shared memory for rpmsg"
+#endif
+
+/* Constants from device tree */
+#define SHM_NODE            DT_CHOSEN(zephyr_ipc_shm)
+#define SHM_START_ADDR      DT_REG_ADDR(SHM_NODE)
+#define SHM_SIZE            DT_REG_SIZE(SHM_NODE)
+/* Stack size for the management thread */
+#define APP_TASK_STACK_SIZE (1024)
+
+#if CONFIG_IPM_MAX_DATA_SIZE > 0
+#define IPM_SEND(dev, w, id, d, s) ipm_send(dev, w, id, d, s)
+#else
+#define IPM_SEND(dev, w, id, d, s) ipm_send(dev, w, id, NULL, 0)
+#endif
+/* Thread definitions */
+K_THREAD_STACK_DEFINE(thread_mng_stack, APP_TASK_STACK_SIZE);
+static struct k_thread thread_mng_data;
+/* OpenAMP components */
+static const struct device *const ipm_handle = DEVICE_DT_GET(DT_CHOSEN(zephyr_ipc));
+static metal_phys_addr_t shm_physmap = SHM_START_ADDR;
+static metal_phys_addr_t rsc_tab_physmap;
+static struct metal_io_region shm_io_data;
+static struct metal_io_region rsc_io_data;
+static struct metal_io_region *shm_io = &shm_io_data;
+static struct metal_io_region *rsc_io = &rsc_io_data;
+static struct rpmsg_virtio_device rvdev;
+static void *rsc_table;
+/* Public variables */
+struct rpmsg_device *rpdev;
+struct rpmsg_endpoint tty_ept;
+K_MSGQ_DEFINE(tty_msgq, sizeof(struct rpmsg_rcv_msg), 16, 4);
+/* 16 messages * 20ms per frames covers 320ms of data */
+K_SEM_DEFINE(data_tty_ready_sem, 0, 1);
+/* Internal semaphore, hidden from other files */
+static K_SEM_DEFINE(data_sem, 0, 1);
+
+static void platform_ipm_callback(const struct device *dev, void *context, uint32_t id,
+				  volatile void *data)
+
+{
+	LOG_DBG("%s: msg received from mb %d", __func__, id);
+	k_sem_give(&data_sem);
+}
+
+int rpmsg_recv_tty_callback(struct rpmsg_endpoint *ept, void *data, size_t len, uint32_t src,
+			    void *priv)
+{
+	struct rpmsg_rcv_msg msg = {.data = data, .len = len};
+
+	rpmsg_hold_rx_buffer(ept, data);
+
+	if (k_msgq_put(&tty_msgq, &msg, K_NO_WAIT) != 0) {
+		LOG_WRN("tty_msgq full, dropping frame len %zu\n", len);
+		rpmsg_release_rx_buffer(ept, data);
+	}
+	return RPMSG_SUCCESS;
+}
+
+static void receive_message(unsigned char **msg, unsigned int *len)
+{
+	if (k_sem_take(&data_sem, K_FOREVER) == 0) {
+		rproc_virtio_notified(rvdev.vdev, VRING1_ID);
+	}
+}
+
+static void new_service_cb(struct rpmsg_device *rdev, const char *name, uint32_t src)
+{
+	LOG_ERR("%s: unexpected ns service receive for name %s", __func__, name);
+}
+
+static int mailbox_notify(void *priv, uint32_t id)
+{
+	ARG_UNUSED(priv);
+	LOG_DBG("%s: msg received", __func__);
+	IPM_SEND(ipm_handle, 0, id, &id, 4);
+	return 0;
+}
+
+static int platform_init(void)
+{
+	int rsc_size;
+	struct metal_init_params metal_params = METAL_INIT_DEFAULTS;
+	int status;
+
+	status = metal_init(&metal_params);
+	if (status) {
+		LOG_ERR("metal_init: failed: %d", status);
+		return -1;
+	}
+
+	metal_io_init(shm_io, (void *)SHM_START_ADDR, &shm_physmap, SHM_SIZE, -1, 0,
+		      addr_translation_get_ops(shm_physmap));
+
+	rsc_table_get(&rsc_table, &rsc_size);
+	rsc_tab_physmap = (uintptr_t)rsc_table;
+	metal_io_init(rsc_io, rsc_table, &rsc_tab_physmap, rsc_size, -1, 0, NULL);
+
+	if (!device_is_ready(ipm_handle)) {
+		LOG_ERR("IPM device is not ready");
+		return -1;
+	}
+
+	ipm_register_callback(ipm_handle, platform_ipm_callback, NULL);
+
+	status = ipm_set_enabled(ipm_handle, 1);
+	if (status) {
+		LOG_ERR("ipm_set_enabled failed");
+		return -1;
+	}
+
+	return 0;
+}
+
+static void cleanup_system(void)
+{
+	ipm_set_enabled(ipm_handle, 0);
+	rpmsg_deinit_vdev(&rvdev);
+	metal_finish();
+}
+
+static struct rpmsg_device *platform_create_rpmsg_vdev(rpmsg_ns_bind_cb ns_cb)
+{
+	struct fw_rsc_vdev_vring *vring_rsc;
+	struct virtio_device *vdev;
+	int ret;
+
+	vdev = rproc_virtio_create_vdev(VIRTIO_DEV_DEVICE, VDEV_ID, rsc_table_to_vdev(rsc_table),
+					rsc_io, NULL, mailbox_notify, NULL);
+	if (!vdev) {
+		LOG_ERR("failed to create vdev");
+		return NULL;
+	}
+
+	rproc_virtio_wait_remote_ready(vdev);
+
+	vring_rsc = rsc_table_get_vring0(rsc_table);
+	ret = rproc_virtio_init_vring(vdev, 0, vring_rsc->notifyid, (void *)vring_rsc->da, rsc_io,
+				      vring_rsc->num, vring_rsc->align);
+	if (ret) {
+		LOG_ERR("failed to init vring 0");
+		goto failed;
+	}
+
+	vring_rsc = rsc_table_get_vring1(rsc_table);
+	ret = rproc_virtio_init_vring(vdev, 1, vring_rsc->notifyid, (void *)vring_rsc->da, rsc_io,
+				      vring_rsc->num, vring_rsc->align);
+	if (ret) {
+		LOG_ERR("failed to init vring 1");
+		goto failed;
+	}
+
+	ret = rpmsg_init_vdev(&rvdev, vdev, ns_cb, shm_io, NULL);
+	if (ret) {
+		LOG_ERR("failed rpmsg_init_vdev");
+		goto failed;
+	}
+
+	return rpmsg_virtio_get_rpmsg_device(&rvdev);
+
+failed:
+	rproc_virtio_remove_vdev(vdev);
+	return NULL;
+}
+
+static void rpmsg_mng_task(void *arg1, void *arg2, void *arg3)
+{
+	ARG_UNUSED(arg1);
+	ARG_UNUSED(arg2);
+	ARG_UNUSED(arg3);
+
+	unsigned char *msg;
+	unsigned int len;
+	int ret = 0;
+
+	LOG_INF("Microspeech sample with OpenAMP started");
+
+	ret = platform_init();
+	if (ret) {
+		LOG_ERR("Failed to initialize platform");
+		goto task_end;
+	}
+
+	rpdev = platform_create_rpmsg_vdev(new_service_cb);
+	if (!rpdev) {
+		LOG_ERR("Failed to create rpmsg virtio device");
+		goto task_end;
+	}
+
+	k_sem_give(&data_tty_ready_sem);
+
+	while (1) {
+		receive_message(&msg, &len);
+	}
+
+task_end:
+	cleanup_system();
+	LOG_INF("Microspeech sample with OpenAMP ended");
+}
+
+/* Public Function */
+void rpmsg_transport_start(void)
+{
+	k_thread_create(&thread_mng_data, thread_mng_stack, APP_TASK_STACK_SIZE, rpmsg_mng_task,
+			NULL, NULL, NULL, K_PRIO_COOP(8), 0, K_NO_WAIT);
+}
diff --git a/samples/modules/tflite-micro/micro_speech/src/transport/rpmsg_transport.h b/samples/modules/tflite-micro/micro_speech/src/transport/rpmsg_transport.h

Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,7 @@ file(`
`17`	`17`	`app_sources`
`18`	`18`	`src/*`
`19`	`19`	`src/inference/*`
	`20`	`+ src/transport/*`
`20`	`21`	`)`
`21`	`22`
`22`	`23`	`target_include_directories(`