Compare commits

...

18 Commits

Author SHA1 Message Date
a8cfd61491 chore: Touchup on NIL comments 2026-07-23 16:22:11 +01:00
14ca6d651b chore: Renamed incphub to libshunt 2026-07-01 17:54:07 +01:00
ff8d7914a9 fix: Increased stack size to allow printf to work 2026-07-01 11:49:50 +01:00
6a71c2863d fix: Added buffer releasing to requester task 2026-07-01 11:37:55 +01:00
c8ebbb2946 fix: Added localhost to intra-routing filter 2026-07-01 11:37:40 +01:00
998bafc53a chore: First tests 2026-06-30 16:05:11 +01:00
6a0df95eeb feat: Changed client to accept messages to any destination 2026-06-30 15:36:44 +01:00
9f7d1aae4f fix: Corrected double dequeue in incphub task 2026-06-30 15:23:47 +01:00
55b3404674 chore: Moved unpacking of NIL/NTL headers for consistent responsibilities 2026-06-30 15:05:09 +01:00
7aa5a91181 fix: Corrected function declarations 2026-06-30 15:04:35 +01:00
bfcd91e61a feat: Implemented await_message_dispatch 2026-06-30 14:46:30 +01:00
7edc2e21da feat: Implemented requeueing of packets to busy tasks 2026-06-30 12:00:11 +01:00
a5f79f5eac feat: Implemented do_message_dispatch 2026-06-30 11:59:19 +01:00
cf4914d939 chore: Unified buffer passing to ignore size 2026-06-29 16:02:07 +01:00
854d6eab36 docs: Clarifications on buffer ownership 2026-06-29 16:00:02 +01:00
24554e9d23 fix: Corrected buffer passed to release 2026-06-29 15:59:22 +01:00
fe09b689c8 feat: Implemented dequeue_message_ingest 2026-06-29 15:20:05 +01:00
628f031f26 feat: Added max_delay and reworked ingest queue 2026-06-29 15:15:58 +01:00
29 changed files with 1049 additions and 737 deletions

View File

@@ -31,7 +31,7 @@ add_executable(${EXECUTABLE} "")
# Set compiler/linker options
target_compile_definitions(${EXECUTABLE} PRIVATE
${COMPILER_DEFINES}
INCPHUB_LOCAL_NIL_ADDR=1
LIBSHUNT_LOCAL_NIL_ADDR=1
)
target_compile_options(${EXECUTABLE} PRIVATE
-Wall

View File

@@ -1,5 +1,5 @@
add_subdirectory(tasks)
add_subdirectory(incphub)
add_subdirectory(libshunt)
add_subdirectory(nil)
add_subdirectory(ntl)

View File

@@ -1,19 +0,0 @@
# INCPHUB
## FreeRTOS requirements
- INCLUDE_xTaskGetCurrentTaskHandle set to 1
- INCLUDE_vTaskSuspend set to 1
- configUSE_TASK_NOTIFICATIONS set to 1
- configTASK_NOTIFICATION_ARRAY_ENTRIES set to >= 2 (0 is for stream/message buffers, 1(configurable) is used by incphub)
- configUSE_MUTEXES set to 1
## Configuration
Set as a macro in compile flags if wishing to override the defaults
- INCPHUB_TASKINDEX_MAILBOX, defaults to 1 (avoid using 0)
## Setup
- incphub_task_preinit MUST be run before any calls to the client, likely even before the scheduler is started

View File

@@ -1,29 +0,0 @@
// incphub.h - Public definitions common to incphub and incphub client
#ifndef INCPHUB_H_
#define INCPHUB_H_
#include "ntl/ntl.h"
#include "nil/nil.h"
/// @brief Success
#define INCPHUB_OK (0)
/// @brief Unspecified error
#define INCPHUB_ERR (1)
/// @brief Invalid argument
#define INCPHUB_EINVAL (2)
/// @brief Resource or entry not found
#define INCPHUB_ENOTFOUND (3)
/// @brief Resource in use
#define INCPHUB_EBUSY (4)
/// @brief Resource temporarily unavailable
#define INCPHUB_EAGAIN (5)
/// @brief Could not allocate memory
#define INCPHUB_ENOMEM (6)
typedef struct {
nil_addr_t addr;
ntl_port_t port;
} incphub_addr_t;
#endif

View File

@@ -1,53 +0,0 @@
// incphub_client.h - Public interface of incphub client side
#ifndef INCPHUB_CLIENT_H_
#define INCPHUB_CLIENT_H_
#include <stddef.h>
#include "incphub/incphub.h"
#include "nil/nil.h"
#include "ntl/ntl.h"
typedef struct incphub_cli_t incphub_cli_t;
/// @brief Initializes an incphub client
/// @param port The port number to assign to the caller
/// @return INCPHUB_OK on success, an error otherwise
int incphub_init_client(incphub_cli_t **cli, ntl_port_t port);
/// @brief Attempts to get a free message buffer from the central storage space
/// @param cli Current client handle
/// @param buffer_ptr Location to return the buffer
/// @param length The minimum size, in bytes, of the buffer get
/// @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
/// @param cli Current client handle
/// @param buffer The buffer to release
/// @return INCPHUB_OK on success, an error otherwise
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 length The message length
/// @param dst_port The destination NTL port
/// @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);
/// @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
/// @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);
#endif

View File

@@ -1,17 +0,0 @@
#ifndef INCPHUB_TASK_H_
#define INCPHUB_TASK_H_
#include "FreeRTOS.h"
#include "task.h"
#include "FreeRTOSConfig.h"
#define INCPHUB_TASK_STACK_DEPTH configMINIMAL_STACK_SIZE
#define INCPHUB_TASK_PRIORITY (configMAX_PRIORITIES - 1)
extern StackType_t _incphub_task_stack[INCPHUB_TASK_STACK_DEPTH];
extern StaticTask_t _incphub_task;
void incphub_task_preinit();
void incphub_task_main(void *params);
#endif

View File

@@ -1,101 +0,0 @@
#include <stdbool.h>
#include "incphub/incphub_client.h"
#include "incphub/incphub_task.h"
#include "incphub/incphub.h"
#include "incphub_config.h"
#include "incphub_shared.h"
#include "ntl/ntl.h"
#include "nil/nil.h"
#include "message_buffer.h"
int incphub_init_client(incphub_cli_t **cli, ntl_port_t port)
{
return register_client(cli, port, xTaskGetCurrentTaskHandle());
}
int incphub_get_msg_buffer(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t length)
{
if (cli == NULL)
return INCPHUB_EINVAL;
if (buffer_ptr == NULL)
return INCPHUB_EINVAL;
// Request extra space for NIL/NTL headers
size_t extended_len = sizeof(nil_header_t) + sizeof(ntl_header_t) + length;
uint8_t *buf;
int status = get_msg_buffer(&buf, extended_len);
if (status != INCPHUB_OK)
return status;
// Reserve space for NIL/NTL headers
buf += sizeof(nil_header_t) + sizeof(ntl_header_t);
// Return buffer
*buffer_ptr = buf;
return INCPHUB_OK;
}
int incphub_release_msg_buffer(incphub_cli_t *cli, uint8_t *buffer)
{
if (cli == NULL)
return INCPHUB_EINVAL;
if (buffer == NULL)
return INCPHUB_EINVAL;
// cli requested for interface consistency only (for now)
(void)cli;
// Unwind buffer pointer
uint8_t *unwound = buffer - sizeof(nil_header_t) + sizeof(ntl_header_t);
// Release correct buffer pointer
return release_msg_buffer(buffer);
}
int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_port_t dst_port)
{
if (cli == NULL)
return INCPHUB_EINVAL;
if (buffer == NULL)
return INCPHUB_EINVAL;
if (dst_port == NTL_PORT_EMPTY)
return INCPHUB_EINVAL;
// 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);
// Populate NIL header
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);
}
int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr)
{
if (cli == NULL)
return INCPHUB_EINVAL;
if (buffer_ptr == NULL)
return INCPHUB_EINVAL;
if (length_ptr == NULL)
return INCPHUB_EINVAL;
if (source_ptr == NULL)
return INCPHUB_EINVAL;
// cli requested for interface consistency only (for now)
(void)cli;
return await_message_dispatch(buffer_ptr, length_ptr, source_ptr);
}

