Adding error handler in PIC state machine and updating client/stream state machines to use the error handler on state transitions (#143)

* The commit includes following -
(1) Refactoring Exponential backoff utility by changing name(s) to use 'RetryStrategy'
(2) Minor logging improvements in  Exponential backoff utility
(3) Adding KVS retry strategy and related callbacks to be used in client and streams
(4) Adding a new error handler in PIC state machine
(5) Updating KVS client and stream state machines to use PIC error handler and exponential backoff retry strategy on encountering error while state transitions

* - Adding client callback to get retry strategy. This is particularly needed for unit testing
- Addressing PR feedback

* - Updating UTs to handle memory leak for offline streaming
- Addressing PR feedback

* Addressing PR feedback

* Update Client.c

trigger travis CI

* All events will have a unique L1 TAGS. (#145)

* Events mkv size bug (#144)

* committing to test out changes, added ability to increment the size of tags after appending additional SimpleTags

* updating logs

* tagsStart initialization was incorrect

* cleanup debug logs

* Unit test and clang sanitization

* accidental leftover comment forward slash

* Memory leak in unit test

* Removing if else that doesn't cover existing cases

* Unnecessary setting of variables that will not be used past this point in the function

Co-authored-by: Hassan Sahibzada <[email protected]>

* Log git hash

Co-authored-by: Hassan Sahibzada <[email protected]>
Co-authored-by: jdelapla <[email protected]>
Co-authored-by: Divya Sampath Kumar <[email protected]>

Author: 4 people

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

@@ -1114,6 +1114,21 @@ struct __StorageInfo {
 
 typedef struct __StorageInfo* PStorageInfo;
 
+/**
+ * Function to create a retry strategy to be used for the client and streams
+ * on detecting errors from service API calls. The retry strategy is a part of
+ * client struct.
+ *
+ * NOTE: This is an optional callback. If not specified, the SDK will use
+ * default retry strategy. Unless there is no specific use case, it is
+ * recommended to leave this callback NULL and let SDK handle the retries.
+ *
+ * @param 1 CLIENT_HANDLE - IN - The client handle.
+ *
+ * @return Status of the callback
+ */
+typedef STATUS (*GetKvsRetryStrategyFn)(CLIENT_HANDLE);
+
 /**
  * Client Info
  */
@@ -1152,6 +1167,12 @@ typedef struct __ClientInfo {
     // period (in hundreds of nanos) at which callback will be fired to check stream
     // clients should set this value to 0.
     UINT64 reservedCallbackPeriod;
+
+    // Retry strategy for the client and all the streams under it
+    KvsRetryStrategy kvsRetryStrategy;
+
+    // Function pointer for application to provide a custom retry strategy
+    GetKvsRetryStrategyFn getKvsRetryStrategyFn;
 } ClientInfo, *PClientInfo;
 
 /**

src/client/src/Client.c

@@ -83,7 +83,6 @@ STATUS checkIntermittentProducerCallback(UINT32 timerId, UINT64 currentTime, UIN
                 }
             }
         }
-
         pKinesisVideoClient->clientCallbacks.unlockMutexFn(pKinesisVideoClient->clientCallbacks.customData, pKinesisVideoClient->base.streamListLock);
     }
 
@@ -94,6 +93,67 @@ STATUS checkIntermittentProducerCallback(UINT32 timerId, UINT64 currentTime, UIN
     return retStatus;
 }
 
+STATUS setupDefaultKvsRetryStrategy(PKinesisVideoClient pKinesisVideoClient) {
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+
+    // Use default as exponential backoff wait
+    pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.createRetryStrategyFn = exponentialBackoffRetryStrategyCreate;
+    pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.freeRetryStrategyFn = exponentialBackoffRetryStrategyFree;
+    pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.executeRetryStrategyFn = getExponentialBackoffRetryStrategyWaitTime;
+    pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.retryStrategyType = KVS_RETRY_STRATEGY_EXPONENTIAL_BACKOFF_WAIT;
+
+CleanUp:
+    LEAVES();
+    return retStatus;
+}
+
+STATUS configureClientWithRetryStrategy(PKinesisVideoClient pKinesisVideoClient) {
+    ENTERS();
+    PRetryStrategy pRetryStrategy = NULL;
+    STATUS retStatus = STATUS_SUCCESS;
+    KVS_RETRY_STRATEGY_TYPE defaultKvsRetryStrategyType = KVS_RETRY_STRATEGY_EXPONENTIAL_BACKOFF_WAIT;
+
+    CHK(pKinesisVideoClient != NULL, STATUS_NULL_ARG);
+
+    // If the callbacks for retry strategy are already set, then use that otherwise
+    // build the client with a default retry strategy.
+    if (pKinesisVideoClient->deviceInfo.clientInfo.getKvsRetryStrategyFn != NULL) {
+        CHK_STATUS(pKinesisVideoClient->deviceInfo.clientInfo.getKvsRetryStrategyFn(TO_CLIENT_HANDLE(pKinesisVideoClient)));
+    } else {
+        CHK_STATUS(setupDefaultKvsRetryStrategy(pKinesisVideoClient));
+    }
+
+    if (pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.createRetryStrategyFn != NULL) {
+        CHK_STATUS(pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.createRetryStrategyFn(
+            NULL /* use default retry strategy configuration */, &pRetryStrategy));
+        pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.pRetryStrategy = pRetryStrategy;
+    }
+
+CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
+STATUS freeClientRetryStrategy(PKinesisVideoClient pKinesisVideoClient) {
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+
+    CHK(pKinesisVideoClient != NULL &&
+            pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.pRetryStrategy != NULL &&
+            pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.freeRetryStrategyFn != NULL, STATUS_SUCCESS);
+
+    CHK_STATUS(pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.freeRetryStrategyFn(
+            &(pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.pRetryStrategy)));
+    pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy.pRetryStrategy = NULL;
+
+CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
 /**
  * Creates a client
  */
@@ -158,6 +218,8 @@ STATUS createKinesisVideoClient(PDeviceInfo pDeviceInfo, PClientCallbacks pClien
     // Copy the structures in their entirety
     MEMCPY(&pKinesisVideoClient->clientCallbacks, pClientCallbacks, SIZEOF(ClientCallbacks));
 
+    CHK_STATUS(configureClientWithRetryStrategy(pKinesisVideoClient));
+
     // Fix-up the defaults if needed
     // IMPORTANT!!! The calloc allocator will zero the memory which will also act as a
     // sentinel value in case of an earlier version of the structure
@@ -1613,6 +1675,9 @@ STATUS freeKinesisVideoClientInternal(PKinesisVideoClient pKinesisVideoClient, S
         gKinesisVideoClient = NULL;
     }
 
+    // Free retry strategy
+    freeClientRetryStrategy(pKinesisVideoClient);
+
     // Release the heap
     if (pKinesisVideoClient->pHeap) {
         heapDebugCheckAllocator(pKinesisVideoClient->pHeap, TRUE);

src/client/src/ClientState.c

@@ -9,24 +9,107 @@
  * Static definitions of the states
  */
 StateMachineState CLIENT_STATE_MACHINE_STATES[] = {
-    {CLIENT_STATE_NEW, CLIENT_STATE_NONE | CLIENT_STATE_NEW, fromNewClientState, executeNewClientState, INFINITE_RETRY_COUNT_SENTINEL,
-     STATUS_INVALID_CLIENT_READY_STATE},
-    {CLIENT_STATE_AUTH, CLIENT_STATE_READY | CLIENT_STATE_NEW | CLIENT_STATE_AUTH, fromAuthClientState, executeAuthClientState,
-     SERVICE_CALL_MAX_RETRY_COUNT, STATUS_CLIENT_AUTH_CALL_FAILED},
-    {CLIENT_STATE_GET_TOKEN, CLIENT_STATE_AUTH | CLIENT_STATE_PROVISION | CLIENT_STATE_GET_TOKEN, fromGetTokenClientState, executeGetTokenClientState,
-     SERVICE_CALL_MAX_RETRY_COUNT, STATUS_GET_CLIENT_TOKEN_CALL_FAILED},
-    {CLIENT_STATE_PROVISION, CLIENT_STATE_AUTH | CLIENT_STATE_PROVISION, fromProvisionClientState, executeProvisionClientState,
-     SERVICE_CALL_MAX_RETRY_COUNT, STATUS_CLIENT_PROVISION_CALL_FAILED},
-    {CLIENT_STATE_CREATE, CLIENT_STATE_PROVISION | CLIENT_STATE_GET_TOKEN | CLIENT_STATE_AUTH | CLIENT_STATE_CREATE, fromCreateClientState,
-     executeCreateClientState, SERVICE_CALL_MAX_RETRY_COUNT, STATUS_CREATE_CLIENT_CALL_FAILED},
-    {CLIENT_STATE_TAG_CLIENT, CLIENT_STATE_CREATE | CLIENT_STATE_TAG_CLIENT | CLIENT_STATE_READY, fromTagClientState, executeTagClientState,
-     SERVICE_CALL_MAX_RETRY_COUNT, STATUS_TAG_CLIENT_CALL_FAILED},
-    {CLIENT_STATE_READY, CLIENT_STATE_GET_TOKEN | CLIENT_STATE_AUTH | CLIENT_STATE_TAG_CLIENT | CLIENT_STATE_CREATE | CLIENT_STATE_READY,
-     fromReadyClientState, executeReadyClientState, INFINITE_RETRY_COUNT_SENTINEL, STATUS_CLIENT_READY_CALLBACK_FAILED},
+    {
+        CLIENT_STATE_NEW,
+        CLIENT_STATE_NONE | CLIENT_STATE_NEW,
+        fromNewClientState,
+        executeNewClientState,
+        defaultClientStateTransitionHook,
+        INFINITE_RETRY_COUNT_SENTINEL,
+        STATUS_INVALID_CLIENT_READY_STATE
+    },
+    {
+        CLIENT_STATE_AUTH,
+        CLIENT_STATE_READY | CLIENT_STATE_NEW | CLIENT_STATE_AUTH,
+        fromAuthClientState,
+        executeAuthClientState,
+        defaultClientStateTransitionHook,
+        SERVICE_CALL_MAX_RETRY_COUNT,
+        STATUS_CLIENT_AUTH_CALL_FAILED
+    },
+    {
+        CLIENT_STATE_GET_TOKEN,
+        CLIENT_STATE_AUTH | CLIENT_STATE_PROVISION | CLIENT_STATE_GET_TOKEN,
+        fromGetTokenClientState,
+        executeGetTokenClientState,
+        defaultClientStateTransitionHook,
+        SERVICE_CALL_MAX_RETRY_COUNT,
+        STATUS_GET_CLIENT_TOKEN_CALL_FAILED
+    },
+    {
+        CLIENT_STATE_PROVISION,
+        CLIENT_STATE_AUTH | CLIENT_STATE_PROVISION,
+        fromProvisionClientState,
+        executeProvisionClientState,
+        defaultClientStateTransitionHook,
+        SERVICE_CALL_MAX_RETRY_COUNT,
+        STATUS_CLIENT_PROVISION_CALL_FAILED
+    },
+    {
+        CLIENT_STATE_CREATE,
+        CLIENT_STATE_PROVISION | CLIENT_STATE_GET_TOKEN | CLIENT_STATE_AUTH | CLIENT_STATE_CREATE,
+        fromCreateClientState,
+        executeCreateClientState,
+        defaultClientStateTransitionHook,
+        SERVICE_CALL_MAX_RETRY_COUNT,
+        STATUS_CREATE_CLIENT_CALL_FAILED
+    },
+    {
+        CLIENT_STATE_TAG_CLIENT,
+        CLIENT_STATE_CREATE | CLIENT_STATE_TAG_CLIENT | CLIENT_STATE_READY,
+        fromTagClientState,
+        executeTagClientState,
+        defaultClientStateTransitionHook,
+        SERVICE_CALL_MAX_RETRY_COUNT,
+        STATUS_TAG_CLIENT_CALL_FAILED
+    },
+    {
+        CLIENT_STATE_READY,
+        CLIENT_STATE_GET_TOKEN | CLIENT_STATE_AUTH | CLIENT_STATE_TAG_CLIENT | CLIENT_STATE_CREATE | CLIENT_STATE_READY,
+        fromReadyClientState,
+        executeReadyClientState,
+        defaultClientStateTransitionHook,
+        INFINITE_RETRY_COUNT_SENTINEL,
+        STATUS_CLIENT_READY_CALLBACK_FAILED
+    }
 };
 
 UINT32 CLIENT_STATE_MACHINE_STATE_COUNT = SIZEOF(CLIENT_STATE_MACHINE_STATES) / SIZEOF(StateMachineState);
 
+STATUS defaultClientStateTransitionHook(
+        UINT64 customData /* customData should be PKinesisVideoClient */,
+        PUINT64 stateTransitionWaitTime) {
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+    PKinesisVideoClient pKinesisVideoClient = NULL;
+    PKvsRetryStrategy pKvsRetryStrategy = NULL;
+    UINT64 retryWaitTime = 0;
+
+    pKinesisVideoClient = CLIENT_FROM_CUSTOM_DATA(customData);
+    CHK(pKinesisVideoClient != NULL && stateTransitionWaitTime != NULL, STATUS_NULL_ARG);
+
+    pKvsRetryStrategy = &(pKinesisVideoClient->deviceInfo.clientInfo.kvsRetryStrategy);
+
+    // result > SERVICE_CALL_RESULT_OK covers case for -
+    // result != SERVICE_CALL_RESULT_NOT_SET and != SERVICE_CALL_RESULT_OK
+    // If we support any other 2xx service call results, the condition
+    // should change to (pKinesisVideoStream->base.result > 299 && ...)
+    CHK(pKinesisVideoClient->base.result > SERVICE_CALL_RESULT_OK &&
+            pKvsRetryStrategy != NULL &&
+            pKvsRetryStrategy->pRetryStrategy != NULL &&
+            pKvsRetryStrategy->executeRetryStrategyFn != NULL, STATUS_SUCCESS);
+
+    DLOGD("KinesisVideoClient base result is [%u]. Executing KVS retry handler of retry strategy type [%u]",
+          pKinesisVideoClient->base.result, pKvsRetryStrategy->retryStrategyType);
+    pKvsRetryStrategy->executeRetryStrategyFn(pKvsRetryStrategy->pRetryStrategy, &retryWaitTime);
+    *stateTransitionWaitTime = retryWaitTime;
+
+CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
 // Helper method for stepping the client state machine
 STATUS stepClientStateMachine(PKinesisVideoClient pKinesisVideoClient)
 {

src/client/src/Include_i.h

@@ -463,8 +463,13 @@ STATUS executeCreateClientState(UINT64, UINT64);
 STATUS executeTagClientState(UINT64, UINT64);
 STATUS executeReadyClientState(UINT64, UINT64);
 
+STATUS defaultClientStateTransitionHook(UINT64, PUINT64);
+
 STATUS checkIntermittentProducerCallback(UINT32, UINT64, UINT64);
 
+STATUS freeClientRetryStrategy(PKinesisVideoClient);
+STATUS configureClientWithRetryStrategy(PKinesisVideoClient);
+
 #ifdef __cplusplus
 }
 #endif

src/client/src/Stream.h

@@ -858,6 +858,8 @@ STATUS executeStreamingStreamState(UINT64, UINT64);
 STATUS executeStoppedStreamState(UINT64, UINT64);
 STATUS executeTagStreamState(UINT64, UINT64);
 
+STATUS defaultStreamStateTransitionHook(UINT64, PUINT64);
+
 #ifdef __cplusplus
 }
 #endif