Skip to content

Commit

Permalink
Merge pull request OpenMathLib#4359 from mseminatore/win_perf
Browse files Browse the repository at this point in the history 
Improve Windows threading performance scaling
  • Loading branch information
@martin-frbg
martin-frbg committed Dec 9, 2023
2 parents 5b09833 + efa9515 commit e60fb0f
Show file tree
Hide file tree
Showing 3 changed files with 114 additions and 68 deletions.
3 changes: 3 additions & 0 deletions CONTRIBUTORS.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -216,3 +216,6 @@ In chronological order:

* Pablo Romero <https://github.com/pablorcum>
* [2022-08] Fix building from sources for QNX

* Mark Seminatore <https://github.com/mseminatore>
* [2023-11-09] Improve Windows threading performance scaling
5 changes: 3 additions & 2 deletions common_thread.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
174 changes: 108 additions & 66 deletions driver/others/blas_server_win32.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,27 +51,30 @@
/* 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;
static CRITICAL_SECTION queue_lock;

/* We need this global for checking if initialization is finished. */
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){

Expand Down Expand Up @@ -202,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);
Expand All @@ -225,29 +224,44 @@ static DWORD WINAPI blas_thread_server(void *arg){
#ifdef SMP_DEBUG
fprintf(STDERR, "Server[%2ld] Waiting for Queue.\n", cpu);
#endif
// event raised when work is added to the queue
WaitForSingleObject(kickoff_event, INFINITE);

do {
action = WaitForMultipleObjects(2, handles, FALSE, INFINITE);
} while ((action != WAIT_OBJECT_0) && (action != WAIT_OBJECT_0 + 1));

if (action == WAIT_OBJECT_0 + 1) break;
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

EnterCriticalSection(&pool.lock);
#if 1
EnterCriticalSection(&queue_lock);

queue = work_queue;
if (queue)
work_queue = work_queue->next;

LeaveCriticalSection(&queue_lock);
#else
volatile blas_queue_t* queue_next;

queue = pool.queue;
if (queue) pool.queue = queue->next;
INT_PTR prev_value;
do {
queue = (volatile blas_queue_t*)work_queue;
if (!queue)
break;

LeaveCriticalSection(&pool.lock);
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 != queue);
#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;

Expand Down Expand Up @@ -331,14 +345,9 @@ static DWORD WINAPI blas_thread_server(void *arg){
#ifdef SMP_DEBUG
fprintf(STDERR, "Server[%2ld] Finished!\n", cpu);
#endif

queue->finished = 1;

EnterCriticalSection(&queue->lock);

queue -> status = BLAS_STATUS_FINISHED;

LeaveCriticalSection(&queue->lock);

SetEvent(queue->finish);
}

/* Shutdown procedure */
Expand Down Expand Up @@ -366,15 +375,16 @@ 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);
thread_target = blas_cpu_number;

pool.shutdown = 0;
pool.queue = NULL;
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]);
Expand Down Expand Up @@ -409,32 +419,39 @@ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){
current = queue;

while (current) {
InitializeCriticalSection(&current -> lock);
current -> finish = CreateEvent(NULL, FALSE, FALSE, NULL);
current -> position = pos;

#ifdef CONSISTENT_FPCSR
__asm__ __volatile__ ("fnstcw %0" : "=m" (current -> x87_mode));
__asm__ __volatile__ ("stmxcsr %0" : "=m" (current -> sse_mode));
#endif

current->finished = 0;
current = current -> next;
pos ++;
}

EnterCriticalSection(&pool.lock);
EnterCriticalSection(&queue_lock);

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;

if (pool.queue) {
current = pool.queue;
while (current -> next) current = current -> next;
current -> next = queue;
} else {
pool.queue = queue;
// add new work to the end
next_item = queue;
}

LeaveCriticalSection(&pool.lock);
LeaveCriticalSection(&queue_lock);

SetEvent(pool.filled);
SetEvent(kickoff_event);

return 0;
}
Expand All @@ -449,21 +466,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 */
Expand Down Expand Up @@ -512,8 +534,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);
Expand All @@ -528,9 +548,6 @@ int BLASFUNC(blas_thread_shutdown)(void){
CloseHandle(blas_threads[i]);
}

CloseHandle(pool.filled);
CloseHandle(pool.killed);

blas_server_avail = 0;
}

Expand All @@ -552,23 +569,48 @@ 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
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;
}

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,
Expand Down

0 comments on commit e60fb0f

Please sign in to comment.