From 5f51811728645baae32ad54f7b34a00259e6e8e8 Mon Sep 17 00:00:00 2001 From: Mark Seminatore Date: Tue, 5 Dec 2023 22:43:36 -0800 Subject: [PATCH 1/3] try at new threading model --- CONTRIBUTORS.md | 3 + common_thread.h | 5 +- driver/others/blas_server_win32.c | 137 ++++++++++++++++-------------- 3 files changed, 78 insertions(+), 67 deletions(-) diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 71df13634..493747052 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -216,3 +216,6 @@ In chronological order: * Pablo Romero * [2022-08] Fix building from sources for QNX + +* Mark Seminatore + * [2023-11-09] Improve Windows threading performance scaling \ No newline at end of file diff --git a/common_thread.h b/common_thread.h index 6e18d2a8e..9e7dae74a 100644 --- a/common_thread.h +++ b/common_thread.h @@ -111,8 +111,9 @@ typedef struct blas_queue { struct blas_queue *next; #if defined( __WIN32__) || defined(__CYGWIN32__) || defined(_WIN32) || defined(__CYGWIN__) - CRITICAL_SECTION lock; - HANDLE finish; + // CRITICAL_SECTION lock; + // HANDLE finish; + volatile int finished; #else pthread_mutex_t lock; pthread_cond_t finished; diff --git a/driver/others/blas_server_win32.c b/driver/others/blas_server_win32.c index 5bdfc1276..464e3fa03 100644 --- a/driver/others/blas_server_win32.c +++ b/driver/others/blas_server_win32.c @@ -51,15 +51,19 @@ /* This is a thread implementation for Win32 lazy implementation */ /* Thread server common information */ -typedef struct{ - CRITICAL_SECTION lock; - HANDLE filled; - HANDLE killed; +//typedef struct{ +// CRITICAL_SECTION lock; +// HANDLE filled; +// HANDLE killed; +// +// blas_queue_t *queue; /* Parameter Pointer */ +// int shutdown; /* server shutdown flag */ +// +//} blas_pool_t; - blas_queue_t *queue; /* Parameter Pointer */ - int shutdown; /* server shutdown flag */ - -} blas_pool_t; +static blas_queue_t *work_queue = NULL; +static HANDLE kickoff_event = NULL; +static CRITICAL_SECTION queue_lock; /* We need this global for checking if initialization is finished. */ int blas_server_avail = 0; @@ -67,7 +71,7 @@ int blas_server_avail = 0; /* Local Variables */ static BLASULONG server_lock = 0; -static blas_pool_t pool; +//static blas_pool_t pool; static HANDLE blas_threads [MAX_CPU_NUMBER]; static DWORD blas_threads_id[MAX_CPU_NUMBER]; @@ -209,7 +213,7 @@ static DWORD WINAPI blas_thread_server(void *arg){ void *buffer, *sa, *sb; blas_queue_t *queue; DWORD action; - HANDLE handles[] = {pool.filled, pool.killed}; + //HANDLE handles[] = {pool.filled, pool.killed}; /* Each server needs each buffer */ buffer = blas_memory_alloc(2); @@ -225,29 +229,38 @@ static DWORD WINAPI blas_thread_server(void *arg){ #ifdef SMP_DEBUG fprintf(STDERR, "Server[%2ld] Waiting for Queue.\n", cpu); #endif - - do { - action = WaitForMultipleObjects(2, handles, FALSE, INFINITE); - } while ((action != WAIT_OBJECT_0) && (action != WAIT_OBJECT_0 + 1)); - - if (action == WAIT_OBJECT_0 + 1) break; + // event raised when work is added to the queue + WaitForSingleObject(kickoff_event, INFINITE); #ifdef SMP_DEBUG fprintf(STDERR, "Server[%2ld] Got it.\n", cpu); #endif - EnterCriticalSection(&pool.lock); +#if 1 + EnterCriticalSection(&queue_lock); - queue = pool.queue; - if (queue) pool.queue = queue->next; + queue = work_queue; + if (queue) + work_queue = work_queue->next; - LeaveCriticalSection(&pool.lock); + LeaveCriticalSection(&queue_lock); +#else + volatile work_queue_t* queue_next; + + INT_PTR prev_value; + do { + queue = (volatile work_queue_t*)work_queue; + if (!queue) + break; + + queue_next = (volatile work_queue_t*)queue->next; + prev_value = WIN_CAS((INT_PTR*)&work_queue, (INT_PTR)queue_next, (INT_PTR)queue); + } while (prev_value != work_item); +#endif if (queue) { int (*routine)(blas_arg_t *, void *, void *, void *, void *, BLASLONG) = queue -> routine; - if (pool.queue) SetEvent(pool.filled); - sa = queue -> sa; sb = queue -> sb; @@ -331,14 +344,6 @@ static DWORD WINAPI blas_thread_server(void *arg){ #ifdef SMP_DEBUG fprintf(STDERR, "Server[%2ld] Finished!\n", cpu); #endif - - EnterCriticalSection(&queue->lock); - - queue -> status = BLAS_STATUS_FINISHED; - - LeaveCriticalSection(&queue->lock); - - SetEvent(queue->finish); } /* Shutdown procedure */ @@ -366,13 +371,10 @@ int blas_thread_init(void){ #endif if (!blas_server_avail){ + // create the kickoff Event + kickoff_event = CreateEvent(NULL, TRUE, FALSE, NULL); - InitializeCriticalSection(&pool.lock); - pool.filled = CreateEvent(NULL, FALSE, FALSE, NULL); - pool.killed = CreateEvent(NULL, TRUE, FALSE, NULL); - - pool.shutdown = 0; - pool.queue = NULL; + InitializeCriticalSection(&queue_lock); for(i = 0; i < blas_cpu_number - 1; i++){ blas_threads[i] = CreateThread(NULL, 0, @@ -409,8 +411,6 @@ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){ current = queue; while (current) { - InitializeCriticalSection(¤t -> lock); - current -> finish = CreateEvent(NULL, FALSE, FALSE, NULL); current -> position = pos; #ifdef CONSISTENT_FPCSR @@ -418,23 +418,32 @@ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){ __asm__ __volatile__ ("stmxcsr %0" : "=m" (current -> sse_mode)); #endif + current->finished = 0; current = current -> next; pos ++; } - EnterCriticalSection(&pool.lock); + EnterCriticalSection(&queue_lock); - if (pool.queue) { - current = pool.queue; - while (current -> next) current = current -> next; - current -> next = queue; - } else { - pool.queue = queue; + if (!work_queue) + { + work_queue = queue; + } + else + { + blas_queue_t *next_item = work_queue; + + // find the end of the work queue + while (next_item) + next_item = next_item->next; + + // add new work to the end + next_item = queue; } - LeaveCriticalSection(&pool.lock); + LeaveCriticalSection(&queue_lock); - SetEvent(pool.filled); + SetEvent(kickoff_event); return 0; } @@ -449,21 +458,26 @@ int exec_blas_async_wait(BLASLONG num, blas_queue_t *queue){ #ifdef SMP_DEBUG fprintf(STDERR, "Waiting Queue ..\n"); #endif + while (!queue->finished) + YIELDING; - WaitForSingleObject(queue->finish, INFINITE); - - CloseHandle(queue->finish); - DeleteCriticalSection(&queue -> lock); - - queue = queue -> next; - num --; + queue = queue->next; + num--; } #ifdef SMP_DEBUG fprintf(STDERR, "Completely Done.\n\n"); #endif + // if work was added to the queue after this batch we can't sleep the worker threads + // by resetting the event + EnterCriticalSection(&queue_lock); - return 0; + if (work_queue == NULL) + ResetEvent(kickoff_event); + + LeaveCriticalSection(&queue_lock); + + return 0; } /* Execute Threads */ @@ -512,8 +526,6 @@ int BLASFUNC(blas_thread_shutdown)(void){ if (blas_server_avail){ - SetEvent(pool.killed); - for(i = 0; i < blas_num_threads - 1; i++){ // Could also just use WaitForMultipleObjects DWORD wait_thread_value = WaitForSingleObject(blas_threads[i], 50); @@ -528,9 +540,6 @@ int BLASFUNC(blas_thread_shutdown)(void){ CloseHandle(blas_threads[i]); } - CloseHandle(pool.filled); - CloseHandle(pool.killed); - blas_server_avail = 0; } @@ -558,13 +567,11 @@ void goto_set_num_threads(int num_threads) //increased_threads = 1; if (!blas_server_avail){ + // create the kickoff Event + kickoff_event = CreateEvent(NULL, TRUE, FALSE, NULL); - InitializeCriticalSection(&pool.lock); - pool.filled = CreateEvent(NULL, FALSE, FALSE, NULL); - pool.killed = CreateEvent(NULL, TRUE, FALSE, NULL); + InitializeCriticalSection(&queue_lock); - pool.shutdown = 0; - pool.queue = NULL; blas_server_avail = 1; } From 4ebf814b4258904d1f48bfe427a6727514d9efa6 Mon Sep 17 00:00:00 2001 From: Mark Seminatore Date: Tue, 5 Dec 2023 23:28:37 -0800 Subject: [PATCH 2/3] fix bug failing to mark task as finished. --- driver/others/blas_server_win32.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/driver/others/blas_server_win32.c b/driver/others/blas_server_win32.c index 464e3fa03..5af1f1a51 100644 --- a/driver/others/blas_server_win32.c +++ b/driver/others/blas_server_win32.c @@ -344,6 +344,9 @@ static DWORD WINAPI blas_thread_server(void *arg){ #ifdef SMP_DEBUG fprintf(STDERR, "Server[%2ld] Finished!\n", cpu); #endif + + queue->finished = 1; + } /* Shutdown procedure */ From edac80d7e8ba97e39002f223628a956456356fa9 Mon Sep 17 00:00:00 2001 From: Mark Seminatore Date: Thu, 7 Dec 2023 14:59:27 -0800 Subject: [PATCH 3/3] some cleanup, dynamically scale threads, add missing WIN_CASE defn --- driver/others/blas_server_win32.c | 70 ++++++++++++++++++++++--------- 1 file changed, 51 insertions(+), 19 deletions(-) diff --git a/driver/others/blas_server_win32.c b/driver/others/blas_server_win32.c index 5af1f1a51..40ff85abc 100644 --- a/driver/others/blas_server_win32.c +++ b/driver/others/blas_server_win32.c @@ -51,15 +51,6 @@ /* This is a thread implementation for Win32 lazy implementation */ /* Thread server common information */ -//typedef struct{ -// CRITICAL_SECTION lock; -// HANDLE filled; -// HANDLE killed; -// -// blas_queue_t *queue; /* Parameter Pointer */ -// int shutdown; /* server shutdown flag */ -// -//} blas_pool_t; static blas_queue_t *work_queue = NULL; static HANDLE kickoff_event = NULL; @@ -71,11 +62,19 @@ int blas_server_avail = 0; /* Local Variables */ static BLASULONG server_lock = 0; -//static blas_pool_t pool; static HANDLE blas_threads [MAX_CPU_NUMBER]; static DWORD blas_threads_id[MAX_CPU_NUMBER]; +static volatile int thread_target; // target num of live threads, volatile for cross-thread reads - +#if defined (__GNUC__) && (__GNUC__ < 6) + #define WIN_CAS(dest, exch, comp) __sync_val_compare_and_swap(dest, comp, exch) +#else + #if defined(_WIN64) + #define WIN_CAS(dest, exch, comp) InterlockedCompareExchange64(dest, exch, comp) + #else + #define WIN_CAS(dest, exch, comp) InterlockedCompareExchange(dest, exch, comp) + #endif +#endif static void legacy_exec(void *func, int mode, blas_arg_t *args, void *sb){ @@ -206,14 +205,10 @@ static void legacy_exec(void *func, int mode, blas_arg_t *args, void *sb){ static DWORD WINAPI blas_thread_server(void *arg){ /* Thread identifier */ -#ifdef SMP_DEBUG BLASLONG cpu = (BLASLONG)arg; -#endif void *buffer, *sa, *sb; blas_queue_t *queue; - DWORD action; - //HANDLE handles[] = {pool.filled, pool.killed}; /* Each server needs each buffer */ buffer = blas_memory_alloc(2); @@ -232,6 +227,12 @@ static DWORD WINAPI blas_thread_server(void *arg){ // event raised when work is added to the queue WaitForSingleObject(kickoff_event, INFINITE); + if (cpu > thread_target - 2) + { + //printf("thread [%d] exiting.\n", cpu); + break; // excess thread, so worker thread exits + } + #ifdef SMP_DEBUG fprintf(STDERR, "Server[%2ld] Got it.\n", cpu); #endif @@ -245,17 +246,17 @@ static DWORD WINAPI blas_thread_server(void *arg){ LeaveCriticalSection(&queue_lock); #else - volatile work_queue_t* queue_next; + volatile blas_queue_t* queue_next; INT_PTR prev_value; do { - queue = (volatile work_queue_t*)work_queue; + queue = (volatile blas_queue_t*)work_queue; if (!queue) break; - queue_next = (volatile work_queue_t*)queue->next; + queue_next = (volatile blas_queue_t*)queue->next; prev_value = WIN_CAS((INT_PTR*)&work_queue, (INT_PTR)queue_next, (INT_PTR)queue); - } while (prev_value != work_item); + } while (prev_value != queue); #endif if (queue) { @@ -377,9 +378,13 @@ int blas_thread_init(void){ // create the kickoff Event kickoff_event = CreateEvent(NULL, TRUE, FALSE, NULL); + thread_target = blas_cpu_number; + InitializeCriticalSection(&queue_lock); for(i = 0; i < blas_cpu_number - 1; i++){ + //printf("thread_init: creating thread [%d]\n", i); + blas_threads[i] = CreateThread(NULL, 0, blas_thread_server, (void *)i, 0, &blas_threads_id[i]); @@ -564,10 +569,36 @@ void goto_set_num_threads(int num_threads) if (num_threads > MAX_CPU_NUMBER) num_threads = MAX_CPU_NUMBER; + if (blas_server_avail && num_threads < blas_num_threads) { + LOCK_COMMAND(&server_lock); + + thread_target = num_threads; + + SetEvent(kickoff_event); + + for (i = num_threads - 1; i < blas_num_threads - 1; i++) { + //printf("set_num_threads: waiting on thread [%d] to quit.\n", i); + + WaitForSingleObject(blas_threads[i], INFINITE); + + //printf("set_num_threads: thread [%d] has quit.\n", i); + + CloseHandle(blas_threads[i]); + } + + blas_num_threads = num_threads; + + ResetEvent(kickoff_event); + + UNLOCK_COMMAND(&server_lock); + } + if (num_threads > blas_num_threads) { LOCK_COMMAND(&server_lock); + thread_target = num_threads; + //increased_threads = 1; if (!blas_server_avail){ // create the kickoff Event @@ -579,6 +610,7 @@ void goto_set_num_threads(int num_threads) } for(i = (blas_num_threads > 0) ? blas_num_threads - 1 : 0; i < num_threads - 1; i++){ + //printf("set_num_threads: creating thread [%d]\n", i); blas_threads[i] = CreateThread(NULL, 0, blas_thread_server, (void *)i,