Websocket Async Messages

[Baldhead]

Hi,

I was having data consistency issues on my websocket server because I was using the same global variables for sending and receiving websocket frames.
static uint8_t buf[125] = { 0 };
static httpd_ws_frame_t ws_pkt;

So I decided to change my logic and use local allocation on the stack (maximum 125 bytes buffer) and also use allocation on the heap when necessary.

The problem is that I don't know if the functions httpd_ws_send_frame_async() and httpd_ws_send_frame() are blocking or not, or if they copy the message or not.

The problem I'm trying to solve:

Actual implementation:
A fota task called this when ws_pkt and buf they were static global.
static uint8_t buf[125] = { 0 };
static httpd_ws_frame_t ws_pkt;

The req from httpd handler is already out of scope here, i.e., the handler function has already returned and the request no longer exists, so it is mandatory to use httpd_ws_send_frame_async().

Code: Select all

void ws_resp_with_message_async(int sockfd, char* message, uint8_t command)
{   
    ESP_LOGI(TAG, LOG_USER2("FUNCTION: %s"), __func__); 
    ESP_LOGI(TAG, LOG_USER2("sockfd: %d"), sockfd);   
    ESP_LOGI(TAG, LOG_USER2("command: %u"), command);
    ESP_LOGI(TAG, LOG_USER2("Message: %s"), message);
    
    size_t strSize = strlen(message);
    init_ws_pkt(1 + strSize);  // command size + error message size 
    ws_pkt.payload[0] = command;               
    
    for (uint32_t i = 0 ; i < strSize ; i++)
    {
        ws_pkt.payload[i+1] = message[i];  // add message after the command.
    }
    
    httpd_ws_send_frame_async(server, sockfd, &ws_pkt);  // Async message, valid after returning to uri handler and probably before returning to uri handler too.   
}

New implementation:

Using malloc allocation, but when free the memory if httpd_ws_send_frame_async() returns immediately without copying the message ???

I welcome suggestions and ideas, please.

Code: Select all

static void ws_async_send_worker(void *arg)
{
    httpd_ws_frame* pointer = (httpd_ws_frame*)arg;
    
    // This is just an example; also need to receive the socket fd extracted from a struct.
    httpd_ws_send_frame_async(server, sockfd, pointer);   
        
    // pointer struct with httpd_ws_frame and socket fd
    free(pointer);  //  Can I free up memory here? If not, how will I know when I can ?   
}


void ws_resp_with_message_async(int sockfd, char* message, uint8_t command)
{
    Dynamically allocate memory to httpd_ws_frame	
    Dynamically allocate memory to buf
    Assigns the buffer to httpd_ws_frame	
    Fill the buffer with the message	
    
    // This is just an example; also need to send the socket fd inside a struct.    
    httpd_queue_work(server, ws_async_send_worker, httpd_ws_frame pointer);  
}

[limpens]

My websocket function is slightly different (iterating over all connections), but previously malloced buffers in a different function are free-ed after completing:

Code: Select all

static void http_ws_async_send(void *arg)
{
    async_resp_arg_t *resp_arg = (async_resp_arg_t *)arg;
    httpd_ws_frame_t ws_pkt = {};

    ws_pkt.payload = (uint8_t *)resp_arg->txbuf;
    ws_pkt.len = resp_arg->len;
    ws_pkt.type = HTTPD_WS_TYPE_TEXT;

    size_t fds = 8;
    int client_fds[8];
    httpd_get_client_list(resp_arg->hd, &fds, client_fds);

    for (int i = 0; i < fds; i++)
    {
        if (httpd_ws_get_fd_info(resp_arg->hd, client_fds[i]) == HTTPD_WS_CLIENT_WEBSOCKET)
        {
            httpd_ws_send_frame_async(resp_arg->hd, client_fds[i], &ws_pkt);
        }
    }

    // cleanup previous allocated buffer(s):
    free(resp_arg->txbuf);
    free(resp_arg);
}

And this gets called by httpd_queue_work just like you do. Never had issues with this method of malloc/free.

Code: Select all

