Fixed race conditions bug in streamers

smistad · smistad · commit 65beae099018 · 2025-08-12T12:38:59.000+02:00
diff --git a/.github/workflows/CI-ubuntu.yml b/.github/workflows/CI-ubuntu.yml
@@ -263,7 +263,7 @@ jobs:
     steps:
       - name: Install dependencies
         run: |
-          sudo apt install libopenslide0 libpocl2t64 xvfb libgl1 libopengl0 libusb-1.0-0 libxcb-xinerama0
+          sudo apt install libopenslide0 libpocl2t64 xvfb libgl1 libopengl0 libusb-1.0-0 libxcb-xinerama0 libxcb-icccm4
       - name: Download artifacts
         uses: actions/download-artifact@v4
         with:
diff --git a/source/FAST/Algorithms/ImagePatch/ImageToBatchGenerator.cpp b/source/FAST/Algorithms/ImagePatch/ImageToBatchGenerator.cpp
@@ -32,7 +32,6 @@ void ImageToBatchGenerator::generateStream() {
     bool firstTime = true;
     while(!lastFrame) {
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
diff --git a/source/FAST/Algorithms/ImagePatch/PatchGenerator.cpp b/source/FAST/Algorithms/ImagePatch/PatchGenerator.cpp
@@ -202,16 +202,13 @@ void PatchGenerator::generateStream() {
                             frameAdded();
                         }
                     } catch(ThreadStopped &e) {
-                        std::unique_lock<std::mutex> lock(m_stopMutex);
                         m_stop = true;
                         break;
                     }
                     previousPatch = patch;
-                    std::unique_lock<std::mutex> lock(m_stopMutex);
                     if(m_stop)
                         break;
                 }
-                std::unique_lock<std::mutex> lock(m_stopMutex);
                 if(m_stop) {
                     //m_streamIsStarted = false;
                     m_firstFrameIsInserted = false;
@@ -293,13 +290,11 @@ void PatchGenerator::generateStream() {
                                 //std::this_thread::sleep_for(std::chrono::seconds(2));
                             }
                         } catch(ThreadStopped &e) {
-                            std::unique_lock<std::mutex> lock(m_stopMutex);
                             m_stop = true;
                             break;
                         }
                         mRuntimeManager->stopRegularTimer("create patch");
                         previousPatch = patch;
-                        std::unique_lock<std::mutex> lock(m_stopMutex);
                         if(m_stop)
                             break;
                     }}}
