Asynchronous get ice config

samples/Common.c

@@ -356,13 +356,51 @@ VOID onIceCandidateHandler(UINT64 customData, PCHAR candidateJson)
     CHK_LOG_ERR(retStatus);
 }
 
+PVOID asyncGetIceConfigInfo(PVOID args)
+{
+    STATUS retStatus = STATUS_SUCCESS;
+    AsyncGetIceStruct* data = (AsyncGetIceStruct*) args;
+    PIceConfigInfo pIceConfigInfo = NULL;
+    UINT32 uriCount = 0;
+    UINT32 i = 0, maxTurnServer = 1;
+
+    if (data != NULL) {
+        /* signalingClientGetIceConfigInfoCount can return more than one turn server. Use only one to optimize
+         * candidate gathering latency. But user can also choose to use more than 1 turn server. */
+        for (uriCount = 0, i = 0; i < maxTurnServer; i++) {
+            /*
+             * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port?transport=udp" then ICE will try TURN over UDP
+             * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port?transport=tcp" then ICE will try TURN over TCP/TLS
+             * if configuration.iceServers[uriCount + 1].urls is "turns:ip:port?transport=udp", it's currently ignored because sdk dont do TURN
+             * over DTLS yet. if configuration.iceServers[uriCount + 1].urls is "turns:ip:port?transport=tcp" then ICE will try TURN over TCP/TLS
+             * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port" then ICE will try both TURN over UDP and TCP/TLS
+             *
+             * It's recommended to not pass too many TURN iceServers to configuration because it will slow down ice gathering in non-trickle mode.
+             */
+            CHK_STATUS(signalingClientGetIceConfigInfo(data->signalingClientHandle, i, &pIceConfigInfo));
+            CHECK(uriCount < MAX_ICE_SERVERS_COUNT);
+            uriCount += pIceConfigInfo->uriCount;
+            CHK_STATUS(addConfigToServerList(&(data->pRtcPeerConnection), pIceConfigInfo));
+        }
+    }
+    *(data->pUriCount) += uriCount;
+
+CleanUp:
+    SAFE_MEMFREE(data);
+    CHK_LOG_ERR(retStatus);
+    return NULL;
+}
+
 STATUS initializePeerConnection(PSampleConfiguration pSampleConfiguration, PRtcPeerConnection* ppRtcPeerConnection)
 {
     ENTERS();
     STATUS retStatus = STATUS_SUCCESS;
     RtcConfiguration configuration;
-    UINT32 i, j, iceConfigCount, uriCount = 0, maxTurnServer = 1;
+#ifndef ENABLE_KVS_THREADPOOL
+    UINT32 i, j, maxTurnServer = 1;
     PIceConfigInfo pIceConfigInfo;
+    UINT32 uriCount = 0;
+#endif
     UINT64 data;
     PRtcCertificate pRtcCertificate = NULL;
 
@@ -385,37 +423,6 @@ STATUS initializePeerConnection(PSampleConfiguration pSampleConfiguration, PRtcP
     SNPRINTF(configuration.iceServers[0].urls, MAX_ICE_CONFIG_URI_LEN, KINESIS_VIDEO_STUN_URL, pSampleConfiguration->channelInfo.pRegion,
              pKinesisVideoStunUrlPostFix);
 
-    if (pSampleConfiguration->useTurn) {
-        // Set the URIs from the configuration
-        CHK_STATUS(signalingClientGetIceConfigInfoCount(pSampleConfiguration->signalingClientHandle, &iceConfigCount));
-
-        /* signalingClientGetIceConfigInfoCount can return more than one turn server. Use only one to optimize
-         * candidate gathering latency. But user can also choose to use more than 1 turn server. */
-        for (uriCount = 0, i = 0; i < maxTurnServer; i++) {
-            CHK_STATUS(signalingClientGetIceConfigInfo(pSampleConfiguration->signalingClientHandle, i, &pIceConfigInfo));
-            for (j = 0; j < pIceConfigInfo->uriCount; j++) {
-                CHECK(uriCount < MAX_ICE_SERVERS_COUNT);
-                /*
-                 * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port?transport=udp" then ICE will try TURN over UDP
-                 * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port?transport=tcp" then ICE will try TURN over TCP/TLS
-                 * if configuration.iceServers[uriCount + 1].urls is "turns:ip:port?transport=udp", it's currently ignored because sdk dont do TURN
-                 * over DTLS yet. if configuration.iceServers[uriCount + 1].urls is "turns:ip:port?transport=tcp" then ICE will try TURN over TCP/TLS
-                 * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port" then ICE will try both TURN over UDP and TCP/TLS
-                 *
-                 * It's recommended to not pass too many TURN iceServers to configuration because it will slow down ice gathering in non-trickle mode.
-                 */
-
-                STRNCPY(configuration.iceServers[uriCount + 1].urls, pIceConfigInfo->uris[j], MAX_ICE_CONFIG_URI_LEN);
-                STRNCPY(configuration.iceServers[uriCount + 1].credential, pIceConfigInfo->password, MAX_ICE_CONFIG_CREDENTIAL_LEN);
-                STRNCPY(configuration.iceServers[uriCount + 1].username, pIceConfigInfo->userName, MAX_ICE_CONFIG_USER_NAME_LEN);
-
-                uriCount++;
-            }
-        }
-    }
-
-    pSampleConfiguration->iceUriCount = uriCount + 1;
-
     // Check if we have any pregenerated certs and use them
     // NOTE: We are running under the config lock
     retStatus = stackQueueDequeue(pSampleConfiguration->pregeneratedCertificates, &data);
@@ -430,6 +437,40 @@ STATUS initializePeerConnection(PSampleConfiguration pSampleConfiguration, PRtcP
     }
 
     CHK_STATUS(createPeerConnection(&configuration, ppRtcPeerConnection));
+
+    if (pSampleConfiguration->useTurn) {
+#ifdef ENABLE_KVS_THREADPOOL
+        pSampleConfiguration->iceUriCount = 1;
+        AsyncGetIceStruct* pAsyncData = NULL;
+
+        pAsyncData = (AsyncGetIceStruct*) MEMCALLOC(1, SIZEOF(AsyncGetIceStruct));
+        pAsyncData->signalingClientHandle = pSampleConfiguration->signalingClientHandle;
+        pAsyncData->pRtcPeerConnection = *ppRtcPeerConnection;
+        pAsyncData->pUriCount = &(pSampleConfiguration->iceUriCount);
+        CHK_STATUS(peerConnectionAsync(asyncGetIceConfigInfo, (PVOID) pAsyncData));
+#else
+
+        /* signalingClientGetIceConfigInfoCount can return more than one turn server. Use only one to optimize
+         * candidate gathering latency. But user can also choose to use more than 1 turn server. */
+        for (uriCount = 0, i = 0; i < maxTurnServer; i++) {
+            /*
+             * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port?transport=udp" then ICE will try TURN over UDP
+             * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port?transport=tcp" then ICE will try TURN over TCP/TLS
+             * if configuration.iceServers[uriCount + 1].urls is "turns:ip:port?transport=udp", it's currently ignored because sdk dont do TURN
+             * over DTLS yet. if configuration.iceServers[uriCount + 1].urls is "turns:ip:port?transport=tcp" then ICE will try TURN over TCP/TLS
+             * if configuration.iceServers[uriCount + 1].urls is "turn:ip:port" then ICE will try both TURN over UDP and TCP/TLS
+             *
+             * It's recommended to not pass too many TURN iceServers to configuration because it will slow down ice gathering in non-trickle mode.
+             */
+            CHK_STATUS(signalingClientGetIceConfigInfo(pSampleConfiguration->signalingClientHandle, i, &pIceConfigInfo));
+            CHECK(uriCount < MAX_ICE_SERVERS_COUNT);
+            uriCount += pIceConfigInfo->uriCount;
+            CHK_STATUS(addConfigToServerList(ppRtcPeerConnection, pIceConfigInfo));
+        }
+        pSampleConfiguration->iceUriCount = uriCount + 1;
+#endif
+    }
+
 CleanUp:
 
     CHK_LOG_ERR(retStatus);

samples/Samples.h

@@ -166,6 +166,15 @@ struct __SampleStreamingSession {
     KvsIceAgentMetrics iceMetrics;
 };
 
+// TODO this should all be in a higher webrtccontext layer above PeerConnection
+// Placing it here now since this is where all the current webrtccontext functions are placed
+typedef struct {
+    SIGNALING_CLIENT_HANDLE signalingClientHandle;
+    PRtcPeerConnection pRtcPeerConnection;
+    PUINT32 pUriCount;
+
+} AsyncGetIceStruct;
+
 VOID sigintHandler(INT32);
 STATUS readFrameFromDisk(PBYTE, PUINT32, PCHAR);
 PVOID sendVideoPackets(PVOID);

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

@@ -236,7 +236,7 @@ extern "C" {
 #define STATUS_TURN_CONNECTION_PEER_NOT_USABLE                             STATUS_ICE_BASE + 0x00000027
 #define STATUS_ICE_SERVER_INDEX_INVALID                                    STATUS_ICE_BASE + 0x00000028
 #define STATUS_ICE_CANDIDATE_STRING_MISSING_TYPE                           STATUS_ICE_BASE + 0x00000029
-#define STATUS_TURN_CONNECTION_ALLOCAITON_FAILED                           STATUS_ICE_BASE + 0x0000002a
+#define STATUS_TURN_CONNECTION_ALLOCATION_FAILED                           STATUS_ICE_BASE + 0x0000002a
 #define STATUS_TURN_INVALID_STATE                                          STATUS_ICE_BASE + 0x0000002b
 /*!@} */
 
@@ -1573,6 +1573,16 @@ typedef struct {
  */
 PUBLIC_API STATUS createPeerConnection(PRtcConfiguration, PRtcPeerConnection*);
 
+/**
+ * @brief Give peer connection an ice config to add to its server list
+ *
+ * @param[in] PRtcPeerConnection* initialized RtcPeerConnection
+ * @param[in] PIceConfigInfo Ice config info to add to this peer connection
+ *
+ * @return STATUS code of the execution. STATUS_SUCCESS on success
+ */
+PUBLIC_API STATUS addConfigToServerList(PRtcPeerConnection*, PIceConfigInfo);
+
 /**
  * @brief Free a RtcPeerConnection
  *
@@ -1649,6 +1659,16 @@ PUBLIC_API STATUS peerConnectionGetLocalDescription(PRtcPeerConnection, PRtcSess
  */
 PUBLIC_API STATUS peerConnectionGetCurrentLocalDescription(PRtcPeerConnection, PRtcSessionDescriptionInit);
 
+/**
+ * Allows use of internal threadpool
+ *
+ * @param[in] startRoutine function pointer to execute in threadpool
+ * @param[in] PVOID void pointer to pass to function pointer
+ *
+ * @return STATUS code of the execution. STATUS_SUCCESS on success
+ */
+PUBLIC_API STATUS peerConnectionAsync(startRoutine fn, PVOID data);
+
 /**
  * @brief Populate the provided answer that contains an RFC 3264 offer
  * with the supported configurations for the session.

src/source/Ice/IceAgent.c

@@ -262,6 +262,82 @@ STATUS freeIceAgent(PIceAgent* ppIceAgent)
     return retStatus;
 }
 
+STATUS iceAgentAddConfig(PIceAgent pIceAgent, PIceConfigInfo pIceConfigInfo)
+{
+    STATUS retStatus = STATUS_SUCCESS;
+    UINT32 i = 0;
+    // used in PROFILE macro
+    UINT64 startTimeInMacro = 0;
+    BOOL locked = FALSE;
+
+    CHK(pIceAgent != NULL && pIceConfigInfo != NULL, STATUS_NULL_ARG);
+
+    for (i = 0; i < pIceConfigInfo->uriCount; i++) {
+        MUTEX_LOCK(pIceAgent->lock);
+        locked = TRUE;
+        PROFILE_CALL_WITH_T_OBJ(retStatus = parseIceServer(&pIceAgent->iceServers[pIceAgent->iceServersCount], (PCHAR) pIceConfigInfo->uris[i],
+                                                           (PCHAR) pIceConfigInfo->userName, (PCHAR) pIceConfigInfo->password),
+                                pIceAgent->iceAgentProfileDiagnostics.iceServerParsingTime[i], "ICE server parsing");
+        MUTEX_UNLOCK(pIceAgent->lock);
+        locked = FALSE;
+
+        if (STATUS_SUCCEEDED(retStatus)) {
+            MUTEX_LOCK(pIceAgent->lock);
+            locked = TRUE;
+            pIceAgent->rtcIceServerDiagnostics[i].port = (INT32) getInt16(pIceAgent->iceServers[i].ipAddress.port);
+            switch (pIceAgent->iceServers[pIceAgent->iceServersCount].transport) {
+                case KVS_SOCKET_PROTOCOL_UDP:
+                    STRCPY(pIceAgent->rtcIceServerDiagnostics[i].protocol, ICE_TRANSPORT_TYPE_UDP);
+                    break;
+                case KVS_SOCKET_PROTOCOL_TCP:
+                    STRCPY(pIceAgent->rtcIceServerDiagnostics[i].protocol, ICE_TRANSPORT_TYPE_TCP);
+                    break;
+                default:
+                    MEMSET(pIceAgent->rtcIceServerDiagnostics[i].protocol, 0, SIZEOF(pIceAgent->rtcIceServerDiagnostics[i].protocol));
+            }
+            STRCPY(pIceAgent->rtcIceServerDiagnostics[i].url, pIceConfigInfo->uris[i]);
+
+            MUTEX_UNLOCK(pIceAgent->lock);
+            locked = FALSE;
+
+            // important to unlock iceAgent lock before calling init relay candidate, since iceAgent APIs are thread safe
+            // if you don't unlock this can lead to a deadlock with the timerqueue.
+            //  init candidate && pairs
+            if (pIceAgent->iceServers[pIceAgent->iceServersCount].isTurn) {
+                if (pIceAgent->iceServers[pIceAgent->iceServersCount].transport == KVS_SOCKET_PROTOCOL_UDP ||
+                    pIceAgent->iceServers[pIceAgent->iceServersCount].transport == KVS_SOCKET_PROTOCOL_NONE) {
+                    CHK_STATUS(iceAgentInitRelayCandidate(pIceAgent, pIceAgent->iceServersCount, KVS_SOCKET_PROTOCOL_UDP));
+                }
+
+                if (pIceAgent->iceServers[pIceAgent->iceServersCount].transport == KVS_SOCKET_PROTOCOL_TCP ||
+                    pIceAgent->iceServers[pIceAgent->iceServersCount].transport == KVS_SOCKET_PROTOCOL_NONE) {
+                    CHK_STATUS(iceAgentInitRelayCandidate(pIceAgent, pIceAgent->iceServersCount, KVS_SOCKET_PROTOCOL_TCP));
+                }
+            }
+
+            MUTEX_LOCK(pIceAgent->lock);
+            locked = TRUE;
+
+            pIceAgent->iceServersCount++;
+
+            MUTEX_UNLOCK(pIceAgent->lock);
+            locked = FALSE;
+
+        } else {
+            DLOGE("Failed to parse ICE servers");
+        }
+    }
+
+CleanUp:
+    CHK_LOG_ERR(retStatus);
+
+    if (locked) {
+        MUTEX_UNLOCK(pIceAgent->lock);
+    }
+
+    return retStatus;
+}
+
 STATUS iceAgentValidateKvsRtcConfig(PKvsRtcConfiguration pKvsRtcConfiguration)
 {
     STATUS retStatus = STATUS_SUCCESS;
@@ -310,8 +386,6 @@ STATUS iceAgentReportNewLocalCandidate(PIceAgent pIceAgent, PIceCandidate pIceCa
     iceAgentLogNewCandidate(pIceCandidate);
 
     CHK_WARN(pIceAgent->iceAgentCallbacks.newLocalCandidateFn != NULL, retStatus, "newLocalCandidateFn callback not implemented");
-    CHK_WARN(!ATOMIC_LOAD_BOOL(&pIceAgent->candidateGatheringFinished), retStatus,
-             "Cannot report new ice candidate because candidate gathering is already finished");
     CHK_STATUS(iceCandidateSerialize(pIceCandidate, serializedIceCandidateBuf, &serializedIceCandidateBufLen));
     pIceAgent->iceAgentCallbacks.newLocalCandidateFn(pIceAgent->iceAgentCallbacks.customData, serializedIceCandidateBuf);
 
@@ -614,9 +688,6 @@ STATUS iceAgentStartGathering(PIceAgent pIceAgent)
                                 "Srflx candidates setup time");
     }
 
-    PROFILE_CALL_WITH_T_OBJ(CHK_STATUS(iceAgentInitRelayCandidates(pIceAgent)), pIceAgent->iceAgentProfileDiagnostics.relayCandidateSetUpTime,
-                            "Relay candidates setup time");
-
     // start listening for incoming data
     CHK_STATUS(connectionListenerStart(pIceAgent->pConnectionListener));
 
@@ -956,6 +1027,8 @@ STATUS iceAgentRestart(PIceAgent pIceAgent, PCHAR localIceUfrag, PCHAR localIceP
     CHK_STATUS(setStateMachineCurrentState(pIceAgent->pStateMachine, ICE_AGENT_STATE_NEW));
 
     ATOMIC_STORE_BOOL(&pIceAgent->processStun, TRUE);
+    // this API does not reset servers, so re-initialize relay candidates now.
+    CHK_STATUS(iceAgentInitRelayCandidates(pIceAgent));
 
 CleanUp:
 
@@ -1857,16 +1930,17 @@ STATUS iceAgentInitRelayCandidate(PIceAgent pIceAgent, UINT32 iceServerIndex, KV
     callback.relayAddressAvailableFn = NULL;
     callback.turnStateFailedFn = turnStateFailedFn;
 
+    MUTEX_LOCK(pIceAgent->lock);
+    locked = TRUE;
+
     CHK_STATUS(createTurnConnection(&pIceAgent->iceServers[iceServerIndex], pIceAgent->timerQueueHandle,
                                     TURN_CONNECTION_DATA_TRANSFER_MODE_SEND_INDIDATION, protocol, &callback, pNewCandidate->pSocketConnection,
                                     pIceAgent->pConnectionListener, &pTurnConnection));
     pNewCandidate->pIceAgent = pIceAgent;
     pNewCandidate->pTurnConnection = pTurnConnection;
 
-    MUTEX_LOCK(pIceAgent->lock);
-    locked = TRUE;
-
     CHK_STATUS(doubleListInsertItemHead(pIceAgent->localCandidates, (UINT64) pNewCandidate));
+    CHK_STATUS(iceAgentReportNewLocalCandidate(pIceAgent, pNewCandidate));
     pNewCandidate = NULL;
 
     /* add existing remote candidates to turn. Need to acquire lock because remoteCandidates can be mutated by

src/source/Ice/IceAgent.h

@@ -448,6 +448,8 @@ STATUS updateSelectedLocalRemoteCandidateStats(PIceAgent);
 
 STATUS getIceAgentStats(PIceAgent, PKvsIceAgentMetrics);
 
+STATUS iceAgentAddConfig(PIceAgent, PIceConfigInfo);
+
 #ifdef __cplusplus
 }
 #endif

src/source/Ice/TurnConnectionStateMachine.c

@@ -364,7 +364,7 @@ STATUS executeAllocationTurnState(UINT64 customData, UINT64 time)
         pTurnConnection->state = TURN_STATE_ALLOCATION;
     } else {
         pTurnConnection->stateTryCount++;
-        CHK(pTurnConnection->stateTryCount < pTurnConnection->stateTryCountMax, STATUS_TURN_CONNECTION_ALLOCAITON_FAILED);
+        CHK(pTurnConnection->stateTryCount < pTurnConnection->stateTryCountMax, STATUS_TURN_CONNECTION_ALLOCATION_FAILED);
     }
     CHK_STATUS(iceUtilsSendStunPacket(pTurnConnection->pTurnPacket, pTurnConnection->longTermKey, ARRAY_SIZE(pTurnConnection->longTermKey),
                                       &pTurnConnection->turnServer.ipAddress, pTurnConnection->pControlChannel, NULL, FALSE));

src/source/Include_i.h

@@ -139,6 +139,11 @@ STATUS generateJSONSafeString(PCHAR, UINT32);
 #include "Ice/NatBehaviorDiscovery.h"
 #include "Srtp/SrtpSession.h"
 #include "Sctp/Sctp.h"
+#include "Signaling/FileCache.h"
+#include "Signaling/Signaling.h"
+#include "Signaling/ChannelInfo.h"
+#include "Signaling/StateMachine.h"
+#include "Signaling/LwsApiCalls.h"
 #include "Rtp/RtpPacket.h"
 #include "Rtcp/RtcpPacket.h"
 #include "Rtcp/RollingBuffer.h"
@@ -154,11 +159,6 @@ STATUS generateJSONSafeString(PCHAR, UINT32);
 #include "Rtp/Codecs/RtpH264Payloader.h"
 #include "Rtp/Codecs/RtpOpusPayloader.h"
 #include "Rtp/Codecs/RtpG711Payloader.h"
-#include "Signaling/FileCache.h"
-#include "Signaling/Signaling.h"
-#include "Signaling/ChannelInfo.h"
-#include "Signaling/StateMachine.h"
-#include "Signaling/LwsApiCalls.h"
 #include "Metrics/Metrics.h"
 
 ////////////////////////////////////////////////////