fix(wifi_remote): Fix server event/command race condtion using eventfd

Author: david-cermak

components/esp_wifi_remote/CMakeLists.txt

@@ -14,7 +14,7 @@ idf_component_register(INCLUDE_DIRS include
                          ${src_wifi_is_remote}
                     PRIV_INCLUDE_DIRS eppp
                     REQUIRES esp_event esp_netif
-                    PRIV_REQUIRES esp_wifi esp-tls)
+                    PRIV_REQUIRES esp_wifi esp-tls vfs)
 
 idf_component_get_property(wifi esp_wifi COMPONENT_LIB)
 target_link_libraries(${wifi} PUBLIC ${COMPONENT_LIB})

components/esp_wifi_remote/eppp/wifi_remote_rpc_impl.hpp

@@ -129,6 +129,15 @@ class RpcEngine {
         return ESP_OK;
     }
 
+    int get_socket_fd()
+    {
+        int sock;
+        if (esp_tls_get_conn_sockfd(tls_, &sock) != ESP_OK) {
+            return -1;
+        }
+        return sock;
+    }
+
     RpcHeader get_header()
     {
         RpcHeader header{};

components/esp_wifi_remote/eppp/wifi_remote_rpc_params.h

@@ -16,7 +16,7 @@ struct esp_wifi_remote_mac_t {
 };
 
 struct esp_wifi_remote_eppp_ip_event {
-    uint32_t id;
+    int32_t id;
     esp_netif_ip_info_t wifi_ip;
     esp_netif_ip_info_t ppp_ip;
     esp_netif_dns_info_t dns;

components/esp_wifi_remote/eppp/wifi_remote_rpc_server.cpp

@@ -15,6 +15,8 @@
 #include "eppp_link.h"
 #include "wifi_remote_rpc_params.h"
 #include "lwip/apps/snmp.h"
+#include "esp_vfs.h"
+#include "esp_vfs_eventfd.h"
 
 extern "C" esp_netif_t *wifi_remote_eppp_init(eppp_type_t role);
 
@@ -32,30 +34,70 @@ const unsigned char key[] = "-----BEGIN PRIVATE KEY-----\n" CONFIG_ESP_WIFI_REMO
 
 using namespace server;
 
+struct Events {
+    api_id type;
+    int32_t id;
+    esp_wifi_remote_eppp_ip_event *ip_data{nullptr};
+    bool clean_ip_data{true};
+    esp_err_t create_ip_data()
+    {
+        ip_data = new (std::nothrow) esp_wifi_remote_eppp_ip_event;
+        return ip_data ? ESP_OK : ESP_ERR_NO_MEM;
+    }
+    ~Events()
+    {
+        if (clean_ip_data) {
+            delete ip_data;
+        }
+    }
+};
+
 class Sync {
     friend class RpcInstance;
 public:
-    void lock()
+    esp_err_t put(Events &ev)
     {
-        xSemaphoreTake(mutex, portMAX_DELAY);
+        ESP_RETURN_ON_FALSE(xQueueSend(queue, &ev, pdMS_TO_TICKS(queue_timeout)), ESP_FAIL, TAG, "Failed to queue event %" PRIi32, ev.id);
+        ev.clean_ip_data = false; // IP data were successfully sent to the queue, will free manually after receiving from it
+        uint64_t event_queued = 1;
+        write(fd, &event_queued, sizeof(event_queued));     // trigger the wait loop that
+        return ESP_OK;
     }
-    void unlock()
+    Events get()
     {
-        xSemaphoreGive(mutex);
+        Events ev{};
+        if (!xQueueReceive(queue, &ev, 0)) {
+            ev.type = api_id::ERROR;
+        }
+        return ev;
     }
     esp_err_t init()
     {
-        mutex = xSemaphoreCreateMutex();
-        return mutex == nullptr ? ESP_ERR_NO_MEM : ESP_OK;
+        queue = xQueueCreate(max_items, sizeof(Events));
+        esp_vfs_eventfd_config_t config = ESP_VFS_EVENTD_CONFIG_DEFAULT();
+        esp_vfs_eventfd_register(&config);
+        fd = eventfd(0, EFD_SUPPORT_ISR);
+        return queue == nullptr || fd < 0 ? ESP_ERR_NO_MEM : ESP_OK;
     }
     ~Sync()
     {
-        if (mutex) {
-            vSemaphoreDelete(mutex);
+        if (queue) {
+            vQueueDelete(queue);
+        }
+        if (fd >= 0) {
+            close(fd);
         }
     }
+    int fd{-1};
+    // Used to trigger task by either an internal event or rpc command
+    static const int NONE = 0;
+    static const int ERROR = 1;
+    static const int EVENT = 2;
+    static const int RPC   = 4;
 private:
-    SemaphoreHandle_t mutex{nullptr};
+    QueueHandle_t  queue{nullptr};
+    const int max_items = 15;
+    const int queue_timeout = 200;
 };
 
 class RpcInstance {
@@ -70,7 +112,7 @@ class RpcInstance {
         ESP_RETURN_ON_ERROR(start_server(), TAG, "Failed to start RPC server");
         ESP_RETURN_ON_ERROR(rpc.init(), TAG, "Failed to init RPC engine");
         ESP_RETURN_ON_ERROR(esp_netif_napt_enable(netif), TAG, "Failed to enable NAPT");
-        ESP_RETURN_ON_ERROR(sync.init(), TAG, "Failed to init locks");
+        ESP_RETURN_ON_ERROR(sync.init(), TAG, "Failed to init event queue");
         ESP_RETURN_ON_ERROR(esp_event_handler_register(WIFI_EVENT, ESP_EVENT_ANY_ID, handler, this), TAG, "Failed to register event");
         ESP_RETURN_ON_ERROR(esp_event_handler_register(IP_EVENT, ESP_EVENT_ANY_ID, handler, this), TAG, "Failed to register event");
         return xTaskCreate(task, "server", 8192, this, 5, nullptr) == pdTRUE ? ESP_OK : ESP_FAIL;
@@ -109,25 +151,24 @@ class RpcInstance {
     esp_err_t wifi_event(int32_t id)
     {
         ESP_LOGI(TAG, "Received WIFI event %" PRIi32, id);
-        std::lock_guard<Sync> lock(sync);
-        ESP_RETURN_ON_ERROR(rpc.send(api_id::WIFI_EVENT, &id), TAG, "Failed to marshall WiFi event");
+        Events ev{api_id::WIFI_EVENT, id, nullptr};
+        ESP_RETURN_ON_ERROR(sync.put(ev), TAG, "Failed to queue WiFi event");
         return ESP_OK;
     }
     esp_err_t ip_event(int32_t id, ip_event_got_ip_t *ip_data)
     {
         ESP_LOGI(TAG, "Received IP event %" PRIi32, id);
-        esp_wifi_remote_eppp_ip_event ip_event{};
-        ip_event.id = id;
+        Events ev{api_id::IP_EVENT, id, nullptr};
         if (ip_data->esp_netif) {
-            // marshall additional data, only if netif available
-            ESP_RETURN_ON_ERROR(esp_netif_get_dns_info(ip_data->esp_netif, ESP_NETIF_DNS_MAIN, &ip_event.dns), TAG, "Failed to get DNS info");
-            ESP_LOGI(TAG, "Main DNS:" IPSTR, IP2STR(&ip_event.dns.ip.u_addr.ip4));
-            memcpy(&ip_event.wifi_ip, &ip_data->ip_info, sizeof(ip_event.wifi_ip));
-            ESP_RETURN_ON_ERROR(esp_netif_get_ip_info(netif, &ip_event.ppp_ip), TAG, "Failed to get IP info");
+            ESP_RETURN_ON_ERROR(ev.create_ip_data(), TAG, "Failed to allocate event data");
+            ev.ip_data->id = id;
+            ESP_RETURN_ON_ERROR(esp_netif_get_dns_info(ip_data->esp_netif, ESP_NETIF_DNS_MAIN, &ev.ip_data->dns), TAG, "Failed to get DNS info");
+            ESP_LOGI(TAG, "Main DNS:" IPSTR, IP2STR(&ev.ip_data->dns.ip.u_addr.ip4));
+            memcpy(&ev.ip_data->wifi_ip, &ip_data->ip_info, sizeof(ev.ip_data->wifi_ip));
+            ESP_RETURN_ON_ERROR(esp_netif_get_ip_info(netif, &ev.ip_data->ppp_ip), TAG, "Failed to get IP info");
             ESP_LOGI(TAG, "IP address:" IPSTR, IP2STR(&ip_data->ip_info.ip));
         }
-        std::lock_guard<Sync> lock(sync);
-        ESP_RETURN_ON_ERROR(rpc.send(api_id::IP_EVENT, &ip_event), TAG, "Failed to marshal IP event");
+        ESP_RETURN_ON_ERROR(sync.put(ev), TAG, "Failed to queue IP event");
         return ESP_OK;
     }
     static void handler(void *ctx, esp_event_base_t base, int32_t id, void *data)
@@ -140,11 +181,83 @@ class RpcInstance {
             instance->ip_event(id, ip_data);
         }
     }
+    int select()
+    {
+        struct timeval timeout = { .tv_sec = 1, .tv_usec = 0};
+        int rpc_sock = rpc.get_socket_fd();
+
+        ESP_RETURN_ON_FALSE(rpc_sock != -1, Sync::ERROR, TAG, "failed ot get rpc socket");
+        fd_set readset;
+        fd_set errset;
+        FD_ZERO(&readset);
+        FD_ZERO(&errset);
+        FD_SET(rpc_sock, &readset);
+        FD_SET(sync.fd, &readset);
+        FD_SET(rpc_sock, &errset);
+        int ret = ::select(std::max(rpc_sock, 5) + 1, &readset, nullptr, &errset, &timeout);
+        if (ret == 0) {
+            ESP_LOGV(TAG, "poll_read: select - Timeout before any socket was ready!");
+            return Sync::NONE;
+        }
+        if (ret < 0) {
+            ESP_LOGE(TAG, "select error: %d", errno);
+            return Sync::ERROR;
+        }
+        if (FD_ISSET(rpc_sock, &errset)) {
+            int sock_errno = 0;
+            uint32_t optlen = sizeof(sock_errno);
+            getsockopt(rpc_sock, SOL_SOCKET, SO_ERROR, &sock_errno, &optlen);
+            ESP_LOGE(TAG, "select failed, socket errno = %d", sock_errno);
+            return Sync::ERROR;
+        }
+        int result = Sync::NONE;
+        if (FD_ISSET(rpc_sock, &readset)) {
+            result |= Sync::RPC;
+        }
+        if (FD_ISSET(sync.fd, &readset)) {
+            result |= Sync::EVENT;
+        }
+        return result;
+    }
+    esp_err_t marshall_events()
+    {
+        api_id type;
+        do {
+            Events ev = sync.get();
+            type = ev.type;
+            if (ev.type == api_id::WIFI_EVENT) {
+                ESP_RETURN_ON_ERROR(rpc.send(api_id::WIFI_EVENT, &ev.id), TAG, "Failed to marshall WiFi event");
+            } else if (ev.type == api_id::IP_EVENT && ev.ip_data) {
+                ESP_RETURN_ON_ERROR(rpc.send(api_id::IP_EVENT, ev.ip_data), TAG, "Failed to marshal IP event");
+            }
+        } while (type != api_id::ERROR);
+        return ESP_OK;
+    }
     esp_err_t perform()
+    {
+        auto res = select();
+        if (res == Sync::ERROR) {
+            return ESP_FAIL;
+        }
+        if (res & Sync::EVENT) {
+            uint64_t data;
+            read(sync.fd, &data, sizeof(data));
+            if (marshall_events() != ESP_OK) {
+                return ESP_FAIL;
+            }
+        }
+        if (res & Sync::RPC) {
+            if (handle_commands() != ESP_OK) {
+                return ESP_FAIL;
+            }
+        }
+        return ESP_OK;
+    }
+
+    esp_err_t handle_commands()
     {
         auto header = rpc.get_header();
         ESP_LOGI(TAG, "Received header id %d", (int) header.id);
-        std::lock_guard<Sync> lock(sync);
         switch (header.id) {
         case api_id::SET_MODE: {
             auto req = rpc.get_payload<wifi_mode_t>(api_id::SET_MODE, header);