chore: Unified buffer passing to ignore size
This commit is contained in:
@@ -18,6 +18,12 @@ Set as a macro in compile flags if wishing to override the defaults
|
||||
|
||||
- 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.
|
||||
|
||||
@@ -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,7 +79,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));
|
||||
|
||||
// Pass to ingest
|
||||
return enqueue_message_ingest((uint8_t *)nil_packet, extended_len, portMAX_DELAY);
|
||||
return enqueue_message_ingest((uint8_t *)nil_packet, portMAX_DELAY);
|
||||
}
|
||||
|
||||
int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr)
|
||||
|
||||
@@ -170,18 +170,15 @@ 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, TickType_t max_delay)
|
||||
int enqueue_message_ingest(uint8_t *buffer, TickType_t max_delay)
|
||||
{
|
||||
//NOTE: Does not need global lock
|
||||
|
||||
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)
|
||||
if (xQueueSendToBack(incphub_ingest_queue, &buffer, max_delay) != pdPASS)
|
||||
{
|
||||
return INCPHUB_EAGAIN;
|
||||
}
|
||||
@@ -189,25 +186,23 @@ int enqueue_message_ingest(uint8_t *buffer, size_t length, TickType_t max_delay)
|
||||
return INCPHUB_OK;
|
||||
}
|
||||
|
||||
int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr, TickType_t max_delay)
|
||||
int dequeue_message_ingest(uint8_t **buffer_ptr, TickType_t max_delay)
|
||||
{
|
||||
//NOTE: Does not need global lock
|
||||
|
||||
if (buffer_ptr == NULL)
|
||||
return INCPHUB_EINVAL;
|
||||
if (length_ptr == NULL)
|
||||
return INCPHUB_EINVAL;
|
||||
|
||||
incphub_msg_t msg = { 0 };
|
||||
uint8_t *buffer;
|
||||
|
||||
// Try to dequeue message
|
||||
if (xQueueReceive(incphub_ingest_queue, &msg, max_delay))
|
||||
if (xQueueReceive(incphub_ingest_queue, &buffer, max_delay))
|
||||
{
|
||||
return INCPHUB_EAGAIN;
|
||||
}
|
||||
|
||||
*buffer_ptr = msg.buffer;
|
||||
*length_ptr = msg.length;
|
||||
// Return buffer
|
||||
*buffer_ptr = buffer;
|
||||
|
||||
return INCPHUB_OK;
|
||||
}
|
||||
|
||||
@@ -20,12 +20,7 @@ struct incphub_cli_t {
|
||||
TaskHandle_t task;
|
||||
};
|
||||
|
||||
typedef struct {
|
||||
uint8_t *buffer;
|
||||
size_t length;
|
||||
} incphub_msg_t;
|
||||
|
||||
// Queue<incphub_msg_t>
|
||||
// Queue<uint8_t*> (queue of buffers)
|
||||
extern QueueHandle_t incphub_ingest_queue;
|
||||
|
||||
#ifndef INCPHUB_CLIENT_H_
|
||||
@@ -64,19 +59,17 @@ 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
|
||||
/// @return INCPHUB_OK on success, an error otherwise
|
||||
int enqueue_message_ingest(uint8_t *buffer, size_t length, TickType_t max_delay);
|
||||
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 resources
|
||||
/// @return INCPHUB_OK on success, an error otherwise
|
||||
int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr, TickType_t maxDelay);
|
||||
int dequeue_message_ingest(uint8_t **buffer_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);
|
||||
|
||||
int await_message_dispatch(uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr);
|
||||
|
||||
|
||||
@@ -29,7 +29,7 @@ QueueHandle_t incphub_ingest_queue;
|
||||
|
||||
// === Private variables ===
|
||||
#define INGEST_QUEUE_LENGTH (16)
|
||||
#define INGEST_QUEUE_ITEM_SIZE (sizeof(incphub_msg_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;
|
||||
|
||||
@@ -39,7 +39,7 @@ static StaticQueue_t _ingest_queue_static;
|
||||
static nil_packet_t *extract_nil_packet(uint8_t *buffer, size_t length);
|
||||
static ntl_packet_t *extract_ntl_packet(nil_packet_t *nil);
|
||||
static void handle_nil_packet(nil_packet_t *nil);
|
||||
static void handle_ntl_packet(ntl_packet_t *ntl, nil_addr_t src_addr);
|
||||
static void handle_ntl_packet(nil_packet_t *nil, ntl_packet_t *ntl);
|
||||
|
||||
|
||||
|
||||
@@ -59,11 +59,10 @@ void incphub_task_main(void *params)
|
||||
{
|
||||
// Dequeue message
|
||||
uint8_t *buffer;
|
||||
size_t length;
|
||||
dequeue_message_ingest(&buffer, &length, portMAX_DELAY);
|
||||
dequeue_message_ingest(&buffer, portMAX_DELAY);
|
||||
|
||||
// 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
|
||||
@@ -129,7 +128,7 @@ static void handle_nil_packet(nil_packet_t *nil)
|
||||
return;
|
||||
}
|
||||
|
||||
handle_ntl_packet(ntl, nil->header.src_addr);
|
||||
handle_ntl_packet(nil, ntl);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -137,7 +136,7 @@ static void handle_nil_packet(nil_packet_t *nil)
|
||||
return;
|
||||
}
|
||||
|
||||
static void handle_ntl_packet(ntl_packet_t *ntl, nil_addr_t src_addr)
|
||||
static void handle_ntl_packet(nil_packet_t *nil, ntl_packet_t *ntl)
|
||||
{
|
||||
ntl_port_t port = ntl_get_dst(&ntl->header);
|
||||
|
||||
@@ -148,8 +147,7 @@ static void handle_ntl_packet(ntl_packet_t *ntl, nil_addr_t src_addr)
|
||||
return;
|
||||
}
|
||||
|
||||
incphub_addr_t sender = { .addr = src_addr, .port = ntl->header.src_port };
|
||||
if (do_message_dispatch(client, ntl->payload, ntl->header.len, sender) != INCPHUB_OK)
|
||||
if (do_message_dispatch(client, (uint8_t*)nil) != INCPHUB_OK)
|
||||
{
|
||||
//TODO: Error
|
||||
return;
|
||||
|
||||
Reference in New Issue
Block a user