feat: Added max_delay and reworked ingest queue

This commit is contained in:
2026-06-29 15:15:58 +01:00
parent 7e033b7779
commit 628f031f26
4 changed files with 53 additions and 9 deletions

View File

@@ -80,7 +80,7 @@ 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)); nil_build_header(&nil_packet->header, NIL_ADDR_LOCALHOST, INCPHUB_LOCAL_NIL_ADDR, length + sizeof(ntl_header_t));
// Pass to ingest // Pass to ingest
return enqueue_message_ingest((uint8_t *)nil_packet, extended_len); return enqueue_message_ingest((uint8_t *)nil_packet, extended_len, portMAX_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)

View File

@@ -172,18 +172,55 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr)
return INCPHUB_OK; return INCPHUB_OK;
} }
int enqueue_message_ingest(uint8_t *buffer, size_t length) int enqueue_message_ingest(uint8_t *buffer, size_t length, TickType_t max_delay)
{ {
(void)buffer; //NOTE: Does not need global lock
(void)length;
if (buffer == NULL)
return INCPHUB_EINVAL;
// Build message definition
incphub_msg_t msg = { .buffer = buffer, .length = length };
// Try to queue message
if (xQueueSendToBack(incphub_ingest_queue, &msg, max_delay) != pdPASS)
{
return INCPHUB_EAGAIN;
}
return INCPHUB_OK;
}
int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr, TickType_t max_delay)
{
//NOTE: Does not need global lock
(void)buffer_ptr;
(void)length_ptr;
(void)max_delay;
//TODO: Implement //TODO: Implement
return INCPHUB_ERR; return INCPHUB_ERR;
} }
int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr) int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer, size_t length, incphub_addr_t source)
{ {
//NOTE: Does not need global lock
(void)cli;
(void)buffer;
(void)length;
(void)source;
//TODO: Implement
return INCPHUB_ERR;
}
int await_message_dispatch(uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr)
{
//NOTE: Does not need global lock
(void)buffer_ptr; (void)buffer_ptr;
(void)length_ptr; (void)length_ptr;
(void)source_ptr;
//TODO: Implement //TODO: Implement
return INCPHUB_ERR; return INCPHUB_ERR;
} }

View File

@@ -20,6 +20,11 @@ struct incphub_cli_t {
TaskHandle_t task; TaskHandle_t task;
}; };
typedef struct {
uint8_t *buffer;
size_t length;
} incphub_msg_t;
// Queue<incphub_msg_t> // Queue<incphub_msg_t>
extern QueueHandle_t incphub_ingest_queue; extern QueueHandle_t incphub_ingest_queue;
@@ -60,14 +65,16 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr);
/// @brief Enqueues a message for ingest by the incphub task /// @brief Enqueues a message for ingest by the incphub task
/// @param buffer The buffer to enqueue /// @param buffer The buffer to enqueue
/// @param length The message length /// @param length The message length
/// @param max_delay The maximum time to wait for resources
/// @return INCPHUB_OK on success, an error otherwise /// @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, size_t length, TickType_t max_delay);
/// @brief Dequeues a message for ingest /// @brief Dequeues a message for ingest
/// @param buffer_ptr Location to return the buffer /// @param buffer_ptr Location to return the buffer
/// @param length_ptr Location to store the message length /// @param length_ptr Location to store the message length
/// @param max_delay The maximum time to wait for resources
/// @return INCPHUB_OK on success, an error otherwise /// @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, size_t *length_ptr, TickType_t maxDelay);
int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer, size_t length, incphub_addr_t source); int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer, size_t length, incphub_addr_t source);

View File

@@ -29,7 +29,7 @@ QueueHandle_t incphub_ingest_queue;
// === Private variables === // === Private variables ===
#define INGEST_QUEUE_LENGTH (16) #define INGEST_QUEUE_LENGTH (16)
#define INGEST_QUEUE_ITEM_SIZE (sizeof(ntl_packet_t *)) #define INGEST_QUEUE_ITEM_SIZE (sizeof(incphub_msg_t))
static uint8_t _ingest_queue_store[INGEST_QUEUE_LENGTH * INGEST_QUEUE_ITEM_SIZE]; static uint8_t _ingest_queue_store[INGEST_QUEUE_LENGTH * INGEST_QUEUE_ITEM_SIZE];
static StaticQueue_t _ingest_queue_static; static StaticQueue_t _ingest_queue_static;
@@ -60,7 +60,7 @@ void incphub_task_main(void *params)
// Dequeue message // Dequeue message
uint8_t *buffer; uint8_t *buffer;
size_t length; size_t length;
dequeue_message_ingest(&buffer, &length); dequeue_message_ingest(&buffer, &length, portMAX_DELAY);
// Get NIL packet // Get NIL packet
nil_packet_t *nil = extract_nil_packet(buffer, length); nil_packet_t *nil = extract_nil_packet(buffer, length);