Fixed bug with streamers never being destroyed

Author: smistad

CMakeLists.txt

@@ -7,7 +7,7 @@ project(FAST)
 
 set(VERSION_MAJOR 4)
 set(VERSION_MINOR 11)
-set(VERSION_PATCH 0)
+set(VERSION_PATCH 1)
 set(VERSION_SO 4)   # SO version, should be incremented by 1 every time the existing API changes
 set(FAST_VERSION "${VERSION_MAJOR}.${VERSION_MINOR}.${VERSION_PATCH}")
 

source/FAST/ProcessObject.cpp

@@ -364,9 +364,6 @@ cl::Program ProcessObject::getOpenCLProgram(
     return program->build(device, buildOptions);
 }
 
-ProcessObject::~ProcessObject() {
-}
-
 int ProcessObject::getNrOfInputConnections() const {
     return mInputConnections.size();
 }

source/FAST/ProcessObject.hpp

@@ -65,7 +65,6 @@ class ProcessObject;
  */
 class FAST_EXPORT  ProcessObject : public AttributeObject {
     public:
-        virtual ~ProcessObject();
         /**
          * @brief Update/Run the pipeline up to this process object.
          *
@@ -128,7 +127,7 @@ class FAST_EXPORT  ProcessObject : public AttributeObject {
         /**
          * @brief Stop a pipeline.
          */
-        void stopPipeline();
+        virtual void stopPipeline();
 
         /**
          * @brief Mark this process object as modified or not.

source/FAST/Streamers/Streamer.cpp

@@ -61,20 +61,44 @@ void Streamer::setStreamingMode(StreamingMode mode) {
 }
 
 DataChannel::pointer Streamer::getOutputPort(uint portID) {
-    if(m_outputPOs.count(portID) == 0) {
+    if(m_outputPOs.count(portID) == 0) { // Doesn't exist
         auto channel = ProcessObject::getOutputPort(portID);
-        m_outputPOs[portID] = RunLambda::create([](DataObject::pointer data) -> DataList {
+        auto PO = RunLambda::create([](DataObject::pointer data) -> DataList {
             return DataList(data);
         });
-        m_outputPOs[portID]->setInputConnection(channel);
-    }
+        PO->setInputConnection(channel);
+        m_outputPOs[portID] = PO;
+        return PO->getOutputPort();
+    } else {
+        auto PO = m_outputPOs[portID].lock();
+        if(!PO) { // Expired, recreate
+            // TODO duplicate code:
+            auto channel = ProcessObject::getOutputPort(portID);
+            auto PO = RunLambda::create([](DataObject::pointer data) -> DataList {
+                return DataList(data);
+            });
+            PO->setInputConnection(channel);
+            m_outputPOs[portID] = PO;
+        }
 
-    return m_outputPOs[portID]->getOutputPort();
+        return PO->getOutputPort();
+    }
 }
 
 bool Streamer::isStopped() {
     std::lock_guard<std::mutex> lock(m_stopMutex);
     return m_stop;
 }
 
+Streamer::~Streamer() noexcept {
+    reportInfo() << "Destroying streamer.." << reportEnd();
+    stop();
+    reportInfo() << "Streamer DESTROYED." << reportEnd();
+}
+
+void Streamer::stopPipeline() {
+    stop();
+    ProcessObject::stopPipeline();
+}
+
 }

source/FAST/Streamers/Streamer.hpp

@@ -35,7 +35,7 @@ class FAST_EXPORT Streamer : public ProcessObject {
     public:
         typedef std::shared_ptr<Streamer> pointer;
         Streamer();
-        virtual ~Streamer() { stop(); };
+        virtual ~Streamer();
         static std::string getStaticNameOfClass() {
             return "Streamer";
         }
@@ -56,6 +56,7 @@ class FAST_EXPORT Streamer : public ProcessObject {
         StreamingMode getStreamingMode() const;
 
         virtual DataChannel::pointer getOutputPort(uint portID = 0) override;
+        void stopPipeline() override;
     protected:
         /**
          * Block until the first data frame has been sent using a condition variable
@@ -87,7 +88,8 @@ class FAST_EXPORT Streamer : public ProcessObject {
         std::unique_ptr<std::thread> m_thread;
         std::condition_variable m_firstFrameCondition;
 
-        std::map<uint, std::shared_ptr<ProcessObject>> m_outputPOs;
+        // 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;
 
 
 };