embedded/FreeRTOS
1
0
Fork 0
mirror of https://github.com/FreeRTOS/FreeRTOS.git synced 2026-08-09 06:33:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add reconnection and resubscribe logic to the MQTT plaintext demo (#271)

Browse Source 
This implements retries for FreeRTOS IoT Beta2 libraries, initializing the backoff to some value returned by the PRNG: uxRand(). If an attempt (connect or subscribe) fails, vTaskDelay is called to make the task sleep for some backoff period that doubles on each retry but is bounded.

Co-authored-by: Muneeb Ahmed <54290492+muneebahmed10@users.noreply.github.com>
This commit is contained in:
Oscar Michael Abrina
2020-09-17 17:12:43 -07:00
committed by GitHub
parent 6a8bb87dec
commit 71f5984776
5 changed files with 548 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

270
FreeRTOS-Plus/Demo/FreeRTOS-IoT-Libraries-LTS-Beta2/mqtt/mqtt_plain_text/DemoTasks/PlaintextMQTTExample.c
View File

@@ -57,6 +57,9 @@
/* MQTT library includes. */
#include "mqtt.h"
/* Retry utilities include. */
#include "retry_utils.h"
/* Transport interface include. */
#include "plaintext_freertos.h"
@@ -163,23 +166,46 @@
*/
static void prvMQTTDemoTask( void * pvParameters );
/**
* @brief Connect to MQTT broker with reconnection retries.
*
* If connection fails, retry is attempted after a timeout.
* Timeout value will exponentially increase until maximum
* timeout value is reached or the number of attempts are exhausted.
*
* @param pxNetworkContext The output parameter to return the created network context.
*
* @return The status of the final connection attempt.
*/
static PlaintextTransportStatus_t prvConnectToServerWithBackoffRetries( NetworkContext_t * pxNetworkContext );
/**
* @brief Sends an MQTT Connect packet over the already connected TCP socket.
*
* @param pxMQTTContext MQTT context pointer.
* @param xNetworkContext network context.
* @param pxNetworkContext Network context.
*
*/
static void prvCreateMQTTConnectionWithBroker( MQTTContext_t * pxMQTTContext,
NetworkContext_t * pxNetworkContext );
/**
* @brief Function to update variable globalSubAckStatus with status
* information from Subscribe ACK. Called by eventCallback after processing
* incoming subscribe echo.
*
* @param Server response to the subscription request.
*/
static void prvUpdateSubAckStatus( MQTTPacketInfo_t * pxPacketInfo );
/**
* @brief Subscribes to the topic as specified in mqttexampleTOPIC at the top of
* this file.
* this file. In the case of a Subscribe ACK failure, then subscription is
* retried using an exponential backoff strategy with jitter.
*
* @param pxMQTTContext MQTT context pointer.
*/
static void prvMQTTSubscribeToTopic( MQTTContext_t * pxMQTTContext );
static void prvMQTTSubscribeWithBackoffRetries( MQTTContext_t * pxMQTTContext );
/**
* @brief Publishes a message mqttexampleMESSAGE on mqttexampleTOPIC topic.
@@ -260,6 +286,18 @@ static uint16_t usSubscribePacketIdentifier;
*/
static uint16_t usUnsubscribePacketIdentifier;
/**
* @brief Status of latest Subscribe ACK;
* it is updated every time the callback function processes a Subscribe ACK.
*/
static MQTTSubAckStatus_t xGlobalSubAckStatus = MQTTSubAckFailure;
/**
* @brief Array to keep subscription topics.
* Used to re-subscribe to topics that failed initial subscription attempts.
*/
static MQTTSubscribeInfo_t xGlobalSubscribeInfo;
/** @brief Static buffer used to hold MQTT messages being sent and received. */
static MQTTFixedBuffer_t xBuffer =
@@ -270,7 +308,7 @@ static MQTTFixedBuffer_t xBuffer =
/*-----------------------------------------------------------*/
/*
/**
* @brief Create the task that demonstrates the Plain text MQTT API Demo.
*/
void vStartSimpleMQTTDemo( void )
@@ -305,15 +343,12 @@ static void prvMQTTDemoTask( void * pvParameters )
{
/****************************** Connect. ******************************/
/* Establish a TCP connection with the MQTT broker. This example connects to
* the MQTT broker as specified in democonfigMQTT_BROKER_ENDPOINT and
* democonfigMQTT_BROKER_PORT at the top of this file. */
LogInfo( ( "Create a TCP connection to %s.\r\n", democonfigMQTT_BROKER_ENDPOINT ) );
xNetworkStatus = Plaintext_FreeRTOS_Connect( &xNetworkContext,
democonfigMQTT_BROKER_ENDPOINT,
democonfigMQTT_BROKER_PORT,
TRANSPORT_SEND_RECV_TIMEOUT_MS,
TRANSPORT_SEND_RECV_TIMEOUT_MS );
/* Attempt to connect to the MQTT broker. If connection fails, retry after
* a timeout. Timeout value will be exponentially increased until the maximum
* number of attempts are reached or the maximum timeout value is reached.
* The function returns a failure status if the TCP connection cannot be established
* to the broker after the configured number of attempts. */
xNetworkStatus = prvConnectToServerWithBackoffRetries( &xNetworkContext );
configASSERT( xNetworkStatus == PLAINTEXT_TRANSPORT_SUCCESS );
/* Sends an MQTT Connect packet over the already connected TCP socket,
@@ -323,25 +358,10 @@ static void prvMQTTDemoTask( void * pvParameters )
/**************************** Subscribe. ******************************/
/* The client is now connected to the broker. Subscribe to the topic
* as specified in mqttexampleTOPIC at the top of this file by sending a
* subscribe packet then waiting for a subscribe acknowledgment (SUBACK).
* This client will then publish to the same topic it subscribed to, so it
* will expect all the messages it sends to the broker to be sent back to it
* from the broker. This demo uses QOS0 in Subscribe, therefore, the Publish
* messages received from the broker will have QOS0. */
LogInfo( ( "Attempt to subscribe to the MQTT topic %s.\r\n", mqttexampleTOPIC ) );
prvMQTTSubscribeToTopic( &xMQTTContext );
/* Process incoming packet from the broker. After sending the subscribe, the
* client may receive a publish before it receives a subscribe ack. Therefore,
* call generic incoming packet processing function. Since this demo is
* subscribing to the topic to which no one is publishing, probability of
* receiving Publish message before subscribe ack is zero; but application
* must be ready to receive any packet. This demo uses the generic packet
* processing function everywhere to highlight this fact. */
xMQTTStatus = MQTT_ProcessLoop( &xMQTTContext, mqttexamplePROCESS_LOOP_TIMEOUT_MS );
configASSERT( xMQTTStatus == MQTTSuccess );
/* If server rejected the subscription request, attempt to resubscribe to topic.
* Attempts are made according to the exponential backoff retry strategy
* implemented in retryUtils. */
prvMQTTSubscribeWithBackoffRetries( &xMQTTContext );
/**************************** Publish and Keep Alive Loop. ******************************/
/* Publish messages with QOS0, send and process Keep alive messages. */
@@ -381,6 +401,9 @@ static void prvMQTTDemoTask( void * pvParameters )
xNetworkStatus = Plaintext_FreeRTOS_Disconnect( &xNetworkContext );
configASSERT( xNetworkStatus == PLAINTEXT_TRANSPORT_SUCCESS );
/* Reset global SUBACK status variable after completion of subscription request cycle. */
xGlobalSubAckStatus = MQTTSubAckFailure;
/* Wait for some time between two iterations to ensure that we do not
* bombard the public test mosquitto broker. */
LogInfo( ( "prvMQTTDemoTask() completed an iteration successfully. Total free heap is %u.\r\n", xPortGetFreeHeapSize() ) );
@@ -391,6 +414,51 @@ static void prvMQTTDemoTask( void * pvParameters )
}
/*-----------------------------------------------------------*/
static PlaintextTransportStatus_t prvConnectToServerWithBackoffRetries( NetworkContext_t * pNetworkContext )
{
PlaintextTransportStatus_t xNetworkStatus;
RetryUtilsStatus_t xRetryUtilsStatus = RetryUtilsSuccess;
RetryUtilsParams_t xReconnectParams;
/* Initialize reconnect attempts and interval. */
xReconnectParams.maxRetryAttempts = MAX_RETRY_ATTEMPTS;
RetryUtils_ParamsReset( &xReconnectParams );
/* Attempt to connect to MQTT broker. If connection fails, retry after
* a timeout. Timeout value will exponentially increase till maximum
* attempts are reached.
*/
do
{
/* Establish a TCP connection with the MQTT broker. This example connects to
* the MQTT broker as specified in democonfigMQTT_BROKER_ENDPOINT and
* democonfigMQTT_BROKER_PORT at the top of this file. */
LogInfo( ( "Create a TCP connection to %s:%d.",
democonfigMQTT_BROKER_ENDPOINT,
democonfigMQTT_BROKER_PORT ) );
xNetworkStatus = Plaintext_FreeRTOS_Connect( pNetworkContext,
democonfigMQTT_BROKER_ENDPOINT,
democonfigMQTT_BROKER_PORT,
TRANSPORT_SEND_RECV_TIMEOUT_MS,
TRANSPORT_SEND_RECV_TIMEOUT_MS );
if( xNetworkStatus != PLAINTEXT_TRANSPORT_SUCCESS )
{
LogWarn( ( "Connection to the broker failed. Retrying connection with backoff and jitter." ) );
xRetryUtilsStatus = RetryUtils_BackoffAndSleep( &xReconnectParams );
}
if( xRetryUtilsStatus == RetryUtilsRetriesExhausted )
{
LogError( ( "Connection to the broker failed, all attempts exhausted." ) );
xNetworkStatus = PLAINTEXT_TRANSPORT_CONNECT_FAILURE;
}
} while( ( xNetworkStatus != PLAINTEXT_TRANSPORT_SUCCESS ) && ( xRetryUtilsStatus == RetryUtilsSuccess ) );
return xNetworkStatus;
}
/*-----------------------------------------------------------*/
static void prvCreateMQTTConnectionWithBroker( MQTTContext_t * pxMQTTContext,
NetworkContext_t * pxNetworkContext )
{
@@ -414,7 +482,7 @@ static void prvCreateMQTTConnectionWithBroker( MQTTContext_t * pxMQTTContext,
configASSERT( xResult == MQTTSuccess );
/* Many fields not used in this demo so start with everything at 0. */
memset( ( void * ) &xConnectInfo, 0x00, sizeof( xConnectInfo ) );
( void ) memset( ( void * ) &xConnectInfo, 0x00, sizeof( xConnectInfo ) );
/* Start with a clean session i.e. direct the MQTT broker to discard any
* previous session data. Also, establishing a connection with clean session
@@ -440,43 +508,90 @@ static void prvCreateMQTTConnectionWithBroker( MQTTContext_t * pxMQTTContext,
NULL,
mqttexampleCONNACK_RECV_TIMEOUT_MS,
&xSessionPresent );
if( xResult != MQTTSuccess )
{
LogError( ( "Connection with MQTT broker failed.\r\n" ) );
}
configASSERT( xResult == MQTTSuccess );
}
/*-----------------------------------------------------------*/
static void prvMQTTSubscribeToTopic( MQTTContext_t * pxMQTTContext )
static void prvUpdateSubAckStatus( MQTTPacketInfo_t * pxPacketInfo )
{
MQTTStatus_t xResult;
MQTTSubscribeInfo_t xMQTTSubscription[ 1 ];
MQTTStatus_t xResult = MQTTSuccess;
uint8_t * pucPayload = NULL;
size_t ulSize = 0;
/***
* For readability, error handling in this function is restricted to the use of
* asserts().
***/
xResult = MQTT_GetSubAckStatusCodes( pxPacketInfo, &pucPayload, &ulSize );
/* MQTT_GetSubAckStatusCodes always returns success if called with packet info
* from the event callback and non-NULL parameters. */
configASSERT( xResult == MQTTSuccess );
/* Demo only subscribes to one topic, so only one status code is returned. */
xGlobalSubAckStatus = pucPayload[ 0 ];
}
/*-----------------------------------------------------------*/
static void prvMQTTSubscribeWithBackoffRetries( MQTTContext_t * pxMQTTContext )
{
MQTTStatus_t xResult = MQTTSuccess;
RetryUtilsStatus_t xRetryUtilsStatus = RetryUtilsSuccess;
RetryUtilsParams_t xRetryParams;
/* Some fields not used by this demo so start with everything at 0. */
( void ) memset( ( void * ) &xMQTTSubscription, 0x00, sizeof( xMQTTSubscription ) );
/* Subscribe to the mqttexampleTOPIC topic filter. This example subscribes to
* only one topic and uses QOS0. */
xMQTTSubscription[ 0 ].qos = MQTTQoS0;
xMQTTSubscription[ 0 ].pTopicFilter = mqttexampleTOPIC;
xMQTTSubscription[ 0 ].topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
( void ) memset( ( void * ) &xGlobalSubscribeInfo, 0x00, sizeof( MQTTSubscribeInfo_t ) );
/* Get a unique packet id. */
usSubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );
/* Send SUBSCRIBE packet. */
xResult = MQTT_Subscribe( pxMQTTContext,
xMQTTSubscription,
sizeof( xMQTTSubscription ) / sizeof( MQTTSubscribeInfo_t ),
usSubscribePacketIdentifier );
/* Subscribe to the mqttexampleTOPIC topic filter. This example subscribes to
* only one topic and uses QOS0. */
xGlobalSubscribeInfo.qos = MQTTQoS0;
xGlobalSubscribeInfo.pTopicFilter = mqttexampleTOPIC;
xGlobalSubscribeInfo.topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
configASSERT( xResult == MQTTSuccess );
/* Initialize retry attempts and interval. */
xRetryParams.maxRetryAttempts = MAX_RETRY_ATTEMPTS;
RetryUtils_ParamsReset( &xRetryParams );
do
{
/* The client is now connected to the broker. Subscribe to the topic
* as specified in mqttexampleTOPIC at the top of this file by sending a
* subscribe packet then waiting for a subscribe acknowledgment (SUBACK).
* This client will then publish to the same topic it subscribed to, so it
* will expect all the messages it sends to the broker to be sent back to it
* from the broker. This demo uses QOS0 in Subscribe, therefore, the Publish
* messages received from the broker will have QOS0. */
LogInfo( ( "Attempt to subscribe to the MQTT topic %s.\r\n", mqttexampleTOPIC ) );
xResult = MQTT_Subscribe( pxMQTTContext,
&xGlobalSubscribeInfo,
sizeof( xGlobalSubscribeInfo ) / sizeof( MQTTSubscribeInfo_t ),
usSubscribePacketIdentifier );
configASSERT( xResult == MQTTSuccess );
LogInfo( ( "SUBSCRIBE sent for topic %s to broker.\n\n", mqttexampleTOPIC ) );
/* Process incoming packet from the broker. After sending the subscribe, the
* client may receive a publish before it receives a subscribe ack. Therefore,
* call generic incoming packet processing function. Since this demo is
* subscribing to the topic to which no one is publishing, probability of
* receiving Publish message before subscribe ack is zero; but application
* must be ready to receive any packet. This demo uses the generic packet
* processing function everywhere to highlight this fact. */
xResult = MQTT_ProcessLoop( pxMQTTContext, mqttexamplePROCESS_LOOP_TIMEOUT_MS );
configASSERT( xResult == MQTTSuccess );
/* Check if recent subscription request has been rejected. #xGlobalSubAckStatus is updated
* in eventCallback to reflect the status of the SUBACK sent by the broker. It represents
* either the QoS level granted by the server upon subscription, or acknowledgement of
* server rejection of the subscription request. */
if( xGlobalSubAckStatus == MQTTSubAckFailure )
{
LogWarn( ( "Server rejected subscription request. Attempting to re-subscribe to topic %s.",
mqttexampleTOPIC ) );
xRetryUtilsStatus = RetryUtils_BackoffAndSleep( &xRetryParams );
}
configASSERT( xRetryUtilsStatus != RetryUtilsRetriesExhausted );
} while( ( xGlobalSubAckStatus == MQTTSubAckFailure ) && ( xRetryUtilsStatus == RetryUtilsSuccess ) );
}
/*-----------------------------------------------------------*/
@@ -485,7 +600,6 @@ static void prvMQTTPublishToTopic( MQTTContext_t * pxMQTTContext )
MQTTStatus_t xResult;
MQTTPublishInfo_t xMQTTPublishInfo;
/***
* For readability, error handling in this function is restricted to the use of
* asserts().
@@ -494,7 +608,7 @@ static void prvMQTTPublishToTopic( MQTTContext_t * pxMQTTContext )
/* Some fields not used by this demo so start with everything at 0. */
( void ) memset( ( void * ) &xMQTTPublishInfo, 0x00, sizeof( xMQTTPublishInfo ) );
/* This demo uses QOS0 */
/* This demo uses QOS0. */
xMQTTPublishInfo.qos = MQTTQoS0;
xMQTTPublishInfo.retain = false;
xMQTTPublishInfo.pTopicName = mqttexampleTOPIC;
@@ -512,25 +626,16 @@ static void prvMQTTPublishToTopic( MQTTContext_t * pxMQTTContext )
static void prvMQTTUnsubscribeFromTopic( MQTTContext_t * pxMQTTContext )
{
MQTTStatus_t xResult;
MQTTSubscribeInfo_t xMQTTSubscription[ 1 ];
/* Some fields not used by this demo so start with everything at 0. */
memset( ( void * ) &xMQTTSubscription, 0x00, sizeof( xMQTTSubscription ) );
/* Unsubscribe to the mqttexampleTOPIC topic filter. */
xMQTTSubscription[ 0 ].qos = MQTTQoS0;
xMQTTSubscription[ 0 ].pTopicFilter = mqttexampleTOPIC;
xMQTTSubscription[ 0 ].topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
/* Get next unique packet identifier */
/* Get next unique packet identifier. */
usUnsubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );
/* Make sure the packet id obtained is valid. */
configASSERT( usUnsubscribePacketIdentifier != 0 );
/* Send UNSUBSCRIBE packet. */
/* Send UNSUBSCRIBE packet. Note that because #xGlobalSubscribeInfo
* was initialized before sending the SUBSCRIBE packet, there is no need
* to initialize it again. */
xResult = MQTT_Unsubscribe( pxMQTTContext,
xMQTTSubscription,
sizeof( xMQTTSubscription ) / sizeof( MQTTSubscribeInfo_t ),
&xGlobalSubscribeInfo,
sizeof( xGlobalSubscribeInfo ) / sizeof( MQTTSubscribeInfo_t ),
usUnsubscribePacketIdentifier );
configASSERT( xResult == MQTTSuccess );
@@ -543,7 +648,20 @@ static void prvMQTTProcessResponse( MQTTPacketInfo_t * pxIncomingPacket,
switch( pxIncomingPacket->type )
{
case MQTT_PACKET_TYPE_SUBACK:
LogInfo( ( "Subscribed to the topic %s.\r\n", mqttexampleTOPIC ) );
/* A SUBACK from the broker, containing the server response to our subscription request, has been received.
* It contains the status code indicating server approval/rejection for the subscription to the single topic
* requested. The SUBACK will be parsed to obtain the status code, and this status code will be stored in global
* variable globalSubAckStatus. */
prvUpdateSubAckStatus( pxIncomingPacket );
if( xGlobalSubAckStatus != MQTTSubAckFailure )
{
LogInfo( ( "Subscribed to the topic %s with maximum QoS %u.\r\n",
mqttexampleTOPIC,
xGlobalSubAckStatus ) );
}
/* Make sure ACK packet identifier matches with Request packet identifier. */
configASSERT( usSubscribePacketIdentifier == usPacketId );
break;

2
FreeRTOS-Plus/Demo/FreeRTOS-IoT-Libraries-LTS-Beta2/mqtt/mqtt_plain_text/WIN32.vcxproj
View File

@@ -159,6 +159,7 @@
<ClCompile Include="..\..\..\..\..\FreeRTOS-Plus\Source\FreeRTOS-Plus-TCP\portable\NetworkInterface\WinPCap\NetworkInterface.c" />
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\transport\src\freertos_sockets_wrapper.c" />
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\transport\src\plaintext_freertos.c" />
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\retry_utils\retry_utils_freertos.c" />
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\standard\mqtt\src\mqtt_lightweight.c" />
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\standard\mqtt\src\mqtt_state.c" />
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\standard\mqtt\src\mqtt.c" />
@@ -193,6 +194,7 @@
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\transport\include\freertos_sockets_wrapper.h" />
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\transport\include\plaintext_freertos.h" />
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\include\transport_interface.h" />
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\include\retry_utils.h" />
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\standard\mqtt\include\mqtt_lightweight.h" />
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\standard\mqtt\include\mqtt_state.h" />
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\standard\mqtt\include\mqtt.h" />

6
FreeRTOS-Plus/Demo/FreeRTOS-IoT-Libraries-LTS-Beta2/mqtt/mqtt_plain_text/WIN32.vcxproj.filters
View File

@@ -126,6 +126,9 @@
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\transport\src\freertos_sockets_wrapper.c">
<Filter>FreeRTOS+\FreeRTOS IoT Libraries\platform</Filter>
</ClCompile>
<ClCompile Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\retry_utils\retry_utils_freertos.c">
<Filter>FreeRTOS+\FreeRTOS IoT Libraries\platform</Filter>
</ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\..\..\..\..\FreeRTOS-Plus\Source\FreeRTOS-Plus-TCP\include\NetworkInterface.h">
@@ -216,6 +219,9 @@
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\include\transport_interface.h">
<Filter>FreeRTOS+\FreeRTOS IoT Libraries\platform</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\include\retry_utils.h">
<Filter>FreeRTOS+\FreeRTOS IoT Libraries\platform</Filter>
</ClInclude>
<ClInclude Include="..\..\..\..\Source\FreeRTOS-IoT-Libraries-LTS-Beta2\c_sdk\platform\freertos\transport\include\freertos_sockets_wrapper.h">
<Filter>FreeRTOS+\FreeRTOS IoT Libraries\platform</Filter>
</ClInclude>

101
FreeRTOS-Plus/Source/FreeRTOS-IoT-Libraries-LTS-Beta2/c_sdk/platform/freertos/retry_utils/retry_utils_freertos.c Normal file
View File

@@ -0,0 +1,101 @@
/*
* Copyright (C) 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
/**
* @file retry_utils_freertos.c
* @brief Utility implementation of backoff logic, used for attempting retries of failed processes.
*/
/* Standard includes. */
#include <stdint.h>
/* Kernel includes. */
#include "FreeRTOS.h"
#include "task.h"
#include "retry_utils.h"
#define _MILLISECONDS_PER_SECOND ( 1000U ) /**< @brief Milliseconds per second. */
extern UBaseType_t uxRand( void );
/*-----------------------------------------------------------*/
RetryUtilsStatus_t RetryUtils_BackoffAndSleep( RetryUtilsParams_t * pRetryParams )
{
RetryUtilsStatus_t status = RetryUtilsRetriesExhausted;
int32_t backOffDelayMs = 0;
/* If pRetryParams->maxRetryAttempts is set to 0, try forever. */
if( ( pRetryParams->attemptsDone < pRetryParams->maxRetryAttempts ) ||
( 0 == pRetryParams->maxRetryAttempts ) )
{
/* Choose a random value for back-off time between 0 and the max jitter value. */
backOffDelayMs = uxRand() % pRetryParams->nextJitterMax;
/* Wait for backoff time to expire for the next retry. */
vTaskDelay( pdMS_TO_TICKS( backOffDelayMs * _MILLISECONDS_PER_SECOND ) );
/* Increment backoff counts. */
pRetryParams->attemptsDone++;
/* Double the max jitter value for the next retry attempt, only
* if the new value will be less than the max backoff time value. */
if( pRetryParams->nextJitterMax < ( MAX_RETRY_BACKOFF_SECONDS / 2U ) )
{
pRetryParams->nextJitterMax += pRetryParams->nextJitterMax;
}
else
{
pRetryParams->nextJitterMax = MAX_RETRY_BACKOFF_SECONDS;
}
status = RetryUtilsSuccess;
}
else
{
/* When max retry attempts are exhausted, let application know by
* returning RetryUtilsRetriesExhausted. Application may choose to
* restart the retry process after calling RetryUtils_ParamsReset(). */
status = RetryUtilsRetriesExhausted;
RetryUtils_ParamsReset( pRetryParams );
}
return status;
}
/*-----------------------------------------------------------*/
void RetryUtils_ParamsReset( RetryUtilsParams_t * pRetryParams )
{
uint32_t jitter = 0;
/* Reset attempts done to zero so that the next retry cycle can start. */
pRetryParams->attemptsDone = 0;
/* Calculate jitter value using picking a random number. */
jitter = ( uxRand() % MAX_JITTER_VALUE_SECONDS );
/* Reset the backoff value to the initial time out value plus jitter. */
pRetryParams->nextJitterMax = INITIAL_RETRY_BACKOFF_SECONDS + jitter;
}
/*-----------------------------------------------------------*/

245
FreeRTOS-Plus/Source/FreeRTOS-IoT-Libraries-LTS-Beta2/c_sdk/platform/include/retry_utils.h Normal file
View File

@@ -0,0 +1,245 @@
/*
* Copyright (C) 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
/**
* @file retry_utils.h
* @brief Declaration of the exponential backoff retry logic utility functions
* and constants.
*/
#ifndef RETRY_UTILS_H_
#define RETRY_UTILS_H_
/* Standard include. */
#include <stdint.h>
/**
* @page retryutils_page Retry Utilities
* @brief An abstraction of utilities for retrying with exponential back off and
* jitter.
*
* @section retryutils_overview Overview
* The retry utilities are a set of APIs that aid in retrying with exponential
* backoff and jitter. Exponential backoff with jitter is strongly recommended
* for retrying failed actions over the network with servers. Please see
* https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/ for
* more information about the benefits with AWS.
*
* Exponential backoff with jitter is typically used when retrying a failed
* connection to the server. In an environment with poor connectivity, a client
* can get disconnected at any time. A backoff strategy helps the client to
* conserve battery by not repeatedly attempting reconnections when they are
* unlikely to succeed.
*
* Before retrying the failed communication to the server there is a quiet period.
* In this quiet period, the task that is retrying must sleep for some random
* amount of seconds between 0 and the lesser of a base value and a predefined
* maximum. The base is doubled with each retry attempt until the maximum is
* reached.<br>
*
* > sleep_seconds = random_between( 0, min( 2<sup>attempts_count</sup> * base_seconds, maximum_seconds ) )
*
* @section retryutils_implementation Implementing Retry Utils
*
* The functions that must be implemented are:<br>
* - @ref RetryUtils_ParamsReset
* - @ref RetryUtils_BackoffAndSleep
*
* The functions are used as shown in the diagram below. This is the exponential
* backoff with jitter loop:
*
* @image html retry_utils_flow.png width=25%
*
* The following steps give guidance on implementing the Retry Utils. An example
* implementation of the Retry Utils for a POSIX platform can be found in file
* @ref retry_utils_posix.c.
*
* -# Implementing @ref RetryUtils_ParamsReset
* @snippet this define_retryutils_paramsreset
*<br>
* This function initializes @ref RetryUtilsParams_t. It is expected to set
* @ref RetryUtilsParams_t.attemptsDone to zero. It is also expected to set
* @ref RetryUtilsParams_t.nextJitterMax to @ref INITIAL_RETRY_BACKOFF_SECONDS
* plus some random amount of seconds, jitter. This jitter is a random number
* between 0 and @ref MAX_JITTER_VALUE_SECONDS. This function must be called
* before entering the exponential backoff with jitter loop using
* @ref RetryUtils_BackoffAndSleep.<br><br>
* Please follow the example below to implement your own @ref RetryUtils_ParamsReset.
* The lines with FIXME comments should be updated.
* @code{c}
* void RetryUtils_ParamsReset( RetryUtilsParams_t * pRetryParams )
* {
* uint32_t jitter = 0;
*
* // Reset attempts done to zero so that the next retry cycle can start.
* pRetryParams->attemptsDone = 0;
*
* // Seed pseudo random number generator with the current time. FIXME: Your
* // system may have another method to retrieve the current time to seed the
* // pseudo random number generator.
* srand( time( NULL ) );
*
* // Calculate jitter value using picking a random number.
* jitter = ( rand() % MAX_JITTER_VALUE_SECONDS );
*
* // Reset the backoff value to the initial time out value plus jitter.
* pRetryParams->nextJitterMax = INITIAL_RETRY_BACKOFF_SECONDS + jitter;
* }
* @endcode<br>
*
* -# Implementing @ref RetryUtils_BackoffAndSleep
* @snippet this define_retryutils_backoffandsleep
* <br>
* When this function is invoked, the calling task is expected to sleep a random
* number of seconds between 0 and @ref RetryUtilsParams_t.nextJitterMax. After
* sleeping this function must double @ref RetryUtilsParams_t.nextJitterMax, but
* not exceeding @ref MAX_RETRY_BACKOFF_SECONDS. When @ref RetryUtilsParams_t.maxRetryAttempts
* are reached this function should return @ref RetryUtilsRetriesExhausted, unless
* @ref RetryUtilsParams_t.maxRetryAttempts is set to zero.
* When @ref RetryUtilsRetriesExhausted is returned the calling application can
* stop trying with a failure, or it can call @ref RetryUtils_ParamsReset again
* and restart the exponential back off with jitter loop.<br><br>
* Please follow the example below to implement your own @ref RetryUtils_BackoffAndSleep.
* The lines with FIXME comments should be updated.
* @code{c}
* RetryUtilsStatus_t RetryUtils_BackoffAndSleep( RetryUtilsParams_t * pRetryParams )
* {
* RetryUtilsStatus_t status = RetryUtilsRetriesExhausted;
* // The quiet period delay in seconds.
* int backOffDelay = 0;
*
* // If pRetryParams->maxRetryAttempts is set to 0, try forever.
* if( ( pRetryParams->attemptsDone < pRetryParams->maxRetryAttempts ) ||
* ( 0 == pRetryParams->maxRetryAttempts ) )
* {
* // Choose a random value for back-off time between 0 and the max jitter value.
* backOffDelay = rand() % pRetryParams->nextJitterMax;
*
* // Wait for backoff time to expire for the next retry.
* ( void ) myThreadSleepFunction( backOffDelay ); // FIXME: Replace with your system's thread sleep function.
*
* // Increment backoff counts.
* pRetryParams->attemptsDone++;
*
* // Double the max jitter value for the next retry attempt, only
* // if the new value will be less than the max backoff time value.
* if( pRetryParams->nextJitterMax < ( MAX_RETRY_BACKOFF_SECONDS / 2U ) )
* {
* pRetryParams->nextJitterMax += pRetryParams->nextJitterMax;
* }
* else
* {
* pRetryParams->nextJitterMax = MAX_RETRY_BACKOFF_SECONDS;
* }
*
* status = RetryUtilsSuccess;
* }
* else
* {
* // When max retry attempts are exhausted, let application know by
* // returning RetryUtilsRetriesExhausted. Application may choose to
* // restart the retry process after calling RetryUtils_ParamsReset().
* status = RetryUtilsRetriesExhausted;
* RetryUtils_ParamsReset( pRetryParams );
* }
*
* return status;
* }
* @endcode
*/
/**
* @brief Max number of retry attempts. Set this value to 0 if the client must
* retry forever.
*/
#define MAX_RETRY_ATTEMPTS 4U
/**
* @brief Initial fixed backoff value in seconds between two successive
* retries. A random jitter value is added to every backoff value.
*/
#define INITIAL_RETRY_BACKOFF_SECONDS 1U
/**
* @brief Max backoff value in seconds.
*/
#define MAX_RETRY_BACKOFF_SECONDS 128U
/**
* @brief Max jitter value in seconds.
*/
#define MAX_JITTER_VALUE_SECONDS 5U
/**
* @brief Status for @ref RetryUtils_BackoffAndSleep.
*/
typedef enum RetryUtilsStatus
{
RetryUtilsSuccess = 0, /**< @brief The function returned successfully after sleeping. */
RetryUtilsRetriesExhausted /**< @brief The function exhausted all retry attempts. */
} RetryUtilsStatus_t;
/**
* @brief Represents parameters required for retry logic.
*/
typedef struct RetryUtilsParams
{
/**
* @brief Max number of retry attempts. Set this value to 0 if the client must
* retry forever.
*/
uint32_t maxRetryAttempts;
/**
* @brief The cumulative count of backoff delay cycles completed
* for retries.
*/
uint32_t attemptsDone;
/**
* @brief The max jitter value for backoff time in retry attempt.
*/
uint32_t nextJitterMax;
} RetryUtilsParams_t;
/**
* @brief Resets the retry timeout value and number of attempts.
* This function must be called by the application before a new retry attempt.
*
* @param[in, out] pRetryParams Structure containing attempts done and timeout
* value.
*/
void RetryUtils_ParamsReset( RetryUtilsParams_t * pRetryParams );
/**
* @brief Simple platform specific exponential backoff function. The application
* must use this function between retry failures to add exponential delay.
* This function will block the calling task for the current timeout value.
*
* @param[in, out] pRetryParams Structure containing retry parameters.
*
* @return #RetryUtilsSuccess after a successful sleep, #RetryUtilsRetriesExhausted
* when all attempts are exhausted.
*/
RetryUtilsStatus_t RetryUtils_BackoffAndSleep( RetryUtilsParams_t * pRetryParams );
#endif /* ifndef RETRY_UTILS_H_ */