Incorporating PIC state machine level retry changes into webrtc signaling state machine (#1326)

* Incorporating PIC state machine level retry changes into webrtc signaling state machine

* Signaling state machine rework (#1323)

* replaced recursive calls to stepSignalingStateMachine with loops

* removed recursive calls to stepSignalingStateMachine

* replaced stepSignalingstatemachine with signalingStateMachineIterator

* removed stepUntil, continueOnReady; removed status from iterator signature; set signalingclient version; added a lock in refreshIceConfiguration

* changed the declaration for i from int to UINT32

* added signaling version; removed unnecessary comment

* removed redeclaration; changed value in an existing macro

* Fix issue with API call failures being treated as success (#1328)

* return proper error, do not reset call result value

* for non retriable failures, set the terminal exit status for state in… (#1320)

* for non retriable failures, set the terminal exit status for state in state machine

* address comments

* adjust tests set retry max to 1

* Update LwsApiCalls.c

trigger travis ci

* Add retry strategy to client info to avoiud changing create signaling channel API signature

* Incorporating PIC state machine level retry changes into webrtc signaling state machine

* Add retry strategy to client info to avoiud changing create signaling channel API signature

* Adding more debug logs in the code

* fix merge conflicts

* PR feedback

Co-authored-by: Niyati Maheshwari <niyatim23@gmail.com>
Co-authored-by: Hassan Sahibzada <hsahibza@amazon.com>

Author: 3 people

CMake/Dependencies/libkvsCommonLws-CMakeLists.txt

@@ -6,7 +6,7 @@ include(ExternalProject)
 
 ExternalProject_Add(libkvsCommonLws-download
     GIT_REPOSITORY    https://github.com/awslabs/amazon-kinesis-video-streams-producer-c.git
-    GIT_TAG           99c1a8cd8cec88f99c9c4ce3944b53ae341d1491
+    GIT_TAG           9a995a5793b4024f19912be9a319993b1e16005c
     PREFIX            ${CMAKE_CURRENT_BINARY_DIR}/build
     CMAKE_ARGS        
       -DCMAKE_INSTALL_PREFIX=${OPEN_SRC_INSTALL_PREFIX}

samples/Common.c

@@ -697,6 +697,29 @@ STATUS lookForSslCert(PSampleConfiguration* ppSampleConfiguration)
     return retStatus;
 }
 
+STATUS setupDefaultSignalingClientRetryStrategy(PSignalingClientInfo pSignalingClientInfo)
+{
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+    PRetryStrategy pRetryStrategy = NULL;
+
+    CHK(pSignalingClientInfo != NULL, STATUS_NULL_ARG);
+
+    pSignalingClientInfo->signalingClientRetryStrategy.retryStrategyType = KVS_RETRY_STRATEGY_EXPONENTIAL_BACKOFF_WAIT;
+    pSignalingClientInfo->signalingClientRetryStrategy.createRetryStrategyFn = exponentialBackoffRetryStrategyCreate;
+    pSignalingClientInfo->signalingClientRetryStrategy.freeRetryStrategyFn = exponentialBackoffRetryStrategyFree;
+    pSignalingClientInfo->signalingClientRetryStrategy.executeRetryStrategyFn = getExponentialBackoffRetryStrategyWaitTime;
+
+    CHK_STATUS(pSignalingClientInfo->signalingClientRetryStrategy.createRetryStrategyFn(NULL /* use default config */, &pRetryStrategy));
+    pSignalingClientInfo->signalingClientRetryStrategy.pRetryStrategy = pRetryStrategy;
+
+    pSignalingClientInfo->signalingClientCreationMaxRetryCount = MAX_CREATE_SIGNALING_CLIENT_RETRIES;
+
+CleanUp:
+    LEAVES();
+    return retStatus;
+}
+
 STATUS createSampleConfiguration(PCHAR channelName, SIGNALING_CHANNEL_ROLE_TYPE roleType, BOOL trickleIce, BOOL useTurn,
                                  PSampleConfiguration* ppSampleConfiguration)
 {
@@ -782,6 +805,8 @@ STATUS createSampleConfiguration(PCHAR channelName, SIGNALING_CHANNEL_ROLE_TYPE
     pSampleConfiguration->clientInfo.version = SIGNALING_CLIENT_INFO_CURRENT_VERSION;
     pSampleConfiguration->clientInfo.loggingLevel = logLevel;
     pSampleConfiguration->clientInfo.cacheFilePath = NULL; // Use the default path
+    CHK_STATUS(setupDefaultSignalingClientRetryStrategy(&pSampleConfiguration->clientInfo));
+
     pSampleConfiguration->iceCandidatePairStatsTimerId = MAX_UINT32;
     pSampleConfiguration->pregenerateCertTimerId = MAX_UINT32;
 
@@ -1063,6 +1088,11 @@ STATUS freeSampleConfiguration(PSampleConfiguration* ppSampleConfiguration)
     SAFE_MEMFREE(pSampleConfiguration->pVideoFrameBuffer);
     SAFE_MEMFREE(pSampleConfiguration->pAudioFrameBuffer);
 
+    if (pSampleConfiguration->clientInfo.signalingClientRetryStrategy.freeRetryStrategyFn != NULL) {
+        CHK_STATUS(pSampleConfiguration->clientInfo.signalingClientRetryStrategy.freeRetryStrategyFn(
+                &(pSampleConfiguration->clientInfo.signalingClientRetryStrategy.pRetryStrategy)));
+    }
+
     if (IS_VALID_CVAR_VALUE(pSampleConfiguration->cvar) && IS_VALID_MUTEX_VALUE(pSampleConfiguration->sampleConfigurationObjLock)) {
         CVAR_BROADCAST(pSampleConfiguration->cvar);
         MUTEX_LOCK(pSampleConfiguration->sampleConfigurationObjLock);

samples/Samples.h

@@ -39,6 +39,7 @@ extern "C" {
 
 #define SAMPLE_HASH_TABLE_BUCKET_COUNT  50
 #define SAMPLE_HASH_TABLE_BUCKET_LENGTH 2
+#define MAX_CREATE_SIGNALING_CLIENT_RETRIES 3
 
 #define IOT_CORE_CREDENTIAL_ENDPOINT ((PCHAR) "AWS_IOT_CORE_CREDENTIAL_ENDPOINT")
 #define IOT_CORE_CERT                ((PCHAR) "AWS_IOT_CORE_CERT")

src/include/com/amazonaws/kinesis/video/webrtcclient/Include.h

@@ -660,6 +660,9 @@ extern "C" {
 typedef UINT64 SIGNALING_CLIENT_HANDLE;
 typedef SIGNALING_CLIENT_HANDLE* PSIGNALING_CLIENT_HANDLE;
 
+typedef KvsRetryStrategy SignalingClientRetryStrategy;
+typedef PKvsRetryStrategy PSignalingClientRetryStrategy;
+
 /**
  * @brief This is a sentinel indicating an invalid handle value
  */
@@ -1177,16 +1180,18 @@ typedef struct {
  * @brief Populate Signaling client with client ID and application log level
  */
 typedef struct {
-    UINT32 version;                                 //!< Version of the structure
-    CHAR clientId[MAX_SIGNALING_CLIENT_ID_LEN + 1]; //!< Client id to use. Defines if the client is a producer/consumer
-    UINT32 loggingLevel;                            //!< Verbosity level for the logging. One of LOG_LEVEL_XXX
-                                                    //!< values or the default verbosity will be assumed. Currently,
-                                                    //!< default value is LOG_LEVEL_WARNING
-    PCHAR cacheFilePath;                            //!< File cache path override. The default
-                                                    //!< path is "./.SignalingCache_vN" which might not work for
-                                                    //!< devices which have read only partition where the code is
-                                                    //!< located. For default value or when file caching is not
-                                                    //!< being used this value can be NULL or point to an EMPTY_STRING.
+    UINT32 version;                                                 //!< Version of the structure
+    CHAR clientId[MAX_SIGNALING_CLIENT_ID_LEN + 1];                 //!< Client id to use. Defines if the client is a producer/consumer
+    UINT32 loggingLevel;                                            //!< Verbosity level for the logging. One of LOG_LEVEL_XXX
+                                                                    //!< values or the default verbosity will be assumed. Currently,
+                                                                    //!< default value is LOG_LEVEL_WARNING
+    PCHAR cacheFilePath;                                            //!< File cache path override. The default
+                                                                    //!< path is "./.SignalingCache_vN" which might not work for
+                                                                    //!< devices which have read only partition where the code is
+                                                                    //!< located. For default value or when file caching is not
+                                                                    //!< being used this value can be NULL or point to an EMPTY_STRING.
+    SignalingClientRetryStrategy signalingClientRetryStrategy;      //!< Retry strategy used while creating signaling client
+    UINT32 signalingClientCreationMaxRetryCount;                    //!< Maximum attempts which createSignalingClientSync API will make on failures to create signaling client
 } SignalingClientInfo, *PSignalingClientInfo;
 
 /**

src/source/Signaling/Client.c

@@ -1,13 +1,37 @@
 #define LOG_CLASS "SignalingClient"
 #include "../Include_i.h"
 
+STATUS validateSignalingClientRetryStrategy(PSignalingClientInfo pClientInfo) {
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+    PSignalingClientRetryStrategy pSignalingClientRetryStrategy;
+
+    CHK(pClientInfo != NULL, STATUS_NULL_ARG);
+
+    pSignalingClientRetryStrategy = &(pClientInfo->signalingClientRetryStrategy);
+
+    CHK(pSignalingClientRetryStrategy->retryStrategyType > KVS_RETRY_STRATEGY_DISABLED &&
+    pSignalingClientRetryStrategy->pRetryStrategy != NULL &&
+    pSignalingClientRetryStrategy->executeRetryStrategyFn != NULL, STATUS_NULL_ARG);
+
+    CHK(pClientInfo->signalingClientCreationMaxRetryCount > 0, STATUS_NOT_IMPLEMENTED);
+
+CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
 STATUS createSignalingClientSync(PSignalingClientInfo pClientInfo, PChannelInfo pChannelInfo, PSignalingClientCallbacks pCallbacks,
                                  PAwsCredentialProvider pCredentialProvider, PSIGNALING_CLIENT_HANDLE pSignalingHandle)
 {
     ENTERS();
     STATUS retStatus = STATUS_SUCCESS;
     PSignalingClient pSignalingClient = NULL;
+    PSignalingClientRetryStrategy pSignalingClientRetryStrategy = NULL;
     SignalingClientInfoInternal signalingClientInfoInternal;
+    UINT32 signalingClientCreationMaxRetryCount;
+    UINT64 signalingClientCreationWaitTime;
 
     DLOGV("Creating Signaling Client Sync");
     CHK(pSignalingHandle != NULL && pClientInfo != NULL, STATUS_NULL_ARG);
@@ -16,7 +40,28 @@ STATUS createSignalingClientSync(PSignalingClientInfo pClientInfo, PChannelInfo
     MEMSET(&signalingClientInfoInternal, 0x00, SIZEOF(signalingClientInfoInternal));
     signalingClientInfoInternal.signalingClientInfo = *pClientInfo;
 
-    CHK_STATUS(createSignalingSync(&signalingClientInfoInternal, pChannelInfo, pCallbacks, pCredentialProvider, &pSignalingClient));
+    CHK_STATUS(validateSignalingClientRetryStrategy(pClientInfo));
+
+    signalingClientCreationMaxRetryCount = pClientInfo->signalingClientCreationMaxRetryCount;
+    pSignalingClientRetryStrategy = &(pClientInfo->signalingClientRetryStrategy);
+
+    while (signalingClientCreationMaxRetryCount > 0) {
+        // Wait before cresting signaling client to ensure the first call from a large
+        // client fleet will be spread across the wait time window.
+        CHK_STATUS(pSignalingClientRetryStrategy->executeRetryStrategyFn(pSignalingClientRetryStrategy->pRetryStrategy, &signalingClientCreationWaitTime));
+        DLOGV("Attempting to back off for [%lf] milliseconds before creating signaling client. Signaling client creation retry count [%d]",
+              signalingClientCreationWaitTime/1000.0, signalingClientCreationMaxRetryCount);
+        THREAD_SLEEP(signalingClientCreationWaitTime);
+
+        retStatus = createSignalingSync(&signalingClientInfoInternal, pChannelInfo, pCallbacks, pCredentialProvider, &pSignalingClient);
+        if (retStatus == STATUS_SUCCESS) {
+            break;
+        }
+        signalingClientCreationMaxRetryCount--;
+    }
+
+    DLOGV("Create signaling client returned [%" PRId64 "].", retStatus);
+    CHK_STATUS(retStatus);
 
     *pSignalingHandle = TO_SIGNALING_CLIENT_HANDLE(pSignalingClient);
 

src/source/Signaling/Signaling.c

@@ -37,9 +37,8 @@ STATUS createSignalingSync(PSignalingClientInfoInternal pClientInfo, PChannelInf
     CHK_STATUS(validateSignalingCallbacks(pSignalingClient, pCallbacks));
     CHK_STATUS(validateSignalingClientInfo(pSignalingClient, pClientInfo));
 
-    pSignalingClient->version = SIGNALING_CLIENT_CURRENT_VERSION;
-
     // Set invalid call times
+    pSignalingClient->version = SIGNALING_CLIENT_CURRENT_VERSION;
     pSignalingClient->describeTime = INVALID_TIMESTAMP_VALUE;
     pSignalingClient->createTime = INVALID_TIMESTAMP_VALUE;
     pSignalingClient->getEndpointTime = INVALID_TIMESTAMP_VALUE;
@@ -74,6 +73,9 @@ STATUS createSignalingSync(PSignalingClientInfoInternal pClientInfo, PChannelInf
     // Store the credential provider
     pSignalingClient->pCredentialProvider = pCredentialProvider;
 
+    // Configure retry strategy for retries on error within signaling state machine
+    CHK_STATUS(configureRetryStrategyForSignalingStateMachine(pSignalingClient));
+
     // Create the state machine
     CHK_STATUS(createStateMachine(SIGNALING_STATE_MACHINE_STATES, SIGNALING_STATE_MACHINE_STATE_COUNT,
                                   CUSTOM_DATA_FROM_SIGNALING_CLIENT(pSignalingClient), signalingGetCurrentTime,
@@ -205,6 +207,8 @@ STATUS freeSignaling(PSignalingClient* ppSignalingClient)
         MUTEX_UNLOCK(pSignalingClient->lwsServiceLock);
     }
 
+    freeClientRetryStrategy(pSignalingClient);
+
     freeStateMachine(pSignalingClient->pStateMachine);
 
     freeChannelInfo(&pSignalingClient->pChannelInfo);
@@ -516,6 +520,52 @@ STATUS validateSignalingClientInfo(PSignalingClient pSignalingClient, PSignaling
     return retStatus;
 }
 
+STATUS configureRetryStrategyForSignalingStateMachine(PSignalingClient pSignalingClient) {
+    ENTERS();
+    PRetryStrategy pRetryStrategy = NULL;
+    STATUS retStatus = STATUS_SUCCESS;
+    KVS_RETRY_STRATEGY_TYPE defaultKvsRetryStrategyType = KVS_RETRY_STRATEGY_EXPONENTIAL_BACKOFF_WAIT;
+
+    CHK(pSignalingClient != NULL, STATUS_NULL_ARG);
+    pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.retryStrategyType = KVS_RETRY_STRATEGY_EXPONENTIAL_BACKOFF_WAIT;
+    pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.createRetryStrategyFn = exponentialBackoffRetryStrategyCreate;
+    pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.freeRetryStrategyFn = exponentialBackoffRetryStrategyFree;
+    pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.executeRetryStrategyFn = getExponentialBackoffRetryStrategyWaitTime;
+
+    CHK_STATUS(pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.createRetryStrategyFn(
+            NULL, &pRetryStrategy));
+
+    if (pRetryStrategy == NULL) {
+        DLOGD("Unable to create exponential backoff retry strategy. This should not happen.");
+    }
+
+    CHK(pRetryStrategy != NULL, STATUS_INTERNAL_ERROR);
+    pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.pRetryStrategy = pRetryStrategy;
+
+    CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
+STATUS freeClientRetryStrategy(PSignalingClient pSignalingClient) {
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+
+    CHK(pSignalingClient != NULL &&
+        pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.freeRetryStrategyFn != NULL, STATUS_SUCCESS);
+
+    CHK_STATUS(pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.freeRetryStrategyFn(
+            &(pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.pRetryStrategy)));
+
+    pSignalingClient->clientInfo.signalingStateMachineRetryStrategy.pRetryStrategy = NULL;
+
+    CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
 STATUS validateIceConfiguration(PSignalingClient pSignalingClient)
 {
     ENTERS();

src/source/Signaling/Signaling.h

@@ -107,6 +107,9 @@ typedef struct {
     SignalingApiCallHookFunc connectPostHookFn;
     SignalingApiCallHookFunc deletePreHookFn;
     SignalingApiCallHookFunc deletePostHookFn;
+
+    // Retry strategy for signaling state machine
+    KvsRetryStrategy signalingStateMachineRetryStrategy;
 } SignalingClientInfoInternal, *PSignalingClientInfoInternal;
 
 /**
@@ -281,6 +284,14 @@ typedef struct {
     UINT64 connectTime;
 } SignalingClient, *PSignalingClient;
 
+static const ExponentialBackoffRetryStrategyConfig DEFAULT_SIGNALING_STATE_MACHINE_EXPONENTIAL_RETRY_CONFIG = {
+        KVS_INFINITE_EXPONENTIAL_RETRIES, /* max retry count */
+        10000, /* max retry wait time milliseconds */
+        300, /* retry factor (base retry wait time milliseconds) */
+        25000, /* minimum time in milliseconds to reset retry state */
+        300 /* jitter factor milliseconds (jitter will be unused for FULL_JITTER variant of exponential backoff algorithm) */
+};
+
 // Public handle to and from object converters
 #define TO_SIGNALING_CLIENT_HANDLE(p)   ((SIGNALING_CLIENT_HANDLE) (p))
 #define FROM_SIGNALING_CLIENT_HANDLE(h) (IS_VALID_SIGNALING_CLIENT_HANDLE(h) ? (PSignalingClient) (h) : NULL)
@@ -299,6 +310,10 @@ STATUS validateSignalingCallbacks(PSignalingClient, PSignalingClientCallbacks);
 STATUS validateSignalingClientInfo(PSignalingClient, PSignalingClientInfoInternal);
 STATUS validateIceConfiguration(PSignalingClient);
 
+STATUS configureRetryStrategyForSignalingStateMachine(PSignalingClient);
+STATUS setupDefaultKvsRetryStrategy(PSignalingClient);
+STATUS freeClientRetryStrategy(PSignalingClient);
+
 STATUS signalingStoreOngoingMessage(PSignalingClient, PSignalingMessage);
 STATUS signalingRemoveOngoingMessage(PSignalingClient, PCHAR);
 STATUS signalingGetOngoingMessage(PSignalingClient, PCHAR, PCHAR, PSignalingMessage*);