Fix shutdown functions

components/rpc/CommunicationRouter.cpp

@@ -34,13 +34,6 @@ CommunicationRouter::~CommunicationRouter() { vTaskDelete(m_router_thread); }
   const auto that = static_cast<CommunicationRouter *>(args);
 
   while (true) {
-        
-    if (std::chrono::system_clock::now() - that->m_last_leader_updated > std::chrono::seconds(15)) {
-        that->m_last_leader_updated = std::chrono::system_clock::now();
-        ESP_LOGI(TAG, "Updating leader");
-        that->update_leader();
-    }
-    
     for (uint8_t channel = 0;
          channel <
          MODULE_TO_NUM_CHANNELS_MAP[that->m_config_manager.get_module_type()];
@@ -95,7 +88,7 @@ void CommunicationRouter::update_leader() {
     } else if (this->m_leader == m_module_id) {
       m_pc_connection->disconnect();
       m_lossless_server->shutdown();
-      m_lossless_server->shutdown();
+      m_lossy_server->shutdown();
     }
   }
 

components/rpc/wireless/TCPServer.cpp

@@ -1,19 +1,19 @@
 #include <memory>
 
-#include "esp_log.h"
-#include "sys/param.h"
-#include "freertos/FreeRTOS.h"
-#include "freertos/task.h"
-#include "esp_event.h"
-#include "esp_netif.h"
-#include "lwip/err.h"
-#include "lwip/sockets.h"
-#include "lwip/sys.h"
-#include "lwip/netdb.h"
-#include "wireless/TCPServer.h"
 #include "bits/shared_ptr_base.h"
 #include "constants/app_comms.h"
 #include "constants/tcp.h"
+#include "esp_event.h"
+#include "esp_log.h"
+#include "esp_netif.h"
+#include "freertos/FreeRTOS.h"
+#include "freertos/task.h"
+#include "lwip/err.h"
+#include "lwip/netdb.h"
+#include "lwip/sockets.h"
+#include "lwip/sys.h"
+#include "sys/param.h"
+#include "wireless/TCPServer.h"
 
 #define TAG "TCPServer"
 
@@ -23,7 +23,9 @@
 //       - authenticate (don't just return true from the auth function)
 //       - tx from board
 
-TCPServer::TCPServer(const int port, const std::shared_ptr<PtrQueue<std::vector<uint8_t>>>& rx_queue) {
+TCPServer::TCPServer(
+    const int port,
+    const std::shared_ptr<PtrQueue<std::vector<uint8_t>>> &rx_queue) {
   this->m_port = port;
   this->m_mutex = xSemaphoreCreateMutex();
   this->m_clients = std::unordered_set<int>();
@@ -41,11 +43,13 @@ TCPServer::~TCPServer() {
 void TCPServer::startup() {
   ESP_LOGI(TAG, "Starting TCP server on port %d", this->m_port);
   if (nullptr != this->m_task || nullptr != this->m_rx_task) {
-        ESP_LOGW(TAG, "Attempted to start TCP server when already started, ignoring start request");
+    ESP_LOGW(TAG, "Attempted to start TCP server when already started, "
+                  "ignoring start request");
     return;
   }
 
-    xTaskCreate(tcp_server_task, "tcp_accept_server", 3072, this, 5, &this->m_task);
+  xTaskCreate(tcp_server_task, "tcp_accept_server", 3072, this, 5,
+              &this->m_task);
   xTaskCreate(socket_monitor_thread, "tcp_rx", 4096, this, 5, &this->m_rx_task);
 }
 
@@ -54,6 +58,8 @@ void TCPServer::shutdown() {
   if (nullptr != this->m_task) {
     vTaskDelete(this->m_task);
     close(this->m_server_sock);
+    this->m_task = nullptr;
+    this->m_server_sock = -1;
   }
 
   if (nullptr != this->m_rx_task) {
@@ -61,6 +67,8 @@ void TCPServer::shutdown() {
     for (const auto sock : this->m_clients) {
       close(sock);
     }
+    this->m_rx_task = nullptr;
+    this->m_clients.clear();
   }
 }
 
@@ -70,7 +78,7 @@ void TCPServer::shutdown() {
   constexpr int keepInterval = KEEPALIVE_INTERVAL;
   constexpr int keepCount = KEEPALIVE_COUNT;
 
-    const auto that = static_cast<TCPServer*>(args);
+  const auto that = static_cast<TCPServer *>(args);
 
   while (true) {
     ESP_LOGI(TAG, "Attempting to start TCP Server on port %d", that->m_port);
@@ -83,18 +91,22 @@ void TCPServer::shutdown() {
     }
 
     constexpr int opt = 1;
-        setsockopt(that->m_server_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+    setsockopt(that->m_server_sock, SOL_SOCKET, SO_REUSEADDR, &opt,
+               sizeof(opt));
     ESP_LOGI(TAG, "Socket created");
 
     sockaddr_in server_addr = {
         .sin_family = AF_INET,
         .sin_port = htons(that->m_port),
-            .sin_addr = {
+        .sin_addr =
+            {
                 .s_addr = htonl(INADDR_ANY),
             },
     };
 
-        int err = bind(that->m_server_sock, reinterpret_cast<struct sockaddr *>(&server_addr), sizeof(server_addr));
+    int err = bind(that->m_server_sock,
+                   reinterpret_cast<struct sockaddr *>(&server_addr),
+                   sizeof(server_addr));
     if (0 != err) {
       ESP_LOGE(TAG, "Socket unable to bind: errno %d\n", errno);
       close(that->m_server_sock);
@@ -117,7 +129,9 @@ void TCPServer::shutdown() {
     while (is_network_connected()) {
       sockaddr_in client_addr{};
       socklen_t addr_len = sizeof(client_addr);
-            int client_sock = accept(that->m_server_sock, reinterpret_cast<struct sockaddr *>(&client_addr), &addr_len);
+      int client_sock =
+          accept(that->m_server_sock,
+                 reinterpret_cast<struct sockaddr *>(&client_addr), &addr_len);
       if (client_sock < 0) {
         ESP_LOGE(TAG, "Unable to accept TCP connection: errno %d\n", errno);
         continue;
@@ -128,10 +142,14 @@ void TCPServer::shutdown() {
         ESP_LOGE(TAG, "Client failed authentication\n");
       }
 
-            setsockopt(client_sock, SOL_SOCKET, SO_KEEPALIVE, &keepAlive, sizeof(int));
+      setsockopt(client_sock, SOL_SOCKET, SO_KEEPALIVE, &keepAlive,
-            setsockopt(client_sock, IPPROTO_TCP, TCP_KEEPIDLE, &keepIdle, sizeof(int));
+                 sizeof(int));
-            setsockopt(client_sock, IPPROTO_TCP, TCP_KEEPINTVL, &keepInterval, sizeof(int));
+      setsockopt(client_sock, IPPROTO_TCP, TCP_KEEPIDLE, &keepIdle,
-            setsockopt(client_sock, IPPROTO_TCP, TCP_KEEPCNT, &keepCount, sizeof(int));
+                 sizeof(int));
+      setsockopt(client_sock, IPPROTO_TCP, TCP_KEEPINTVL, &keepInterval,
+                 sizeof(int));
+      setsockopt(client_sock, IPPROTO_TCP, TCP_KEEPCNT, &keepCount,
+                 sizeof(int));
 
       xSemaphoreTake(that->m_mutex, portMAX_DELAY);
       that->m_clients.emplace(client_sock);
@@ -159,7 +177,8 @@ void TCPServer::shutdown() {
     }
     for (const auto sock : that->m_clients) {
       FD_SET(sock, &readfds);
-            if (sock > max_fd) max_fd = sock;
+      if (sock > max_fd)
+        max_fd = sock;
     }
     xSemaphoreGive(that->m_mutex);
 
@@ -177,7 +196,9 @@ void TCPServer::shutdown() {
 
           uint32_t msg_size = 0;
           if (int len = recv(sock, &msg_size, 4, MSG_WAITALL); len < 0) {
-                        ESP_LOGE(TAG, "Error occurred during receiving msg length: errno %d\n", errno);
+            ESP_LOGE(TAG,
+                     "Error occurred during receiving msg length: errno %d\n",
+                     errno);
             to_remove.emplace_back(sock);
             continue;
           } else if (0 == len) {
@@ -194,7 +215,8 @@ void TCPServer::shutdown() {
           auto buffer = std::make_unique<std::vector<uint8_t>>();
           buffer->resize(MIN(MAX_RX_BUFFER_SIZE, msg_size));
 
-                    if (int len = recv(sock, buffer->data(), msg_size, MSG_WAITALL); len < 0) {
+          if (int len = recv(sock, buffer->data(), msg_size, MSG_WAITALL);
+              len < 0) {
             ESP_LOGE(TAG, "Error occurred during receiving: errno %d\n", errno);
             to_remove.emplace_back(sock);
           } else if (0 == len) {

components/rpc/wireless/UDPServer.cpp

@@ -1,21 +1,20 @@
-#include <memory>
 #include <cstring>
+#include <memory>
 
-#include "esp_log.h"
-#include "sys/param.h"
-#include "freertos/FreeRTOS.h"
-#include "freertos/task.h"
-#include "esp_event.h"
-#include "esp_netif.h"
-#include "lwip/err.h"
-#include "lwip/sockets.h"
-#include "lwip/sys.h"
-#include "lwip/netdb.h"
-#include "wireless/UDPServer.h"
 #include "bits/shared_ptr_base.h"
 #include "constants/app_comms.h"
 #include "constants/udp.h"
+#include "esp_event.h"
+#include "esp_log.h"
 #include "esp_netif.h"
+#include "freertos/FreeRTOS.h"
+#include "freertos/task.h"
+#include "lwip/err.h"
+#include "lwip/netdb.h"
+#include "lwip/sockets.h"
+#include "lwip/sys.h"
+#include "sys/param.h"
+#include "wireless/UDPServer.h"
 
 #define TAG "UDPServer"
 
@@ -23,7 +22,9 @@
 
 // todo: - authenticate
 
-UDPServer::UDPServer(const int rx_port, const int tx_port, const std::shared_ptr<PtrQueue<std::vector<uint8_t>>>& rx_queue) {
+UDPServer::UDPServer(
+    const int rx_port, const int tx_port,
+    const std::shared_ptr<PtrQueue<std::vector<uint8_t>>> &rx_queue) {
   this->m_rx_port = rx_port;
   this->m_tx_port = tx_port;
   this->m_rx_task = nullptr;
@@ -32,14 +33,13 @@ UDPServer::UDPServer(const int rx_port, const int tx_port, const std::shared_ptr
   this->m_tx_server_sock = 0;
 }
 
-UDPServer::~UDPServer() {
+UDPServer::~UDPServer() { this->shutdown(); }
-    this->shutdown();
-}
 
 void UDPServer::startup() {
   ESP_LOGI(TAG, "Starting UDP server on port %d", this->m_rx_port);
   if (nullptr != this->m_rx_task) {
-        ESP_LOGW(TAG, "Attempted to start UDP server when already started, ignoring start request");
+    ESP_LOGW(TAG, "Attempted to start UDP server when already started, "
+                  "ignoring start request");
     return;
   }
 
@@ -50,13 +50,18 @@ void UDPServer::shutdown() {
   ESP_LOGI(TAG, "Shutting down UDP server");
   if (nullptr != this->m_rx_task) {
     vTaskDelete(this->m_rx_task);
+    close(this->m_rx_server_sock);
+    close(this->m_tx_server_sock);
+    this->m_rx_task = nullptr;
+    this->m_rx_server_sock = -1;
+    this->m_tx_server_sock = -1;
   }
 }
 
 [[noreturn]] void UDPServer::socket_monitor_thread(void *args) {
   const auto that = static_cast<UDPServer *>(args);
 
-    while(true) {
+  while (true) {
     ESP_LOGI(TAG, "Attempting to start UDP Server on %d", that->m_rx_port);
 
     if (!is_network_connected()) {
@@ -85,7 +90,8 @@ void UDPServer::shutdown() {
     }
 
     int reuse = 1;
-        if (setsockopt(that->m_rx_server_sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
+    if (setsockopt(that->m_rx_server_sock, SOL_SOCKET, SO_REUSEADDR, &reuse,
+                   sizeof(reuse)) < 0) {
       ESP_LOGE(TAG, "Failed to set SO_REUSEADDR. Error %d", errno);
       close(that->m_rx_server_sock);
       close(that->m_tx_server_sock);
@@ -98,7 +104,8 @@ void UDPServer::shutdown() {
     saddr.sin_family = AF_INET;
     saddr.sin_port = htons(that->m_rx_port);
     saddr.sin_addr.s_addr = htonl(INADDR_ANY);
-        int ret = bind(that->m_rx_server_sock, (struct sockaddr *)&saddr, sizeof(struct sockaddr_in));
+    int ret = bind(that->m_rx_server_sock, (struct sockaddr *)&saddr,
+                   sizeof(struct sockaddr_in));
     if (ret < 0) {
       ESP_LOGE(TAG, "Failed to bind socket. Error %d", errno);
       close(that->m_rx_server_sock);
@@ -112,7 +119,8 @@ void UDPServer::shutdown() {
     struct ip_mreq imreq = {};
     imreq.imr_multiaddr.s_addr = inet_addr(RECV_MCAST);
     imreq.imr_interface.s_addr = htonl(INADDR_ANY);
-        if(setsockopt(that->m_rx_server_sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &imreq, sizeof(struct ip_mreq)) < 0) {
+    if (setsockopt(that->m_rx_server_sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,
+                   &imreq, sizeof(struct ip_mreq)) < 0) {
       ESP_LOGE(TAG, "Failed to set IP_ADD_MEMBERSHIP. Error %d", errno);
       close(that->m_rx_server_sock);
       close(that->m_tx_server_sock);
@@ -123,17 +131,19 @@ void UDPServer::shutdown() {
     }
 
     uint32_t msg_size;
-        while(is_network_connected()) {
+    while (is_network_connected()) {
       auto buffer = std::make_unique<std::vector<uint8_t>>();
       buffer->resize(MAX_RX_BUFFER_SIZE + 4);
 
       ESP_LOGI(TAG, "Before rx");
-            if (int len = recvfrom(that->m_rx_server_sock, buffer->data(), MAX_RX_BUFFER_SIZE, 0, nullptr, nullptr); len < 0) {
+      if (int len = recvfrom(that->m_rx_server_sock, buffer->data(),
+                             MAX_RX_BUFFER_SIZE, 0, nullptr, nullptr);
+          len < 0) {
         ESP_LOGE(TAG, "Error occurred during receiving: errno %d", errno);
       } else if (len < 4 || len > MAX_RX_BUFFER_SIZE) {
         ESP_LOGE(TAG, "Got illegal message size");
       } else {
-            	msg_size = *reinterpret_cast<uint32_t*>(buffer->data());
+        msg_size = *reinterpret_cast<uint32_t *>(buffer->data());
         if (msg_size > len - 4) {
           ESP_LOGW(TAG, "Message size incorrect");
           continue;
@@ -189,9 +199,7 @@ int UDPServer::send_msg(char *buffer, const uint32_t length) const {
   sockaddr_in mcast_dest = {
       .sin_family = AF_INET,
       .sin_port = htons(m_tx_port),
-    	.sin_addr = {
+      .sin_addr = {.s_addr = inet_addr(SEND_MCAST)},
-			.s_addr = inet_addr(SEND_MCAST)
-		},
   };
 
   uint32_t size = length;