View File

@@ -1,30 +0,0 @@
// incphub_config.h - Default configurations for incphub
#ifndef INCPHUB_CONFIG_H_
#define INCPHUB_CONFIG_H_
#ifndef INCPHUB_TASKINDEX_MAILBOX
/// @brief Index of direct-to-task notification for mailbox to use
#define INCPHUB_TASKINDEX_MAILBOX 1
#endif
#ifndef INCPHUB_MAX_CLIENTS
/// @brief Number of client slots to allocate
#define INCPHUB_MAX_CLIENTS 16
#endif
#ifndef INCPHUB_MSG_BUFFER_SIZE
/// @brief Message buffer size in bytes. Must allow space for NIL and NTL headers
#define INCPHUB_MSG_BUFFER_SIZE 32
#endif
#ifndef INCPHUB_MSG_BUFFER_COUNT
/// @brief Number of message buffers to allocate
#define INCPHUB_MSG_BUFFER_COUNT 8
#endif
#ifndef INCPHUB_LOCAL_NIL_ADDR
#error "Local NIL address must be specified with INCPHUB_LOCAL_NIL_ADDR"
#endif
#endif

View File

@@ -1,242 +0,0 @@
#include "incphub_shared.h"
#include <string.h>
#include <stdbool.h>
#include "FreeRTOS.h"
#include "semphr.h"
#include "task.h"
#include "incphub/incphub.h"
#include "incphub_config.h"
#include "nil/nil.h"
// === Private variables ===
// Global lock
static SemaphoreHandle_t _incphub_lock;
static StaticSemaphore_t _incphub_lock_store;
// Clients
static incphub_cli_t _clients[INCPHUB_MAX_CLIENTS];
// Message buffers
// INCPHUB_MSG_BUFFER_COUNT buffers of INCPHUB_MSG_BUFFER_SIZE bytes
static uint8_t _msg_buffers[INCPHUB_MSG_BUFFER_COUNT][INCPHUB_MSG_BUFFER_SIZE];
static bool _msg_buffer_used[INCPHUB_MSG_BUFFER_COUNT];
// === Private function declarations ===
static bool global_lock(TickType_t max_delay);
static void global_unlock();
/// @remark Returns NULL if not found
static incphub_cli_t *find_empty_client();
/// @remark Returns NULL if not found
static incphub_cli_t *find_client_with_port(ntl_port_t port);
/// @remark Returns NULL if not found
static uint8_t *find_unused_buffer();
/// @remark Returns -1 if not found
static int find_index_of_buffer(uint8_t *buffer);
// === Public functions ===
int internal_init()
{
_incphub_lock = xSemaphoreCreateMutexStatic(&_incphub_lock_store);
memset(_clients, 0, sizeof(_clients));
memset(_msg_buffer_used, 0, sizeof(_msg_buffer_used));
return INCPHUB_OK;
}
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;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return INCPHUB_EAGAIN;
// Prevent double binding of ports
if (find_client_with_port(port) != NULL)
{
global_unlock();
return INCPHUB_EBUSY;
}
incphub_cli_t *client = find_empty_client();
if (client == NULL)
{
global_unlock();
return INCPHUB_EBUSY;
}
// Populate client data
client->port = port;
client->task = task;
// Save to passed pointer
*cli_ptr = client;
global_unlock();
return INCPHUB_OK;
}
int get_msg_buffer(uint8_t **buffer_ptr, size_t len)
{
if (buffer_ptr == NULL)
return INCPHUB_EINVAL;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return INCPHUB_EAGAIN;
//TODO: Allow buffer to be given by another layer (below NIL/NTL)
uint8_t *buf = find_unused_buffer();
if (buf == NULL)
{
global_unlock();
return INCPHUB_ENOMEM;
}
// Return buffer
*buffer_ptr = buf;
global_unlock();
return INCPHUB_OK;
}
int release_msg_buffer(uint8_t *buffer)
{
if (buffer == NULL)
return INCPHUB_EINVAL;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return INCPHUB_EAGAIN;
int index = find_index_of_buffer(buffer);
// Handle invalid buffers
if (index == -1)
{
global_unlock();
return INCPHUB_EINVAL;
}
// Handle not-in-use buffers
if (_msg_buffer_used[index] == false)
{
global_unlock();
return INCPHUB_EINVAL;
}
// Mark buffer as free
_msg_buffer_used[index] = false;
global_unlock();
return INCPHUB_OK;
}
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;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return INCPHUB_EAGAIN;
// Map port to client
incphub_cli_t *client = find_client_with_port(port);
if (client == NULL)
{
global_unlock();
return INCPHUB_ENOTFOUND;
}
// Return client
*client_ptr = client;
global_unlock();
return INCPHUB_OK;
}
int enqueue_message_ingest(uint8_t *buffer, size_t length)
{
(void)buffer;
(void)length;
//TODO: Implement
return INCPHUB_ERR;
}
int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr)
{
(void)buffer_ptr;
(void)length_ptr;
//TODO: Implement
return INCPHUB_ERR;
}
// === Private function definitions ===
static bool global_lock(TickType_t max_delay)
{
return xSemaphoreTake(_incphub_lock, max_delay) == pdTRUE;
}
static void global_unlock()
{
xSemaphoreGive(_incphub_lock);
}
static incphub_cli_t *find_empty_client()
{
return find_client_with_port(NTL_PORT_EMPTY);
}
static incphub_cli_t *find_client_with_port(ntl_port_t port)
{
for (int i = 0; i < INCPHUB_MAX_CLIENTS; ++i)
{
incphub_cli_t *client = &_clients[i];
if (client->port == port)
return client;
}
return NULL;
}
static uint8_t *find_unused_buffer()
{
for (int i = 0; i < INCPHUB_MSG_BUFFER_COUNT; ++i)
{
if (!_msg_buffer_used[i])
return _msg_buffers[i];
}
return NULL;
}
static int find_index_of_buffer(uint8_t *buffer)
{
for (int i = 0; i < INCPHUB_MSG_BUFFER_COUNT; ++i)
{
if (buffer = _msg_buffers[i])
return i;
}
return -1;
}

