Browse Source

try at new threading model

tags/v0.3.26
Mark Seminatore 1 year ago
parent
commit
5f51811728
3 changed files with 79 additions and 68 deletions
  1. +3
    -0
      CONTRIBUTORS.md
  2. +3
    -2
      common_thread.h
  3. +73
    -66
      driver/others/blas_server_win32.c

+ 3
- 0
CONTRIBUTORS.md View File

@@ -216,3 +216,6 @@ 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-11-09] Improve Windows threading performance scaling

+ 3
- 2
common_thread.h View File

@@ -111,8 +111,9 @@ typedef struct blas_queue {
struct blas_queue *next; struct blas_queue *next;


#if defined( __WIN32__) || defined(__CYGWIN32__) || defined(_WIN32) || defined(__CYGWIN__) #if defined( __WIN32__) || defined(__CYGWIN32__) || defined(_WIN32) || defined(__CYGWIN__)
CRITICAL_SECTION lock;
HANDLE finish;
// CRITICAL_SECTION lock;
// HANDLE finish;
volatile int finished;
#else #else
pthread_mutex_t lock; pthread_mutex_t lock;
pthread_cond_t finished; pthread_cond_t finished;


+ 73
- 66
driver/others/blas_server_win32.c View File

@@ -51,15 +51,19 @@
/* This is a thread implementation for Win32 lazy implementation */ /* This is a thread implementation for Win32 lazy implementation */


/* Thread server common information */ /* 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;
//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. */ /* We need this global for checking if initialization is finished. */
int blas_server_avail = 0; int blas_server_avail = 0;
@@ -67,7 +71,7 @@ int blas_server_avail = 0;
/* Local Variables */ /* Local Variables */
static BLASULONG server_lock = 0; static BLASULONG server_lock = 0;


static blas_pool_t pool;
//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];


@@ -209,7 +213,7 @@ static DWORD WINAPI blas_thread_server(void *arg){
void *buffer, *sa, *sb; void *buffer, *sa, *sb;
blas_queue_t *queue; blas_queue_t *queue;
DWORD action; DWORD action;
HANDLE handles[] = {pool.filled, pool.killed};
//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);
@@ -225,29 +229,38 @@ static DWORD WINAPI blas_thread_server(void *arg){
#ifdef SMP_DEBUG #ifdef SMP_DEBUG
fprintf(STDERR, "Server[%2ld] Waiting for Queue.\n", cpu); fprintf(STDERR, "Server[%2ld] Waiting for Queue.\n", cpu);
#endif #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 #ifdef SMP_DEBUG
fprintf(STDERR, "Server[%2ld] Got it.\n", cpu); fprintf(STDERR, "Server[%2ld] Got it.\n", cpu);
#endif #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) { 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;


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

EnterCriticalSection(&queue->lock);

queue -> status = BLAS_STATUS_FINISHED;

LeaveCriticalSection(&queue->lock);

SetEvent(queue->finish);
} }


/* Shutdown procedure */ /* Shutdown procedure */
@@ -366,13 +371,10 @@ int blas_thread_init(void){
#endif #endif


if (!blas_server_avail){ 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++){ for(i = 0; i < blas_cpu_number - 1; i++){
blas_threads[i] = CreateThread(NULL, 0, blas_threads[i] = CreateThread(NULL, 0,
@@ -409,8 +411,6 @@ int exec_blas_async(BLASLONG pos, blas_queue_t *queue){
current = queue; current = queue;


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


#ifdef CONSISTENT_FPCSR #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)); __asm__ __volatile__ ("stmxcsr %0" : "=m" (current -> sse_mode));
#endif #endif


current->finished = 0;
current = current -> next; current = current -> next;
pos ++; 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;


LeaveCriticalSection(&pool.lock);
// find the end of the work queue
while (next_item)
next_item = next_item->next;


SetEvent(pool.filled);
// add new work to the end
next_item = queue;
}

LeaveCriticalSection(&queue_lock);

SetEvent(kickoff_event);


return 0; return 0;
} }
@@ -449,21 +458,26 @@ int exec_blas_async_wait(BLASLONG num, blas_queue_t *queue){
#ifdef SMP_DEBUG #ifdef SMP_DEBUG
fprintf(STDERR, "Waiting Queue ..\n"); fprintf(STDERR, "Waiting Queue ..\n");
#endif #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 #ifdef SMP_DEBUG
fprintf(STDERR, "Completely Done.\n\n"); fprintf(STDERR, "Completely Done.\n\n");
#endif #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 */ /* Execute Threads */
@@ -512,8 +526,6 @@ int BLASFUNC(blas_thread_shutdown)(void){


if (blas_server_avail){ if (blas_server_avail){


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
DWORD wait_thread_value = WaitForSingleObject(blas_threads[i], 50); DWORD wait_thread_value = WaitForSingleObject(blas_threads[i], 50);
@@ -528,9 +540,6 @@ int BLASFUNC(blas_thread_shutdown)(void){
CloseHandle(blas_threads[i]); CloseHandle(blas_threads[i]);
} }


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

blas_server_avail = 0; blas_server_avail = 0;
} }


@@ -558,13 +567,11 @@ void goto_set_num_threads(int num_threads)


//increased_threads = 1; //increased_threads = 1;
if (!blas_server_avail){ 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; blas_server_avail = 1;
} }




Loading…
Cancel
Save