floraachy
/
OpenBLAS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "Improve Windows threading performance scaling"

This commit is contained in:
Martin Kroeker 2023-06-28 09:45:17 +02:00 committed by GitHub
parent b4f233e881
commit 616fdea82a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 61 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
CONTRIBUTORS.md
View File

@ -216,8 +216,3 @@ In chronological order:
* Pablo Romero <https://github.com/pablorcum> * Pablo Romero <https://github.com/pablorcum>
* [2022-08] Fix building from sources for QNX * [2022-08] Fix building from sources for QNX
* Mark Seminatore <https://github.com/mseminatore>
* [2023-06-23] Fix bounds issue in goto_set_num_threads
* [2023-06-23] Improve Windows threading performance scaling

102
driver/others/blas_server_win32.c
View File

@ -50,19 +50,11 @@
/* This is a thread implementation for Win32 lazy implementation */ /* This is a thread implementation for Win32 lazy implementation */
#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
/* Thread server common information */ /* Thread server common information */
typedef struct{ typedef struct{
HANDLE taskSemaphore; CRITICAL_SECTION lock;
HANDLE filled;
HANDLE killed;
blas_queue_t *queue; /* Parameter Pointer */ blas_queue_t *queue; /* Parameter Pointer */
int shutdown; /* server shutdown flag */ int shutdown; /* server shutdown flag */
@ -79,6 +71,8 @@ static blas_pool_t pool;
static HANDLE blas_threads [MAX_CPU_NUMBER]; static HANDLE blas_threads [MAX_CPU_NUMBER];
static DWORD blas_threads_id[MAX_CPU_NUMBER]; static DWORD blas_threads_id[MAX_CPU_NUMBER];
static void legacy_exec(void *func, int mode, blas_arg_t *args, void *sb){ static void legacy_exec(void *func, int mode, blas_arg_t *args, void *sb){
if (!(mode & BLAS_COMPLEX)){ if (!(mode & BLAS_COMPLEX)){
@ -204,6 +198,7 @@ static void legacy_exec(void *func, int mode, blas_arg_t *args, void *sb){
/* This is a main routine of threads. Each thread waits until job is */ /* This is a main routine of threads. Each thread waits until job is */
/* queued. */ /* queued. */
static DWORD WINAPI blas_thread_server(void *arg){ static DWORD WINAPI blas_thread_server(void *arg){
/* Thread identifier */ /* Thread identifier */
@ -212,7 +207,9 @@ static DWORD WINAPI blas_thread_server(void *arg){
#endif #endif
void *buffer, *sa, *sb; void *buffer, *sa, *sb;
volatile blas_queue_t *queue; blas_queue_t *queue;
DWORD action;
HANDLE handles[] = {pool.filled, pool.killed};
/* Each server needs each buffer */ /* Each server needs each buffer */
buffer = blas_memory_alloc(2); buffer = blas_memory_alloc(2);
@ -229,32 +226,28 @@ static DWORD WINAPI blas_thread_server(void *arg){
fprintf(STDERR, "Server[%2ld] Waiting for Queue.\n", cpu); fprintf(STDERR, "Server[%2ld] Waiting for Queue.\n", cpu);
#endif #endif
// all worker threads wait on the semaphore do {
WaitForSingleObject(pool.taskSemaphore, INFINITE); action = WaitForMultipleObjects(2, handles, FALSE, INFINITE);
} while ((action != WAIT_OBJECT_0) && (action != WAIT_OBJECT_0 + 1));
// kill the thread if we are shutting down the server if (action == WAIT_OBJECT_0 + 1) break;
if (pool.shutdown)
break;
#ifdef SMP_DEBUG #ifdef SMP_DEBUG
fprintf(STDERR, "Server[%2ld] Got it.\n", cpu); fprintf(STDERR, "Server[%2ld] Got it.\n", cpu);
#endif #endif
// grab a queued task and update the list EnterCriticalSection(&pool.lock);
volatile blas_queue_t* queue_next;
INT_PTR prev_value;
do {
queue = (volatile blas_queue_t*)pool.queue;
if (!queue)
break;
queue_next = (volatile blas_queue_t*)queue->next; queue = pool.queue;
prev_value = WIN_CAS((INT_PTR*)&pool.queue, (INT_PTR)queue_next, (INT_PTR)queue); if (queue) pool.queue = queue->next;
} while (prev_value != queue);
LeaveCriticalSection(&pool.lock);
if (queue) { if (queue) {
int (*routine)(blas_arg_t *, void *, void *, void *, void *, BLASLONG) = queue -> routine; int (*routine)(blas_arg_t *, void *, void *, void *, void *, BLASLONG) = queue -> routine;
if (pool.queue) SetEvent(pool.filled);
sa = queue -> sa; sa = queue -> sa;
sb = queue -> sb; sb = queue -> sb;
@ -339,8 +332,13 @@ static DWORD WINAPI blas_thread_server(void *arg){
fprintf(STDERR, "Server[%2ld] Finished!\n", cpu); fprintf(STDERR, "Server[%2ld] Finished!\n", cpu);
#endif #endif
// mark our sub-task as complete EnterCriticalSection(&queue->lock);
InterlockedDecrement(&queue->status);
queue -> status = BLAS_STATUS_FINISHED;
LeaveCriticalSection(&queue->lock);
SetEvent(queue->finish);
} }
/* Shutdown procedure */ /* Shutdown procedure */
@ -369,7 +367,9 @@ static DWORD WINAPI blas_thread_server(void *arg){
if (!blas_server_avail){ if (!blas_server_avail){
pool.taskSemaphore = CreateSemaphore(NULL, 0, blas_cpu_number - 1, NULL); InitializeCriticalSection(&pool.lock);
pool.filled = CreateEvent(NULL, FALSE, FALSE, NULL);
pool.killed = CreateEvent(NULL, TRUE, FALSE, NULL);
pool.shutdown = 0; pool.shutdown = 0;
pool.queue = NULL; pool.queue = NULL;
@ -395,6 +395,7 @@ static DWORD WINAPI blas_thread_server(void *arg){
exec_blas ... returns after jobs are finished. exec_blas ... returns after jobs are finished.
*/ */
int exec_blas_async(BLASLONG pos, blas_queue_t *queue){ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){
#if defined(SMP_SERVER) #if defined(SMP_SERVER)
@ -408,7 +409,8 @@ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){
current = queue; current = queue;
while (current) { while (current) {
current->status = 1; InitializeCriticalSection(&current -> lock);
current -> finish = CreateEvent(NULL, FALSE, FALSE, NULL);
current -> position = pos; current -> position = pos;
#ifdef CONSISTENT_FPCSR #ifdef CONSISTENT_FPCSR
@ -420,10 +422,19 @@ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){
pos ++; pos ++;
} }
pool.queue = queue; EnterCriticalSection(&pool.lock);
// start up worker threads if (pool.queue) {
ReleaseSemaphore(pool.taskSemaphore, pos - 1, NULL); current = pool.queue;
while (current -> next) current = current -> next;
current -> next = queue;
} else {
pool.queue = queue;
}
LeaveCriticalSection(&pool.lock);
SetEvent(pool.filled);
return 0; return 0;
} }
@ -439,9 +450,10 @@ int exec_blas_async_wait(BLASLONG num, blas_queue_t *queue){
fprintf(STDERR, "Waiting Queue ..\n"); fprintf(STDERR, "Waiting Queue ..\n");
#endif #endif
// spin-wait on each sub-task to finish WaitForSingleObject(queue->finish, INFINITE);
while (*((volatile int*)&queue->status))
YIELDING; CloseHandle(queue->finish);
DeleteCriticalSection(&queue -> lock);
queue = queue -> next; queue = queue -> next;
num --; num --;
@ -489,21 +501,18 @@ int exec_blas(BLASLONG num, blas_queue_t *queue){
/* Shutdown procedure, but user don't have to call this routine. The */ /* Shutdown procedure, but user don't have to call this routine. The */
/* kernel automatically kill threads. */ /* kernel automatically kill threads. */
int BLASFUNC(blas_thread_shutdown)(void){ int BLASFUNC(blas_thread_shutdown)(void){
int i; int i;
#ifdef SMP_DEBUG
fprintf(STDERR, "blas_thread_shutdown..\n");
#endif
if (!blas_server_avail) return 0; if (!blas_server_avail) return 0;
LOCK_COMMAND(&server_lock); LOCK_COMMAND(&server_lock);
if (blas_server_avail){ if (blas_server_avail){
pool.shutdown = 1; SetEvent(pool.killed);
for(i = 0; i < blas_num_threads - 1; i++){ for(i = 0; i < blas_num_threads - 1; i++){
// Could also just use WaitForMultipleObjects // Could also just use WaitForMultipleObjects
@ -519,7 +528,8 @@ int BLASFUNC(blas_thread_shutdown)(void){
CloseHandle(blas_threads[i]); CloseHandle(blas_threads[i]);
} }
CloseHandle(pool.taskSemaphore); CloseHandle(pool.filled);
CloseHandle(pool.killed);
blas_server_avail = 0; blas_server_avail = 0;
} }
@ -549,14 +559,16 @@ void goto_set_num_threads(int num_threads)
//increased_threads = 1; //increased_threads = 1;
if (!blas_server_avail){ if (!blas_server_avail){
pool.taskSemaphore = CreateSemaphore(NULL, 0, blas_cpu_number - 1, NULL); InitializeCriticalSection(&pool.lock);
pool.filled = CreateEvent(NULL, FALSE, FALSE, NULL);
pool.killed = CreateEvent(NULL, TRUE, FALSE, NULL);
pool.shutdown = 0; pool.shutdown = 0;
pool.queue = NULL; pool.queue = NULL;
blas_server_avail = 1; blas_server_avail = 1;
} }
for(i = blas_num_threads; i < num_threads - 1; i++){ for(i = blas_num_threads - 1; i < num_threads - 1; i++){
blas_threads[i] = CreateThread(NULL, 0, blas_threads[i] = CreateThread(NULL, 0,
blas_thread_server, (void *)i, blas_thread_server, (void *)i,