MPI

components/rpc/CMakeLists.txt

@@ -1,3 +1,3 @@
-idf_component_register(SRCS "rpc.cpp" "WifiManager.cpp" "mDNSDiscoveryService.cpp" "TCPServer.cpp"
-                       PRIV_REQUIRES driver esp_event nvs_flash esp_netif esp_wifi espressif__mdns constants config
+idf_component_register(SRCS "WifiManager.cpp" "mDNSDiscoveryService.cpp" "TCPServer.cpp" "CommunicationRouter.cpp" "MessagingInterface.cpp"
+                       PRIV_REQUIRES driver esp_event nvs_flash esp_netif esp_wifi espressif__mdns constants config flatbuffers
                        INCLUDE_DIRS "include")

components/rpc/CommunicationRouter.cpp

@@ -0,0 +1,19 @@
+#include "CommunicationRouter.h"
+
+#include <iostream>
+
+CommunicationRouter::~CommunicationRouter() {
+    vQueueDelete(m_tcp_rx_queue);
+}
+
+[[noreturn]] void CommunicationRouter::router_thread(void *args) {
+    const auto that = static_cast<CommunicationRouter *>(args);
+
+    // todo: change to queue set
+    char buffer[512];
+    while (true) {
+        xQueueReceive(that->m_tcp_rx_queue, buffer, portMAX_DELAY);
+        that->m_rx_callback(buffer, 512);
+        std::cout << "callback" << std::endl;
+    }
+}

components/rpc/MessagingInterface.cpp

@@ -2,3 +2,51 @@
 // Created by Johnathon Slightham on 2025-05-25.
 //
 
+#include "MessagingInterface.h"
+
+#include "MPIMessageBuilder.h"
+
+MessagingInterface::~MessagingInterface() {
+
+}
+
+int MessagingInterface::send(char* buffer, int size, int destination, int tag, bool durable) {
+    return 0;
+}
+
+int MessagingInterface::broadcast(char* buffer, int size, int root, bool durable) {
+
+    return 0;
+}
+
+int MessagingInterface::recv(char* buffer, int size, int source, const int tag) {
+    checkOrInsertTag(tag);
+
+    // todo: the buffer needs to be large enough, this copies into the buffer...
+    // todo: handle the source
+    xQueueReceive(m_tag_to_queue.at(tag), buffer, portMAX_DELAY);
+
+    return 0;
+}
+
+int MessagingInterface::sendrecv(char* send_buffer, int send_size, int dest, int send_tag, char* recv_buffer, int recv_size, int recv_tag) {
+
+    return 0;
+}
+
+// todo: when handleRecv returns, remove from queue (from router)
+void MessagingInterface::handleRecv(const char* recv_buffer, int recv_size) {
+    const auto mpi_message = Flatbuffers::MPIMessageBuilder::parse_mpi_message(reinterpret_cast<const uint8_t *>(recv_buffer));
+
+    checkOrInsertTag(mpi_message->tag());
+
+    xQueueSendToBack(m_tag_to_queue.at(mpi_message->tag()), mpi_message->payload(), 0);
+}
+
+void MessagingInterface::checkOrInsertTag(const uint8_t tag) {
+    xSemaphoreTake(m_map_semaphore, portMAX_DELAY);
+    if (!m_tag_to_queue.contains(tag)) {
+        m_tag_to_queue[tag] = xQueueCreate(MPI_QUEUE_SIZE, MAX_MPI_BUFFER_SIZE);
+    }
+    xSemaphoreGive(m_map_semaphore);
+}

components/rpc/TCPServer.cpp

@@ -20,15 +20,17 @@
 //       - authenticate (don't just return true from the auth function)
 //       - tx from board
 
-TCPServer::TCPServer(const int port) {
+TCPServer::TCPServer(const int port, QueueHandle_t rx_queue) {
     this->m_port = port;
     this->m_mutex = xSemaphoreCreateMutex();
     this->m_clients = std::unordered_set<int>();
     this->m_task = nullptr;
     this->m_rx_task = nullptr;
     this->m_tx_task = nullptr;
+    this->m_rx_queue = rx_queue;
+    this->m_server_sock = 0;
 
-    xTaskCreate(tcp_server_task, "tcp_accept_server", 4096, this, 5, &this->m_task);
+    xTaskCreate(tcp_server_task, "tcp_accept_server", 2048, this, 5, &this->m_task);
     xTaskCreate(socket_monitor_thread, "tcp_rx", 4096, this, 5, &this->m_rx_task);
 }
 
@@ -45,7 +47,7 @@ TCPServer::~TCPServer() {
     constexpr int keepInterval = KEEPALIVE_INTERVAL;
     constexpr int keepCount = KEEPALIVE_COUNT;
 
-    auto that = static_cast<TCPServer*>(args);
+    const auto that = static_cast<TCPServer*>(args);
 
     while (true) {
         printf("Attempting to start TCP Server on port %d", that->m_port);
@@ -119,7 +121,7 @@ TCPServer::~TCPServer() {
 }
 
 [[noreturn]] void TCPServer::socket_monitor_thread(void *args) {
-    auto that = static_cast<TCPServer *>(args);
+    const auto that = static_cast<TCPServer *>(args);
 
     while (true) {
         fd_set readfds;
@@ -152,9 +154,8 @@ TCPServer::~TCPServer() {
                         close(sock);
                         to_remove.emplace_back(sock);
                     } else {
-                        // todo: send to rx queue instead of printing
-                        buffer[len] = 0; // temp: Null-terminate whatever is received and treat it like a string
-                        printf("TCP Server Received %d bytes: %s\n", len, buffer);
+                        printf("TCP Server Received %d bytes\n", len);
+                        xQueueSendToBack(that->m_rx_queue, buffer, 0);
                     }
                 }
             }

components/rpc/WifiManager.cpp

@@ -21,7 +21,7 @@ WifiManager::WifiManager() {
     this->m_task = nullptr;
     this->m_netif = nullptr;
 
-    xTaskCreate(reinterpret_cast<TaskFunction_t>(s_manage), "wifi_task", 4096, this, 5, &m_task);
+    xTaskCreate(reinterpret_cast<TaskFunction_t>(s_manage), "wifi_task", 3096, this, 5, &m_task);
 }
 
 WifiManager::~WifiManager() {

components/rpc/include/CommunicationRouter.h

@@ -5,12 +5,34 @@
 #ifndef COMMUNICATIONROUTER_H
 #define COMMUNICATIONROUTER_H
 
+#include <functional>
+#include <memory>
 
+#include "TCPServer.h"
+#include "constants/tcp.h"
+#include "constants/app_comms.h"
 
-class CommunicationRouter : ICommunicationRouter {
+class CommunicationRouter {
+public:
+    explicit CommunicationRouter(const std::function<void(char*, int)> &rx_callback)
+        : m_tcp_rx_queue(xQueueCreate(RX_QUEUE_SIZE, MAX_RX_BUFFER_SIZE)),
+            m_rx_callback(rx_callback),
+            m_tcp_server(std::make_unique<TCPServer>(TCP_PORT, m_tcp_rx_queue)) {
+        xTaskCreate(router_thread, "communication_router", 2048, this, 3, &this->m_router_thread);
+    }
+
+    ~CommunicationRouter();
+
+    [[noreturn]] static void router_thread(void *args);
+
+    // todo: does this really need to be here (so i can access from thread)?
+    QueueHandle_t m_tcp_rx_queue;
+    std::function<void(char*, int)> m_rx_callback;
+private:
+
+    TaskHandle_t m_router_thread;
+    std::unique_ptr<TCPServer> m_tcp_server;
 
 };
 
-
-
 #endif //COMMUNICATIONROUTER_H

components/rpc/include/ICommunicationRouter.h

@@ -1,12 +0,0 @@
-//
-// Created by Johnathon Slightham on 2025-05-25.
-//
-
-#ifndef ICOMMUNICATIONROUTER_H
-#define ICOMMUNICATIONROUTER_H
-
-class ICommunicationRouter {
-
-}
-
-#endif //ICOMMUNICATIONROUTER_H

components/rpc/include/MessagingInterface.h

@@ -5,12 +5,35 @@
 #ifndef MESSAGINGINTERFACE_H
 #define MESSAGINGINTERFACE_H
 
+#include <memory>
+#include <unordered_map>
 
+#include "CommunicationRouter.h"
 
 class MessagingInterface {
+public:
+    MessagingInterface()
+        : m_mpi_rx_queue(xQueueCreate(MAX_RX_BUFFER_SIZE, RX_QUEUE_SIZE)),
+            m_router(std::make_unique<CommunicationRouter>([this](char* buffer, const int size) { handleRecv(buffer, size); })),
+            m_map_semaphore(xSemaphoreCreateMutex()) {};
 
+    ~MessagingInterface();
+
+    int send(char* buffer, int size, int destination, int tag, bool durable);
+    int broadcast(char* buffer, int size, int root, bool durable);
+    int recv(char* buffer, int size, int source, int tag);
+    int sendrecv(char* send_buffer, int send_size, int dest, int send_tag, char* recv_buffer, int recv_size, int recv_tag);
+
+private:
+    void handleRecv(const char* recv_buffer, int recv_size);
+
+    void checkOrInsertTag(uint8_t tag);
+
+    uint16_t sequence_number = 0;
+    QueueHandle_t m_mpi_rx_queue; // todo: maybe move this down classes more
+    std::unique_ptr<CommunicationRouter> m_router;
+    SemaphoreHandle_t m_map_semaphore;
+    std::unordered_map<uint8_t, QueueHandle_t> m_tag_to_queue;
 };
 
-
-
 #endif //MESSAGINGINTERFACE_H

components/rpc/include/TCPServer.h

@@ -14,7 +14,7 @@
 
 class TCPServer : IRPCServer {
 public:
-    explicit TCPServer(int port);
+    TCPServer(int port, QueueHandle_t rx_queue);
     ~TCPServer();
 
 private:
@@ -31,6 +31,8 @@ private:
     TaskHandle_t m_rx_task;
     TaskHandle_t m_tx_task;
 
+    QueueHandle_t m_rx_queue;
+
     SemaphoreHandle_t m_mutex;
     std::unordered_set<int> m_clients;
 };

components/rpc/include/rpc.h

@@ -1 +0,0 @@
-void func(void);

components/rpc/rpc.cpp

@@ -1,7 +0,0 @@
-#include <stdio.h>
-#include "rpc.h"
-
-void func(void)
-{
-
-}