FS-10167: Fixed bug in windows test_thread_pools test related to the thread pool…

FS-10167: Fixed bug in windows test_thread_pools test related to the thread pool itself and ks_q cleanup

libs/libks/src/include/ks_time.h

@@ -44,6 +44,7 @@ KS_BEGIN_EXTERN_C
 #define ks_time_nsec(time) (((time) % KS_USEC_PER_SEC) * 1000)
 #define ks_sleep_ms(_t) ks_sleep(_t * 1000)
 
+KS_DECLARE(void) ks_time_init(void);
 KS_DECLARE(ks_time_t) ks_time_now(void);
 KS_DECLARE(ks_time_t) ks_time_now_sec(void);
 KS_DECLARE(void) ks_sleep(ks_time_t microsec);

libs/libks/src/ks.c

@@ -64,6 +64,9 @@ KS_DECLARE(ks_status_t) ks_global_set_cleanup(ks_pool_cleanup_fn_t fn, void *arg
 KS_DECLARE(ks_status_t) ks_init(void)
 {
 	unsigned int pid = 0;
+
+	ks_time_init();
+
 #ifdef __WINDOWS__
 	pid = _getpid();
 #else

libs/libks/src/ks_mutex.c

@@ -383,7 +383,8 @@ KS_DECLARE(ks_status_t) ks_cond_wait(ks_cond_t *cond)
 KS_DECLARE(ks_status_t) ks_cond_timedwait(ks_cond_t *cond, ks_time_t ms)
 {
 #ifdef WIN32
-	if(!SleepConditionVariableCS(&cond->cond, &cond->mutex->mutex, (DWORD)ms)) {
+	BOOL res = SleepConditionVariableCS(&cond->cond, &cond->mutex->mutex, (DWORD)ms);
+	if (!res) {
 		if (GetLastError() == ERROR_TIMEOUT) {
 			return KS_STATUS_TIMEOUT;
 		} else {

libs/libks/src/ks_q.c

@@ -67,6 +67,10 @@ static void ks_q_cleanup(ks_pool_t *mpool, void *ptr, void *arg, int type, ks_po
 
 	switch(action) {
 	case KS_MPCL_ANNOUNCE:
+		if (q->active) {
+			ks_q_flush(q);
+			ks_q_term(q);
+		}
 		break;
 	case KS_MPCL_TEARDOWN:
 		np = q->head;
@@ -133,7 +137,7 @@ KS_DECLARE(ks_size_t) ks_q_term(ks_q_t *q)
 	active = q->active;
 	q->active = 0;
 	ks_mutex_unlock(q->list_mutex);
-
+	
 	if (active) {
 		ks_q_wake(q);
 	}
@@ -227,6 +231,7 @@ static ks_status_t do_push(ks_q_t *q, void *ptr)
 	ks_qnode_t *node;
 
 	ks_mutex_lock(q->list_mutex);
+
 	if (!q->active) {
 		ks_mutex_unlock(q->list_mutex);
 		return KS_STATUS_INACTIVE;

libs/libks/src/ks_thread_pool.c

@@ -129,13 +129,20 @@ static void *worker_thread(ks_thread_t *thread, void *data)
 		ks_status_t status;
 		
 		status = ks_q_pop_timeout(tp->q, &pop, 1000);
+		if (status == KS_STATUS_BREAK) {
+			if (tp->state != TP_STATE_RUNNING) {
+				break;
+			}
+			continue;
+		}
+		
 		/*
 		ks_log(KS_LOG_DEBUG, "WORKER %d idle_sec %d running %d dying %d total %d max %d\n", 
 			   my_id, idle_sec, tp->running_thread_count, tp->dying_thread_count, tp->thread_count, tp->max);		
 		*/		
 		check_queue(tp, KS_FALSE);
-		
-		if (status == KS_STATUS_TIMEOUT || status == KS_STATUS_BREAK) {
+
+		if (status == KS_STATUS_TIMEOUT) { // || status == KS_STATUS_BREAK) {
 			idle_sec++;
 
 			if (idle_sec >= tp->idle_sec) {
@@ -148,7 +155,6 @@ static void *worker_thread(ks_thread_t *thread, void *data)
 				ks_mutex_unlock(tp->mutex);
 
 				if (die) {
-					ks_log(KS_LOG_DEBUG, "WORKER %d IDLE TIMEOUT\n", my_id);
 					break;
 				}
 			}
@@ -156,8 +162,8 @@ static void *worker_thread(ks_thread_t *thread, void *data)
 			continue;
 		}
 
-		if ((status != KS_STATUS_SUCCESS && status != KS_STATUS_BREAK) || !pop) {
-			ks_log(KS_LOG_DEBUG, "WORKER %d POP FAIL %d %p\n", my_id, status, (void *)pop);
+		if ((status != KS_STATUS_SUCCESS && status != KS_STATUS_BREAK)) {
+			ks_log(KS_LOG_ERROR, "WORKER %d POP FAIL %d %p\n", my_id, status, (void *)pop);
 			break;
 		}
 

libs/libks/src/ks_time.c

@@ -40,7 +40,7 @@ static DWORD win32_last_get_time_tick = 0;
 
 static uint8_t win32_use_qpc = 0;
 static uint64_t win32_qpc_freq = 0;
-static int timer_init;
+
 static inline void win32_init_timers(void)
 {
 	OSVERSIONINFOEX version_info; /* Used to fetch current OS version from Windows */
@@ -86,18 +86,17 @@ static inline void win32_init_timers(void)
 	}
 
 	LeaveCriticalSection(&timer_section);
+}
 
-	timer_init = 1;
+KS_DECLARE(void) ks_time_init(void)
+{
+	win32_init_timers();
 }
 
 KS_DECLARE(ks_time_t) ks_time_now(void)
 {
 	ks_time_t now;
 	
-	if (!timer_init) {
-		win32_init_timers();
-	}
-
 	if (win32_use_qpc) {
 		/* Use QueryPerformanceCounter */
 		uint64_t count = 0;
@@ -133,10 +132,6 @@ KS_DECLARE(ks_time_t) ks_time_now_sec(void)
 {
 	ks_time_t now;
 	
-	if (!timer_init) {
-		win32_init_timers();
-	}
-
 	if (win32_use_qpc) {
 		/* Use QueryPerformanceCounter */
 		uint64_t count = 0;

libs/libks/test/test_thread_pools.c

@@ -16,7 +16,7 @@ static void *test1_thread(ks_thread_t *thread, void *data)
 {
 	struct x *mydata = (struct x *) data;
 
-	ks_log(KS_LOG_DEBUG, "Thread %d\n", mydata->i);
+	//ks_log(KS_LOG_DEBUG, "Thread %d\n", mydata->i);
 	ks_sleep(100000);
 	ks_pool_free(mydata->pool, &mydata);
 	return NULL;
@@ -37,6 +37,7 @@ static int test1()
 		struct x *data = ks_pool_alloc(pool, sizeof(*data));
 		data->i = i;
 		data->pool = pool;
+		ks_sleep(1);
 		ks_thread_pool_add_job(tp, test1_thread, data);
 	}