Created the FramerateSynchronizer object

Author: smistad

source/FAST/CMakeLists.txt

@@ -31,11 +31,12 @@ fast_add_sources(
     Attribute.cpp
     Attribute.hpp
     ProcessObjectRegistry.hpp
-    PipelineSynchronizer.cpp
-    PipelineSynchronizer.hpp
+    FramerateSynchronizer.cpp
+    FramerateSynchronizer.hpp
     DataStream.cpp
     DataStream.hpp
 )
+fast_add_process_object(FramerateSynchronizer FramerateSynchronizer.hpp)
 if(FAST_MODULE_Visualization)
     fast_add_sources(
         Pipeline.hpp

source/FAST/FramerateSynchronizer.cpp

@@ -0,0 +1,100 @@
+#include "FramerateSynchronizer.hpp"
+
+namespace fast {
+
+FramerateSynchronizer::FramerateSynchronizer(int priorityPort) {
+    m_priorityPort = priorityPort;
+}
+
+/*
+uint FramerateSynchronizer::addInputConnection(DataChannel::pointer channel) {
+    uint nr = getNrOfInputConnections();
+    createInputPort<DataObject>(nr, false);
+    createOutputPort<DataObject>(nr);
+    setInputConnection(nr, channel);
+    return nr;
+}*/
+
+void FramerateSynchronizer::execute() {
+    if(!mInputConnections.empty()) {
+        // First time
+        m_parents = mInputConnections;
+        mInputConnections.clear(); // Severe the connections
+
+        // Start one thread per input
+        for(const auto& parent : m_parents) {
+            uint port = parent.first;
+            DataChannel::pointer channel = parent.second;
+            //channel->setMaximumNumberOfFrames(1);
+            auto thread = new std::thread([port, channel, this]() {
+                //std::cout << "Started thread" << std::endl;
+                try {
+                    while(true) {
+                        channel->getProcessObject()->run();
+                        DataObject::pointer data = channel->getNextFrame();
+                        //std::cout << "Adding data to port " << port << std::endl;
+                        {
+                            std::lock_guard<std::mutex> lock(m_latestDataMutex);
+                            m_latestData[port] = data;
+                            m_newData = true;
+                        }
+                        m_dataCV.notify_one();
+                        if(data->isLastFrame())
+                            break;
+                    }
+                } catch(ThreadStopped &e) {
+                }
+                //std::cout << "THREAD STOPPED" << std::endl;
+            });
+            m_threads.push_back(thread);
+        }
+    }
+
+    std::unique_lock<std::mutex> lock(m_latestDataMutex);
+    if(m_priorityPort >= 0) {
+        // Wait until priority port has new data
+        while(m_latestData.size() != m_parents.size() || m_latestData.count(m_priorityPort) == 0) {
+            //std::cout << "Waiting for new data on priority port" << std::endl;
+            m_dataCV.wait(lock);
+        }
+    } else {
+        while(!m_newData || m_latestData.size() != m_parents.size()) {
+            //std::cout << "Waiting for new data" << std::endl;
+            m_dataCV.wait(lock);
+        }
+    }
+    auto dataToAdd = m_latestData;
+    m_newData = false;
+    lock.unlock();
+    //std::cout << "Got data on priority port" << std::endl;
+    for(const auto& parent : m_parents) {
+        //std::cout << "adding output data to " << parent.first << ": " << dataToAdd[parent.first] << std::endl;
+        addOutputData(parent.first, dataToAdd[parent.first]);
+    }
+    if(m_priorityPort >= 0) {
+        lock.lock();
+        m_latestData.erase(m_priorityPort);
+        lock.unlock();
+    }
+    //std::cout << "Done adding data." << std::endl;
+    mIsModified = true; // Always rerun
+}
+
+FramerateSynchronizer::~FramerateSynchronizer() {
+    //std::cout << "In destructor" << std::endl;
+    for(auto parent : m_parents) {
+        parent.second->stop();
+    }
+    for(auto thread : m_threads) {
+        thread->join();
+        delete thread;
+    }
+}
+
+void FramerateSynchronizer::setInputConnection(uint portID, DataChannel::pointer port) {
+    createInputPort<DataObject>(portID, false);
+    createOutputPort<DataObject>(portID);
+    ProcessObject::setInputConnection(portID, port);
+}
+
+}

source/FAST/FramerateSynchronizer.hpp

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <FAST/ProcessObject.hpp>
+#include <FAST/Streamers/Streamer.hpp>
+
+namespace fast {
+
+/**
+ * @brief Synchronize multiple process objects
+ * This PO takes in N input connections and creates N output connections.
+ * It keeps the last frame of every connection, and every time a connection
+ * has a new data frame, it send out the latest frame to all output connections.
+ */
+class FAST_EXPORT FramerateSynchronizer : public ProcessObject {
+    FAST_PROCESS_OBJECT(FramerateSynchronizer)
+    public:
+        FAST_CONSTRUCTOR(FramerateSynchronizer, int, priorityPort, = -1);
+        ~FramerateSynchronizer();
+        void setInputConnection(uint portID, DataChannel::pointer port) override;
+    protected:
+        void execute() override;
+
+        std::unordered_map<uint, std::shared_ptr<DataObject>> m_latestData;
+
+        int m_priorityPort = -1;
+        std::map<uint, DataChannel::pointer> m_parents;
+        std::mutex m_latestDataMutex;
+        std::condition_variable m_dataCV;
+        bool m_newData = false;
+        std::vector<std::thread*> m_threads;
+};
+
+}

source/FAST/PipelineSynchronizer.cpp

@@ -11,7 +11,7 @@ fast_add_test_sources(
     Algorithms/DoubleFilterTests.cpp
     SceneGraphTests.cpp
     UtilityTests.cpp
-    PipelineSynchronizerTests.cpp
+        FramerateSynchronizerTests.cpp
     PipelineTests.cpp
 )
 if(FAST_MODULE_Visualization)

source/FAST/PipelineSynchronizer.hpp

@@ -0,0 +1,78 @@
+#include <FAST/Testing.hpp>
+#include <FAST/FramerateSynchronizer.hpp>
+#include "DummyObjects.hpp"
+
+using namespace fast;
+
+TEST_CASE("Framerate synchronizer - two streams at very different rates", "[fast][FramerateSynchronizer]") {
+    const int frames = 20;
+    auto streamer1 = DummyStreamer::New();
+    streamer1->setStreamingMode(StreamingMode::NewestFrameOnly);
+    streamer1->setTotalFrames(frames);
+    streamer1->setSleepTime(100);
+
+    auto streamer2 = DummyStreamer::New();
+    streamer2->setStreamingMode(StreamingMode::NewestFrameOnly);
+    streamer2->setTotalFrames(frames);
+    streamer2->setSleepTime(200);
+
+    auto synchronizer = FramerateSynchronizer::create();
+    synchronizer->connect(0, streamer1);
+    synchronizer->connect(1, streamer2);
+
+    auto port1 = synchronizer->getOutputPort(0);
+    auto port2 = synchronizer->getOutputPort(1);
+
+    int previousID1 = -1;
+    int previousID2 = -1;
+    for(int i = 0; i < 10; ++i) {
+        std::cout << "Iteration: " << i << std::endl;
+        synchronizer->run();
+        auto data1 = port1->getNextFrame<DummyDataObject>();
+        auto data2 = port2->getNextFrame<DummyDataObject>();
+        //CHECK(data1->getID() == i);
+        CHECK(((int)data1->getID() != previousID1 || (int)data2->getID() != previousID2));
+        previousID1 = data1->getID();
+        previousID2 = data2->getID();
+        std::cout << previousID1 << " " << previousID2 << std::endl;
+        if(data1->isLastFrame() || data2->isLastFrame())
+            break;
+    }
+}
+
+TEST_CASE("Framerate synchronizer - two streams at very different rates and priority port", "[fast][FramerateSynchronizer]") {
+    const int frames = 20;
+    auto streamer1 = DummyStreamer::New();
+    streamer1->setStreamingMode(StreamingMode::NewestFrameOnly);
+    streamer1->setTotalFrames(frames);
+    streamer1->setSleepTime(100);
+
+    auto streamer2 = DummyStreamer::New();
+    streamer2->setStreamingMode(StreamingMode::NewestFrameOnly);
+    streamer2->setTotalFrames(frames);
+    streamer2->setSleepTime(200);
+
+    auto synchronizer = FramerateSynchronizer::create(0);
+    synchronizer->connect(0, streamer1);
+    synchronizer->connect(1, streamer2);
+
+    auto port1 = synchronizer->getOutputPort(0);
+    auto port2 = synchronizer->getOutputPort(1);
+
+    int previousID1 = -1;
+    int previousID2 = -1;
+    for(int i = 0; i < 10; ++i) {
+        std::cout << "Iteration: " << i << std::endl;
+        synchronizer->run();
+        auto data1 = port1->getNextFrame<DummyDataObject>();
+        auto data2 = port2->getNextFrame<DummyDataObject>();
+        //CHECK(data1->getID() == i);
+        CHECK((int)data1->getID() > previousID1);
+        CHECK((int)data2->getID() >= previousID2);
+        previousID1 = data1->getID();
+        previousID2 = data2->getID();
+        std::cout << previousID1 << " " << previousID2 << std::endl;
+        if(data1->isLastFrame() || data2->isLastFrame())
+            break;
+    }
+}