From e588882eb9d47ef2dd1477b817798f054a26d4d5 Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Mon, 10 Nov 2025 18:13:54 +1100 Subject: [PATCH 1/9] fix: Remove unused threadState code flagged by AddressSanitizer AddressSanitizer identified a new-delete-type-mismatch in threadState. The feature is unused anywhere in the codebase; removing it eliminates the bug and the dead code. Signed-off-by: Mark Reid --- src/lib/mu/Mu/Mu/Thread.h | 4 --- src/lib/mu/Mu/Thread.cpp | 56 --------------------------------------- 2 files changed, 60 deletions(-) diff --git a/src/lib/mu/Mu/Mu/Thread.h b/src/lib/mu/Mu/Mu/Thread.h index de072d582..bd8b4dc1b 100644 --- a/src/lib/mu/Mu/Mu/Thread.h +++ b/src/lib/mu/Mu/Mu/Thread.h @@ -349,9 +349,6 @@ namespace Mu void setProcess(Process* p) { _process = p; } - void* threadState(); - size_t threadStateSize(); - private: Process* _process; const Node* _rootNode; @@ -370,7 +367,6 @@ namespace Mu StackPointer _bottomOfStack; bool _applicationThread; bool _suspended; - size_t* _threadState; const Node* _continuation; pthread_mutex_t _controlMutex; diff --git a/src/lib/mu/Mu/Thread.cpp b/src/lib/mu/Mu/Thread.cpp index 5e15448b6..ca5316843 100644 --- a/src/lib/mu/Mu/Thread.cpp +++ b/src/lib/mu/Mu/Thread.cpp @@ -80,19 +80,10 @@ namespace Mu , _continuation(0) , _applicationThread(appthread) , _returnValueType(0) - , _threadState(0) { GarbageCollector::init(); _stack.reserve(_stackCapacity); -#ifdef PLATFORM_DARWIN - _threadState = new size_t[sizeof(DARWIN_THREAD_STATE)]; -#endif - -#ifdef PLATFORM_LINUX - _threadState = (size_t*)(new ucontext_t); -#endif - if (isApplicationThread()) { _id = pthread_self(); @@ -130,14 +121,6 @@ namespace Mu _process->removeThread(this); _process = 0; - -#ifdef PLATFORM_DARWIN - delete[] _threadState; -#endif - -#ifdef PLATFORM_LINUX - delete _threadState; -#endif } void* Thread::trampoline(void* arg) @@ -488,45 +471,6 @@ namespace Mu } } - void* Thread::threadState() - { -#ifdef PLATFORM_DARWIN - if (_alive) - { - mach_msg_type_number_t thread_state_count = DARWIN_THREAD_STATE_COUNT; - kern_return_t krc; - if ((krc = thread_get_state(pthread_mach_thread_np(_id), // <-- can't use self here - DARWIN_THREAD_STATE, (natural_t*)_threadState, &thread_state_count)) - != KERN_SUCCESS) - { - return _threadState; - } - else - { - cerr << "Thread::threadState(): error obtaining thread state" << endl; - } - } -#endif - - return 0; - } - - size_t Thread::threadStateSize() - { -#if 0 -#ifdef PLATFORM_DARWIN - return sizeof(darwin_thread_state); -#endif - -#ifdef PLATFORM_LINUX - return sizeof(ucontext); -#endif - -#endif - /* AJG - baaaaad ! */ - return sizeof(int); - } - void Thread::jump(int returnValue, int n, Value v) { if (_jumpPoints.size()) From 21094b824d6a76a5ee4f5380efdb67a90c372e45 Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Tue, 11 Nov 2025 15:47:12 +1100 Subject: [PATCH 2/9] fix: Defer thread start until after construction in TwkMediaLibrary Starting a thread in the constructor risks accessing members still being initialized. Identified by ThreadSanitizer. Add an explicit start() method called post-construction and remove the unnecessary threadTrampoline wrapper. Signed-off-by: Mark Reid --- src/lib/app/RvCommon/RvConsoleApplication.cpp | 1 + src/lib/app/TwkMediaLibrary/Library.cpp | 9 +++++---- src/lib/app/TwkMediaLibrary/TwkMediaLibrary/Library.h | 3 +-- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/lib/app/RvCommon/RvConsoleApplication.cpp b/src/lib/app/RvCommon/RvConsoleApplication.cpp index 68f0f5ffb..b863d8c22 100644 --- a/src/lib/app/RvCommon/RvConsoleApplication.cpp +++ b/src/lib/app/RvCommon/RvConsoleApplication.cpp @@ -16,6 +16,7 @@ namespace Rv , m_pyLibrary(INTERNAL_APPLICATION_NAME) { m_pyLibrary.setName("py-media-library"); + m_pyLibrary.start(); } } // namespace Rv diff --git a/src/lib/app/TwkMediaLibrary/Library.cpp b/src/lib/app/TwkMediaLibrary/Library.cpp index 890ef8a50..e092044ae 100644 --- a/src/lib/app/TwkMediaLibrary/Library.cpp +++ b/src/lib/app/TwkMediaLibrary/Library.cpp @@ -346,7 +346,6 @@ namespace TwkMediaLibrary , m_appName(appName) , m_name(typeName) , m_taskStop(false) - , m_taskThread(threadTrampoline, this) { // globalLibraryMap[typeName] = this; } @@ -359,8 +358,10 @@ namespace TwkMediaLibrary m_taskCond.notify_one(); } - m_taskThread.join(); - + if (m_taskThread.joinable()) + { + m_taskThread.join(); + } // // willDeleteSignal() should be sent by derived classes. Its sent // here as a stop-gap but if this is used is almost certain going @@ -636,7 +637,7 @@ namespace TwkMediaLibrary const Task* Library::referencesToNodeASync(const Node*, PropertyQueryResultFunction) const { return 0; } - void Library::threadTrampoline(Library* library) { library->threadMain(); } + void Library::start() { m_taskThread = Thread(&Library::threadMain, this); } void Library::addTask(Task* item) const { diff --git a/src/lib/app/TwkMediaLibrary/TwkMediaLibrary/Library.h b/src/lib/app/TwkMediaLibrary/TwkMediaLibrary/Library.h index a1bffc5fb..bf5644093 100644 --- a/src/lib/app/TwkMediaLibrary/TwkMediaLibrary/Library.h +++ b/src/lib/app/TwkMediaLibrary/TwkMediaLibrary/Library.h @@ -817,9 +817,8 @@ namespace TwkMediaLibrary // ASync/Parallel Task API // - static void threadTrampoline(Library*); void threadMain(); - + void start(); void addTask(Task*) const; void cancelTask(const Task*) const; void cancelTasks(const TaskVector&) const; From 51868924e5e8950742f2dfd65365ea89b0c45b17 Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Wed, 12 Nov 2025 09:43:27 +1100 Subject: [PATCH 3/9] refactor: Remove unused boost::thread_group includes Signed-off-by: Mark Reid --- src/bin/apps/rv/main.cpp | 3 +-- src/bin/apps/rvpkg/main.cpp | 3 +-- src/bin/imgtools/rvio/main.cpp | 1 - src/bin/imgtools/rvls/main.cpp | 1 - src/bin/nsapps/RV/main.cpp | 3 +-- src/lib/app/RvApp/Options.cpp | 1 + src/lib/image/TwkMovie/Movie.cpp | 1 - src/lib/ip/IPCore/IPCore/Session.h | 1 + .../BlackMagicDevices/BlackMagicDevices/DeckLinkVideoDevice.h | 2 -- src/plugins/output/NDI/NDI/NDIVideoDevice.h | 1 - 10 files changed, 5 insertions(+), 12 deletions(-) diff --git a/src/bin/apps/rv/main.cpp b/src/bin/apps/rv/main.cpp index a24e8e820..ae1016b12 100644 --- a/src/bin/apps/rv/main.cpp +++ b/src/bin/apps/rv/main.cpp @@ -80,7 +80,6 @@ #include #include #include -#include #include #include #include @@ -122,7 +121,7 @@ extern void rvThirdPartyCustomization(TwkApp::Bundle& bundle, char* licarg); // #error ********* NO VERSION INFORMATION *********** #else #ifndef _MSC_VER -#warning********* NO VERSION INFORMATION *********** +#warning ********* NO VERSION INFORMATION *********** #endif #endif #endif diff --git a/src/bin/apps/rvpkg/main.cpp b/src/bin/apps/rvpkg/main.cpp index 329da7804..bd067c710 100644 --- a/src/bin/apps/rvpkg/main.cpp +++ b/src/bin/apps/rvpkg/main.cpp @@ -40,7 +40,6 @@ #include #include #include -#include #include #include #include @@ -71,7 +70,7 @@ // #error ********* NO VERSION INFORMATION *********** #else #ifndef _MSC_VER -#warning********* NO VERSION INFORMATION *********** +#warning ********* NO VERSION INFORMATION *********** #endif #endif #endif diff --git a/src/bin/imgtools/rvio/main.cpp b/src/bin/imgtools/rvio/main.cpp index a49d2b9f3..2bee038d7 100644 --- a/src/bin/imgtools/rvio/main.cpp +++ b/src/bin/imgtools/rvio/main.cpp @@ -74,7 +74,6 @@ #include #include #include -#include #include #include #include diff --git a/src/bin/imgtools/rvls/main.cpp b/src/bin/imgtools/rvls/main.cpp index 4cc7d6b4d..e45782aee 100644 --- a/src/bin/imgtools/rvls/main.cpp +++ b/src/bin/imgtools/rvls/main.cpp @@ -36,7 +36,6 @@ #include #include #include -#include #include #include #include diff --git a/src/bin/nsapps/RV/main.cpp b/src/bin/nsapps/RV/main.cpp index 7824bb510..8185e6bbb 100644 --- a/src/bin/nsapps/RV/main.cpp +++ b/src/bin/nsapps/RV/main.cpp @@ -55,7 +55,6 @@ #include #include #include -#include #include #include #include @@ -83,7 +82,7 @@ extern void rvThirdPartyCustomization(TwkApp::Bundle& bundle, char* licarg); #if NDEBUG // #error ********* NO VERSION INFORMATION *********** #else -#warning********* NO VERSION INFORMATION *********** +#warning ********* NO VERSION INFORMATION *********** #endif #endif diff --git a/src/lib/app/RvApp/Options.cpp b/src/lib/app/RvApp/Options.cpp index 631f7bfff..dc27fa257 100644 --- a/src/lib/app/RvApp/Options.cpp +++ b/src/lib/app/RvApp/Options.cpp @@ -42,6 +42,7 @@ #include #include #include +#include extern void TwkMovie_GenericIO_setDebug(bool); extern void TwkFB_GenericIO_setDebug(bool); diff --git a/src/lib/image/TwkMovie/Movie.cpp b/src/lib/image/TwkMovie/Movie.cpp index 673d85fd6..2ca1bc16d 100644 --- a/src/lib/image/TwkMovie/Movie.cpp +++ b/src/lib/image/TwkMovie/Movie.cpp @@ -15,7 +15,6 @@ #include #include #include -#include #include #include #include diff --git a/src/lib/ip/IPCore/IPCore/Session.h b/src/lib/ip/IPCore/IPCore/Session.h index 41c17fc85..00a8ce1b6 100644 --- a/src/lib/ip/IPCore/IPCore/Session.h +++ b/src/lib/ip/IPCore/IPCore/Session.h @@ -24,6 +24,7 @@ #include #include #include +#include namespace IPCore { diff --git a/src/plugins/output/BlackMagicDevices/BlackMagicDevices/DeckLinkVideoDevice.h b/src/plugins/output/BlackMagicDevices/BlackMagicDevices/DeckLinkVideoDevice.h index 0fae0a070..250571337 100644 --- a/src/plugins/output/BlackMagicDevices/BlackMagicDevices/DeckLinkVideoDevice.h +++ b/src/plugins/output/BlackMagicDevices/BlackMagicDevices/DeckLinkVideoDevice.h @@ -29,7 +29,6 @@ #include #include -#include #include #include @@ -96,7 +95,6 @@ namespace BlackMagicDevices using GLFence = TwkGLF::GLFence; using GLFBO = TwkGLF::GLFBO; using BufferVector = std::vector; - using ThreadGroup = stl_ext::thread_group; using AudioBuffer = std::vector; using StereoFrameMap = std::map>; using DLVideoFrameDeque = std::deque; diff --git a/src/plugins/output/NDI/NDI/NDIVideoDevice.h b/src/plugins/output/NDI/NDI/NDIVideoDevice.h index 50a642088..4e2a05f5d 100644 --- a/src/plugins/output/NDI/NDI/NDIVideoDevice.h +++ b/src/plugins/output/NDI/NDI/NDIVideoDevice.h @@ -35,7 +35,6 @@ #include #include -#include #include namespace NDI From 0a9dd4652e350fe0848362ed49f814ffdad904f6 Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Wed, 12 Nov 2025 10:02:09 +1100 Subject: [PATCH 4/9] refactor: Replace boost::mutex with std::mutex in ALSAAudioRenderer Signed-off-by: Mark Reid --- .../ALSAAudioModule/ALSAAudioRenderer.h | 3 +- .../ALSAAudioModule/ALSAAudioRenderer.cpp | 35 ++++++++++--------- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/src/lib/audio/ALSAAudioModule/ALSAAudioModule/ALSAAudioRenderer.h b/src/lib/audio/ALSAAudioModule/ALSAAudioModule/ALSAAudioRenderer.h index d8334e3ce..5d0600668 100644 --- a/src/lib/audio/ALSAAudioModule/ALSAAudioModule/ALSAAudioRenderer.h +++ b/src/lib/audio/ALSAAudioModule/ALSAAudioModule/ALSAAudioRenderer.h @@ -13,6 +13,7 @@ #include #include +#include namespace IPCore { @@ -92,7 +93,7 @@ namespace IPCore snd_pcm_uframes_t m_bufferSize; pthread_t m_audioThread; bool m_audioThreadRunning; - pthread_mutex_t m_runningLock; + std::mutex m_runningLock; }; } // namespace IPCore diff --git a/src/lib/audio/ALSAAudioModule/ALSAAudioRenderer.cpp b/src/lib/audio/ALSAAudioModule/ALSAAudioRenderer.cpp index 7a3b32141..73b96588d 100644 --- a/src/lib/audio/ALSAAudioModule/ALSAAudioRenderer.cpp +++ b/src/lib/audio/ALSAAudioModule/ALSAAudioRenderer.cpp @@ -229,7 +229,6 @@ namespace IPCore , m_audioThreadRunning(false) { snd_lib_error_set_handler(alsaErrorHandler); - pthread_mutex_init(&m_runningLock, 0); if (m_parameters.rate == 0) m_parameters.rate = TWEAK_AUDIO_DEFAULT_SAMPLE_RATE; @@ -263,7 +262,6 @@ namespace IPCore { AudioRenderer::stop(); shutdown(); - pthread_mutex_destroy(&m_runningLock); } void ALSAAudioRenderer::createDeviceList() @@ -629,7 +627,7 @@ namespace IPCore snd_pcm_hw_params_set_periods(pcm, params, periods, 0); snd_pcm_hw_params_set_buffer_size(pcm, params, (period_size * periods) >> 2); - + if (snd_pcm_hw_params(pcm, params) >= 0 && rrate == rate) { rates.push_back((unsigned int)rate); @@ -882,10 +880,11 @@ namespace IPCore if (!m_pcm) return; - m_threadGroup.lock(m_runningLock); - m_audioThreadRunning = true; - m_audioThread = pthread_self(); - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + m_audioThreadRunning = true; + m_audioThread = pthread_self(); + } // // Get the current state. (yeah, I know it was just set above) @@ -1123,9 +1122,10 @@ namespace IPCore if (pcmState == SND_PCM_STATE_RUNNING) snd_pcm_drop(m_pcm); - m_threadGroup.lock(m_runningLock); - m_audioThreadRunning = false; - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + m_audioThreadRunning = false; + } if (debug) cout << "DEBUG: audio thread exiting" << endl; @@ -1180,12 +1180,14 @@ namespace IPCore const bool holdOpen = m_parameters.holdOpen; m_parameters.holdOpen = false; + bool isrunning = false; AudioRenderer::stop(); - m_threadGroup.lock(m_runningLock); - bool isrunning = m_audioThreadRunning; - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + isrunning = m_audioThreadRunning; + } if (isrunning) m_threadGroup.control_wait(true, 1.0); @@ -1199,9 +1201,10 @@ namespace IPCore // the audio device until after m_pcm has been zeroed. // snd_pcm_t* tmp = m_pcm; - m_threadGroup.lock(m_runningLock); - m_pcm = 0; - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + m_pcm = 0; + } snd_pcm_close(tmp); } From 5cf2bf195e87a0a0f073c0699f49790d9e496dfa Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Wed, 12 Nov 2025 10:09:00 +1100 Subject: [PATCH 5/9] refactor: Replace boost::mutex with std::mutex in ALSASafeAudioRenderer Signed-off-by: Mark Reid --- .../ALSASafeAudioRenderer.h | 3 +- .../ALSASafeAudioRenderer.cpp | 38 ++++++++++--------- 2 files changed, 23 insertions(+), 18 deletions(-) diff --git a/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioModule/ALSASafeAudioRenderer.h b/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioModule/ALSASafeAudioRenderer.h index fde02710b..42b3ea729 100644 --- a/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioModule/ALSASafeAudioRenderer.h +++ b/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioModule/ALSASafeAudioRenderer.h @@ -13,6 +13,7 @@ #include #include +#include namespace IPCore { @@ -92,7 +93,7 @@ namespace IPCore snd_pcm_uframes_t m_bufferSize; pthread_t m_audioThread; bool m_audioThreadRunning; - pthread_mutex_t m_runningLock; + std::mutex m_runningLock; }; } // namespace IPCore diff --git a/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioRenderer.cpp b/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioRenderer.cpp index f3b464358..b2740e61c 100644 --- a/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioRenderer.cpp +++ b/src/lib/audio/ALSASafeAudioModule/ALSASafeAudioRenderer.cpp @@ -230,7 +230,6 @@ namespace IPCore , m_audioThreadRunning(false) { snd_lib_error_set_handler(alsaErrorHandler); - pthread_mutex_init(&m_runningLock, 0); if (m_parameters.rate == 0) m_parameters.rate = TWEAK_AUDIO_DEFAULT_SAMPLE_RATE; @@ -264,7 +263,6 @@ namespace IPCore { AudioRenderer::stop(); shutdown(); - pthread_mutex_destroy(&m_runningLock); } void ALSASafeAudioRenderer::createDeviceList() @@ -639,7 +637,7 @@ namespace IPCore snd_pcm_hw_params_set_periods(pcm, params, periods, 0); snd_pcm_hw_params_set_buffer_size(pcm, params, (period_size * periods) >> 2); - + if (snd_pcm_hw_params(pcm, params) >= 0 && rrate == rate) { rates.push_back((unsigned int)rate); @@ -890,10 +888,11 @@ namespace IPCore if (!m_pcm) return; - m_threadGroup.lock(m_runningLock); - m_audioThreadRunning = true; - m_audioThread = pthread_self(); - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + m_audioThreadRunning = true; + m_audioThread = pthread_self(); + } // // Get the current state. (yeah, I know it was just set above) @@ -1130,10 +1129,10 @@ namespace IPCore return; if (pcmState == SND_PCM_STATE_RUNNING) snd_pcm_drop(m_pcm); - - m_threadGroup.lock(m_runningLock); - m_audioThreadRunning = false; - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + m_audioThreadRunning = false; + } if (debug) cout << "DEBUG: audio thread exiting" << endl; @@ -1188,12 +1187,14 @@ namespace IPCore const bool holdOpen = m_parameters.holdOpen; m_parameters.holdOpen = false; + bool isrunning = false; AudioRenderer::stop(); - m_threadGroup.lock(m_runningLock); - bool isrunning = m_audioThreadRunning; - m_threadGroup.unlock(m_runningLock); + { + std::lock_guard guard(m_runningLock); + isrunning = m_audioThreadRunning; + } if (isrunning) m_threadGroup.control_wait(true, 1.0); @@ -1207,9 +1208,12 @@ namespace IPCore // the audio device until after m_pcm has been zeroed. // snd_pcm_t* tmp = m_pcm; - m_threadGroup.lock(m_runningLock); - m_pcm = 0; - m_threadGroup.unlock(m_runningLock); + + { + std::lock_guard guard(m_runningLock); + m_pcm = 0; + } + snd_pcm_close(tmp); } From ad4ae837dcc7a0342221d83ec016232e5156a59b Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Wed, 12 Nov 2025 10:28:26 +1100 Subject: [PATCH 6/9] refactor: Replace boost::mutex with std::mutex in ThreadedMovie m_runLock was not being destroyed previously. Signed-off-by: Mark Reid --- src/lib/image/TwkMovie/ThreadedMovie.cpp | 78 +++++++++---------- .../image/TwkMovie/TwkMovie/ThreadedMovie.h | 5 +- 2 files changed, 40 insertions(+), 43 deletions(-) diff --git a/src/lib/image/TwkMovie/ThreadedMovie.cpp b/src/lib/image/TwkMovie/ThreadedMovie.cpp index 005beed50..08257cd5c 100644 --- a/src/lib/image/TwkMovie/ThreadedMovie.cpp +++ b/src/lib/image/TwkMovie/ThreadedMovie.cpp @@ -33,8 +33,6 @@ namespace TwkMovie // if (!m_movie->isThreadSafe()) throw runtime_exception(); m_info = movies.front()->info(); m_threadSafe = false; - pthread_mutex_init(&m_mapLock, 0); - pthread_mutex_init(&m_runLock, 0); m_threadData.resize(movies.size()); for (size_t i = 0; i < m_threadData.size(); i++) @@ -60,13 +58,11 @@ namespace TwkMovie for (size_t q = 0; q < i->second.size(); q++) delete i->second[q]; } - - pthread_mutex_destroy(&m_mapLock); } - void ThreadedMovie::lock() { m_threadGroup.lock(m_mapLock); } + void ThreadedMovie::lock() { m_mapLock.lock(); } - void ThreadedMovie::unlock() { m_threadGroup.unlock(m_mapLock); } + void ThreadedMovie::unlock() { m_mapLock.unlock(); } void ThreadedMovie::threadMain() { @@ -83,46 +79,46 @@ namespace TwkMovie ThreadData* td = 0; pthread_t self = pthread_self(); - m_threadGroup.lock(m_runLock); - - // - // If we already have a ThreadData for this thread use it - // - - for (size_t i = 0; !td && i < m_threadData.size(); i++) { - ThreadData& d = m_threadData[i]; + std::lock_guard guard(m_runLock); - if (d.init && pthread_equal(d.thread, self)) + // + // If we already have a ThreadData for this thread use it + // + + for (size_t i = 0; !td && i < m_threadData.size(); i++) { - d.running = true; - td = &d; - } - } + ThreadData& d = m_threadData[i]; - // - // If we don't have a ThreadData find an available one and use - // that - // + if (d.init && pthread_equal(d.thread, self)) + { + d.running = true; + td = &d; + } + } - bool first = false; + // + // If we don't have a ThreadData find an available one and use + // that + // - for (size_t i = 0; !td && i < m_threadData.size(); i++) - { - ThreadData& d = m_threadData[i]; + bool first = false; - if (!d.init) + for (size_t i = 0; !td && i < m_threadData.size(); i++) { - d.init = true; - first = true; - d.thread = self; - d.running = true; - td = &d; + ThreadData& d = m_threadData[i]; + + if (!d.init) + { + d.init = true; + first = true; + d.thread = self; + d.running = true; + td = &d; + } } } - m_threadGroup.unlock(m_runLock); - if (m_initialize) m_initialize(); @@ -188,12 +184,12 @@ namespace TwkMovie } } while (1); - m_threadGroup.lock(m_runLock); - td->running = false; - - bool allFramesDone = (m_currentIndex >= m_frames.size()); - - m_threadGroup.unlock(m_runLock); + bool allFramesDone; + { + std::lock_guard guard(m_runLock); + td->running = false; + allFramesDone = (m_currentIndex >= m_frames.size()); + } if (allFramesDone && m_finalize != nullptr) { diff --git a/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h b/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h index 18d71d0b0..50ad09a4b 100644 --- a/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h +++ b/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h @@ -12,6 +12,7 @@ #include #include #include +#include namespace TwkMovie { @@ -98,8 +99,8 @@ namespace TwkMovie Movies m_movies; ThreadGroup m_threadGroup; FBMap m_map; - pthread_mutex_t m_mapLock; - pthread_mutex_t m_runLock; + std::mutex m_mapLock; + std::mutex m_runLock; Frames m_frames; ThreadDataVector m_threadData; int m_currentIndex; From 12d7220ff91520744d332543a193e0bd2682f39e Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Wed, 12 Nov 2025 10:52:03 +1100 Subject: [PATCH 7/9] refactor: Adopt std::lock_guard for RAII locking in ThreadedMovie Signed-off-by: Mark Reid --- src/lib/image/TwkMovie/ThreadedMovie.cpp | 139 ++++++++++-------- .../image/TwkMovie/TwkMovie/ThreadedMovie.h | 19 --- 2 files changed, 74 insertions(+), 84 deletions(-) diff --git a/src/lib/image/TwkMovie/ThreadedMovie.cpp b/src/lib/image/TwkMovie/ThreadedMovie.cpp index 08257cd5c..4a227b0cc 100644 --- a/src/lib/image/TwkMovie/ThreadedMovie.cpp +++ b/src/lib/image/TwkMovie/ThreadedMovie.cpp @@ -60,10 +60,6 @@ namespace TwkMovie } } - void ThreadedMovie::lock() { m_mapLock.lock(); } - - void ThreadedMovie::unlock() { m_mapLock.unlock(); } - void ThreadedMovie::threadMain() { const size_t threads = m_threadGroup.num_threads(); @@ -124,64 +120,69 @@ namespace TwkMovie do { - lock(); - const size_t current = m_currentIndex; - const size_t requested = m_requestIndex; + size_t current, requested; + int frame; + bool exists; - if (current - requested < threads * 2 && current < m_frames.size()) { - // - // Bump the current index for the next thread - // + std::lock_guard guard(m_mapLock); + current = m_currentIndex; + requested = m_requestIndex; - m_currentIndex++; - const int frame = m_frames[current]; - bool exists = m_map.count(frame) > 0; - unlock(); - - if (!exists) + if (current - requested < threads * 2 && current < m_frames.size()) { - td->request.frame = frame; - td->request.missing = false; - FrameBufferVector fbs; - - try - { - // cout << "thread " << td->id << " @ frame " << - // td->request.frame << endl; - td->movie->imagesAtFrame(td->request, fbs); - } - catch (std::exception& exc) - { - cerr << "WARNING: an exception was raised evaluting " - "frame " - << frame << ":" << endl; - cerr << exc.what() << endl; - unlock(); - break; - } - - lock(); - m_map[frame] = fbs; - unlock(); + // + // Bump the current index for the next thread + // + + m_currentIndex++; + frame = m_frames[current]; + exists = m_map.count(frame) > 0; } else { - // cout << "thread " << td->id << " @ frame " << - // td->request.frame - //<< " already in cache" + // cout << "thread " << td->id << " finished, current = " << + // current + //<< ", requested = " << requested //<< endl; + break; + } + } + + if (!exists) + { + td->request.frame = frame; + td->request.missing = false; + FrameBufferVector fbs; + + try + { + // cout << "thread " << td->id << " @ frame " << + // td->request.frame << endl; + td->movie->imagesAtFrame(td->request, fbs); + } + catch (std::exception& exc) + { + cerr << "WARNING: an exception was raised evaluting " + "frame " + << frame << ":" << endl; + cerr << exc.what() << endl; + break; + } + + { + std::lock_guard guard(m_mapLock); + m_map[frame] = std::move(fbs); } } else { - unlock(); - // cout << "thread " << td->id << " finished, current = " << - // current - //<< ", requested = " << requested + // cout << "thread " << td->id << " @ frame " << + // td->request.frame + //<< " already in cache" //<< endl; - break; } + } while (1); bool allFramesDone; @@ -246,13 +247,15 @@ namespace TwkMovie int frame = request.frame; const size_t n = m_threadGroup.num_threads(); + size_t current, requested; dispatchAll(); - lock(); - size_t current = m_currentIndex; - size_t requested = m_requestIndex; - unlock(); + { + std::lock_guard guard(m_mapLock); + current = m_currentIndex; + requested = m_requestIndex; + } #if 0 if (frame != m_frames[requested]) @@ -266,18 +269,23 @@ namespace TwkMovie for (size_t count = 0; true; count++) { - lock(); - FBMap::iterator i = m_map.find(frame); - FBMap::iterator e = m_map.end(); - unlock(); + bool found = false; + { + std::lock_guard guard(m_mapLock); + FBMap::iterator i = m_map.find(frame); + FBMap::iterator e = m_map.end(); - if (i != e) + if (i != e) + { + fbs = i->second; + m_map.erase(i); + // cout << "consumed frame " << frame << endl; + found = true; + } + } + + if (found) { - fbs = i->second; - lock(); - m_map.erase(i); - // cout << "consumed frame " << frame << endl; - unlock(); dispatchAll(); break; } @@ -299,9 +307,10 @@ namespace TwkMovie } } - lock(); - m_requestIndex++; - unlock(); + { + std::lock_guard guard(m_mapLock); + m_requestIndex++; + } } void ThreadedMovie::identifiersAtFrame(const ReadRequest& request, IdentifierVector& ids) diff --git a/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h b/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h index 50ad09a4b..d9d67ed1c 100644 --- a/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h +++ b/src/lib/image/TwkMovie/TwkMovie/ThreadedMovie.h @@ -76,25 +76,6 @@ namespace TwkMovie void dispatchAll(); - protected: - void lock(); - void unlock(); - - struct Lock - { - Lock(ThreadedMovie* m) - : movie(m) - { - movie->lock(); - } - - ~Lock() { movie->unlock(); } - - ThreadedMovie* movie; - }; - - friend class ThreadedMovie::Lock; - private: Movies m_movies; ThreadGroup m_threadGroup; From ba2c664e7dd1e4d9bfe248a36147b17535e3d8a2 Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Wed, 12 Nov 2025 12:27:56 +1100 Subject: [PATCH 8/9] refactor: Replace boost::mutex with std::mutex in MovieFBWriter Signed-off-by: Mark Reid --- src/lib/image/MovieFB/MovieFBWriter.cpp | 71 ++++++++++--------------- 1 file changed, 28 insertions(+), 43 deletions(-) diff --git a/src/lib/image/MovieFB/MovieFBWriter.cpp b/src/lib/image/MovieFB/MovieFBWriter.cpp index 258667e83..0dabb7d59 100644 --- a/src/lib/image/MovieFB/MovieFBWriter.cpp +++ b/src/lib/image/MovieFB/MovieFBWriter.cpp @@ -22,6 +22,9 @@ #include #include #include +#include +#include +#include namespace TwkMovie { @@ -92,15 +95,12 @@ namespace TwkMovie { public: WriteTaskManager(int size); - ~WriteTaskManager(); const WriteTask& taskByIndex(int index); void finishTask(int index); int nextReadyTaskIndex(); void addTask(WriteTask& t); void waitAll(); - void lock(); - void unlock(); void threadMain(int threadNumber); void dispatchThreads(); @@ -124,8 +124,8 @@ namespace TwkMovie private: vector tasks; stl_ext::thread_group threadGroup; - pthread_mutex_t managerLock; - pthread_cond_t waitCond; + std::mutex managerLock; + std::condition_variable waitCond; vector threadData; // Indicates that a task is currently being added when non 0 @@ -141,26 +141,13 @@ namespace TwkMovie for (int i = 0; i < size; ++i) threadData.push_back(ThreadData(this, i)); - - pthread_mutex_init(&managerLock, 0); - pthread_cond_init(&waitCond, 0); - } - - WriteTaskManager::~WriteTaskManager() - { - pthread_mutex_destroy(&managerLock); - pthread_cond_destroy(&waitCond); } - void WriteTaskManager::lock() { threadGroup.lock(managerLock); } - - void WriteTaskManager::unlock() { threadGroup.unlock(managerLock); } - int WriteTaskManager::nextReadyTaskIndex() { int ret = NO_MORE_TASKS; - lock(); + std::lock_guard guard(managerLock); for (int i = 0; i < tasks.size(); ++i) { if (tasks[i].state == WriteTask::Ready) @@ -174,7 +161,6 @@ namespace TwkMovie { ret = A_TASK_IS_CURRENTLY_BEING_ADDED; } - unlock(); DB("nextReadyTaskIndex " << ret); @@ -185,20 +171,19 @@ namespace TwkMovie { DB("finishTask " << index); - lock(); - - // - // Mark task complete - // - tasks[index].state = WriteTask::Finished; - - // - // Signal main thread that there's room for more tasks - // - stl_ext::thread_group::signal(waitCond); + { + std::lock_guard guard(managerLock); - unlock(); + // + // Mark task complete + // + tasks[index].state = WriteTask::Finished; + // + // Signal main thread that there's room for more tasks + // + } + waitCond.notify_one(); DB("finishTask " << index << " complete"); } @@ -258,13 +243,14 @@ namespace TwkMovie { bool addedTask = false; - lock(); - currentlyAddingATask++; - unlock(); + { + std::lock_guard guard(managerLock); + currentlyAddingATask++; + } while (!addedTask) { - lock(); + std::unique_lock lock(managerLock); for (int i = 0; i < tasks.size(); ++i) { @@ -284,8 +270,8 @@ namespace TwkMovie // Wait for worker thread to signal that there's room for more // tasks // - static const size_t waitCondTimeOutInSecs = 5; - if (!stl_ext::thread_group::wait_cond_time(waitCond, managerLock, waitCondTimeOutInSecs * 1e6)) + auto timeout = std::chrono::seconds(5); + if (waitCond.wait_for(lock, timeout) == std::cv_status::timeout) { DB("addTask waiting-timed out"); @@ -302,15 +288,14 @@ namespace TwkMovie dispatchThreads(); } } - - unlock(); } dispatchThreads(); - lock(); - currentlyAddingATask--; - unlock(); + { + std::lock_guard guard(managerLock); + currentlyAddingATask--; + } } void WriteTaskManager::dispatchThreads() From ef9862deec1f8dbd235ad727ee047579d81adc26 Mon Sep 17 00:00:00 2001 From: Mark Reid Date: Thu, 13 Nov 2025 13:39:26 +1100 Subject: [PATCH 9/9] refactor: Replace boost::thread_group with std::thread Rewrite stl_ext::thread_group to use C++ standard library threading primitives instead of raw pthreads wrappers. - Replace manual pthread_mutex/cond lock/unlock/signal/broadcast helpers with std::mutex, std::condition_variable, and std::lock_guard - Replace the complex worker-wait/signal dispatch model with a std::queue work queue that workers pull from - Simplify thread_main from a three-phase wait/jump/cleanup cycle to a straightforward dequeue-and-execute loop - Merge dispatch() and maybe_dispatch() into a single internal_dispatch() that pushes to the queue - Simplify destructor: set m_running=false, notify all, join. No more noop dispatch workaround or manual mutex/cond teardown - Replace platform-specific gettimeofday/GetSystemTimeAsFileTime with std::chrono - Remove cancel() (pthread_cancel) which was unsafe - Net reduction of ~600 lines Signed-off-by: Mark Reid --- src/lib/base/stl_ext/stl_ext/thread_group.h | 135 ++- src/lib/base/stl_ext/thread_group.cpp | 857 ++++---------------- src/lib/ip/IPCore/IPGraph.cpp | 2 +- 3 files changed, 202 insertions(+), 792 deletions(-) diff --git a/src/lib/base/stl_ext/stl_ext/thread_group.h b/src/lib/base/stl_ext/stl_ext/thread_group.h index 5ab3a75c0..2c944ec05 100644 --- a/src/lib/base/stl_ext/stl_ext/thread_group.h +++ b/src/lib/base/stl_ext/stl_ext/thread_group.h @@ -9,6 +9,10 @@ #define __stl_ext__thread_group__h__ #include #include +#include +#include +#include +#include #include namespace stl_ext @@ -92,18 +96,19 @@ namespace stl_ext bool control_wait(bool allThreads = true, double timeout_seconds = 0.0); // control thread API - // - // Call dispatch for each work thread you want to dispatch. Its an - // error to call this if there are not available threads. + // Queue a job to be executed by a worker thread asynchronously. The + // job will be placed in the work queue and executed by the next + // available worker. If all workers are busy, the job will wait + // in the queue until a worker becomes available. // - void dispatch(thread_function, void*); + void dispatch(thread_function func, void* data) { internal_dispatch(func, data, false); } // // Dispatch, but if all threads are busy returns false otherwise true // - bool maybe_dispatch(thread_function, void*, bool async = false); + bool maybe_dispatch(thread_function func, void* data) { return internal_dispatch(func, data, true); } // // Use the function to add additional threads. Once a thread is added @@ -119,40 +124,18 @@ namespace stl_ext // Number of worker threads in the thread group. // - size_t num_threads() const { return _num_threads; } + size_t num_threads() const { return m_num_threads; } // // set debugging messages on/off // - void debug(bool b) { _debugging = b; } - - static void debug_all(bool b) { _debug_all = b; } - - // - // Cancel a thread if its running. USE WITH CAUTION - // - - void cancel(pthread_t); + void debug(bool b) { m_debugging = b; } - // - // Pthreads wrappers - // - - static void lock(pthread_mutex_t&); - static void unlock(pthread_mutex_t&); - static void wait_cond(pthread_cond_t&, pthread_mutex_t&); - static bool wait_cond_time(pthread_cond_t&, pthread_mutex_t&, size_t usec); - static void signal(pthread_cond_t&); - static void broadcast(pthread_cond_t&); + static void debug_all(bool b) { DEBUG_ALL = b; } void debug(const char*, ...); - //---------------------------------------------------------------------- - // - // Worker thread API - // - // // Restart any waiting workers, using the func/data pair that // was provided to them originally via dispatch or @@ -163,66 +146,52 @@ namespace stl_ext void awaken_all_workers(); private: - static void* thread_main(void*); - static void thread_cleanup(void*); - - void worker_wait(); // worker thread API - void worker_jump(); - void worker_cleanup(); - void release_control(); - void release_worker(); - - typedef std::vector thread_vector; - - struct thread_package + struct WorkerState { - // thread_package () {} - // thread_package (thread_group* g, thread_function f, void* d) : - // group(g), func(f), data(d) {} - - thread_group* group; - thread_function func; - void* data; + size_t worker_id = 0; + thread_group* group = nullptr; + thread_function original_func = nullptr; // Function to recall + void* original_data = nullptr; // Data to recall }; - typedef std::vector package_vector; - - bool running() { return _running; } - - private: - thread_api _api; - pthread_mutex_t _wait_mutex; - pthread_cond_t _wait_cond; - - pthread_mutex_t _worker_mutex; - pthread_cond_t _worker_cond; - - pthread_mutex_t _control_mutex; - pthread_cond_t _control_cond; - - pthread_mutex_t _debug_mutex; - - pthread_attr_t _thread_attr; + struct Job + { + thread_function func; // Function to execute + void* data; // Data to pass to the function + bool recall; // If true, use the thread's original func/data + }; - bool _wait_flag; - thread_vector _threads; - package_vector _packages; + // + // Main loop for each worker thread. + // - int _num_finished; - int _num_threads; - thread_function _func; - void* _data; - bool _recall; - bool _debugging; - bool _running; - pthread_t _join_thread; + void thread_main(WorkerState* state); - bool _dispatching; - size_t _default_stack_size; - static bool _debug_all; + bool internal_dispatch(thread_function func, void* data, bool maybe); - pthread_key_t _funcKey; - pthread_key_t _dataKey; + private: + thread_api m_api; + pthread_attr_t m_thread_attr; + std::vector m_threads; + size_t m_default_stack_size = 0; + + // Debug logging + static bool DEBUG_ALL; + static std::mutex DEBUG_MUTEX; + int m_group_id; + bool m_debugging = false; + + // Work queue and synchronization + int m_num_threads = 0; + std::atomic m_num_jobs{0}; + std::atomic m_running{true}; + std::queue m_job_queue; + std::mutex m_queue_mutex; + std::condition_variable m_queue_cond; + + // For control_wait synchronization + std::mutex m_finished_mutex; + std::condition_variable m_finished_cond; }; } // namespace stl_ext diff --git a/src/lib/base/stl_ext/thread_group.cpp b/src/lib/base/stl_ext/thread_group.cpp index 761eb500c..6009174f7 100644 --- a/src/lib/base/stl_ext/thread_group.cpp +++ b/src/lib/base/stl_ext/thread_group.cpp @@ -13,805 +13,246 @@ #include #include #include +#include #include #include - -#ifdef PLATFORM_WINDOWS -#include -#include -#else -#include -#include -#endif +#include namespace stl_ext { using namespace std; - bool thread_group::_debug_all = false; - -#ifdef PLATFORM_WINDOWS - static int currentTime(struct timeval* tv) - { - union - { - // AJG - this used to be a LONG_LONG - __int64 ns100; // time since 1 Jan 1601 in 100ns units - FILETIME ft; - } now; - - GetSystemTimeAsFileTime(&now.ft); - tv->tv_usec = (long)((now.ns100 / 10LL) % 1000000LL); - tv->tv_sec = (long)((now.ns100 - 116444736000000000LL) / 10000000LL); - return (0); - } -#endif - -#if defined(PLATFORM_APPLE_MACH_BSD) - static int currentTime(struct timeval* tp) { return gettimeofday(tp, 0); } -#endif + bool thread_group::DEBUG_ALL = false; + std::mutex thread_group::DEBUG_MUTEX; thread_group::thread_group(int num_threads, int stack_mult, thread_api* api, const func_vector* funcs, const data_vector* datas) - : _num_finished(0) - , _num_threads(0) - , _data(0) - , _func(0) - , _recall(false) - , _debugging(false) - , _running(true) - , _wait_flag(true) - , _dispatching(false) - , _default_stack_size(0) - , _api() { if (api) - _api = *api; - _debugging = _debug_all; - pthread_mutex_init(&_wait_mutex, 0); - pthread_mutex_init(&_worker_mutex, 0); - pthread_mutex_init(&_control_mutex, 0); - pthread_mutex_init(&_debug_mutex, 0); + m_api = *api; - pthread_cond_init(&_wait_cond, 0); - pthread_cond_init(&_worker_cond, 0); - pthread_cond_init(&_control_cond, 0); + static std::atomic next_group_id{0}; + m_group_id = ++next_group_id; + m_debugging = DEBUG_ALL; - memset(&_thread_attr, 0, sizeof(pthread_attr_t)); - pthread_attr_init(&_thread_attr); - pthread_attr_getstacksize(&_thread_attr, &_default_stack_size); + pthread_attr_init(&m_thread_attr); + pthread_attr_getstacksize(&m_thread_attr, &m_default_stack_size); - pthread_key_create(&_funcKey, 0); - pthread_key_create(&_dataKey, 0); - - add_thread(num_threads, stack_mult, funcs, datas); + if (num_threads) + add_thread(num_threads, stack_mult, funcs, datas); } - static void noop(void*) {} - thread_group::~thread_group() { - control_wait(); - _running = false; - - debug("~thread_group: control shutting down threads"); - - while (_num_threads > 0) + m_running = false; + m_queue_cond.notify_all(); + for (auto& t : m_threads) { - debug("~thread_group: control dispatch with noop, num_threads = %d", _num_threads); - dispatch(noop, 0); - debug("~thread_group: control joining worker: %d", _num_threads); - - // void *retval = 0; - lock(_wait_mutex); - unlock(_wait_mutex); - lock(_worker_mutex); - unlock(_worker_mutex); - lock(_control_mutex); - unlock(_control_mutex); - - // - // We have to do this memcpy to a size_t type - // do guard against a zero value for _join_thread. - // - size_t threadID = 0; - memcpy(&threadID, &_join_thread, std::min(sizeof(threadID), sizeof(_join_thread))); - if (threadID) + if (int err = m_api.join(t, NULL)) { - if (int err = _api.join(_join_thread, NULL)) - { - pthread_t thread = pthread_self(); - printf("~thread_group: %p -- join: %s", thread, strerror(err)); - } - } - /* AJG - This WILL break something */ - //_join_thread = 0; - debug("~thread_group: control successfully joined with worker"); - } - - debug("~thread_group: control destroying state"); - - // - // need to wait until its ok to destroy all of this stuff. - // - debug("~thread_group: deleting keys"); - pthread_key_delete(_funcKey); - pthread_key_delete(_dataKey); - - debug("~thread_group: destroying mutexes"); - pthread_mutex_destroy(&_wait_mutex); - pthread_mutex_destroy(&_worker_mutex); - pthread_mutex_destroy(&_control_mutex); - - debug("~thread_group: destroying _wait_cond"); - pthread_cond_destroy(&_wait_cond); - debug("~thread_group: destroying _worker_cond"); - pthread_cond_destroy(&_worker_cond); - debug("~thread_group: destroying _control_cond"); - pthread_cond_destroy(&_control_cond); - - debug("~thread_group: destroying attr"); - pthread_attr_destroy(&_thread_attr); - - debug("~thread_group: destroying debug"); - pthread_mutex_destroy(&_debug_mutex); - } - - void thread_group::debug(const char* c, ...) - { - if (_debugging) - { - va_list ap; - va_start(ap, c); - lock(_debug_mutex); - pthread_t thread = pthread_self(); - int worker_num = -1; - - for (int i = 0; i < _threads.size(); i++) - { - if (pthread_equal(thread, _threads[i])) - { - worker_num = i; - break; - } + pthread_t thread = pthread_self(); + printf("~thread_group: %p -- join: %s", thread, strerror(err)); } - -#ifndef PLATFORM_LINUX - struct timeval tv; - int t = currentTime(&tv); - - fprintf(stderr, "%d %d %p/%p/%d/%lu: ", (int)tv.tv_sec, (int)tv.tv_usec, this, thread, worker_num + 1, _threads.size()); -#endif - vfprintf(stderr, c, ap); - fprintf(stderr, "\n"); - fflush(stderr); - unlock(_debug_mutex); } + pthread_attr_destroy(&m_thread_attr); + debug("~thread_group: all threads joined"); } - // - // This is the "main" function of each of the worker threads. The thread - // is not marked detached until its about to exit - // - - void thread_group::thread_cleanup(void* arg) - { - thread_group* group = (thread_group*)arg; - group->debug("worker clean up after cancellation"); - group->worker_cleanup(); - } - - void* thread_group::thread_main(void* arg) + void thread_group::debug(const char* fmt, ...) { - thread_package pack = *((thread_package*)arg); - pack.group->debug("thread_main: worker in main, thread_package: %p", arg); - -#ifdef PLATFORM_APPLE_MACH_BSD - // - // Set the initial name to thread_group to indicate that the thread - // has not yet been used. - // - - pthread_setname_np("thread_group"); -#endif - pack.group->debug("thread_main: pthread_setspecific: %p %p %p %p", pack.group->_funcKey, (void*)pack.func, pack.group->_dataKey, - pack.data); + if (!m_debugging) + return; - pthread_setspecific(pack.group->_funcKey, (void*)pack.func); - pthread_setspecific(pack.group->_dataKey, pack.data); + auto now = std::chrono::system_clock::now(); + auto seconds = std::chrono::duration_cast(now.time_since_epoch()).count(); + auto microseconds = std::chrono::duration_cast(now.time_since_epoch()).count() % 1000000; - // pthread_cleanup_push(thread_cleanup, arg); + std::lock_guard lock(DEBUG_MUTEX); - while (pack.group->running()) - { - pack.group->debug("thread_main: running"); - pack.group->worker_wait(); - pack.group->worker_jump(); - } - - // pthread_cleanup_pop(1); - - // pack.group->debug("worker returning"); - pack.group->debug("thread_main: worker returning"); - return 0; + fprintf(stderr, "%" PRId64 " %06" PRId64, static_cast(seconds), static_cast(microseconds)); + fprintf(stderr, " [thread_group %d] ", m_group_id); + va_list args; + va_start(args, fmt); + vfprintf(stderr, fmt, args); + va_end(args); + fprintf(stderr, "\n"); + fflush(stderr); } - void thread_group::add_thread(int num_to_add, size_t stack_mult, const func_vector* funcs, const data_vector* datas) + void thread_group::thread_main(WorkerState* state) { - debug("add_thread"); - if (num_to_add) + debug("worker %zu: started", state->worker_id); + while (m_running) { - debug("add_thread: resizing _threads and _packages to %d + %d, " - "_default_stack_size: %d, stack_mult: %d", - _threads.size(), num_to_add, _default_stack_size, stack_mult); - size_t s = _threads.size(); - _threads.resize(s + num_to_add); - _packages.resize(_threads.size()); - pthread_attr_setstacksize(&_thread_attr, _default_stack_size * stack_mult); - - if (funcs) - assert(funcs->size() == num_to_add); - if (funcs) + Job job; { - debug("add_thread: funcs size %d", funcs->size()); + std::unique_lock lock(m_queue_mutex); + m_queue_cond.wait(lock, [this] { return !m_job_queue.empty() || !m_running; }); + if (!m_running && m_job_queue.empty()) + break; + job = m_job_queue.front(); + m_job_queue.pop(); } - else + + // If this is a recall job, restore the original function/data for this worker + if (job.recall) { - debug("add_thread: funcs is NULL"); + job.func = state->original_func; + job.data = state->original_data; + // awaken_all_workers expects workers to have been created with + // a function; a null original_func would decrement m_num_jobs + // without doing any work, corrupting the job counter. + assert(job.func); + debug("worker %zu: recalling original func/data", state->worker_id); } - for (int i = s; i < _threads.size(); i++) - { - thread_package& pack = _packages[i]; - pack.group = this; - pack.func = (funcs) ? (*funcs)[i - s] : 0; - pack.data = (funcs) ? (*datas)[i - s] : 0; - memset(&_threads[i], 0, sizeof(pthread_t)); - if (int err = _api.create(&_threads[i], &_thread_attr, thread_main, &pack)) + // Execute the job if valid + if (job.func) + { + debug("worker %zu: running func %p with data %p", state->worker_id, (void*)job.func, job.data); + try { - printf("add_thread: Error trying to create thread %d: %d", i, err); - abort(); + job.func(job.data); + } + catch (...) + { + debug("worker %zu: caught exception", state->worker_id); } - _num_threads++; - debug("add_thread: added worker #%d", _num_threads); } - debug("add_thread: control_wait"); - control_wait(); - debug("add_thread done."); - } - } - - void thread_group::lock(pthread_mutex_t& mutex) - { - if (int err = pthread_mutex_lock(&mutex)) - { - pthread_t thread = pthread_self(); - printf("%p -- unable to lock: %s", thread, strerror(err)); - fflush(stdout); - } - } - - void thread_group::unlock(pthread_mutex_t& mutex) - { - if (int err = pthread_mutex_unlock(&mutex)) - { - pthread_t thread = pthread_self(); - printf("%p -- unable to unlock: %s", thread, strerror(err)); - fflush(stdout); - } - } - void thread_group::signal(pthread_cond_t& cond) - { - if (int err = pthread_cond_signal(&cond)) - { - pthread_t thread = pthread_self(); - printf("%p -- signal: %s", thread, strerror(err)); - fflush(stdout); - } - } - - void thread_group::broadcast(pthread_cond_t& cond) - { - if (int err = pthread_cond_broadcast(&cond)) - { - pthread_t thread = pthread_self(); - printf("%p -- broadcast: %s\n", thread, strerror(err)); - fflush(stdout); + // Even though m_num_jobs is atomic, we lock m_finished_mutex here + // to synchronize with threads waiting on m_finished_cond. + // This ensures that decrement and notification are not separated, + // avoiding missed wakeups or spurious waits. + { + std::lock_guard lock(m_finished_mutex); + m_num_jobs--; + m_finished_cond.notify_all(); + } + debug("worker %zu: job finished, jobs left: %d ", state->worker_id, m_num_jobs.load()); } - } - void thread_group::wait_cond(pthread_cond_t& cond, pthread_mutex_t& mutex) - { - if (int err = pthread_cond_wait(&cond, &mutex)) - { - pthread_t thread = pthread_self(); - printf("%p -- cond_wait: %s\n", thread, strerror(err)); - fflush(stdout); - } + debug("worker %zu: returning", state->worker_id); } - bool thread_group::wait_cond_time(pthread_cond_t& cond, pthread_mutex_t& mutex, size_t usec) + void thread_group::add_thread(int num_to_add, size_t stack_mult, const func_vector* funcs, const data_vector* datas) { - timespec ts; - -#ifdef PLATFORM_LINUX - clock_gettime(CLOCK_REALTIME, &ts); - ts.tv_sec += usec / 1000000; - size_t r = (usec % 1000000) * 1000 + ts.tv_nsec; - - if (r > 1000000000) - { - ts.tv_sec += r / 1000000000; - ts.tv_nsec = r % 1000000000; - } - -#else - struct timeval tv; - currentTime(&tv); + debug("add_thread: resizing _threads to %d + %d, default_stack_size: %d, stack_mult: %d", m_threads.size(), num_to_add, + m_default_stack_size, stack_mult); - ts.tv_sec = tv.tv_sec; - size_t us = usec + tv.tv_usec; + pthread_attr_setstacksize(&m_thread_attr, m_default_stack_size * stack_mult); - if (us > 1000000) + if (funcs) { - ts.tv_sec += us / 1000000; - ts.tv_nsec = (us % 1000000) * 1000; + assert(funcs->size() == num_to_add); + debug("add_thread: funcs size %d", funcs->size()); } else { - ts.tv_nsec = us * 1000; + debug("add_thread: funcs is NULL"); } -#endif - if (int err = pthread_cond_timedwait(&cond, &mutex, &ts)) + for (int i = 0; i < num_to_add; ++i) { - if (err != ETIMEDOUT) + pthread_t thread; + int ret; + WorkerState* state = new WorkerState{}; + state->worker_id = m_threads.size(); + state->original_func = (funcs && funcs->size() > i) ? (*funcs)[i] : nullptr; + state->original_data = (datas && datas->size() > i) ? (*datas)[i] : nullptr; + state->group = this; + + ret = m_api.create( + &thread, &m_thread_attr, + [](void* arg) -> void* + { + WorkerState* state = static_cast(arg); + state->group->thread_main(state); + delete state; + return nullptr; + }, + state); + + if (ret != 0) { - pthread_t thread = pthread_self(); - printf("ERROR: %p -- cond_wait_timed: %s\n", thread, strerror(err)); - fflush(stdout); + printf("add_thread: Error trying to create thread %d: %d", i, ret); + abort(); } - // fprintf (stderr, "err %d %s\n", err, strerror(err)); - return false; + m_threads.push_back(thread); + m_num_threads++; } - - return true; - } - - void thread_group::release_control() - { - lock(_wait_mutex); - // debug("worker releasing control thread"); - debug("release_control: _wait_mutex locked, signaling _wait_cond"); - signal(_wait_cond); - unlock(_wait_mutex); - debug("release_control: _wait_mutex unlocked"); - } - - void thread_group::release_worker() - { - lock(_worker_mutex); - // debug("control releasing a worker thread"); - debug("release_worker: _worker_mutex locked, signaling _worker_cond"); - _wait_flag = false; - signal(_worker_cond); - unlock(_worker_mutex); - debug("release_worker: _worker_mutex unlocked"); + debug("add_thread: added %d threads, total now %d", num_to_add, m_num_threads); } bool thread_group::control_wait(bool allThreads, double seconds) { - bool timedout = false; - - if (_num_finished != _num_threads) - { - lock(_wait_mutex); - debug("control_wait: _wait_mutex locked"); - - // - // POSIX says that you can get a spurious wake up from a - // condition variable. We need to keep checking in that case. - // + if (m_num_threads <= 0) + return true; - int num_finished = _num_finished; + std::unique_lock lock(m_finished_mutex); + int current_num_jobs = m_num_jobs.load(); + // Predicate: all jobs finished (if allThreads), or any job finished + auto predicate = [this, allThreads, current_num_jobs]() -> bool + { if (allThreads) - { - while (_num_finished != _num_threads) - { - debug("control_wait: control waiting on %d threads, " - "timeout %g seconds", - (_num_threads - _num_finished), seconds); - - if (seconds) - { - timedout = !wait_cond_time(_wait_cond, _wait_mutex, size_t(seconds * 1000000.0)); - if (timedout) - { - debug("control_wait: control wait timed out, %d " - "threads remaining, _wait_cond and _wait_mutex", - (_num_threads - _num_finished)); - break; - } - } - else - { - wait_cond(_wait_cond, _wait_mutex); - debug("control_wait: control wait_cond returned, " - "_wait_cond and _wait_mutex"); - } - } - } + return m_num_jobs <= 0; else - { - while (_num_finished == num_finished) - { - debug("control_wait: control waiting on any thread, " - "timeout %g seconds, _wait_cond and _wait_mutex", - seconds); + return m_num_jobs != current_num_jobs || m_num_jobs <= 0; + }; - if (seconds) - { - timedout = !wait_cond_time(_wait_cond, _wait_mutex, size_t(seconds * 1000000.0)); - if (timedout) - { - debug("control_wait: control wait on any thread " - "timed out"); - break; - } - } - else - { - wait_cond(_wait_cond, _wait_mutex); - } - } - } - - unlock(_wait_mutex); - debug("control_wait: control done waiting, _wait_mutex unlocked"); - } - else + // Wait with timeout if specified, otherwise wait indefinitely + if (seconds > 0.0) { - debug("control did not need to wait"); - } - - return !timedout; - } - - void thread_group::cancel(pthread_t th) - { - lock(_worker_mutex); - int f = _num_finished; - unlock(_worker_mutex); - if (f != _num_threads) - pthread_cancel(th); - } - - void thread_group::worker_wait() - { - lock(_worker_mutex); - debug("worker_wait: _worker_mutex is locked"); - ++_num_finished; - - if (_num_finished == _num_threads) - { - debug("worker_wait: workers are all waiting"); - release_control(); - } - else - { - if (_num_finished < 0 || _num_threads < 0) + auto timeout = std::chrono::steady_clock::now() + std::chrono::duration(seconds); + while (!predicate()) { - _wait_flag = true; - unlock(_worker_mutex); - debug("worker_wait: ERROR: _num_finished %d _num_threads %d", _num_finished, _num_threads); - return; - } - debug("worker_wait: %d workers left", _num_threads - _num_finished); - } - - while (_wait_flag) - { - debug("worker_wait: worker waiting on _worker_cond with " - "_worker_mutex"); - wait_cond(_worker_cond, _worker_mutex); - debug("worker_wait: worker awoke"); - } - - _wait_flag = true; - _num_finished--; - unlock(_worker_mutex); - - debug("worker_wait: worker done waiting, _worker_mutex unlocked"); - } - - void thread_group::worker_jump() - { - if (running()) - { - lock(_control_mutex); - debug("worker_jump: worker loading jump point, recall %d", _recall); - - if (_recall) - { - _recall = false; - - thread_function savedFunc = (thread_function)pthread_getspecific(_funcKey); - void* savedData = pthread_getspecific(_dataKey); - - if (savedFunc) - { - // We want to just awaken this thread, using the func - // and data that were originally given it by dispatch. - // So recall them from thread-specific data. - // - _func = savedFunc; - _data = savedData; - debug("worker_jump: re-using data %x", _data); - } - else + if (m_finished_cond.wait_until(lock, timeout) == std::cv_status::timeout) { - // An attempt was made to awaken this thread before - // it was run the first time, so go back to sleep. - // - debug("worker_jump: worker returning to wait"); - signal(_control_cond); - unlock(_control_mutex); - return; + debug("control_wait: timed out waiting for %s job(s)", allThreads ? "all" : "any"); + return false; } } - else - { - // This is the first time we've dispatched this thread, so - // store func/data for future awakenings. - // - pthread_setspecific(_funcKey, (void*)_func); - pthread_setspecific(_dataKey, _data); - debug("worker_jump: setting data %x", _data); - } - - thread_function func = _func; - void* data = _data; - // assert(_dispatching); - assert(_func); - _func = 0; - _data = 0; - - debug("worker_jump: worker signaling control"); - signal(_control_cond); - unlock(_control_mutex); - - debug("worker_jump: worker jumping"); - - try - { - (*func)(data); - } - catch (...) - { - debug("worker caught C++ exception"); - } - - debug("worker_jump: worker finished, returning to wait"); } else { - lock(_control_mutex); - _num_threads--; - _func = 0; - _join_thread = pthread_self(); - debug("worker_jump: worker decremented, num_threads is now %d. " - "_control_mutex locked", - _num_threads); - debug("worker_jump: worker signaling _control_cond"); - signal(_control_cond); - unlock(_control_mutex); - debug("worker_jump: worker exiting, _control_mutex unlocked"); + m_finished_cond.wait(lock, predicate); } + return true; } - void thread_group::worker_cleanup() - { - lock(_worker_mutex); - _wait_flag = false; - unlock(_worker_mutex); - worker_wait(); - - lock(_control_mutex); - _num_threads--; - _join_thread = pthread_self(); - debug("worker_cleanup: worker decremented, num_threads is now %d, " - "_control_mutex locked", - _num_threads); - debug("worker_cleanup: worker signaling _control_cond"); - signal(_control_cond); - unlock(_control_mutex); - debug("worker_cleanup: worker finishing, _control_mutex unlocked"); - } - - void thread_group::dispatch(thread_group::thread_function func, void* data) - { - debug("dispatch: control dispatching worker, func %p", func); - - if (_num_finished == 0 && running()) - { - debug("dispatch: control: thread_group: can't dispatch: no idle " - "workers"); - // - // There's nothing to do, no need to abort() ... - // abort(); - // - return; - } - - lock(_control_mutex); - - _dispatching = true; - assert(_func == 0); - - if (!func) - { - // - // This will tell the worker to recall it's original func/data. - // - debug("dispatch: worker recall original stuff %p %p", _func, _data); - _recall = true; - - // We need to set _func to something, so that the wait - // loop below functions as intended. - // - _func = (thread_function)0xdeadc0de; - _data = 0; - } - else - { - _recall = false; - _func = func; - _data = data; - } - release_worker(); - debug("dispatch: control waiting for worker to load"); - - while (_func) - { - wait_cond(_control_cond, _control_mutex); - } - - debug("dispatch: control done waiting for worker to dispatch"); - _dispatching = false; - unlock(_control_mutex); - } - - bool thread_group::maybe_dispatch(thread_group::thread_function func, void* data, bool async) + bool thread_group::internal_dispatch(thread_function func, void* data, bool maybe) { - bool ret = false; - debug("control maybe dispatching worker num_finished %d", _num_finished); - - lock(_control_mutex); - - lock(_worker_mutex); - bool workersWaiting = (0 != _num_finished); - unlock(_worker_mutex); - - if (workersWaiting && _func == 0) + bool dispatched = false; { - _dispatching = true; - if (!func) - { - // - // This will tell the worker to recall it's original func/data. - // - _recall = true; - - // We need to set _func to something, so that the wait - // loop below functions as intended. - // - _func = (thread_function)0xdeadc0de; - _data = 0; - } - else + std::lock_guard lock(m_queue_mutex); + if (!maybe || m_num_jobs < m_num_threads) { - _recall = false; - _func = func; - _data = data; + m_job_queue.push(Job{func, data, func ? false : true}); + m_num_jobs++; + dispatched = true; } - release_worker(); - debug("control waiting for worker to load"); - - while (!async && _func) - { - wait_cond(_control_cond, _control_mutex); - } - - debug("control done waiting for worker to dispatch"); - _dispatching = false; - ret = true; + } + if (dispatched) + { + debug("internal_dispatch: job dispatched"); + m_queue_cond.notify_one(); } - unlock(_control_mutex); - return ret; + return dispatched; } void thread_group::awaken_all_workers() { - debug("control awakening workers"); - - // - // We'll check again in loop below with proper locking, but can bail - // now if all threads are busy. - // - if (_num_finished == 0) - return; - - // - // In theory, we could sit here forever, awakening worker - // threads that immediately sleep again. So don't try more - // times than we have threads. - // - int attemptCount = _num_threads; - bool workersWaiting; - - do { - //////////////////////////////////////////////////////// - // - // Deadlock danger. Note that if someone else locks - // these mutexs in the opposite order, we have a - // possible deadlock. - // - // But we (and dispatch and redispatch) are already - // two-level locking in this order, since - // release_worker will lock _worker_mutex while - // _control_mutex is locked (see below). - // - // Further, if we don't do this, and release - // _worker_mutex before we lock _control_mutex, there - // is a danger that _num_finished will drop to 0 after - // we release _worker_mutex, so there will be no - // workers in worker_wait when we call wait_cond below, - // in which case it will block until a worker enters - // worker_wait, which could take forever. - // - //////////////////////////////////////////////////////// - - lock(_control_mutex); - - lock(_worker_mutex); - workersWaiting = (0 != _num_finished); - unlock(_worker_mutex); - - if (workersWaiting) + std::lock_guard lock(m_queue_mutex); + while (m_num_jobs < m_num_threads) { - if (_func != 0) - { - unlock(_control_mutex); - return; - } - _dispatching = true; - - // This will tell the worker to recall it's original func/data. - // - _recall = true; - - // We need to set _func to something, so that the wait - // loop below functions as intended. - // - _func = (thread_function)0xdeadc0de; - - release_worker(); - - debug("control waiting for awakened worker to load"); - while (_func) - { - wait_cond(_control_cond, _control_mutex); - } - debug("control done waiting for worker to reawaken"); - - _dispatching = false; + m_job_queue.push(Job{nullptr, nullptr, true}); + m_num_jobs++; } - unlock(_control_mutex); - } while (workersWaiting && --attemptCount); + m_queue_cond.notify_all(); + } + debug("awaken_all_workers: recall jobs enqueued"); } - } // namespace stl_ext diff --git a/src/lib/ip/IPCore/IPGraph.cpp b/src/lib/ip/IPCore/IPGraph.cpp index 9998e2d94..b42e1305b 100644 --- a/src/lib/ip/IPCore/IPGraph.cpp +++ b/src/lib/ip/IPCore/IPGraph.cpp @@ -2741,7 +2741,7 @@ IPGraph::findNodesByAbstractPath(int frame, if (ok) { - m_audioThreadGroup.maybe_dispatch(evalAudioThreadTrampoline, this, true); + m_audioThreadGroup.maybe_dispatch(evalAudioThreadTrampoline, this); } }