void http_ws_send_status(httpd_handle_t server, void *Param)
{
    async_resp_arg_t *resp_arg = (async_resp_arg_t *)malloc(sizeof(async_resp_arg_t));
    ws_status_t *ws_status = (ws_status_t *)Param;
    cJSON *wsDoc = cJSON_CreateObject();

    if (ws_status->event == event_status)
    {
        char uptime_buf[64] = {};
        wifi_ap_record_t ap;
        esp_wifi_sta_get_ap_info(&ap);

        uint64_t microSinceStartup = esp_timer_get_time();
        uint32_t secSinceStartup = (microSinceStartup / 1000000);
        time_t epoch = secSinceStartup;
        struct tm *tm_T = gmtime(&epoch);
        sprintf(uptime_buf, "%d %s, %d %s, %d %s, %d %s", tm_T->tm_yday, (tm_T->tm_yday > 1) ? "days" : "day", tm_T->tm_hour, (tm_T->tm_hour > 1) ? "hours" : "hour",
            tm_T->tm_min, (tm_T->tm_min > 1) ? "minutes" : "minute", tm_T->tm_sec, (tm_T->tm_sec > 1) ? "seconds" : "second");

        cJSON_AddStringToObject(wsDoc, "f", "status");
        cJSON_AddItemToObject(wsDoc, "rssi", cJSON_CreateNumber(ap.rssi));
        cJSON_AddStringToObject(wsDoc, "uptime", uptime_buf);
        cJSON_AddItemToObject(wsDoc, "voltage", cJSON_CreateNumber(ws_status->voltage));
    }
    else if (ws_status->event == event_ota)
    {
        cJSON_AddStringToObject(wsDoc, "f", "ota");
        cJSON_AddItemToObject(wsDoc, "progress", cJSON_CreateNumber(ws_status->progress));
        cJSON_AddItemToObject(wsDoc, "status", cJSON_CreateString(ws_status->status));
    }

    char *json = cJSON_Print(wsDoc);
    if (json)
    {
        resp_arg->hd = server;
        resp_arg->txbuf = json;
        resp_arg->len = strlen(json);

        httpd_queue_work(server, http_ws_async_send, resp_arg);
    }
    cJSON_Delete(wsDoc);
}

[MicroController]

The problem is that I don't know if the functions httpd_ws_send_frame_async() and httpd_ws_send_frame() are blocking or not, or if they copy the message or not.

Both functions operate synchronously. They return when the WS frame has been 'sent', i.e. its data was handed over to the network stack.
So you can discard/re-use your httpd_ws_frame_t immediately afterwards.

[Bryght-Richard]

Agree with Microcontroller, and will also mention that

Code: Select all

httpd_ws_send_frame()

and

Code: Select all

httpd_ws_send_frame_async()

are not thread-safe, so the serialization is up to you. Using httpd_queue_work is a good way to handle it. Otherwise, calling the above functions at the same time may cross the streams of (header,payload) between packets, corrupt the flow, and often drop the connection.

[Baldhead]

@MicroController
@cermak

Now that i changed my reception handler to eliminate static global variables, this error appears when I receive a ping from my app on my phone. Previously, the same ping sent by the app worked perfectly.

Aside from the ping, everything seems to be working fine, both sending and receiving. The ping is received 30 seconds after the client connects. I don't know if this happens with the other control frames, but it probably does.

This error alone When I just wait for the ping without sending anything other than opening the connection and receiving the server state.
W (42702) httpd_ws: httpd_ws_recv_frame: WS frame is not properly masked.

or this error When I send other commands after opening the connection and receiving the server state:
W (145027) httpd_ws: httpd_ws_recv_frame: WS frame is not properly masked.
W (145034) WSS_SERVER: httpd_ws_recv_frame() nao conseguiu pegar o tamanho da mensagem recebida. Error: ESP_ERR_INVALID_STATE

Problem here.
Stack variables and calling httpd_ws_recv_frame() two times:

Code: Select all