View File

@@ -1,76 +0,0 @@
// incphub_sharedl.h - Internal mechanisms shared by incphub and incphub_client
#ifndef INCPHUB_SHARED_H_
#define INCPHUB_SHARED_H_
#include "FreeRTOS.h"
#include "task.h"
#include "queue.h"
#include "nil/nil.h"
#include "ntl/ntl.h"
#include "incphub/incphub.h"
struct incphub_cli_t {
/// @remark If set to NTL_PORT_EMPTY denotes an empty client
ntl_port_t port;
TaskHandle_t task;
};
// Queue<incphub_msg_t>
extern QueueHandle_t incphub_ingest_queue;
#ifndef INCPHUB_CLIENT_H_
typedef struct incphub_cli_t incphub_cli_t;
#endif
/// @brief Initializes the internal state of incphub
/// @return INCPHUB_OK on success, an error otherwise
int internal_init();
/// @brief Registers a task as an incphub client
/// @param cli_ptr Location to return client
/// @param port The port to register to
/// @param task The task to register as the handler
/// @return INCPHUB_OK on success, an error otherwise
int register_client(incphub_cli_t **cli_ptr, ntl_port_t port, TaskHandle_t task);
/// @brief Attempts to get a free message buffer from the central storage space
/// @param buffer_ptr Location to return the buffer
/// @param len The minimum size, in bytes, of the buffer get
/// @return INCPHUB_OK on success, an error otherwise
int get_msg_buffer(uint8_t **buffer_ptr, size_t len);
/// @brief Releases a buffer previously acquired with get_msg_buffer
/// @param buffer The buffer to release
/// @return INCPHUB_OK on success, an error otherwise
int release_msg_buffer(uint8_t *buffer);
/// @brief Attempts to get the client assigned to the given port
/// @param port The port to look for
/// @param client_ptr Location to return the client
/// @return INCPHUB_OK on success, an error otherwise
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
/// @return INCPHUB_OK on success, an error otherwise
int enqueue_message_ingest(uint8_t *buffer, size_t length);
/// @brief Dequeues a message for ingest
/// @param buffer_ptr Location to return the buffer
/// @param length_ptr Location to store the message length
/// @return INCPHUB_OK on success, an error otherwise
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);
int await_message_dispatch(uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr);
#endif

View File

@@ -1,159 +0,0 @@
#include "incphub/incphub_task.h"
#include <stdint.h>
#include "FreeRTOS.h"
#include "queue.h"
#include "incphub/incphub.h"
#include "incphub_config.h"
#include "incphub_shared.h"
#include "ntl/ntl.h"
#include "nil/nil.h"
// === Static task space ===
StackType_t _incphub_task_stack[INCPHUB_TASK_STACK_DEPTH];
StaticTask_t _incphub_task;
// === Public variables ===
// Queue<nil_packet_t*> Packets awaiting to be accepted by incphub
QueueHandle_t incphub_ingest_queue;
// === Private variables ===
#define INGEST_QUEUE_LENGTH (16)
#define INGEST_QUEUE_ITEM_SIZE (sizeof(ntl_packet_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 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);
// === Public functions ===
void incphub_task_preinit()
{
internal_init();
incphub_ingest_queue =
xQueueCreateStatic(INGEST_QUEUE_LENGTH, INGEST_QUEUE_ITEM_SIZE, _ingest_queue_store, &_ingest_queue_static);
}
void incphub_task_main(void *params)
{
(void)params;
while (1)
{
// Dequeue message
uint8_t *buffer;
size_t length;
dequeue_message_ingest(&buffer, &length);
// Get NIL packet
nil_packet_t *nil = extract_nil_packet(buffer, length);
if (nil == NULL)
{
//TODO: Error
continue;
}
// Process or route
handle_nil_packet(nil);
}
}
// === Private functions ===
static 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))
{
//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)
{
ntl_packet_t *ntl = extract_ntl_packet(nil);
if (ntl == NULL)
{
//TODO: Error
return;
}
handle_ntl_packet(ntl, nil->header.src_addr);
return;
}
//TODO: Route to external interface
return;
}
static void handle_ntl_packet(ntl_packet_t *ntl, nil_addr_t src_addr)
{
ntl_port_t port = ntl_get_dst(&ntl->header);
incphub_cli_t *client;
if (get_client_with_port(port, &client) != INCPHUB_OK)
{
//TODO: Error
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)
{
//TODO: Error
return;
}
return;
}

