From 12dc701604bbb23e729717fbbc02bbadba6d5d08 Mon Sep 17 00:00:00 2001 From: Uladzimir Karpenka Date: Thu, 9 Jul 2026 23:59:20 +0300 Subject: [PATCH] analyzer: block the dispatcher on a semaphore, stop spawning a thread per frame Two pieces of pure overhead, neither of them doing any DSP. The POSIX QueueUserWorkItem() shim spawned a thread and immediately joined it. That is a synchronous call -- the parallelism the Windows thread pool provides is absent here either way -- so it only bought a pthread_create() and its stack mmap, ~15 us, for every FFT frame. Call the function directly. (The Windows build is untouched and still gets its thread pool; a real pool for POSIX would be a separate change, and only pays off for num_stitch > 1.) The dispatcher thread ran `for (each ss, LO) {...} Sleep(1);`, and Sleep(1) is usleep(1000), so it woke 1000 times a second to re-read the same flags. That burned 0.63% of a core even with no samples arriving. Give it a semaphore instead, signalled by the four Spectrum*() entry points when new samples land, and by SetAnalyzer/DestroyAnalyzer after they raise end_dispatcher so the blocking wait always has a way out. SetAnalyzer holds SetAnalyzerSection while it waits for the dispatcher to quit, and the dispatcher never takes that section, so signalling from under it is safe. Measured on an Apple M1 Pro; 16384-point complex FFT, 1024-sample buffers, ~40 pixel frames/s, CPU of all threads via getrusage: idle, no samples at all 0.63% of a core -> 0.00% under load 3.54% -> 2.65% Output pixels are bit-identical. Stressed with 3 create/destroy cycles and 24 on-the-fly SetAnalyzer reconfigurations while a second thread fed samples: no deadlock. Co-Authored-By: Claude Opus 4.8 --- analyzer.c | 20 +++++++++++++++++++- analyzer.h | 3 +++ linux_port.c | 14 +++++++++++--- 3 files changed, 33 insertions(+), 4 deletions(-) diff --git a/analyzer.c b/analyzer.c index 00dc36f..94598ed 100644 --- a/analyzer.c +++ b/analyzer.c @@ -1087,7 +1087,15 @@ void __cdecl sendbuf(void *arg) LeaveCriticalSection(&(a->BufferControlSection[a->ss][a->LO])); } } - Sleep(1); + // + // Block until a Spectrum*() call announces new samples. This used to be + // Sleep(1), i.e. 1000 wakeups per second spent re-reading the same flags + // -- 0.67% of a core even with no data arriving at all. + // + // Whoever sets end_dispatcher also signals the semaphore, so the wait + // below always has a way out. + // + WaitForSingleObject(a->Sem_BuffReady, INFINITE); } InterlockedBitTestAndReset(&a->dispatcher, 0); _endthread(); @@ -1202,6 +1210,9 @@ void SetAnalyzer ( int disp, // display identifier EnterCriticalSection(&a->SetAnalyzerSection); a->end_dispatcher = 1; + // wake the dispatcher out of its blocking wait so it can observe the flag; + // it does not take SetAnalyzerSection, so holding it here is safe + ReleaseSemaphore(a->Sem_BuffReady, 1, 0); while (InterlockedAnd(&a->dispatcher, 1)) Sleep(1); a->stop = 1; @@ -1347,6 +1358,7 @@ void XCreateAnalyzer( int disp, a->hSnapEvent[i][j] = CreateEvent(NULL, FALSE, FALSE, TEXT("snap")); a->snap[i][j] = 0; } + a->Sem_BuffReady = CreateSemaphore(0, 0, 1000, 0); InitializeCriticalSectionAndSpinCount(&a->ResampleSection, 0); InitializeCriticalSectionAndSpinCount(&a->SetAnalyzerSection, 0); InitializeCriticalSectionAndSpinCount(&a->StitchSection, 0); @@ -1434,6 +1446,7 @@ void DestroyAnalyzer(int disp) int i, j; a->end_dispatcher = 1; + ReleaseSemaphore(a->Sem_BuffReady, 1, 0); while (InterlockedAnd(&a->dispatcher, 1)) Sleep(1); @@ -1497,6 +1510,7 @@ void DestroyAnalyzer(int disp) for (i = 0; i < a->max_stitch; i++) for (j = 0; j < a->max_num_fft; j++) CloseHandle(a->hSnapEvent[i][j]); + CloseHandle(a->Sem_BuffReady); _aligned_free ((void *) a->pnum_threads); @@ -1633,6 +1647,7 @@ void CloseBuffer(int disp, int ss, int LO) if((a->IQin_index[ss][LO] += a->buff_size) >= a->bsize) //REQUIRES buff_size IS A SUB-MULTIPLE OF SIZE OF INPUT SAMPLE BUFFS! a->IQin_index[ss][LO] = 0; + ReleaseSemaphore(a->Sem_BuffReady, 1, 0); // new samples: let the dispatcher run if (!InterlockedAnd(&a->dispatcher, 1)) { InterlockedBitTestAndSet (&a->dispatcher, 0); @@ -1672,6 +1687,7 @@ void Spectrum(int disp, int ss, int LO, dINREAL* pI, dINREAL* pQ) if((a->IQin_index[ss][LO] += a->buff_size) >= a->bsize) //REQUIRES buff_size IS A SUB-MULTIPLE OF SIZE OF INPUT SAMPLE BUFFS! a->IQin_index[ss][LO] = 0; + ReleaseSemaphore(a->Sem_BuffReady, 1, 0); // new samples: let the dispatcher run if (!InterlockedAnd(&a->dispatcher, 1)) { InterlockedBitTestAndSet(&a->dispatcher, 0); @@ -1717,6 +1733,7 @@ void Spectrum2(int run, int disp, int ss, int LO, dINREAL* pbuff) if((a->IQin_index[ss][LO] += a->buff_size) >= a->bsize) //REQUIRES buff_size IS A SUB-MULTIPLE OF SIZE OF INPUT SAMPLE BUFFS! a->IQin_index[ss][LO] = 0; + ReleaseSemaphore(a->Sem_BuffReady, 1, 0); // new samples: let the dispatcher run if (!InterlockedAnd(&a->dispatcher, 1)) { InterlockedBitTestAndSet(&a->dispatcher, 0); @@ -1763,6 +1780,7 @@ void Spectrum0(int run, int disp, int ss, int LO, double* pbuff) if((a->IQin_index[ss][LO] += a->buff_size) >= a->bsize) //REQUIRES buff_size IS A SUB-MULTIPLE OF SIZE OF INPUT SAMPLE BUFFS! a->IQin_index[ss][LO] = 0; + ReleaseSemaphore(a->Sem_BuffReady, 1, 0); // new samples: let the dispatcher run if (!InterlockedAnd(&a->dispatcher, 1)) { InterlockedBitTestAndSet(&a->dispatcher, 0); diff --git a/analyzer.h b/analyzer.h index 9c6675c..8ef2e75 100644 --- a/analyzer.h +++ b/analyzer.h @@ -121,6 +121,9 @@ typedef struct _dp HANDLE hSnapEvent[dMAX_STITCH][dMAX_NUM_FFT]; // mutex handles; mutexes will be used to signal a snap is complete double *snap_buff[dMAX_STITCH][dMAX_NUM_FFT]; // pointers to buffers for the snap + HANDLE Sem_BuffReady; // signalled when input samples arrive, so the + // dispatcher can block instead of polling + CRITICAL_SECTION PB_ControlsSection[dMAX_PIXOUTS]; CRITICAL_SECTION SetAnalyzerSection; CRITICAL_SECTION BufferControlSection[dMAX_STITCH][dMAX_NUM_FFT]; diff --git a/linux_port.c b/linux_port.c index 8e40a91..b7988db 100644 --- a/linux_port.c +++ b/linux_port.c @@ -39,9 +39,17 @@ john.d.melton@googlemail.com #if defined(linux) || defined(__APPLE__) void QueueUserWorkItem(void *function,void *context,int flags) { - pthread_t t; - pthread_create(&t, NULL, function, context); - pthread_join(t, NULL); + // + // The Windows call queues the work item on a thread pool and returns at + // once, so callers get their items run in parallel. This shim spawned a + // thread and immediately joined it, which is a plain synchronous call that + // happens to cost a thread creation (~15 us) and delivers no parallelism. + // Call the function directly: same ordering, no thread. + // + // The cast matches the one pthread_create() performed here before. + // + (void)flags; + ((void *(*)(void *))function)(context); } static inline void init_crit_section(pthread_mutex_t *mutex) {