Add API for specifying thread stack size

CMakeLists.txt

@@ -58,6 +58,11 @@ function(enableSanitizer SANITIZER)
     set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fsanitize=${SANITIZER}" PARENT_SCOPE)
 endfunction()
 
+if(KVS_DEFAULT_STACK_SIZE)
+  message(STATUS "Building with default stack size")
+  add_compile_definitions(KVS_DEFAULT_STACK_SIZE_BYTES=${KVS_DEFAULT_STACK_SIZE})
+endif()
+
 if(ADDRESS_SANITIZER)
   enableSanitizer("address")
 endif()

README.md

@@ -55,6 +55,7 @@ You can pass the following options to `cmake ..`
 * `-DUNDEFINED_BEHAVIOR_SANITIZER` Build with UndefinedBehaviorSanitizer
 * `-DBUILD_DEBUG_HEAP` Build debug heap with guard bands and validation. This is ONLY intended for low-level debugging purposes. Default is OFF
 * `-DALIGNED_MEMORY_MODEL` Build for aligned memory model only devices. Default is OFF.
+* `-DKVS_DEFAULT_STACK_SIZE` -- Value in bytes for default stack size of threads. Pthread minimum for Unix applications is 16 Kib.
 
 ### Build
 To build the library run make in the build directory you executed CMake.

src/common/include/com/amazonaws/kinesis/video/common/CommonDefs.h

@@ -721,6 +721,7 @@ typedef VOID (*unlockMutex)(MUTEX);
 typedef BOOL (*tryLockMutex)(MUTEX);
 typedef VOID (*freeMutex)(MUTEX);
 typedef STATUS (*createThread)(PTID, startRoutine, PVOID);
+typedef STATUS (*createThreadWithParams)(PTID, startRoutine, SIZE_T, PVOID);
 typedef STATUS (*joinThread)(TID, PVOID*);
 typedef VOID (*threadSleep)(UINT64);
 typedef VOID (*threadSleepUntil)(UINT64);
@@ -763,6 +764,7 @@ extern unlockMutex globalUnlockMutex;
 extern tryLockMutex globalTryLockMutex;
 extern freeMutex globalFreeMutex;
 extern createThread globalCreateThread;
+extern createThreadWithParams globalCreateThreadWithParams;
 extern joinThread globalJoinThread;
 extern threadSleep globalThreadSleep;
 extern threadSleepUntil globalThreadSleepUntil;
@@ -1032,12 +1034,14 @@ extern PUBLIC_API atomicXor globalAtomicXor;
 //
 // Thread functionality
 //
-#define THREAD_CREATE      globalCreateThread
-#define THREAD_JOIN        globalJoinThread
-#define THREAD_SLEEP       globalThreadSleep
-#define THREAD_SLEEP_UNTIL globalThreadSleepUntil
-#define THREAD_CANCEL      globalCancelThread
-#define THREAD_DETACH      globalDetachThread
+// Wrappers around OS specific utilities for threads. Takes arguments as given.
+#define THREAD_CREATE             globalCreateThread
+#define THREAD_CREATE_WITH_PARAMS globalCreateThreadWithParams
+#define THREAD_JOIN               globalJoinThread
+#define THREAD_SLEEP              globalThreadSleep
+#define THREAD_SLEEP_UNTIL        globalThreadSleepUntil
+#define THREAD_CANCEL             globalCancelThread
+#define THREAD_DETACH             globalDetachThread
 
 //
 // Static initializers

src/utils/src/Thread.c

@@ -37,7 +37,7 @@ PUBLIC_API DWORD WINAPI startWrapperRoutine(LPVOID data)
     return retVal;
 }
 
-PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID args)
+PUBLIC_API STATUS defaultCreateThreadWithParams(PTID pThreadId, startRoutine start, SIZE_T stackSize, PVOID args)
 {
     STATUS retStatus = STATUS_SUCCESS;
     HANDLE threadHandle;
@@ -51,7 +51,7 @@ PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID
     pWrapper->storedArgs = args;
     pWrapper->storedStartRoutine = start;
 
-    threadHandle = CreateThread(NULL, 0, startWrapperRoutine, pWrapper, 0, NULL);
+    threadHandle = CreateThread(NULL, stackSize, startWrapperRoutine, pWrapper, 0, NULL);
     CHK(threadHandle != NULL, STATUS_CREATE_THREAD_FAILED);
 
     *pThreadId = (TID) threadHandle;
@@ -64,6 +64,21 @@ PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID
     return retStatus;
 }
 
+PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID args)
+{
+    STATUS retStatus = STATUS_SUCCESS;
+
+#if defined(KVS_DEFAULT_STACK_SIZE_BYTES)
+    CHK_STATUS(defaultCreateThreadWithParams(pThreadId, start, (SIZE_T) KVS_DEFAULT_STACK_SIZE_BYTES, args));
+#else
+    CHK_STATUS(defaultCreateThreadWithParams(pThreadId, start, 0, args));
+#endif
+
+CleanUp:
+
+    return retStatus;
+}
+
 PUBLIC_API STATUS defaultJoinThread(TID threadId, PVOID* retVal)
 {
     STATUS retStatus = STATUS_SUCCESS;
@@ -150,7 +165,7 @@ PUBLIC_API TID defaultGetThreadId()
     return (TID) pthread_self();
 }
 
-PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID args)
+PUBLIC_API STATUS defaultCreateThreadWithParams(PTID pThreadId, startRoutine start, SIZE_T stackSize, PVOID args)
 {
     STATUS retStatus = STATUS_SUCCESS;
     pthread_t threadId;
@@ -159,14 +174,14 @@ PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID
 
     CHK(pThreadId != NULL, STATUS_NULL_ARG);
 
-#ifdef CONSTRAINED_DEVICE
     pthread_attr_t attr;
-    pAttr = &attr;
-    result = pthread_attr_init(pAttr);
-    CHK_ERR(result == 0, STATUS_THREAD_ATTR_INIT_FAILED, "pthread_attr_init failed with %d", result);
-    result = pthread_attr_setstacksize(&attr, THREAD_STACK_SIZE_ON_CONSTRAINED_DEVICE);
-    CHK_ERR(result == 0, STATUS_THREAD_ATTR_SET_STACK_SIZE_FAILED, "pthread_attr_setstacksize failed with %d", result);
-#endif
+    if (stackSize != 0) {
+        pAttr = &attr;
+        result = pthread_attr_init(pAttr);
+        CHK_ERR(result == 0, STATUS_THREAD_ATTR_INIT_FAILED, "pthread_attr_init failed with %d", result);
+        result = pthread_attr_setstacksize(&attr, stackSize);
+        CHK_ERR(result == 0, STATUS_THREAD_ATTR_SET_STACK_SIZE_FAILED, "pthread_attr_setstacksize failed with %d", result);
+    }
 
     result = pthread_create(&threadId, pAttr, start, args);
     switch (result) {
@@ -198,6 +213,23 @@ PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID
     return retStatus;
 }
 
+PUBLIC_API STATUS defaultCreateThread(PTID pThreadId, startRoutine start, PVOID args)
+{
+    STATUS retStatus = STATUS_SUCCESS;
+
+#if defined(KVS_DEFAULT_STACK_SIZE_BYTES)
+    CHK_STATUS(defaultCreateThreadWithParams(pThreadId, start, (SIZE_T) KVS_DEFAULT_STACK_SIZE_BYTES, args));
+#elif defined(CONSTRAINED_DEVICE)
+    CHK_STATUS(defaultCreateThreadWithParams(pThreadId, start, (SIZE_T) THREAD_STACK_SIZE_ON_CONSTRAINED_DEVICE, args));
+#else
+    CHK_STATUS(defaultCreateThreadWithParams(pThreadId, start, 0, args));
+#endif
+
+CleanUp:
+
+    return retStatus;
+}
+
 PUBLIC_API STATUS defaultJoinThread(TID threadId, PVOID* retVal)
 {
     STATUS retStatus = STATUS_SUCCESS;
@@ -330,6 +362,7 @@ PUBLIC_API VOID defaultThreadSleepUntil(UINT64 time)
 getTId globalGetThreadId = defaultGetThreadId;
 getTName globalGetThreadName = defaultGetThreadName;
 createThread globalCreateThread = defaultCreateThread;
+createThreadWithParams globalCreateThreadWithParams = defaultCreateThreadWithParams;
 threadSleep globalThreadSleep = defaultThreadSleep;
 threadSleepUntil globalThreadSleepUntil = defaultThreadSleepUntil;
 joinThread globalJoinThread = defaultJoinThread;

tst/utils/Thread.cpp

@@ -117,3 +117,58 @@ TEST_F(ThreadFunctionalityTest, ThreadCreateAndCancel)
 
     MUTEX_FREE(gThreadMutex);
 }
+
+TEST_F(ThreadFunctionalityTest, ThreadCreateAndReleaseSimpleCheckWithStack)
+{
+    UINT64 index;
+    TID threads[TEST_THREAD_COUNT];
+    gThreadMutex = MUTEX_CREATE(FALSE);
+    srand(GETTIME());
+    SIZE_T threadStack = 64 * 1024;
+    struct sleep_times st[TEST_THREAD_COUNT];
+
+    gThreadCount = 0;
+
+    // Create the threads
+    for (index = 0; index < TEST_THREAD_COUNT; index++) {
+        st[index].threadVisited = FALSE;
+        st[index].threadCleared = FALSE;
+        st[index].threadSleepTime = index * HUNDREDS_OF_NANOS_IN_A_MILLISECOND;
+        EXPECT_EQ(STATUS_SUCCESS, THREAD_CREATE_WITH_PARAMS(&threads[index], testThreadRoutine, threadStack, (PVOID)&st[index]));
+    }
+
+    // Await for the threads to finish
+    for (index = 0; index < TEST_THREAD_COUNT; index++) {
+        EXPECT_EQ(STATUS_SUCCESS, THREAD_JOIN(threads[index], NULL));
+    }
+
+    MUTEX_LOCK(gThreadMutex);
+    EXPECT_EQ(0, gThreadCount);
+    MUTEX_UNLOCK(gThreadMutex);
+
+    for (index = 0; index < TEST_THREAD_COUNT; index++) {
+        EXPECT_TRUE(st[index].threadVisited) << "Thread didn't visit index " << index;
+        EXPECT_TRUE(st[index].threadCleared) << "Thread didn't clear index " << index;
+    }
+
+    MUTEX_FREE(gThreadMutex);
+}
+
+TEST_F(ThreadFunctionalityTest, NegativeTest)
+{
+    UINT64 index;
+    TID threads[TEST_THREAD_COUNT];
+    gThreadMutex = MUTEX_CREATE(FALSE);
+    SIZE_T threadStack = 16 * 1024;
+    struct sleep_times st[TEST_THREAD_COUNT];
+
+    gThreadCount = 0;
+    EXPECT_NE(STATUS_SUCCESS, THREAD_CREATE_WITH_PARAMS(NULL, testThreadRoutine, threadStack, NULL));
+    EXPECT_NE(STATUS_SUCCESS, THREAD_CREATE(NULL, testThreadRoutine, NULL));
+
+    MUTEX_LOCK(gThreadMutex);
+    EXPECT_EQ(0, gThreadCount);
+    MUTEX_UNLOCK(gThreadMutex);
+
+    MUTEX_FREE(gThreadMutex);
+}