chore: Moved unpacking of NIL/NTL headers for consistent responsibilities

app/src/incphub/README.md

@@ -17,3 +17,19 @@ Set as a macro in compile flags if wishing to override the defaults
 ## Setup
 
 - incphub_task_preinit MUST be run before any calls to the client, likely even before the scheduler is started
+
+## IPC flow
+
+Ingest (accepting of messages into incphub) is done via queue (many to one)
+
+Dispatch (internal routing to local task) is done via direct-to-task notifications (see INCPHUB_TASKINDEX_MAILBOX) by passing the message buffer pointer
+
+## Buffer ownership
+
+Buffers are allocated with incphub_get_msg_buffer, at which point they belong to the requesting task.
+
+When passed to incphub_send_XXX functions, they become owned by incphub
+
+When dispatched internally, the receiving task becomes the owner and must invoke incphub_release_msg_buffer when done
+
+When routed externally, after calling the transmitter function, incphub releases the buffer

app/src/incphub/include/incphub/incphub.h

@@ -20,6 +20,8 @@
 #define INCPHUB_EAGAIN (5)
 /// @brief Could not allocate memory
 #define INCPHUB_ENOMEM (6)
+/// @brief Packet failed integrity checks
+#define INCPHUB_BAD_PACKET (7)
 
 typedef struct {
 	nil_addr_t addr;

app/src/incphub/include/incphub/incphub_client.h

@@ -5,6 +5,8 @@
 
 #include <stddef.h>
 
+#include "FreeRTOS.h"
+
 #include "incphub/incphub.h"
 
 #include "nil/nil.h"
@@ -28,7 +30,7 @@ int incphub_init_client(incphub_cli_t **cli, ntl_port_t port);
 /// @return INCPHUB_OK on success, an error otherwise
 int incphub_get_msg_buffer(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t length);
 
-/// @brief Releases a buffer previously acquired with incphub_get_msg_buffer
+/// @brief Releases a buffer received from incphub_await_msg
 /// @param cli Current client handle
 /// @param buffer The buffer to release
 /// @return INCPHUB_OK on success, an error otherwise
@@ -36,18 +38,20 @@ int incphub_release_msg_buffer(incphub_cli_t *cli, uint8_t *buffer);
 
 /// @brief Attempts to send a message to the local interface
 /// @param cli Current client handle
-/// @param buffer The message buffer
+/// @param buffer The message buffer (gets consumed)
 /// @param length The message length
 /// @param dst_port The destination NTL port
+/// @param max_delay The maximum time to wait for a resources, in ticks
 /// @return INCPHUB_OK on success, an error otherwise
-int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_port_t dst_port);
+int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_port_t dst_port, TickType_t max_delay);
 
 /// @brief Awaits the receival of a message
 /// @param cli Current client handle
 /// @param buffer_ptr Location to return the buffer
 /// @param length_ptr Location to return the length
 /// @param source_ptr Location to return the message sender
+/// @param max_delay The maximum time to wait for a message, in ticks
 /// @return INCPHUB_OK on success, an error otherwise
-int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr);
+int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr, TickType_t max_delay);
 
 #endif

app/src/incphub/src/incphub_client.c

@@ -56,10 +56,10 @@ int incphub_release_msg_buffer(incphub_cli_t *cli, uint8_t *buffer)
 	uint8_t *unwound = buffer - sizeof(nil_header_t) + sizeof(ntl_header_t);
 
 	// Release correct buffer pointer
-	return release_msg_buffer(buffer);
+	return release_msg_buffer(unwound);
 }
 
