FreeRTOS socket manager task

socket_task.c

#include "socket_task.h"


/*
 *******************************************************************************
 *                             Symbolic Constants                              *
 *******************************************************************************
*/


// Flags indicating on which socket events we want select to wait on
#define SOCK_SELECT_FLAGS       (eSELECT_READ | eSELECT_EXCEPT)


// Maximum amount of bytes that can be received from a socket on a call
#define SOCK_MAX_RECV_SIZE      64


/*
 *******************************************************************************
 *                              Global Variables                               *
 *******************************************************************************
*/


// Socket table
sock_t g_socket_table[MAX_SOCKET_COUNT];


// Socket table length
int g_tab_len;


/*
 *******************************************************************************
 *                        Internal Function Definitions                        *
 *******************************************************************************
*/


/* Call order for socket functions
 * 
 * 1. init_socket()    <----
 * 2. connect_socket()     |
 * 3. close_socket()       |
 * 4. unset_socket()       |
 * 5. ----------------------
*/


// Initializes a socket (but does not connect it)
void init_socket (int index) {

    // Extract entry
    sock_t entry = g_socket_table[index];

    // Initialize a TCP/IP stream socket
    Socket_t sock = FreeRTOS_socket(FREERTOS_AF_INET, FREERTOS_SOCK_STREAM,
        FREERTOS_IPPROTO_TCP);

    // Check if the socket was created
    if (sock == FREERTOS_INVALID_SOCKET) {
        ERR("task_sock_manager: Invalid socket", ESP_FAIL);
        return;
    }

    // Update the table with the TCP socket
    g_socket_table[index].sock = sock;
}


// Connects a socket. Returns zero on success, nonzero on error
int connect_socket (int index) {
    struct freertos_sockaddr addr_descr;

    // Extract entry
    sock_t entry = g_socket_table[index];

    // Configure remote connection information
    addr_descr.sin_addr = entry.addr;
    addr_descr.sin_port = entry.port;

    // Return connection outcome
    return (int)FreeRTOS_connect(entry.sock, &addr_descr, sizeof(addr_descr));
}


// Closes a socket
void close_socket (int index) {

    // Extract entry
    sock_t entry = g_socket_table[index];

    // If the socket is not active, then return now
    if (entry.sock == NULL) {
        return;
    }


    // Otherwise shutdown the socket
    if (FreeRTOS_shutdown(entry.sock, FREERTOS_SHUT_RDWR) != 0) {
        ERR("task_socket_manager: Couldn't shutdown socket", ESP_FAIL);
        return;
    }

    // Mark the socket as erroneous (to be unset on next recv() - see below)
    g_socket_table[index].err = 1;

    /* From this point on you need to close the socket. However you must wait
     * until the socket is ready to be closed. This entails calling
     * FreeRTOS_recv() until you get FREERTOS_EINVAL
     *
     * Then, you may finally 
     * 1. close the socket with FreeRTOS_closesocket()
     * 2. remove the socket from the socket-set
     * 3. unset the table entry
    */
}


// Unsets a socket
void unset_socket (int index, SocketSet_t *socket_set_p) {

    // Extract entry
    sock_t entry = g_socket_table[index];

    // Close the socket
    FreeRTOS_closesocket(entry.sock);

    // Remove socket from select set
    FreeRTOS_FD_CLR(entry.sock, *socket_set_p, SOCK_SELECT_FLAGS);

    // Reset socket value
    g_socket_table[index].sock = NULL;

    // Reset socket error
    g_socket_table[index].err = 0x0;
}


// Receives data from a socket, and sends it to the appropriate task if enabled
void recv_socket (int index, SocketSet_t *socket_set_p) {
    int32_t ret;
    esp_err_t err;
    static uint8_t recv_buffer[SOCK_MAX_RECV_SIZE];

    // Extract entry
    sock_t entry = g_socket_table[index];

    // If an error occurred, take action depending on status
    if ((ret = FreeRTOS_recv(entry.sock, (void *)recv_buffer, SOCK_MAX_RECV_SIZE,
        0x0)) < 0) {

        // If socket already experienced an error, unset it. Otherwise close it
        if (entry.err != 0) {
            unset_socket(index, socket_set_p);
        } else {
            close_socket(index);
        }

        return;
    }

    // Otherwise data was read. If the socket is already in err state, ignore it
    if (entry.err != 0) {
        return;
    }

    // If the task didn't register a handler, ignore the data and return now
    if (entry.recv_queue == NULL) {
        return;
    }

    // Otherwise place the data on the receive queue
    if ((err = ipc_enqueue(entry.recv_queue, index, ret, recv_buffer)) 
        != ESP_OK) {
        ERR("task_sock_manager: Could not enqueue received data", err);
    }

    // Notify tasks that data is ready (so they should check their queues)
    xEventGroupSetBits(g_event_group, FLAG_SOCK_RECV_MSG);
}


