Add support for cancelling enqueued inferences
If an enqueued inference is cancelled it is simply removed from the
queue. Both a cancel inference response with status done and a
inference response with status cancelled are sent back.
Also override the new operators to call in the FreeRTOS allocator
instead of malloc/free.
Change-Id: I243e678aa6b996084c9b9be1d1b00ffcecc75bc9
diff --git a/applications/message_handler/main.cpp b/applications/message_handler/main.cpp
index dde5dc5..b527840 100644
--- a/applications/message_handler/main.cpp
+++ b/applications/message_handler/main.cpp
@@ -71,10 +71,6 @@
namespace {
-SemaphoreHandle_t messageNotify;
-QueueHandle_t inferenceInputQueue;
-QueueHandle_t inferenceOutputQueue;
-
// Mailbox driver
#ifdef MHU_V2
Mailbox::MHUv2 mailbox(MHU_TX_BASE_ADDRESS, MHU_RX_BASE_ADDRESS); // txBase, rxBase
@@ -87,6 +83,26 @@
} // namespace
/****************************************************************************
+ * Override new operators to call in FreeRTOS allocator
+ ****************************************************************************/
+
+void *operator new(size_t size) {
+ return pvPortMalloc(size);
+}
+
+void *operator new[](size_t size) {
+ return pvPortMalloc(size);
+}
+
+void operator delete(void *ptr) {
+ vPortFree(ptr);
+}
+
+void operator delete[](void *ptr) {
+ vPortFree(ptr);
+}
+
+/****************************************************************************
* Mutex & Semaphore
****************************************************************************/
@@ -150,6 +166,24 @@
* Application
****************************************************************************/
+struct TaskParams {
+ TaskParams() :
+ messageNotify(xSemaphoreCreateBinary()),
+ inferenceInputQueue(std::make_shared<Queue<ethosu_core_inference_req>>()),
+ inferenceOutputQueue(xQueueCreate(10, sizeof(ethosu_core_inference_rsp))) {}
+
+ SemaphoreHandle_t messageNotify;
+ // Used to pass inference requests to the inference runner task
+ std::shared_ptr<Queue<ethosu_core_inference_req>> inferenceInputQueue;
+ // Queue for message responses to the remote host
+ QueueHandle_t inferenceOutputQueue;
+};
+
+struct InferenceTaskParams {
+ TaskParams *taskParams;
+ uint8_t *arena;
+};
+
namespace {
#ifdef MHU_IRQ
@@ -160,21 +194,27 @@
void inferenceTask(void *pvParameters) {
printf("Starting inference task\n");
+ InferenceTaskParams *params = reinterpret_cast<InferenceTaskParams *>(pvParameters);
- uint8_t *arena = reinterpret_cast<uint8_t *>(pvParameters);
- InferenceHandler process(arena, arenaSize, inferenceInputQueue, inferenceOutputQueue, messageNotify);
+ InferenceHandler process(params->arena,
+ arenaSize,
+ params->taskParams->inferenceInputQueue,
+ params->taskParams->inferenceOutputQueue,
+ params->taskParams->messageNotify);
+
process.run();
}
-void messageTask(void *) {
- printf("Starting input message task\n");
+void messageTask(void *pvParameters) {
+ printf("Starting message task\n");
+ TaskParams *params = reinterpret_cast<TaskParams *>(pvParameters);
IncomingMessageHandler process(*inputMessageQueue.toQueue(),
*outputMessageQueue.toQueue(),
mailbox,
- inferenceInputQueue,
- inferenceOutputQueue,
- messageNotify);
+ params->inferenceInputQueue,
+ params->inferenceOutputQueue,
+ params->messageNotify);
#ifdef MHU_IRQ
// Register mailbox interrupt handler
@@ -196,21 +236,22 @@
return 1;
}
- // Create message queues for inter process communication
- messageNotify = xSemaphoreCreateBinary();
- inferenceInputQueue = xQueueCreate(10, sizeof(ethosu_core_inference_req));
- inferenceOutputQueue = xQueueCreate(10, sizeof(ethosu_core_inference_rsp));
+ TaskParams taskParams;
- // Task for handling incoming messages from the remote host
- ret = xTaskCreate(messageTask, "messageTask", 1024, nullptr, 2, nullptr);
+ // Task for handling incoming /outgoing messages from the remote host
+ ret = xTaskCreate(messageTask, "messageTask", 1024, &taskParams, 2, nullptr);
if (ret != pdPASS) {
printf("Failed to create 'messageTask'\n");
return ret;
}
+ InferenceTaskParams infParams[NUM_PARALLEL_TASKS];
+
// One inference task for each NPU
for (size_t n = 0; n < NUM_PARALLEL_TASKS; n++) {
- ret = xTaskCreate(inferenceTask, "inferenceTask", 8 * 1024, &tensorArena[n], 3, nullptr);
+ infParams[n].taskParams = &taskParams;
+ infParams[n].arena = reinterpret_cast<uint8_t *>(&tensorArena[n]);
+ ret = xTaskCreate(inferenceTask, "inferenceTask", 8 * 1024, &infParams[n], 3, nullptr);
if (ret != pdPASS) {
printf("Failed to create 'inferenceTask%d'\n", n);
return ret;