diff --git a/source/FAST/Algorithms/Interleave/Interleave.cpp b/source/FAST/Algorithms/Interleave/Interleave.cpp
@@ -31,7 +31,6 @@ void Interleave::generateStream() {
     auto previousTime = std::chrono::high_resolution_clock::now();
     while(!lastFrame) {
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
@@ -129,7 +128,6 @@ void InterleavePlayback::generateStream() {
         pause = getPause();
 
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
diff --git a/source/FAST/Algorithms/TissueMicroArrayExtractor/TissueMicroArrayExtractor.cpp b/source/FAST/Algorithms/TissueMicroArrayExtractor/TissueMicroArrayExtractor.cpp
@@ -113,7 +113,6 @@ void TissueMicroArrayExtractor::generateStream() {
                 frameAdded();
             }
         } catch(ThreadStopped &e) {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             m_stop = true;
             break;
         }
diff --git a/source/FAST/ProcessObject.cpp b/source/FAST/ProcessObject.cpp
@@ -195,11 +195,10 @@ void ProcessObject::addOutputData(uint portID, DataObject::pointer data, bool pr
 
     // Add it to all output connections, if any connections exist
     if(mOutputConnections.count(portID) > 0) {
-        for(auto output : mOutputConnections.at(portID)) {
-            if(!output.expired()) {
-                DataChannel::pointer port = output.lock();
+        for(const auto& output : mOutputConnections.at(portID)) {
+            auto port = output.lock();
+            if(port)
                 port->addFrame(data);
-            }
         }
     }
 }
diff --git a/source/FAST/Python/Tests/test_data_stream_object.py b/source/FAST/Python/Tests/test_data_stream_object.py
@@ -16,16 +16,16 @@ def test_data_stream_empty_generator():
     assert counter == 0
 
 
-# def test_data_stream_empty_file_streamer():
-#     streamer = fast.ImageFileStreamer \
-#         .create(fast.Config.getTestDataPath() + "/US/Heart/ApicalFourChamber/US-2D_#.mhd",)
-#     streamer.setStartNumber(200) # Incorrect start number, will result in no frames
-#
-#     counter = 0
-#     with pytest.raises(RuntimeError):
-#         for image in fast.DataStream(streamer):
-#             counter += 1
-#     assert counter == 0
+def test_data_stream_empty_file_streamer():
+    streamer = fast.ImageFileStreamer \
+        .create(fast.Config.getTestDataPath() + "/US/Heart/ApicalFourChamber/US-2D_#.mhd",)
+    streamer.setStartNumber(200) # Incorrect start number, will result in no frames
+
+    counter = 0
+    with pytest.raises(RuntimeError):
+        for image in fast.DataStream(streamer):
+            counter += 1
+    assert counter == 0
 
 
 def test_data_stream_single():
diff --git a/source/FAST/Streamers/DicomMultiFrameStreamer.cpp b/source/FAST/Streamers/DicomMultiFrameStreamer.cpp
@@ -112,7 +112,6 @@ void DicomMultiFrameStreamer::generateStream() {
         pause = getPause();
 
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
diff --git a/source/FAST/Streamers/FileStreamer.cpp b/source/FAST/Streamers/FileStreamer.cpp
@@ -102,6 +102,7 @@ void FileStreamer::setFilenameFormats(std::vector<std::string> strs) {
 void FileStreamer::generateStream() {
     if(getNrOfFrames() == 0) {
         stopWithError("No frames were created by the FileStreamer");
+        return;
     }
 
     // Read timestamp file if available
@@ -140,7 +141,6 @@ void FileStreamer::generateStream() {
         pause = getPause();
 
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
@@ -153,6 +153,7 @@ void FileStreamer::generateStream() {
         std::string filename = getFilename(i, currentSequence);
         try {
             reportInfo() << "Filestreamer reading " << filename << reportEnd();
+            // If this causes pure virtual method called error, it means derived class forgot to call stop() in destructor
             DataObject::pointer dataFrame = getDataFrame(filename);
 
             // Timing
diff --git a/source/FAST/Streamers/ImageFileStreamer.hpp b/source/FAST/Streamers/ImageFileStreamer.hpp
@@ -69,6 +69,7 @@ class FAST_EXPORT ImageFileStreamer : public FileStreamer {
          */
         void setGrayscale(bool grayscale);
         void loadAttributes() override;
+        ~ImageFileStreamer() { stop(); }; // FileStreamer derived classes must call stop() in destructor or you will get a pure virtual method called error
     protected:
         DataObject::pointer getDataFrame(std::string filename) override;
 
diff --git a/source/FAST/Streamers/ManualImageStreamer.cpp b/source/FAST/Streamers/ManualImageStreamer.cpp
@@ -60,7 +60,6 @@ void ManualImageStreamer::generateStream() {
     int currentSequence = 0;
     while(true) {
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
diff --git a/source/FAST/Streamers/MeshFileStreamer.hpp b/source/FAST/Streamers/MeshFileStreamer.hpp
@@ -28,6 +28,7 @@ class FAST_EXPORT  MeshFileStreamer : public FileStreamer {
                          std::string, filenameFormat,,
                          bool, loop, = false
         );
+        ~MeshFileStreamer() { stop(); }; // FileStreamer derived classes must call stop() in destructor or you will get a pure virtual method called error
     protected:
         DataObject::pointer getDataFrame(std::string filename);
         MeshFileStreamer();
diff --git a/source/FAST/Streamers/OpenIGTLinkStreamer.cpp b/source/FAST/Streamers/OpenIGTLinkStreamer.cpp
@@ -139,8 +139,11 @@ void OpenIGTLinkStreamer::updateFirstFrameSetFlag() {
     bool allHaveGotData = true;
     for(auto port : mOutputConnections) {
         for(auto output : port.second) {
-            if(output.lock()->getSize() == 0)
-                allHaveGotData = false;
+            auto channel = output.lock();
+            if(channel) {
+                if(channel->getSize() == 0)
+                    allHaveGotData = false;
+            }
         }
     }
 
@@ -186,7 +189,6 @@ void OpenIGTLinkStreamer::generateStream() {
 
     while(true) {
         {
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
diff --git a/source/FAST/Streamers/RealSenseStreamer.cpp b/source/FAST/Streamers/RealSenseStreamer.cpp
@@ -147,7 +147,6 @@ void RealSenseStreamer::generateStream() {
     while(true) {
         {
             // Check if stop signal is sent
-            std::unique_lock<std::mutex> lock(m_stopMutex);
             if(m_stop) {
                 m_streamIsStarted = false;
                 m_firstFrameIsInserted = false;
diff --git a/source/FAST/Streamers/Streamer.cpp b/source/FAST/Streamers/Streamer.cpp
@@ -34,18 +34,37 @@ Streamer::Streamer() {
 }
 
 void Streamer::stop() {
+    std::lock_guard<std::mutex> lock(m_stopMutex);
+    if(!m_thread) // Already stopped
+        return;
     reportInfo() << "Stopping in streamer.." << reportEnd();
-    {
-        std::unique_lock<std::mutex> lock(m_stopMutex);
-        m_stop = true;
-        m_streamIsStarted = false;
-        m_firstFrameIsInserted = false;
+    m_stop = true;
+    // Thread might be locked due th QueueDataChannel, we have to signal them to stop:
+    for(const auto& output : mOutputConnections) {
+        for(const auto& channel : output.second) {
+            auto channelLocked = channel.lock();
+            if(channelLocked) { // Make sure it is still valid, before calling stop
+                channelLocked->stop();
+            }
+        }
     }
-    if(m_thread) {
+    reportInfo() << "frameadd.." << reportEnd();
+    frameAdded(); // Unblock in execute, if needed
+    reportInfo() << "done" << reportEnd();
+    // Join thread and reset
+    if(std::this_thread::get_id() != m_thread->get_id()) {
+        // join can't be called from the streaming thread
+        reportInfo() << "join" << reportEnd();
         m_thread->join();
-        m_thread = nullptr;
-        reportInfo() << "Streamer thread for " << getNameOfClass() << " returned" << reportEnd();
+        reportInfo() << "reset" << reportEnd();
+        m_thread.reset(); // This will destroy existing thread object, and will crash program if it happens inside thread
+        reportInfo() << "reset done" << reportEnd();
+    } else {
+        m_thread->detach(); // Test
     }
+    m_streamIsStarted = false;
+    m_firstFrameIsInserted = false;
+    reportInfo() << "Streamer::stop() done." << reportEnd();
 }
 
 void Streamer::setMaximumNrOfFrames(int frames) {
@@ -86,11 +105,10 @@ DataChannel::pointer Streamer::getOutputPort(uint portID) {
 }
 
 bool Streamer::isStopped() {
-    std::lock_guard<std::mutex> lock(m_stopMutex);
     return m_stop;
 }
 
-Streamer::~Streamer() noexcept {
+Streamer::~Streamer() {
     reportInfo() << "Destroying streamer.." << reportEnd();
     stop();
     reportInfo() << "Streamer DESTROYED." << reportEnd();
@@ -101,7 +119,9 @@ void Streamer::stopWithError(std::string message, int outputPort) {
     if(outputPort < 0)
         outputPort = 0;
     for(const auto& channel : mOutputConnections[outputPort]) {
-        channel.lock()->stop();
+        auto channelLocked = channel.lock();
+        if(channelLocked) // Make sure it is still valid, before calling stop
+            channelLocked->stop();
     }
 }
 
diff --git a/source/FAST/Streamers/Streamer.hpp b/source/FAST/Streamers/Streamer.hpp
@@ -44,7 +44,10 @@ class FAST_EXPORT Streamer : public ProcessObject {
         }
 
         /**
-         * Stop the stream
+         * @brief Stop the stream
+         *
+         * The streaming thread has finished when this method returns.
+         * This method should not be called from the streaming thread.
          */
         virtual void stop();
 
@@ -81,17 +84,19 @@ class FAST_EXPORT Streamer : public ProcessObject {
 
         bool m_firstFrameIsInserted = false;
         bool m_streamIsStarted = false;
-        bool m_stop = false;
         StreamingMode m_streamingMode = StreamingMode::ProcessAllFrames;
 
         std::mutex m_firstFrameMutex;
-        std::mutex m_stopMutex;
         std::unique_ptr<std::thread> m_thread;
         std::condition_variable m_firstFrameCondition;
 
         // Must be weak_ptr or be we get a circular dependency and streamers are never destroyed here
         std::map<uint, std::weak_ptr<ProcessObject>> m_outputPOs;
 
+        std::atomic_bool m_stop = false;
+    private:
+        std::mutex m_stopMutex; // Used to ensure stop() is only called by one thread at a time
+
 
 };
 
diff --git a/source/FAST/Streamers/UFFStreamer.cpp b/source/FAST/Streamers/UFFStreamer.cpp
@@ -435,7 +435,6 @@ void UFFStreamer::generateStream() {
         pause = getPause();
 
         {
-            std::lock_guard<std::mutex> lock(m_stopMutex);
             if(m_stop) break;
         }
 
diff --git a/source/FAST/Tests/DummyObjects.cpp b/source/FAST/Tests/DummyObjects.cpp
@@ -30,7 +30,6 @@ void DummyStreamer::generateStream() {
     std::cout << "DummyStreamer thread started" << std::endl;
     for(int i = 0; i < mFramesToGenerate; ++i) {
         {
-            std::lock_guard<std::mutex> lock(m_stopMutex);
             if(m_stop)
                 break;
         }

Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,6 @@ void ImageToBatchGenerator::generateStream() {`
`32`	`32`	`bool firstTime = true;`
`33`	`33`	`while(!lastFrame) {`
`34`	`34`	`{`
`35`		`- std::unique_lock<std::mutex> lock(m_stopMutex);`
`36`	`35`	`if(m_stop) {`
`37`	`36`	`m_streamIsStarted = false;`
`38`	`37`	`m_firstFrameIsInserted = false;`
Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,6 @@ void Interleave::generateStream() {`
`31`	`31`	`auto previousTime = std::chrono::high_resolution_clock::now();`
`32`	`32`	`while(!lastFrame) {`
`33`	`33`	`{`
`34`		`- std::unique_lock<std::mutex> lock(m_stopMutex);`
`35`	`34`	`if(m_stop) {`
`36`	`35`	`m_streamIsStarted = false;`
`37`	`36`	`m_firstFrameIsInserted = false;`
`@@ -129,7 +128,6 @@ void InterleavePlayback::generateStream() {`
`129`	`128`	`pause = getPause();`
`130`	`129`
`131`	`130`	`{`
`132`		`- std::unique_lock<std::mutex> lock(m_stopMutex);`
`133`	`131`	`if(m_stop) {`
`134`	`132`	`m_streamIsStarted = false;`
`135`	`133`	`m_firstFrameIsInserted = false;`
Original file line number	Diff line number	Diff line change
`@@ -113,7 +113,6 @@ void TissueMicroArrayExtractor::generateStream() {`
`113`	`113`	`frameAdded();`
`114`	`114`	`}`
`115`	`115`	`} catch(ThreadStopped &e) {`
`116`		`- std::unique_lock<std::mutex> lock(m_stopMutex);`
`117`	`116`	`m_stop = true;`
`118`	`117`	`break;`
`119`	`118`	`}`
Original file line number	Diff line number	Diff line change
`@@ -195,11 +195,10 @@ void ProcessObject::addOutputData(uint portID, DataObject::pointer data, bool pr`
`195`	`195`
`196`	`196`	`// Add it to all output connections, if any connections exist`
`197`	`197`	`if(mOutputConnections.count(portID) > 0) {`
`198`		`- for(auto output : mOutputConnections.at(portID)) {`
`199`		`- if(!output.expired()) {`
`200`		`- DataChannel::pointer port = output.lock();`
	`198`	`+ for(const auto& output : mOutputConnections.at(portID)) {`
	`199`	`+ auto port = output.lock();`
	`200`	`+ if(port)`
`201`	`201`	`port->addFrame(data);`
`202`		`- }`
`203`	`202`	`}`
`204`	`203`	`}`
`205`	`204`	`}`
Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,6 @@ void DicomMultiFrameStreamer::generateStream() {`
`112`	`112`	`pause = getPause();`
`113`	`113`
`114`	`114`	`{`
`115`		`- std::unique_lock<std::mutex> lock(m_stopMutex);`
`116`	`115`	`if(m_stop) {`
`117`	`116`	`m_streamIsStarted = false;`
`118`	`117`	`m_firstFrameIsInserted = false;`