View File

@@ -1,7 +1,7 @@
target_sources(${EXECUTABLE} PRIVATE
src/incphub_client.c
src/incphub_task.c
src/incphub_shared.c
src/shunt_client.c
src/shunt_task.c
src/shunt_shared.c
)
target_include_directories(${EXECUTABLE} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)

View File

@@ -0,0 +1,37 @@
# LibShunt
Shunts messages, like trains
## FreeRTOS requirements
- INCLUDE_xTaskGetCurrentTaskHandle set to 1
- INCLUDE_vTaskSuspend set to 1
- configUSE_TASK_NOTIFICATIONS set to 1
- configTASK_NOTIFICATION_ARRAY_ENTRIES set to >= 2 (0 is for stream/message buffers, 1(configurable) is used by libshunt)
- configUSE_MUTEXES set to 1
## Configuration
Set as a macro in compile flags if wishing to override the defaults
- LIBSHUNT_TASKINDEX_MAILBOX, defaults to 1 (avoid using 0)
## Setup
- libshunt_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 libshunt) is done via queue (many to one)
Dispatch (internal routing to local task) is done via direct-to-task notifications (see LIBSHUNT_TASKINDEX_MAILBOX) by passing the message buffer pointer
## Buffer ownership
Buffers are allocated with shunt_get_msg_buffer, at which point they belong to the requesting task.
When passed to libshunt_send_XXX functions, they become owned by libshunt
When dispatched internally, the receiving task becomes the owner and must invoke shunt_release_msg_buffer when done
When routed externally, after calling the transmitter function, libshunt releases the buffer

View File

@@ -0,0 +1,31 @@
// libshunt.h - Public definitions common to all libshunt components
#ifndef LIBSHUNT_H_
#define LIBSHUNT_H_
#include "ntl/ntl.h"
#include "nil/nil.h"
/// @brief Success
#define LIBSHUNT_OK (0)
/// @brief Unspecified error
#define LIBSHUNT_ERR (1)
/// @brief Invalid argument
#define LIBSHUNT_EINVAL (2)
/// @brief Resource or entry not found
#define LIBSHUNT_ENOTFOUND (3)
/// @brief Resource in use
#define LIBSHUNT_EBUSY (4)
/// @brief Resource temporarily unavailable
#define LIBSHUNT_EAGAIN (5)
/// @brief Could not allocate memory
#define LIBSHUNT_ENOMEM (6)
/// @brief Packet failed integrity checks
#define LIBSHUNT_BAD_PACKET (7)
typedef struct {
nil_addr_t addr;
ntl_port_t port;
} shunt_addr_t;
#endif

View File