/*
 *******************************************************************************
 *                        External Function Definitions                        *
 *******************************************************************************
*/


int task_sock_manager_register (uint32_t addr, uint16_t port, 
    QueueHandle_t recv_queue) {

    // If no room remains in the table, return an invalid index
    if (g_tab_len >= MAX_SOCKET_COUNT) {
        return -1;
    } 

    // Register the new table entry
    g_socket_table[g_tab_len] = (sock_t){
        .sock = NULL,                       // Socket_t is a pointer to void
        .addr = addr,
        .port = port,
        .err  = 0x0,
        .recv_queue = recv_queue;
    };

    // Return index (post-incremented)
    return g_tab_len++;
}


void task_sock_manager (void *args) {
    uint32_t flags;
    SocketSet_t socket_set;
    uint8_t active_connections = 0;
    task_queue_msg_t queue_msg;
    const TickType_t flag_block_time = 8; // How long to wait for flags to be set
    const TickType_t sock_block_time = 8; // How long to wait for data in sockets

    // Create socket-set for select (selSELECT_QUEUE_SIZE = 6 events max)
    socket_set = FreeRTOS_CreateSocketSet(selSELECT_QUEUE_SIZE);

    // Init all sockets in the socket-table to the set (hence start task last)
    for (int i = 0; i < g_tab_len; ++i) {
        init_socket(i, &socket_set);
    }


    do {

        // Block until something needs to be sent (idea: variable block time)
        flags = xEventGroupWaitBits(g_event_group, FLAG_SOCK_SEND_MSG,
            pdTRUE, pdFALSE, flag_block_time);

        // Try to send everything in the queue
        while (uxQueueMessagesWaiting(g_sock_tx_queue) > 0) {

            // Dequeue next message
            xQueueReceive(g_sock_tx_queue, (void *)&queue_msg,
                TASK_QUEUE_MAX_TICKS);

            // Check the ID 
            if (queue_msg.id < 0 || queue_msg.id >= g_tab_len) {
                ERR("task_socket_manager: Invalid index for send", ESP_FAIL);
                continue;
            }

            // Create the socket if needed
            if (g_socket_table[queue_msg.id].sock == NULL) {
                if (connect_socket(queue_msg.id) != 0) {
                    ERR("task_socket_manager: Connect socket failed", ESP_FAIL);
                    continue;
                } else {

                    // Add socket to listen set
                    FreeRTOS_FD_SET(g_sock_tx_queue[queue_msg.id].sock,
                        socket_set, SOCK_SELECT_FLAGS);

                    // Increment active connections
                    active_connections++;
                }
            }

            // Close socket if size to send is zero or a socket error occurred
            if (queue_msg.size == 0 || 
                FreeRTOS_send(g_sock_tx_queue[queue_msg.id].sock, 
                (void *)queue_msg.data, queue_msg.size, 0x0) < 0) {
                ERR("task_socket_manager: Couldn't send on socket", ESP_FAIL);

                // Close socket on error, then decrement active connections
                close_socket();
                active_connections--;
            }
        }


        // Do not run select unless there are active connections
        if (active_connections == 0) {
            continue;
        }

        // Continue if no events occurred within the blocking interval
        if (FreeRTOS_select(socket_set, sock_block_time) == 0) {
            continue;
        }

        // Otherwise check all the sockets and take action on possible events
        for (int i = 0; i < g_tab_len; ++i) {

            // Extract set flags
            BaseType_t sock_flags = FreeRTOS_FD_ISSET(g_socket_table[i].sock, 
                socket_set);

            // If an exception occurred
            if (sock_flags & eSELECT_EXCEPT) {
                close_socket(i);
            }

            // If a read event occurred
            if (sock_flags & eSELECT_READ) {
                recv_socket(i, &socket_set);
            }

        }

    } while (1);

}