static esp_err_t get_clients_message_handler(httpd_req_t* req)
{       
    ESP_LOGI(TAG, LOG_USER("FUNCTION: %s"), __func__ );
    ESP_LOGI(TAG, LOG_USER("[%s]req->method = %d\n"), __func__, req->method);    
    
    if (req->method == HTTP_GET)
    {
        char message[50];
        int sockfd = httpd_req_to_sockfd(req);

        ESP_LOGI(TAG, LOG_USER("New Connection Was Opened"));              
        ESP_LOGI(TAG, LOG_USER("New Connection Socket: %d\n"), sockfd);

        bool code = ws_server_users_login(req, message, sizeof(message));
        if (code) 
        {           
            add_ws_client_to_list(sockfd);  
            resp_Ws_Error_Message_Sync(req, message, command_error);              
            return ESP_OK;
        }
        else
        {
            ESP_LOGI(TAG, LOG_USER("Message: %s\n"), message);
            resp_Ws_Error_Message_Sync(req, message, command_error);          
            ESP_LOGI(TAG, LOG_USER("SEND CLOSE FRAME"));                        
            esp_err_t ret = send_close_frame( unexpected_condition_code, message, sockfd );  // send_close_frame( normal_closure, NULL, sockfd );
            ESP_LOGI(TAG, LOG_USER("Send Close Frame: %s\n"), esp_err_to_name(ret));
            return ESP_FAIL;                     
        }        
    }

    httpd_ws_frame_t ws_pkt;
    memset(&ws_pkt, 0, sizeof(httpd_ws_frame_t));

    esp_err_t ret = httpd_ws_recv_frame(req, &ws_pkt, 0);
    if (ret != ESP_OK)
    {
        ESP_LOGW(TAG, "httpd_ws_recv_frame() nao conseguiu pegar o tamanho da mensagem recebida. Error: %s", esp_err_to_name(ret));
        return ret;
    }

    if (ws_pkt.len > 125)
    {
        ESP_LOGW(TAG, "httpd_ws_recv_frame(), ws_pkt.len > 125 bytes. Rejeitando mensagem. ws_pkt.len: %d", ws_pkt.len);
        return ESP_OK;
        //return ESP_ERR_INVALID_SIZE;
    }
     
    // Garante que o buffer tenha pelo menos 1 byte para evitar erro de compilação/execução se len for 0
    size_t buf_capacity = (ws_pkt.len > 0) ? ws_pkt.len : 1;
    uint8_t buf[buf_capacity];
    memset(buf, 0, buf_capacity);
    ws_pkt.payload = buf;
  
    // Passa buf_capacity para garantir que o driver tenha onde escrever se len > 0
    // Se len for 0, o driver apenas finaliza a leitura do cabeçalho
    ret = httpd_ws_recv_frame(req, &ws_pkt, ws_pkt.len);   
    if (ret != ESP_OK)
    {
    #if webSocketDebug 
        ESP_LOGI(TAG, LOG_USER("httpd_ws_recv_frame failed with: %s\n"), esp_err_to_name(ret));
    #endif
        return ret;
    }

#if webSocketDebug 
    printFrame(&ws_pkt, "Received Client Frame", req);    
#endif

#if enable_handle_ws_control_frames
    if (ws_pkt.type == HTTPD_WS_TYPE_PING || ws_pkt.type == HTTPD_WS_TYPE_PONG || ws_pkt.type == HTTPD_WS_TYPE_CLOSE)    
        return control_frames_handler(req, &ws_pkt);  // Control frames = HTTPD_WS_TYPE_PING or HTTPD_WS_TYPE_PONG or HTTPD_WS_TYPE_CLOSE.                
#endif    

    // Other frames types = HTTPD_WS_TYPE_BINARY or HTTPD_WS_TYPE_TEXT or HTTPD_WS_TYPE_CONTINUE.  
    if (ws_pkt.type != HTTPD_WS_TYPE_BINARY)  // if( ws_pkt.type != HTTPD_WS_TYPE_BINARY || ws_pkt.type != HTTPD_WS_TYPE_TEXT )
    {
        int fd = httpd_req_to_sockfd(req);
        ESP_LOGI(TAG, LOG_USER("HTTPD_WS_TYPE != HTTPD_WS_TYPE_BINARY: %d"), ws_pkt.type);
        ESP_LOGI(TAG, LOG_USER("SOCKET: %d\n"), fd);
        resp_Ws_Error_Message_Sync(req, "WebSocket Type Not Binary!!!", command_error);
        return ESP_FAIL;  
    } 
    
    // PROTEÇÃO
    // Só entra aqui se for BINARY e tiver pelo menos 1 byte de conteúdo.
    // Isso impede que frames BINARY vazios ou resíduos do buffer estático
    // disparem o ponteiro de função.
    if (ws_pkt.len > 0) 
    {
        uint8_t comando = ws_pkt.payload[0];
        if (comando < web_comm_t_array_size_t) 
        {    
            // Antes de executar, você pode limpar o buffer para a PRÓXIMA vez
            // Mas o mais importante é que agora 'comando' é garantidamente deste frame.
            ret = func_pointer[comando](req, comando, &ws_pkt);
            
            #if webSocketDebug
                ESP_LOGI(TAG, LOG_USER("ret select_func = %d\n"), ret);  
            #endif

            return ESP_OK;
        }
        else
        {
            #if webSocketDebug
                ESP_LOGE(TAG, "Error: Function Index out of allowable range!!! ws_pkt.payload[0] = %u\n", ws_pkt.payload[0]);
            #endif
            
            char err_msg[100];    
            snprintf(err_msg, sizeof(err_msg), "Error: Function Index out of allowable range!!! ws_pkt.payload[0] = %u", ws_pkt.payload[0]);
            resp_Ws_Error_Message_Sync(req, err_msg, command_error);        
            return ESP_FAIL; 
        }
    } 
    else if (ws_pkt.len == 0)
    {
        ESP_LOGW(TAG, "Frame binário vazio recebido, ignorando processamento de comando.");
        return ESP_OK; 
    }

    return ESP_OK;
}