@@ -0,0 +1,57 @@
// shunt_client.h - Public interface of libshunt client side
#ifndef LIBSHUNT_CLIENT_H_
#define LIBSHUNT_CLIENT_H_
#include <stddef.h>
#include "FreeRTOS.h"
#include "libshunt/libshunt.h"
#include "nil/nil.h"
#include "ntl/ntl.h"
typedef struct shunt_cli_t shunt_cli_t;
/// @brief Initializes an libshunt client
/// @param port The port number to assign to the caller
/// @return LIBSHUNT_OK on success, an error otherwise
int shunt_init_client(shunt_cli_t **cli, ntl_port_t port);
/// @brief Attempts to get a free message buffer from the central storage space
/// @param cli Current client handle
/// @param buffer_ptr Location to return the buffer
/// @param length The minimum size, in bytes, of the buffer get
/// @return LIBSHUNT_OK on success, an error otherwise
int shunt_get_msg_buffer(shunt_cli_t *cli, uint8_t **buffer_ptr, size_t length);
/// @brief Releases a buffer received from shunt_await_msg
/// @param cli Current client handle
/// @param buffer The buffer to release
/// @return LIBSHUNT_OK on success, an error otherwise
int shunt_release_msg_buffer(shunt_cli_t *cli, uint8_t *buffer);
/// @brief Attempts to send a message
/// @param cli Current client handle
/// @param buffer The message buffer (gets consumed)
/// @param length The message length
/// @param dst_addr The destination address
/// @param max_delay The maximum time to wait for a resources, in ticks
/// @return LIBSHUNT_OK on success, an error otherwise
int shunt_send_msg(shunt_cli_t *cli, uint8_t *buffer, size_t length, shunt_addr_t dst, 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 LIBSHUNT_OK on success, an error otherwise
int shunt_await_msg(shunt_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, shunt_addr_t *source_ptr, TickType_t max_delay);
#endif

View File

@@ -0,0 +1,19 @@
// shunt_task.h - Public interface to launch the libshunt FreeRTOS task
#ifndef LIBSHUNT_TASK_H_
#define LIBSHUNT_TASK_H_
#include "FreeRTOS.h"
#include "task.h"
#include "FreeRTOSConfig.h"
#define LIBSHUNT_TASK_STACK_DEPTH configMINIMAL_STACK_SIZE
#define LIBSHUNT_TASK_PRIORITY (configMAX_PRIORITIES - 1)
extern StackType_t _libshunt_task_stack[LIBSHUNT_TASK_STACK_DEPTH];
extern StaticTask_t _libshunt_task;
void libshunt_task_preinit();
void libshunt_task_main(void *params);
#endif

View File

@@ -0,0 +1,129 @@
#include <stdbool.h>
#include "libshunt/shunt_client.h"
#include "libshunt/shunt_task.h"
#include "libshunt/libshunt.h"
#include "shunt_config.h"
#include "shunt_shared.h"
#include "ntl/ntl.h"
#include "nil/nil.h"
#include "message_buffer.h"
int shunt_init_client(shunt_cli_t **cli, ntl_port_t port)
{
return register_client(cli, port, xTaskGetCurrentTaskHandle());
}
int shunt_get_msg_buffer(shunt_cli_t *cli, uint8_t **buffer_ptr, size_t length)
{
if (cli == NULL)
return LIBSHUNT_EINVAL;
if (buffer_ptr == NULL)
return LIBSHUNT_EINVAL;
// Request extra space for NIL/NTL headers
size_t extended_len = sizeof(nil_header_t) + sizeof(ntl_header_t) + length;
uint8_t *buf;
int status = get_msg_buffer(&buf, extended_len);
if (status != LIBSHUNT_OK)
return status;
// Reserve space for NIL/NTL headers
buf += sizeof(nil_header_t) + sizeof(ntl_header_t);
// Return buffer
*buffer_ptr = buf;
return LIBSHUNT_OK;
}
int shunt_release_msg_buffer(shunt_cli_t *cli, uint8_t *buffer)
{
if (cli == NULL)
return LIBSHUNT_EINVAL;
if (buffer == NULL)
return LIBSHUNT_EINVAL;
// cli requested for interface consistency only (for now)
(void)cli;
// Unwind buffer pointer
uint8_t *unwound = buffer - sizeof(nil_header_t) + sizeof(ntl_header_t);
// Release correct buffer pointer
return release_msg_buffer(unwound);
}
int shunt_send_msg(shunt_cli_t *cli, uint8_t *buffer, size_t length, shunt_addr_t dst, TickType_t max_delay)
{
if (cli == NULL)
return LIBSHUNT_EINVAL;
if (buffer == NULL)
return LIBSHUNT_EINVAL;
if (dst.port == NTL_PORT_EMPTY)
return LIBSHUNT_EINVAL;
// Unwind buffer pointer
ntl_packet_t *ntl_packet = (ntl_packet_t*)((uint8_t*)buffer - sizeof(ntl_header_t));
nil_packet_t *nil_packet = (nil_packet_t*)((uint8_t*)ntl_packet - sizeof(ntl_header_t));
// Populate NTL header
ntl_build_header(&ntl_packet->header, dst.port, cli->port, length);
// Populate NIL header
nil_build_header(&nil_packet->header, dst.addr, LIBSHUNT_LOCAL_NIL_ADDR, length + sizeof(ntl_header_t));
// Pass to ingest
return enqueue_message_ingest((uint8_t *)nil_packet, max_delay);
}
int shunt_await_msg(shunt_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, shunt_addr_t *source_ptr, TickType_t max_delay)
{
if (cli == NULL)
return LIBSHUNT_EINVAL;
if (buffer_ptr == NULL)
return LIBSHUNT_EINVAL;
if (length_ptr == NULL)
return LIBSHUNT_EINVAL;
if (source_ptr == NULL)
return LIBSHUNT_EINVAL;
// cli requested for interface consistency only (for now)
(void)cli;
uint8_t *buffer;
int err = await_message_dispatch(&buffer, max_delay);
if (err != LIBSHUNT_OK)
{
return err;
}
// Extract header information
nil_packet_t *nil = extract_nil_packet(buffer, LIBSHUNT_MSG_BUFFER_SIZE);
if (nil == NULL)
{
return LIBSHUNT_BAD_PACKET;
}
ntl_packet_t *ntl = extract_ntl_packet(nil);
if (nil == NULL)
{
return LIBSHUNT_BAD_PACKET;
}
// Build source spec
shunt_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 LIBSHUNT_OK;
}

View File

@@ -0,0 +1,30 @@
// shunt_config.h - Default configurations for libshunt
#ifndef LIBSHUNT_CONFIG_H_
#define LIBSHUNT_CONFIG_H_
#ifndef LIBSHUNT_TASKINDEX_MAILBOX
/// @brief Index of direct-to-task notification for mailbox to use
#define LIBSHUNT_TASKINDEX_MAILBOX 1
#endif
#ifndef LIBSHUNT_MAX_CLIENTS
/// @brief Number of client slots to allocate
#define LIBSHUNT_MAX_CLIENTS 8
#endif
#ifndef LIBSHUNT_MSG_BUFFER_SIZE
/// @brief Message buffer size in bytes. Must allow space for NIL and NTL headers
#define LIBSHUNT_MSG_BUFFER_SIZE 32
#endif
#ifndef LIBSHUNT_MSG_BUFFER_COUNT
/// @brief Number of message buffers to allocate
#define LIBSHUNT_MSG_BUFFER_COUNT 8
#endif
#ifndef LIBSHUNT_LOCAL_NIL_ADDR
#error "Local NIL address must be specified with LIBSHUNT_LOCAL_NIL_ADDR"
#endif
#endif

View File

@@ -0,0 +1,311 @@
#include "shunt_shared.h"
#include <string.h>
#include <stdbool.h>
#include <stdint.h>
#include "FreeRTOS.h"
#include "semphr.h"
#include "task.h"
#include "libshunt/libshunt.h"
#include "shunt_config.h"
#include "nil/nil.h"
#if UINTPTR_MAX > UINT32_MAX
#error "Pointers of 32-bits or less are required for libshunt mailbox to function correctly."
#endif
// === Private variables ===
// Global lock
static SemaphoreHandle_t _libshunt_lock;
static StaticSemaphore_t _libshunt_lock_store;
// Clients
static shunt_cli_t _clients[LIBSHUNT_MAX_CLIENTS];
// Message buffers
// LIBSHUNT_MSG_BUFFER_COUNT buffers of LIBSHUNT_MSG_BUFFER_SIZE bytes
static uint8_t _msg_buffers[LIBSHUNT_MSG_BUFFER_COUNT][LIBSHUNT_MSG_BUFFER_SIZE];
static bool _msg_buffer_used[LIBSHUNT_MSG_BUFFER_COUNT];
// === Private function declarations ===
static bool global_lock(TickType_t max_delay);
static void global_unlock();
/// @remark Returns NULL if not found
static shunt_cli_t *find_empty_client();
/// @remark Returns NULL if not found
static shunt_cli_t *find_client_with_port(ntl_port_t port);
/// @remark Returns NULL if not found
static uint8_t *find_unused_buffer();
/// @remark Returns -1 if not found
static int find_index_of_buffer(uint8_t *buffer);
// === Public functions ===
int internal_init()
{
_libshunt_lock = xSemaphoreCreateMutexStatic(&_libshunt_lock_store);
memset(_clients, 0, sizeof(_clients));
memset(_msg_buffer_used, 0, sizeof(_msg_buffer_used));
return LIBSHUNT_OK;
}
int register_client(shunt_cli_t **cli_ptr, ntl_port_t port, TaskHandle_t task)
{
if (port == NTL_PORT_EMPTY)
return LIBSHUNT_EINVAL;
if (cli_ptr == NULL)
return LIBSHUNT_EINVAL;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return LIBSHUNT_EAGAIN;
// Prevent double binding of ports
if (find_client_with_port(port) != NULL)
{
global_unlock();
return LIBSHUNT_EBUSY;
}
shunt_cli_t *client = find_empty_client();
if (client == NULL)
{
global_unlock();
return LIBSHUNT_EBUSY;
}
// Populate client data
client->port = port;
client->task = task;
// Save to passed pointer
*cli_ptr = client;
global_unlock();
return LIBSHUNT_OK;
}
int get_msg_buffer(uint8_t **buffer_ptr, size_t len)
{
if (buffer_ptr == NULL)
return LIBSHUNT_EINVAL;
if (len > LIBSHUNT_MSG_BUFFER_SIZE)
return LIBSHUNT_ENOMEM;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return LIBSHUNT_EAGAIN;
//TODO: Allow buffer to be given by another layer (below NIL/NTL)
uint8_t *buf = find_unused_buffer();
if (buf == NULL)
{
global_unlock();
return LIBSHUNT_ENOMEM;
}
// Return buffer
*buffer_ptr = buf;
global_unlock();
return LIBSHUNT_OK;
}
int release_msg_buffer(uint8_t *buffer)
{
if (buffer == NULL)
return LIBSHUNT_EINVAL;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return LIBSHUNT_EAGAIN;
int index = find_index_of_buffer(buffer);
// Handle invalid buffers
if (index == -1)
{
global_unlock();
return LIBSHUNT_EINVAL;
}
// Handle not-in-use buffers
if (_msg_buffer_used[index] == false)
{
global_unlock();
return LIBSHUNT_EINVAL;
}
// Mark buffer as free
_msg_buffer_used[index] = false;
global_unlock();
return LIBSHUNT_OK;
}
int get_client_with_port(ntl_port_t port, shunt_cli_t **client_ptr)
{
if (port == NTL_PORT_EMPTY)
return LIBSHUNT_EINVAL;
if (client_ptr == NULL)
return LIBSHUNT_EINVAL;
// == Enter ==
if (!global_lock(portMAX_DELAY))
return LIBSHUNT_EAGAIN;
// Map port to client
shunt_cli_t *client = find_client_with_port(port);
if (client == NULL)
{
global_unlock();
return LIBSHUNT_ENOTFOUND;
}
// Return client
*client_ptr = client;
global_unlock();
return LIBSHUNT_OK;
}
int enqueue_message_ingest(uint8_t *buffer, TickType_t max_delay)
{
//NOTE: Does not need global lock
if (buffer == NULL)
return LIBSHUNT_EINVAL;
// Try to queue message
if (xQueueSendToBack(libshunt_ingest_queue, &buffer, max_delay) != pdPASS)
{
return LIBSHUNT_EAGAIN;
}
return LIBSHUNT_OK;
}
int dequeue_message_ingest(uint8_t **buffer_ptr, TickType_t max_delay)
{
//NOTE: Does not need global lock
if (buffer_ptr == NULL)
return LIBSHUNT_EINVAL;
uint8_t *buffer;
// Try to dequeue message
if (xQueueReceive(libshunt_ingest_queue, &buffer, max_delay) == pdFALSE)
{
return LIBSHUNT_EAGAIN;
}
// Return buffer
*buffer_ptr = buffer;
return LIBSHUNT_OK;
}
int do_message_dispatch(shunt_cli_t *cli, uint8_t *buffer)
{
//NOTE: Does not need global lock
if (cli == NULL)
return LIBSHUNT_EINVAL;
if (buffer == NULL)
return LIBSHUNT_EINVAL;
// Try to notify
if (xTaskNotifyIndexed(cli->task, LIBSHUNT_TASKINDEX_MAILBOX, (uint32_t)buffer, eSetValueWithoutOverwrite) == pdFALSE)
{
// Failed because previous message has not yet been processed
return LIBSHUNT_EAGAIN;
}
return LIBSHUNT_OK;
}
int await_message_dispatch(uint8_t **buffer_ptr, TickType_t max_delay)
{
//NOTE: Does not need global lock
if (buffer_ptr == NULL)
return LIBSHUNT_EINVAL;
// Get raw msg buffer
uint8_t *buffer;
if (xTaskNotifyWaitIndexed(LIBSHUNT_TASKINDEX_MAILBOX, 0, 0, (uint32_t*)&buffer, max_delay) == pdFALSE)
{
// Nothing pending
return LIBSHUNT_EAGAIN;
}
// Return buffer
*buffer_ptr = buffer;
return LIBSHUNT_OK;
}
// === Private function definitions ===
static bool global_lock(TickType_t max_delay)
{
return xSemaphoreTake(_libshunt_lock, max_delay) == pdTRUE;
}
static void global_unlock()
{
xSemaphoreGive(_libshunt_lock);
}
static shunt_cli_t *find_empty_client()
{
return find_client_with_port(NTL_PORT_EMPTY);
}
static shunt_cli_t *find_client_with_port(ntl_port_t port)
{
for (int i = 0; i < LIBSHUNT_MAX_CLIENTS; ++i)
{
shunt_cli_t *client = &_clients[i];
if (client->port == port)
return client;
}
return NULL;
}
static uint8_t *find_unused_buffer()
{
for (int i = 0; i < LIBSHUNT_MSG_BUFFER_COUNT; ++i)
{
if (!_msg_buffer_used[i])
return _msg_buffers[i];
}
return NULL;
}
static int find_index_of_buffer(uint8_t *buffer)
{
for (int i = 0; i < LIBSHUNT_MSG_BUFFER_COUNT; ++i)
{
if (buffer == _msg_buffers[i])
return i;
}
return -1;
}

View File

@@ -0,0 +1,129 @@
// shunt_sharedl.h - Internal mechanisms shared by libshunt components
#ifndef LIBSHUNT_SHARED_H_
#define LIBSHUNT_SHARED_H_
#include "FreeRTOS.h"
#include "task.h"
#include "queue.h"
#include "nil/nil.h"
#include "ntl/ntl.h"
#include "libshunt/libshunt.h"
struct shunt_cli_t {
/// @remark If set to NTL_PORT_EMPTY denotes an empty client
ntl_port_t port;
TaskHandle_t task;
};
// Queue<uint8_t*> (queue of buffers)
extern QueueHandle_t libshunt_ingest_queue;
#ifndef LIBSHUNT_CLIENT_H_
typedef struct shunt_cli_t shunt_cli_t;
#endif
/// @brief Initializes the internal state of libshunt
/// @return LIBSHUNT_OK on success, an error otherwise
int internal_init();
/// @brief Registers a task as an libshunt client
/// @param cli_ptr Location to return client
/// @param port The port to register to
/// @param task The task to register as the handler
/// @return LIBSHUNT_OK on success, an error otherwise
int register_client(shunt_cli_t **cli_ptr, ntl_port_t port, TaskHandle_t task);
/// @brief Attempts to get a free message buffer from the central storage space
/// @param buffer_ptr Location to return the buffer
/// @param len The minimum size, in bytes, of the buffer get
/// @return LIBSHUNT_OK on success, an error otherwise
int get_msg_buffer(uint8_t **buffer_ptr, size_t len);
/// @brief Releases a buffer previously acquired with get_msg_buffer
/// @param buffer The buffer to release
/// @return LIBSHUNT_OK on success, an error otherwise
int release_msg_buffer(uint8_t *buffer);
/// @brief Attempts to get the client assigned to the given port
/// @param port The port to look for
/// @param client_ptr Location to return the client
/// @return LIBSHUNT_OK on success, an error otherwise
int get_client_with_port(ntl_port_t port, shunt_cli_t **client_ptr);
/// @brief Enqueues a message for ingest by the libshunt task
/// @param buffer The buffer to enqueue
/// @param max_delay The maximum time to wait for resources, in ticks
/// @return LIBSHUNT_OK on success, an error otherwise
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 max_delay The maximum time to wait for a message, in ticks
/// @return LIBSHUNT_OK on success, an error otherwise
int dequeue_message_ingest(uint8_t **buffer_ptr, TickType_t max_delay);
/// @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 LIBSHUNT_OK on success, an error otherwise
int do_message_dispatch(shunt_cli_t *cli, uint8_t *buffer);
/// @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 LIBSHUNT_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

View File

@@ -0,0 +1,129 @@
#include "libshunt/shunt_task.h"
#include <stdint.h>
#include "FreeRTOS.h"
#include "queue.h"
#include "libshunt/libshunt.h"
#include "shunt_config.h"
#include "shunt_shared.h"
#include "ntl/ntl.h"
#include "nil/nil.h"
// === Static task space ===
StackType_t _libshunt_task_stack[LIBSHUNT_TASK_STACK_DEPTH];
StaticTask_t _libshunt_task;
// === Public variables ===
// Queue<nil_packet_t*> Packets awaiting to be accepted by libshunt
QueueHandle_t libshunt_ingest_queue;
// === Private variables ===
#define INGEST_QUEUE_LENGTH (16)
#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 int handle_nil_packet(nil_packet_t *nil);
static int handle_ntl_packet(nil_packet_t *nil, ntl_packet_t *ntl);
// === Public functions ===
void libshunt_task_preinit()
{
internal_init();
libshunt_ingest_queue =
xQueueCreateStatic(INGEST_QUEUE_LENGTH, INGEST_QUEUE_ITEM_SIZE, _ingest_queue_store, &_ingest_queue_static);
}
void libshunt_task_main(void *params)
{
(void)params;
int err;
while (1)
{
// Dequeue message
uint8_t *buffer;
err = dequeue_message_ingest(&buffer, portMAX_DELAY);
// Get NIL packet
nil_packet_t *nil = extract_nil_packet(buffer, LIBSHUNT_MSG_BUFFER_SIZE);
if (nil == NULL)
{
//TODO: Error
continue;
}
// Process or route
err = handle_nil_packet(nil);
if (err == LIBSHUNT_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 != LIBSHUNT_OK)
{
// Failed for some other reason
// Do not retry
continue;
}
}
}
// === Private functions ===
static int handle_nil_packet(nil_packet_t *nil)
{
if (nil->header.dst_addr == LIBSHUNT_LOCAL_NIL_ADDR
|| nil->header.dst_addr == NIL_ADDR_LOCALHOST)
{
ntl_packet_t *ntl = extract_ntl_packet(nil);
if (ntl == NULL)
{
//TODO: Error
return LIBSHUNT_BAD_PACKET;
}
return handle_ntl_packet(nil, ntl);
}
//TODO: Route to external interface
return LIBSHUNT_ERR;
}
static int handle_ntl_packet(nil_packet_t *nil, ntl_packet_t *ntl)
{
int err;
ntl_port_t port = ntl_get_dst(&ntl->header);
shunt_cli_t *client;
err = get_client_with_port(port, &client);
if (err != LIBSHUNT_OK)
{
//TODO: Error
return err;
}
return do_message_dispatch(client, (uint8_t*)nil);
}

View File

@@ -4,16 +4,20 @@
#include "task.h"
#include "tasks/heartbeat_task.h"
#include "incphub/incphub_task.h"
#include "tasks/sensor_task.h"
#include "tasks/requester_task.h"
#include "libshunt/shunt_task.h"
int my_main()
{
// Init
incphub_task_preinit();
libshunt_task_preinit();
// Create tasks
xTaskCreateStatic(heartbeat_task_main, "heartbeat", HEARTBEAT_TASK_STACK_DEPTH, NULL, HEARTBEAT_TASK_PRIORITY, _heartbeat_task_stack, &_heartbeat_task);
xTaskCreateStatic(incphub_task_main, "incphub", INCPHUB_TASK_STACK_DEPTH, NULL, INCPHUB_TASK_PRIORITY, _incphub_task_stack, &_incphub_task);
xTaskCreateStatic(sensor_task_main, "sensor", SENSOR_TASK_STACK_DEPTH, NULL, SENSOR_TASK_PRIORITY, _sensor_task_stack, &_sensor_task);
xTaskCreateStatic(requester_task_main, "requester", REQUESTER_TASK_STACK_DEPTH, NULL, REQUESTER_TASK_PRIORITY, _requester_task_stack, &_requester_task);
xTaskCreateStatic(libshunt_task_main, "shunt", LIBSHUNT_TASK_STACK_DEPTH, NULL, LIBSHUNT_TASK_PRIORITY, _libshunt_task_stack, &_libshunt_task);
// Run
vTaskStartScheduler();

View File

@@ -12,7 +12,7 @@ typedef uint32_t ntl_chk_t;
// === Special Port Values ===
// === Special Address Values ===
#define NIL_ADDR_LOCALHOST ((nil_addr_t)0)

View File

@@ -1,4 +1,6 @@
target_sources(${EXECUTABLE} PRIVATE
tasks.c
heartbeat_task.c
sensor_task.c
requester_task.c
)

View File

@@ -1,5 +1,5 @@
#ifndef _HEARTBEAT_TASK_H_
#define _HEARTBEAT_TASK_H_
#ifndef HEARTBEAT_TASK_H_
#define HEARTBEAT_TASK_H_
#include "FreeRTOS.h"
#include "task.h"

View File

@@ -0,0 +1,73 @@
#include "requester_task.h"
#include <stdio.h>
#include <string.h>
#include "stm32f4xx_hal.h"
#include "stm32f4xx_hal_uart.h"
#include "usart.h"
#include "libshunt/shunt_client.h"
StackType_t _requester_task_stack[REQUESTER_TASK_STACK_DEPTH];
StaticTask_t _requester_task;
#define RANDOM_COUNT 4
uint32_t random_values[RANDOM_COUNT] = { 411, 69, 420, 0xdeadbeef };
void requester_task_main(void *params)
{
(void)params;
TickType_t last_wake = xTaskGetTickCount();
int random_index = 0;
uint8_t *buffer;
size_t length;
shunt_addr_t source;
// Init libshunt client
shunt_cli_t *cli;
shunt_init_client(&cli, 70);
while (1)
{
// Avoid spamming the sensor task
xTaskDelayUntil(&last_wake, 50);
// Generate request
uint32_t request = random_values[random_index++];
random_index %= RANDOM_COUNT;
// Get buffer
shunt_get_msg_buffer(cli, &buffer, sizeof(uint32_t));
// Populate buffer
*(uint32_t*)buffer = request;
// Send to sendor task
shunt_addr_t destination = { .addr = NIL_ADDR_LOCALHOST, .port = 69 };
shunt_send_msg(cli, buffer, sizeof(uint32_t), destination, portMAX_DELAY);
// Await reply (assumes no other task will send to this one)
shunt_await_msg(cli, &buffer, &length, &source, portMAX_DELAY);
// Ignore messages of different size
if (length != sizeof(uint32_t))
continue;
// Read reply message
uint32_t num = *(uint32_t*)buffer;
// Release buffer
shunt_release_msg_buffer(cli, buffer);
// Profit (send to UART to ensure it worked)
char strbuf[32];
snprintf(strbuf, sizeof(strbuf), "%lu\r\n", num);
HAL_UART_Transmit(&huart2, (uint8_t*)strbuf, strlen(strbuf), 9999);
}
}

View File

@@ -0,0 +1,16 @@
#ifndef REQUSTER_TASK_H_
#define REQUSTER_TASK_H_
#include "FreeRTOS.h"
#include "task.h"
#include "FreeRTOSConfig.h"
#define REQUESTER_TASK_STACK_DEPTH 256
#define REQUESTER_TASK_PRIORITY (configMAX_PRIORITIES - 1)
extern StackType_t _requester_task_stack[REQUESTER_TASK_STACK_DEPTH];
extern StaticTask_t _requester_task;
void requester_task_main(void *params);
#endif

View File

@@ -0,0 +1,55 @@
#include "sensor_task.h"
#include "stm32f4xx_hal.h"
#include "stm32f4xx_hal_gpio.h"
#include "gpio.h"
#include "libshunt/shunt_client.h"
StackType_t _sensor_task_stack[SENSOR_TASK_STACK_DEPTH];
StaticTask_t _sensor_task;
void sensor_task_main(void *params)
{
(void)params;
uint8_t *buffer;
size_t length;
shunt_addr_t source;
// Init libshunt client
shunt_cli_t *cli;
shunt_init_client(&cli, 69);
while (1)
{
// Get a message (assume it is a uint32_t)
shunt_await_msg(cli, &buffer, &length, &source, portMAX_DELAY);
// Ignore messages of different size
if (length != sizeof(uint32_t))
continue;
// Read incoming message
uint32_t num = *(uint32_t*)buffer;
// Release buffer once read
shunt_release_msg_buffer(cli, buffer);
// Process message
num++;
// Request buffer for reply
shunt_get_msg_buffer(cli, &buffer, sizeof(uint32_t));
// Prepare reply message
*(uint32_t*)buffer = num;
// Send reply
shunt_send_msg(cli, buffer, sizeof(uint32_t), source, portMAX_DELAY);
}
}

View File

@@ -0,0 +1,16 @@
#ifndef SENSOR_TASK_H_
#define SENSOR_TASK_H_
#include "FreeRTOS.h"
#include "task.h"
#include "FreeRTOSConfig.h"
#define SENSOR_TASK_STACK_DEPTH configMINIMAL_STACK_SIZE
#define SENSOR_TASK_PRIORITY (configMAX_PRIORITIES - 1)
extern StackType_t _sensor_task_stack[SENSOR_TASK_STACK_DEPTH];
extern StaticTask_t _sensor_task;
void sensor_task_main(void *params);
#endif