-int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_port_t dst_port)
+int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_port_t dst_port, TickType_t max_delay)
 {
 	if (cli == NULL)
 		return INCPHUB_EINVAL;
@@ -71,7 +71,6 @@ int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_p
 	// Unwind buffer pointer
 	ntl_packet_t *ntl_packet = (ntl_packet_t*)(buffer - sizeof(ntl_header_t));
 	nil_packet_t *nil_packet = (nil_packet_t*)(ntl_packet - sizeof(ntl_header_t));
-	size_t extended_len = sizeof(nil_header_t) + sizeof(ntl_header_t) + length;
 
 	// Populate NTL header
 	ntl_build_header(&ntl_packet->header, dst_port, cli->port, length);
@@ -80,10 +79,10 @@ int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_p
 	nil_build_header(&nil_packet->header, NIL_ADDR_LOCALHOST, INCPHUB_LOCAL_NIL_ADDR, length + sizeof(ntl_header_t));
 
 	// Pass to ingest
-	return enqueue_message_ingest((uint8_t *)nil_packet, extended_len);
+	return enqueue_message_ingest((uint8_t *)nil_packet, max_delay);
 }
 
-int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr)
+int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr, TickType_t max_delay)
 {
 	if (cli == NULL)
 		return INCPHUB_EINVAL;
@@ -97,5 +96,34 @@ int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_p
 	// cli requested for interface consistency only (for now)
 	(void)cli;
 
-	return await_message_dispatch(buffer_ptr, length_ptr, source_ptr);
+	uint8_t *buffer;
+	int err = await_message_dispatch(&buffer, max_delay);
+	if (err != INCPHUB_OK)
+	{
+		return err;
+	}
+
+	// Extract header information
+    nil_packet_t *nil = extract_nil_packet(buffer, INCPHUB_MSG_BUFFER_SIZE);
+    if (nil == NULL)
+    {
+        return INCPHUB_BAD_PACKET;
+    }
+    ntl_packet_t *ntl = extract_ntl_packet(nil);
+    if (nil == NULL)
+    {
+        return INCPHUB_BAD_PACKET;
+    }
+
+    // Build source spec
+    incphub_addr_t source = { 0 };
+    source.addr = nil->header.src_addr;
+    source.port = ntl->header.src_port;
+
+    // Return values
+    *buffer_ptr = ntl->payload;
+    *length_ptr = ntl->header.len;
+    *source_ptr = source;
+
+	return INCPHUB_OK;
 }

app/src/incphub/src/incphub_config.h

@@ -10,7 +10,7 @@
 
 #ifndef INCPHUB_MAX_CLIENTS
 /// @brief Number of client slots to allocate
-#define INCPHUB_MAX_CLIENTS 16
+#define INCPHUB_MAX_CLIENTS 8
 #endif
 
 #ifndef INCPHUB_MSG_BUFFER_SIZE

app/src/incphub/src/incphub_shared.c

@@ -2,6 +2,7 @@
 
 #include <string.h>
 #include <stdbool.h>
+#include <stdint.h>
 
 #include "FreeRTOS.h"
 #include "semphr.h"
@@ -13,6 +14,13 @@
 #include "nil/nil.h"
 
 
+
+#if UINTPTR_MAX > UINT32_MAX
+#error "Pointers of 32-bits or less are required for incphub mailbox to function correctly."
+#endif
+
+
+
 // === Private variables ===
 
 // Global lock
@@ -56,7 +64,6 @@ int register_client(incphub_cli_t **cli_ptr, ntl_port_t port, TaskHandle_t task)
 {
     if (port == NTL_PORT_EMPTY)
         return INCPHUB_EINVAL;
-
     if (cli_ptr == NULL)
         return INCPHUB_EINVAL;
 
@@ -93,6 +100,8 @@ int get_msg_buffer(uint8_t **buffer_ptr, size_t len)
 {
     if (buffer_ptr == NULL)
         return INCPHUB_EINVAL;
+    if (len > INCPHUB_MSG_BUFFER_SIZE)
+        return INCPHUB_ENOMEM;
 
     // == Enter ==
     if (!global_lock(portMAX_DELAY))
@@ -149,7 +158,6 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr)
 {
     if (port == NTL_PORT_EMPTY)
         return INCPHUB_EINVAL;
-
     if (client_ptr == NULL)
         return INCPHUB_EINVAL;
 
@@ -172,20 +180,81 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr)
     return INCPHUB_OK;
 }
 