That's how it worked with static variables:

Code: Select all

static uint8_t buf[128] = { 0 };
static httpd_ws_frame_t ws_pkt;

static esp_err_t get_clients_message_handler(httpd_req_t* req)
{       
    ESP_LOGI(TAG, LOG_USER("FUNCTION: %s"), __func__ );
    ESP_LOGI(TAG, LOG_USER("[%s]req->method = %d\n"), __func__, req->method);    
    
    if (req->method == HTTP_GET)
    {
        char message[50];
        int sockfd = httpd_req_to_sockfd(req);

        ESP_LOGI(TAG, LOG_USER("New Connection Was Opened"));              
        ESP_LOGI(TAG, LOG_USER("New Connection Socket: %d\n"), sockfd);

        bool code = ws_server_users_login(req, message, sizeof(message));
        if (code) 
        {           
            add_ws_client_to_list(sockfd);  
            resp_Ws_Error_Message_Sync(req, message, command_error);              
            return ESP_OK;
        }
        else
        {
            ESP_LOGI(TAG, LOG_USER("Message: %s\n"), message);
            resp_Ws_Error_Message_Sync(req, message, command_error);          
            ESP_LOGI(TAG, LOG_USER("SEND CLOSE FRAME"));                        
            esp_err_t ret = send_close_frame( unexpected_condition_code, message, sockfd );  // send_close_frame( normal_closure, NULL, sockfd );
            ESP_LOGI(TAG, LOG_USER("Send Close Frame: %s\n"), esp_err_to_name(ret));
            return ESP_FAIL;                     
        }        
    }
    
	memset(&ws_pkt, 0, sizeof(httpd_ws_frame_t));
    ws_pkt.payload = buf;
  
    esp_err_t ret = httpd_ws_recv_frame(req, &ws_pkt, sizeof(buf));   
    if (ret != ESP_OK)
    {
    #if webSocketDebug 
        ESP_LOGI(TAG, LOG_USER("httpd_ws_recv_frame failed with: %s\n"), esp_err_to_name(ret));
    #endif
        return ret;
    }

#if webSocketDebug 
    printFrame(&ws_pkt, "Received Client Frame", req);    
#endif

#if enable_handle_ws_control_frames
    if (ws_pkt.type == HTTPD_WS_TYPE_PING || ws_pkt.type == HTTPD_WS_TYPE_PONG || ws_pkt.type == HTTPD_WS_TYPE_CLOSE)    
        return control_frames_handler(req, &ws_pkt);  // Control frames = HTTPD_WS_TYPE_PING or HTTPD_WS_TYPE_PONG or HTTPD_WS_TYPE_CLOSE.                
#endif    

    // Other frames types = HTTPD_WS_TYPE_BINARY or HTTPD_WS_TYPE_TEXT or HTTPD_WS_TYPE_CONTINUE.  
    if (ws_pkt.type != HTTPD_WS_TYPE_BINARY)  // if( ws_pkt.type != HTTPD_WS_TYPE_BINARY || ws_pkt.type != HTTPD_WS_TYPE_TEXT )
    {
        int fd = httpd_req_to_sockfd(req);
        ESP_LOGI(TAG, LOG_USER("HTTPD_WS_TYPE != HTTPD_WS_TYPE_BINARY: %d"), ws_pkt.type);
        ESP_LOGI(TAG, LOG_USER("SOCKET: %d\n"), fd);
        resp_Ws_Error_Message_Sync(req, "WebSocket Type Not Binary!!!", command_error);
        return ESP_FAIL;  
    }   

    // The first payload byte is a command, the rest of the bytes are data.
    if( (ws_pkt.payload[0] >= 0) && (ws_pkt.payload[0] < web_comm_t_array_size_t) )  // func_index = ws_pkt.payload[0].
    {    
        ret = func_pointer[ ws_pkt.payload[0] ]( req, ws_pkt.payload[0] );      
    #if webSocketDebug
        ESP_LOGI(TAG, LOG_USER("ret select_func = %d\n"), ret);  
    #endif
        return ESP_OK;  // return ret;
    }
    else
    {
    #if webSocketDebug
    	ESP_LOGE(TAG, "Error: Function Index out of allowable range!!! ws_pkt.payload[0] = %u\n", ws_pkt.payload[0]);
    #endif
        char err_msg[100];    
        snprintf(err_msg, sizeof(err_msg), "Error: Function Index out of allowable range!!! ws_pkt.payload[0] = %u", ws_pkt.payload[0]);
        resp_Ws_Error_Message_Sync(req, err_msg, command_error);        
        return ESP_FAIL;   
    }           
}

