Board to PC

components/rpc/CommunicationRouter.cpp

@@ -5,6 +5,7 @@
 #include "freertos/queue.h"
 
 CommunicationRouter::~CommunicationRouter() {
+    vTaskDelete(m_router_thread);
     vQueueDelete(m_tcp_rx_queue);
 }
 
@@ -19,3 +20,7 @@ CommunicationRouter::~CommunicationRouter() {
         std::cout << "callback" << std::endl;
     }
 }
+
+int CommunicationRouter::send_msg(char* buffer, const size_t length) const {
+    return this->m_tcp_server->send_msg(buffer, length);
+}

components/rpc/MessagingInterface.cpp

@@ -4,21 +4,31 @@
 
 #include "MessagingInterface.h"
 
+#include <ConfigManager.h>
 #include <freertos/queue.h>
 #include <freertos/semphr.h>
 
 #include "MPIMessageBuilder.h"
 
 MessagingInterface::~MessagingInterface() {
+    vQueueDelete(m_mpi_rx_queue);
+    vSemaphoreDelete(m_map_semaphore);
 
+    for (const auto [_tag, queue] : m_tag_to_queue) {
+        vQueueDelete(queue);
+    }
 }
 
 int MessagingInterface::send(char* buffer, int size, int destination, int tag, bool durable) {
+    Flatbuffers::MPIMessageBuilder builder;
+    const auto [mpi_buffer, mpi_size] = builder.build_mpi_message(Messaging::MessageType_PTP, ConfigManager::get_module_id(), destination, sequence_number++, durable, tag, std::vector<uint8_t>(buffer, buffer + size));
+
+    m_router->send_msg(static_cast<char *>(mpi_buffer), mpi_size);
     return 0;
 }
 
 int MessagingInterface::broadcast(char* buffer, int size, int root, bool durable) {
-
+    // todo: impl
     return 0;
 }
 

components/rpc/TCPServer.cpp

@@ -9,6 +9,7 @@
 #include "lwip/sockets.h"
 #include "lwip/sys.h"
 #include <lwip/netdb.h>
+#include <iostream>
 
 #include "TCPServer.h"
 
@@ -20,13 +21,12 @@
 //       - authenticate (don't just return true from the auth function)
 //       - tx from board
 
-TCPServer::TCPServer(const int port, QueueHandle_t rx_queue) {
+TCPServer::TCPServer(const int port, const 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;
 
@@ -37,7 +37,6 @@ TCPServer::TCPServer(const int port, QueueHandle_t rx_queue) {
 TCPServer::~TCPServer() {
     vTaskDelete(this->m_task);
     vTaskDelete(this->m_rx_task);
-    vTaskDelete(this->m_tx_task);
     vSemaphoreDelete(this->m_mutex);
 }
 
@@ -149,6 +148,7 @@ TCPServer::~TCPServer() {
 
                     if (len < 0) {
                         printf("Error occurred during receiving: errno %d\n", errno);
+                        to_remove.emplace_back(sock);
                     } else if (0 == len) {
                         printf("Connection closed\n");
                         close(sock);
@@ -198,3 +198,19 @@ bool TCPServer::authenticate_client(int sock) {
     // todo: authentication (wait for a passphrase from the client)
     return 0;
 }
+
+int TCPServer::send_msg(char *buffer, size_t length) const {
+    // todo: should we assign a unique rank to each pc?
+
+    if (!is_network_connected()) {
+        return -1;
+    }
+
+    for (const auto client_sock : m_clients) {
+        std::cout << "sending tcp" << std::endl;
+        send(client_sock, buffer, length, 0);
+    }
+
+    return 0;
+}
+

components/rpc/include/CommunicationRouter.h

@@ -28,6 +28,8 @@ public:
 
     [[noreturn]] static void router_thread(void *args);
 
+    int send_msg(char* buffer, size_t length) const;
+
     // 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;

components/rpc/include/TCPServer.h

@@ -16,6 +16,7 @@ class TCPServer : IRPCServer {
 public:
     TCPServer(int port, QueueHandle_t rx_queue);
     ~TCPServer();
+    int send_msg(char* buffer, size_t length) const;
 
 private:
     bool authenticate_client(int client_sock);
@@ -29,7 +30,6 @@ private:
 
     TaskHandle_t m_task;
     TaskHandle_t m_rx_task;
-    TaskHandle_t m_tx_task;
 
     QueueHandle_t m_rx_queue;