-int enqueue_message_ingest(uint8_t *buffer, size_t length)
+int enqueue_message_ingest(uint8_t *buffer, TickType_t max_delay)
 {
-    (void)buffer;
+    //NOTE: Does not need global lock
-    (void)length;
+
-    //TODO: Implement
+    if (buffer == NULL)
-    return INCPHUB_ERR;
+        return INCPHUB_EINVAL;
+
+    // Try to queue message
+    if (xQueueSendToBack(incphub_ingest_queue, &buffer, max_delay) != pdPASS)
+    {
+        return INCPHUB_EAGAIN;
+    }
+
+    return INCPHUB_OK;
 }
 
-int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr)
+int dequeue_message_ingest(uint8_t **buffer_ptr, TickType_t max_delay)
 {
-    (void)buffer_ptr;
+    //NOTE: Does not need global lock
-    (void)length_ptr;
+
-    //TODO: Implement
+    if (buffer_ptr == NULL)
-    return INCPHUB_ERR;
+        return INCPHUB_EINVAL;
+
+    uint8_t *buffer;
+
+    // Try to dequeue message
+    if (xQueueReceive(incphub_ingest_queue, &buffer, max_delay))
+    {
+        return INCPHUB_EAGAIN;
+    }
+
+    // Return buffer
+    *buffer_ptr = buffer;
+
+    return INCPHUB_OK;
+}
+
+int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer)
+{
+    //NOTE: Does not need global lock
+
+    if (cli == NULL)
+        return INCPHUB_EINVAL;
+    if (buffer == NULL)
+        return INCPHUB_EINVAL;
+
+    // Try to notify
+    if (xTaskNotifyIndexed(cli->task, INCPHUB_TASKINDEX_MAILBOX, (uint32_t)buffer, eSetValueWithoutOverwrite) == pdFALSE)
+    {
+        // Failed because previous message has not yet been processed
+        return INCPHUB_EAGAIN;
+    }
+
+    return INCPHUB_OK;
+}
+
+int await_message_dispatch(uint8_t **buffer_ptr, TickType_t max_delay)
+{
+    //NOTE: Does not need global lock
+
+    if (buffer_ptr == NULL)
+        return INCPHUB_EINVAL;
+
+    // Get raw msg buffer
+    uint8_t *buffer;
+    if (xTaskNotifyWaitIndexed(INCPHUB_TASKINDEX_MAILBOX, 0, 0, (uint32_t*)&buffer, max_delay) == pdFALSE)
+    {
+        // Nothing pending
+        return INCPHUB_EAGAIN;
+    }
+
+    // Return buffer
+    *buffer_ptr = buffer;
+
+    return INCPHUB_OK;
 }
 
 
@@ -234,7 +303,7 @@ static int find_index_of_buffer(uint8_t *buffer)
 {
     for (int i = 0; i < INCPHUB_MSG_BUFFER_COUNT; ++i)
     {
-        if (buffer = _msg_buffers[i])
+        if (buffer == _msg_buffers[i])
             return i;
     }
 

app/src/incphub/src/incphub_shared.h

@@ -20,7 +20,7 @@ struct incphub_cli_t {
 	TaskHandle_t task;
 };
 
-// Queue<incphub_msg_t>
+// Queue<uint8_t*> (queue of buffers)
 extern QueueHandle_t incphub_ingest_queue;
 
 #ifndef INCPHUB_CLIENT_H_
@@ -59,18 +59,71 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr);
 
 /// @brief Enqueues a message for ingest by the incphub task
 /// @param buffer The buffer to enqueue
-/// @param length The message length
+/// @param max_delay The maximum time to wait for resources, in ticks
 /// @return INCPHUB_OK on success, an error otherwise
-int enqueue_message_ingest(uint8_t *buffer, size_t length);
+int enqueue_message_ingest(uint8_t *buffer, TickType_t max_delay);
 
 /// @brief Dequeues a message for ingest
 /// @param buffer_ptr Location to return the buffer
