MQTTAgent Publish is slow as compare to FreeRTOS MQTT

Partners & Sponsors AWS
21

@txf For optimal performance, it would be best to set your esp_tls connection to non-blocking mode and use select() to wait for an incoming message of any size rather than the default behavior of esp_tls_conn_read which attempts to fill the entire buffer up until the timeout.

When initializing your esp_tls connection, be set to set the non_block field to true.

You can get the current socket handle using the esp_tls_get_conn_sockfd() function call and from there, do something like the following:

int32_t espTlsTransportRecv(NetworkContext_t* pxNetworkContext,
                            void* pvData, size_t uxDataLen)
{
    int32_t lBytesRead = -1;

    if( ( pvData != NULL ) &&
        ( uxDataLen > 0 ) &&
        ( pxNetworkContext != NULL ) &&
        ( pxNetworkContext->pxTls != NULL ) )
    {
        int lSockFd = -1;

        // TODO: Is a semaphore really necessary here?
        xSemaphoreTake(pxNetworkContext->xTlsContextSemaphore, portMAX_DELAY);
        {
            lSockFd = esp_tls_get_conn_sockfd(pxNetworkContext->pxTls);
        }
        xSemaphoreGive(pxNetworkContext->xTlsContextSemaphore);

        if( lSockFd >= 0 )
        {
            int lResult = -1;
            fd_set xReadSet;
            fd_set xErrorSet;
            FD_SET(sockFd, &xReadSet);
            FD_SET(sockFd, &xErrorSet);

            // TODO: set a timeout for the select call
            lResult = select(sockFd + 1, &xReadSet, NULL, &xErrorSet, NULL );

            if( ( lResult > 0 ) &&
                FD_ISSET( sockFd, &xReadSet ) )
            {
                xSemaphoreTake(pxNetworkContext->xTlsContextSemaphore, portMAX_DELAY);
                {
                    lBytesRead = ( int32_t ) esp_tls_conn_read(pxNetworkContext->pxTls, pvData, uxDataLen);
                }
                xSemaphoreGive(pxNetworkContext->xTlsContextSemaphore);
            }
            else if( ( lResult == -1 ) &&
                     FD_ISSET( sockFd, &xErrorSet ) )
            {
                /* Socket Error */
                lBytesRead = -1;
            }
            else
            {
                lBytesRead = 0;
            }
        }
    }
    if( ( lBytesRead == ESP_TLS_ERR_SSL_WANT_WRITE ) || 
        ( lBytesRead == ESP_TLS_ERR_SSL_WANT_READ) ) {
        lBytesRead = 0;
    }
    return lBytesRead;
}

Be sure to use the latest version of coreMQTT for best results. Previous versions had a bug which prevented proper handling of messages that arrive across multiple calls to the transport receive function.

22

@PaulB-AWS Thanks.

Unfortunately, using this method, I can’t even establish a connection to the broker.

I had to fix up some of the functions and variables:

int32_t espTlsTransportRecv(NetworkContext_t* pxNetworkContext,
                            void* pvData, size_t uxDataLen)
{
    int32_t lBytesRead = -1;

    if( ( pvData != NULL ) &&
        ( uxDataLen > 0 ) &&
        ( pxNetworkContext != NULL ) &&
        ( pxNetworkContext->pxTls != NULL ) )
    {
        int lSockFd = -1;

        // TODO: Is a semaphore really necessary here?
        xSemaphoreTake(pxNetworkContext->xTlsContextSemaphore, portMAX_DELAY);
        {
            /*lSockFd =*/ esp_tls_get_conn_sockfd(pxNetworkContext->pxTls,&lSockFd);
        }
        xSemaphoreGive(pxNetworkContext->xTlsContextSemaphore);

        if( lSockFd >= 0 )
        {
            int lResult = -1;
            fd_set xReadSet;
            fd_set xErrorSet;
            FD_SET(lSockFd, &xReadSet);
            FD_SET(lSockFd, &xErrorSet);
            struct timeval test = {.tv_sec=1, .tv_usec=0};
            // TODO: set a timeout for the select call
            lResult = select(lSockFd + 1, &xReadSet, NULL, &xErrorSet, &test);

            if( ( lResult > 0 ) &&
                FD_ISSET( lSockFd, &xReadSet ) )
            {
                xSemaphoreTake(pxNetworkContext->xTlsContextSemaphore, portMAX_DELAY);
                {
                    lBytesRead = ( int32_t ) esp_tls_conn_read(pxNetworkContext->pxTls, pvData, uxDataLen);
                }
                xSemaphoreGive(pxNetworkContext->xTlsContextSemaphore);
            }
            else if( ( lResult == -1 ) &&
                     FD_ISSET( lSockFd, &xErrorSet ) )
            {
                /* Socket Error */
                lBytesRead = -1;
            }
            else
            {
                lBytesRead = 0;
            }
        }
    }
    if( ( lBytesRead == ESP_TLS_ERR_SSL_WANT_WRITE ) || 
        ( lBytesRead == ESP_TLS_ERR_SSL_WANT_READ) ) {
        lBytesRead = 0;
    }
    return lBytesRead;
}
23

Just bumping the thread with some additional info, in case somebody stumbles upon it in a search.

The non-blocking solution provided above by @PaulB-AWS does not appear to be necessary.

All that is required is to set the .non_block flag in the tls config struct. and increase the timeout to something large like 10000 (which is necessary to prevent excessive timeouts during connection).

However, a word of warning, if you use theOTA lib, you cannot make multiple block requests as those will constantly bring the connection down. It works with one block request at a time.

Presumably if your message rate is similar to that of OTA, non_block will have issues with that too.