Note:
.handle_ws_control_frames = true,
ESP-IDF v5.5.1
ESP32-S3

[MicroController]

calling the above functions at the same time may cross the streams

Don't cross the streams! - It would be bad!

IAintAfraid.png (11.19 KiB) Viewed 2803 times

[Baldhead]

calling the above functions at the same time may cross the streams

Don't cross the streams! - It would be bad!

IAintAfraid.png

Now the problem is another.

Should I open another issue ?

When I close the app on my phone, a server error also appears, but I think that's a different problem.
In the mobile app I use okhttp.

The problem that i want solve now is ping sent by celular app is broken my handler.
But how i said in above post, the ping works ok when i used static global variables and dont call httpd_ws_recv_frame() two times.

To test, simply copy and paste my handler and enable `.handle_ws_control_frames = true`.

[Bryght-Richard]

Now that i changed my reception handler to eliminate static global variables, this error appears when I receive a ping from my app on my phone. Previously, the same ping sent by the app worked perfectly.

It looks like two things changed: you're using a stack variable now, but also the stack-variable is a variable length array(VLA). A few times I've run into issues with VLA support before, though haven't used them on Espressif. You might check if it works better or differently with a fixed-sized array on the stack.

[Bryght-Richard]

One other thought: for HTTPD_WS_TYPE_PING or HTTPD_WS_TYPE_PONG, do they usually have a payload? If not, calling httpd_ws_recv_frame() with max_len = 0 a second time might mess up the parser. Seems like if you receive a frame with zero-byte payload on the first call, perhaps the second call(to get the payload) should be skipped entirely, or else it may corrupt the stream by reading a byte or two wrongly.

[Baldhead]

Agree with Microcontroller, and will also mention that

Code: Select all

httpd_ws_send_frame()

and

Code: Select all

httpd_ws_send_frame_async()

are not thread-safe, so the serialization is up to you. Using httpd_queue_work is a good way to handle it. Otherwise, calling the above functions at the same time may cross the streams of (header,payload) between packets, corrupt the flow, and often drop the connection.

So i need to add queue work in the reception handler too ?