-/// @param length_ptr Location to store the message length
+/// @param max_delay The maximum time to wait for a message, in ticks
 /// @return INCPHUB_OK on success, an error otherwise
-int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr);
+int dequeue_message_ingest(uint8_t **buffer_ptr, TickType_t max_delay);
 
-int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer, size_t length, incphub_addr_t source);
+/// @brief Sends a message for dispatch to the corresponding client
+/// @param cli The client to send the message to
+/// @param buffer The message buffer to send
+/// @return INCPHUB_OK on success, an error otherwise
+int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer);
 
-int await_message_dispatch(uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr);
+/// @brief Blocks until a message is received for dispatch, or until a given timeout
+/// @param buffer_ptr Location to return the message buffer
+/// @param max_delay The maximum time to wait for a message, in ticks
+/// @return INCPHUB_OK on success, an error otherwise
+/// @remark Buffers obatined from this function must be release with release_msg_buffer once no longer in use
+int await_message_dispatch(uint8_t **buffer_ptr, TickType_t max_delay);
+
+/// @brief Extracts a NIL packet from a buffer
+/// @param buffer The byte buffer to read from
+/// @param length The maximum length to read from
+/// @return Pointer to the NIL packet in the buffer or NULL if the packet is invalid
+static inline nil_packet_t *extract_nil_packet(uint8_t *buffer, size_t length)
+{
+	// Ensure message is big enough to hold a packet
+	if (length < sizeof(nil_packet_t))
+	{
+		return NULL;
+	}
+
+	nil_packet_t *nil = (nil_packet_t *)buffer;
+	// Ensure length field is not going to cause buffer overflow
+	if (length < sizeof(nil_packet_t) + nil->header.len )
+	{
+		return NULL;
+	}
+
+	return nil;
+}
+
+/// @brief Extracts an NTL packet from a buffer
+/// @param buffer The byte buffer to read from
+/// @param length The maximum length to read from
+/// @return Pointer to the NTL packet in the buffer or NULL if the packet is invalid
+static inline ntl_packet_t *extract_ntl_packet(nil_packet_t *nil)
+{
+	// Ensure message is big enough to hold a packet
+	if (nil->header.len < sizeof(ntl_packet_t))
+	{
+		return NULL;
+	}
+
+	ntl_packet_t *ntl = (ntl_packet_t *)nil->payload;
+	// Ensure length field is not going to cause buffer overflow
+	if (nil->header.len < sizeof(ntl_packet_t) + ntl->header.len)
+	{
+		return NULL;
+	}
+
+	return ntl;
+}
 
 #endif

app/src/incphub/src/incphub_task.c

@@ -29,17 +29,15 @@ QueueHandle_t incphub_ingest_queue;
 
 // === Private variables ===
 #define INGEST_QUEUE_LENGTH    (16)
-#define INGEST_QUEUE_ITEM_SIZE (sizeof(ntl_packet_t *))
+#define INGEST_QUEUE_ITEM_SIZE (sizeof(uint8_t*))
 static uint8_t _ingest_queue_store[INGEST_QUEUE_LENGTH * INGEST_QUEUE_ITEM_SIZE];
 static StaticQueue_t _ingest_queue_static;
 
 
 
 // === Private functions ===
-static nil_packet_t *extract_nil_packet(uint8_t *buffer, size_t length);
+static int handle_nil_packet(nil_packet_t *nil);
-static ntl_packet_t *extract_ntl_packet(nil_packet_t *nil);
+static int handle_ntl_packet(nil_packet_t *nil, ntl_packet_t *ntl);
-static void handle_nil_packet(nil_packet_t *nil);
-static void handle_ntl_packet(ntl_packet_t *ntl, nil_addr_t src_addr);
 
 
 
