Fix linux threadpool (#262)

* fix wrap around issue

* enable other tests

* iwyu fix

linux/src/threadpool_linux.c

@@ -176,7 +176,7 @@ static int threadpool_work_func(void* param)
 
                         /* Codes_SRS_THREADPOOL_LINUX_07_078: [ threadpool_work_func shall increment the current consume index by calling interlocked_increment_64. ]*/
                         /* Codes_SRS_THREADPOOL_LINUX_07_079: [ threadpool_work_func shall get the next waiting task consume index from incremented consume index modulo current task array size. ]*/
-                        current_index = interlocked_increment_64(&threadpool->consume_idx) - 1 % existing_count;
+                        current_index = (interlocked_increment_64(&threadpool->consume_idx) - 1) % existing_count;
                         /* Codes_SRS_THREADPOOL_LINUX_07_080: [ If consume index has task state TASK_WAITING, threadpool_work_func shall set the task state to TASK_WORKING. ]*/
                         TASK_RESULT curr_task_result = interlocked_compare_exchange(&threadpool->task_array[current_index].task_state, TASK_WORKING, TASK_WAITING);
                         if (TASK_WAITING == curr_task_result)
@@ -518,7 +518,7 @@ int threadpool_schedule_work(THANDLE(THREADPOOL) threadpool, THREADPOOL_WORK_FUN
                 int32_t existing_count = interlocked_add(&threadpool_ptr->task_array_size, 0);
 
                 /* Codes_SRS_THREADPOOL_LINUX_07_034: [ threadpool_schedule_work shall increment the insert_pos. ]*/
-                int64_t insert_pos = interlocked_increment_64(&threadpool_ptr->insert_idx) - 1 % existing_count;
+                int64_t insert_pos = (interlocked_increment_64(&threadpool_ptr->insert_idx) - 1) % existing_count;
 
                 /* Codes_SRS_THREADPOOL_LINUX_07_035: [ If task state is TASK_NOT_USED, threadpool_schedule_work shall set the current task state to TASK_INITIALIZING. ]*/
                 int32_t task_state = interlocked_compare_exchange(&threadpool_ptr->task_array[insert_pos].task_state, TASK_INITIALIZING, TASK_NOT_USED);

linux/tests/threadpool_linux_int/threadpool_linux_int.c

@@ -5,6 +5,7 @@
 #include <stdlib.h>
 #include <time.h>
 #include <stdbool.h>
+#include <string.h>
 
 #include "c_logging/logger.h"
 
@@ -20,6 +21,7 @@
 #include "c_pal/threadapi.h"
 #include "c_pal/interlocked.h"
 #include "c_pal/sync.h"
+#include "c_pal/interlocked_hl.h"
 #include "c_pal/execution_engine.h"
 #include "c_pal/execution_engine_linux.h"
 #include "c_pal/thandle.h" // IWYU pragma: keep
@@ -33,6 +35,14 @@ typedef struct WAIT_WORK_CONTEXT_TAG
     volatile_atomic int32_t wait_event;
 } WAIT_WORK_CONTEXT;
 
+typedef struct WRAP_DATA_TAG
+{
+    volatile_atomic int32_t* counter;
+    char mem[10];
+} WRAP_DATA;
+
+TEST_DEFINE_ENUM_TYPE(INTERLOCKED_HL_RESULT, INTERLOCKED_HL_RESULT_VALUES);
+
 BEGIN_TEST_SUITE(TEST_SUITE_NAME_FROM_CMAKE)
 
 TEST_SUITE_INITIALIZE(suite_init)
@@ -86,6 +96,17 @@ static void threadpool_task_wait_random(void* parameter)
     wake_by_address_single(thread_counter);
 }
 
+
+static void threadpool_long_task(void* context)
+{
+    WRAP_DATA* data = context;
+    ASSERT_ARE_EQUAL(int, 0, strcmp(data->mem, "READY"));
+    strcpy(data->mem, "DONE");
+    (void)interlocked_increment(data->counter);
+    wake_by_address_single(data->counter);
+    free(data);
+}
+
 static void work_function(void* context)
 {
     (void)interlocked_increment(&g_call_count);
@@ -256,6 +277,37 @@ TEST_FUNCTION(threadpool_chaos_knight)
     execution_engine_dec_ref(execution_engine);
 }
 
+#define WRAP_TEST_WORK_ITEMS 10000
+
+TEST_FUNCTION(threadpool_force_wrap_around)
+{
+    // assert
+    EXECUTION_ENGINE_HANDLE execution_engine = execution_engine_create(NULL);
+
+    THANDLE(THREADPOOL) threadpool = threadpool_create(execution_engine);
+    ASSERT_IS_NOT_NULL(threadpool);
+    ASSERT_ARE_EQUAL(int, 0, threadpool_open(threadpool));
+
+    volatile_atomic int32_t counter;
+    interlocked_exchange(&counter, 0);
+
+    for (uint32_t index = 0; index < WRAP_TEST_WORK_ITEMS; index++)
+    {
+        WRAP_DATA* data = malloc(sizeof(WRAP_DATA));
+        data->counter = &counter;
+        strcpy(data->mem, "READY");
+        ASSERT_ARE_EQUAL(int, 0, threadpool_schedule_work(threadpool, threadpool_long_task, data));
+    }
+
+    // assert
+    ASSERT_ARE_EQUAL(INTERLOCKED_HL_RESULT, INTERLOCKED_HL_OK, InterlockedHL_WaitForValue(&counter, WRAP_TEST_WORK_ITEMS, UINT32_MAX));
+
+    // cleanup
+    threadpool_close(threadpool);
+    THANDLE_ASSIGN(THREADPOOL)(&threadpool, NULL);
+    execution_engine_dec_ref(execution_engine);
+}
+
 TEST_FUNCTION(one_start_timer_works_runs_once)
 {
     // assert