@@ -54,16 +52,16 @@ void incphub_task_preinit()
 void incphub_task_main(void *params)
 {
     (void)params;
+	int err;
 
     while (1)
     {
 		// Dequeue message
 		uint8_t *buffer;
-		size_t length;
+		dequeue_message_ingest(&buffer, portMAX_DELAY);
-		dequeue_message_ingest(&buffer, &length);
 
 		// Get NIL packet
-		nil_packet_t *nil = extract_nil_packet(buffer, length);
+		nil_packet_t *nil = extract_nil_packet(buffer, INCPHUB_MSG_BUFFER_SIZE);
 		if (nil == NULL)
 		{
 			//TODO: Error
@@ -71,54 +69,34 @@ void incphub_task_main(void *params)
 		}
 
 		// Process or route
-		handle_nil_packet(nil);
+		err = handle_nil_packet(nil);
+		if (err == INCPHUB_EAGAIN)
+		{
+			// Failed but only temporarily
+			// Could be target task not ready
+			// Attempt requeue to allow retries
+			enqueue_message_ingest(buffer, 0);
+
+			// If failed to requeue (no space), discard message
+			// This avoids clogging the system due to a single unresponsive task
+			continue;
+		}
+		else if (err != INCPHUB_OK)
+		{
+			// Failed for some other reason
+			// Do not retry
+			continue;
+		}
+
+		// Successfully processed, no need to retry by requeueing
+		dequeue_message_ingest(&buffer, portMAX_DELAY);
     }
 }
 
 
 
 // === Private functions ===
-static nil_packet_t *extract_nil_packet(uint8_t *buffer, size_t length)
+static int handle_nil_packet(nil_packet_t *nil)
-{
-	// Ensure message is big enough to hold a packet
-	if (length < sizeof(nil_packet_t))
-	{
-		//TODO: Error
-		return NULL;
-	}
-
-	nil_packet_t *nil = (nil_packet_t *)buffer;
-	// Ensure length field is not going to cause buffer overflow
-	if (length < sizeof(nil_packet_t) + nil->header.len )
-	{
-		//TODO: Error
-		return NULL;
-	}
-
-	return nil;
-}
-
-static ntl_packet_t *extract_ntl_packet(nil_packet_t *nil)
-{
-	// Ensure message is big enough to hold a packet
-	if (nil->header.len < sizeof(ntl_packet_t))
-	{
-		//TODO: Error
-		return NULL;
-	}
-
-	ntl_packet_t *ntl = (ntl_packet_t *)nil->payload;
-	// Ensure length field is not going to cause buffer overflow
-	if (nil->header.len < sizeof(ntl_packet_t) + ntl->header.len)
-	{
-		//TODO: Error
-		return NULL;
-	}
-
-	return ntl;
-}
-
-static void handle_nil_packet(nil_packet_t *nil)
 {
 	if (nil->header.dst_addr == INCPHUB_LOCAL_NIL_ADDR)
 	{
@@ -126,34 +104,28 @@ static void handle_nil_packet(nil_packet_t *nil)
 		if (ntl == NULL)
 		{
 			//TODO: Error
-			return;
+			return INCPHUB_BAD_PACKET;
 		}
 
-		handle_ntl_packet(ntl, nil->header.src_addr);
+		return handle_ntl_packet(nil, ntl);
-		return;
 	}
 
 	//TODO: Route to external interface
-	return;
+	return INCPHUB_ERR;
 }
 
-static void handle_ntl_packet(ntl_packet_t *ntl, nil_addr_t src_addr)
+static int handle_ntl_packet(nil_packet_t *nil, ntl_packet_t *ntl)
 {
+	int err;
 	ntl_port_t port = ntl_get_dst(&ntl->header);
 
 	incphub_cli_t *client;
-	if (get_client_with_port(port, &client) != INCPHUB_OK)
+	err = get_client_with_port(port, &client);
+	if (err != INCPHUB_OK)
 	{
 		//TODO: Error
-		return;
+		return err;
 	}
 
-	incphub_addr_t sender = { .addr = src_addr, .port = ntl->header.src_port };
+	return do_message_dispatch(client, (uint8_t*)nil);
-	if (do_message_dispatch(client, ntl->payload, ntl->header.len, sender) != INCPHUB_OK)
-	{
-		//TODO: Error
-		return;
-	}
-
-	return;
 }