control module pipeline fixes & added control module UTs

base/CMakeLists.txt

@@ -627,6 +627,7 @@ SET(UT_FILES
 	test/overlaymodule_tests.cpp
 	test/testSignalGeneratorSrc_tests.cpp
 	test/audioToTextXform_tests.cpp
+	test/simpleControlModuleTests.cpp
 	${ARM64_UT_FILES}
 	${CUDA_UT_FILES}
 )

base/include/AbsControlModule.h

@@ -18,27 +18,35 @@ class AbsControlModule : public Module {
 	bool term();
 	bool enrollModule(std::string role, boost::shared_ptr<Module> module);
 	boost::shared_ptr<Module> getModuleofRole(std::string role);
+	std::string printStatus();
 	virtual void handleMp4MissingVideotrack(std::string previousVideoFile, std::string nextVideoFile) {}
 	virtual void handleMMQExport(Command cmd, bool priority = false) {}
 	virtual void handleMMQExportView(uint64_t startTS, uint64_t endTS = 9999999999999, bool playabckDirection = true, bool Mp4ReaderExport = false, bool priority = false) {}
 	virtual void handleSendMMQTSCmd(uint64_t  mmqBeginTS, uint64_t mmqEndTS, bool priority = false) {}
 	virtual void handleLastGtkGLRenderTS(uint64_t  latestGtkGlRenderTS, bool priority) {}
 	virtual void handleGoLive(bool goLive, bool priority) {}
 	virtual void handleDecoderSpeed(DecoderPlaybackSpeed cmd, bool priority) {}
-	boost::container::deque<boost::shared_ptr<Module>> pipelineModules;
-	std::map<std::string, boost::shared_ptr<Module>> moduleRoles;
+	// Note: weak pointers to avoid cyclic dependency and mem leaks
+	std::map<std::string, boost::weak_ptr<Module>> moduleRoles;
   	virtual void handleError(const APErrorObject &error) {}
-  	virtual void handleHealthCallback(const APHealthObject &healthObj) {}
-
-
+	virtual void handleHealthCallback(const APHealthObject& healthObj);
+	/**
+	 * @brief Register external function to be triggered on every health callBack that control modules recieves from the modules.
+	 * For eg. In SimpleControlModule, this extention is called at the end of handleHealthCallback function.
+	 * @param function with signature void f(const APHealthObject*, unsigned short)
+	 * @return nothing.
+	 */
+	void registerHealthCallbackExtention(
+		boost::function<void(const APHealthObject*, unsigned short)> callbackFunction);
 protected:
 	bool process(frame_container& frames);
 	bool handleCommand(Command::CommandType type, frame_sp& frame);
 	bool handlePropsChange(frame_sp& frame);
 	virtual void sendEOS() {}
 	virtual void sendEOS(frame_sp& frame) {}
 	virtual void sendEOPFrame() {}
-
+	std::vector<std::string> serializeControlModule();
+	boost::function<void(const APHealthObject*, unsigned short)> healthCallbackExtention;
 private:
 	class Detail;
 	boost::shared_ptr<Detail> mDetail;

base/include/Module.h

@@ -21,6 +21,8 @@
 #include "BufferMaker.h"
 #include "APCallback.h"
 
+#define DUMMY_CTRL_EOP_PIN "dummy_ctrl_eop_pin"
+
 using namespace std;
 
 class FrameContainerQueue;

base/include/PipeLine.h

@@ -27,7 +27,8 @@ class PipeLine {
 	Status myStatus;
 	typedef boost::shared_ptr<Module> item_type;
 	typedef boost::container::deque< item_type > container_type;
-
+	boost::shared_ptr<Module> controlModule = nullptr;
+
 	std::string mName;
 	container_type modules;
 	bool validate();
@@ -47,8 +48,8 @@ class PipeLine {
 	void stop();
 	void term();
 	void wait_for_all(bool ignoreStatus = false);
-	void interrup_wait_for_all();
-	void flushAllQueues();
+	void interrupt_wait_for_all();
+	void flushAllQueues(bool flushControlModuleQ=false);
 	const char* getStatus();
 };
 

base/include/SimpleControlModule.h

@@ -15,13 +15,10 @@ class SimpleControlModule : public AbsControlModule
 
 	~SimpleControlModule()
 	{
-
 	}
-
-	void handleError(const APErrorObject &error);
-    void handleHealthCallback(const APHealthObject &healthObj);
-
-	// ErrorCallbacks
+	std::string printStatus();
+	void handleError(const APErrorObject& error) override;
+	void handleHealthCallback(const APHealthObject& healthObj) override;
 protected:
 	void sendEOS();
 	void sendEOS(frame_sp& frame);

base/src/AbsControlModule.cpp

@@ -4,6 +4,7 @@
 #include "Module.h"
 #include "Command.h"
 #include "PipeLine.h"
+#include "boost/algorithm/string/join.hpp"
 
 class AbsControlModule::Detail
 {
@@ -24,6 +25,7 @@ AbsControlModule::AbsControlModule(AbsControlModuleProps _props)
 {
 	mDetail.reset(new Detail(_props));
 }
+
 AbsControlModule::~AbsControlModule() {}
 
 bool AbsControlModule::handleCommand(Command::CommandType type, frame_sp& frame)
@@ -56,6 +58,12 @@ bool AbsControlModule::process(frame_container& frames)
 	return true;
 }
 
+/**
+ * @brief Enroll your module to use healthcallback, errorcallback and other control module functions
+ * @param boost::shared_ptr<Module> the module to be registered
+ * @param role unique string for role of the module
+ * @return bool.
+ */
 bool AbsControlModule::enrollModule(std::string role, boost::shared_ptr<Module> module)
 {
 	if (moduleRoles.find(role) != moduleRoles.end())
@@ -84,11 +92,54 @@ boost::shared_ptr<Module> AbsControlModule::getModuleofRole(std::string role)
 	boost::shared_ptr<Module> moduleWithRole = nullptr;
 	try
 	{
-		moduleWithRole = moduleRoles[role];
+		moduleWithRole = moduleRoles[role].lock();
 	}
 	catch (std::out_of_range)
 	{
 		LOG_ERROR << "no module with the role <" << role << "> registered with the control module.";
 	}
 	return moduleWithRole;
+}
+
+void AbsControlModule::registerHealthCallbackExtention(
+	boost::function<void(const APHealthObject*, unsigned short)> callbackFunction)
+{
+	healthCallbackExtention = callbackFunction;
+};
+
+void AbsControlModule::handleHealthCallback(const APHealthObject& healthObj)
+{
+	LOG_INFO << "Health Callback from  module " << healthObj.getModuleId();
+	if (!healthCallbackExtention.empty())
+	{
+		LOG_INFO << "Calling the registered Health Callback Extention...";
+		healthCallbackExtention(&healthObj, 1);
+	}
+}
+
+std::vector<std::string> AbsControlModule::serializeControlModule()
+{
+	std::string spacedLineFmt = "\t-->";
+	std::vector<std::string> status;
+	status.push_back("Module <" + this->getId() + "> \n");
+	status.push_back("Enrolled Modules \n");
+	for (auto it : moduleRoles)
+	{
+		status.push_back("module <" + it.second.lock()->getId() + "> role <" + it.first + ">\n");
+		std::string cbStatus = "registered for...\n";
+		if (it.second.lock()->getProps().enableHealthCallBack)
+		{
+			cbStatus += spacedLineFmt + "health callbacks \n";
+		}
+		cbStatus += spacedLineFmt + "error callbacks \n";
+		status.push_back(spacedLineFmt + cbStatus);
+	}
+	return status;
+}
+
+std::string AbsControlModule::printStatus()
+{ 
+	auto ser = boost::algorithm::join(serializeControlModule(), "|");
+	LOG_INFO << ser;
+	return ser;
 }

base/src/Module.cpp

@@ -178,7 +178,10 @@ Module::Module(Kind nature, string name, ModuleProps _props)
   mPropsChangeMetadata.reset(
       new FrameMetadata(FrameMetadata::FrameType::PROPS_CHANGE));
 }
-Module::~Module() {}
+Module::~Module() 
+{
+    LOG_INFO << "Module destructor <" << myId << ">";
+}
 
 bool Module::term()
 {
@@ -1295,6 +1298,16 @@ bool Module::step()
     {
       throw AIPException(CTRL_MODULE_INVALID_STATE, "Unexpected: " + std::to_string(frames.size()) + " frames remain unprocessed in control module.");
     }
+    if (mPlay)
+    {
+        mProfiler->startProcessingLap();
+        ret = stepNonSource(frames);
+        mProfiler->endLap(mQue->size());
+    }
+    else
+    {
+        ret = true;
+    }
   }
   else
   {
@@ -1563,6 +1576,14 @@ bool Module::addEoPFrame(frame_container &frames)
     frames.insert(make_pair(me.first, frame));
   }
 
+  if (myNature == CONTROL)
+  {
+      auto frame = frame_sp(new EoPFrame());
+      auto metadata = framemetadata_sp(new FrameMetadata(FrameMetadata::GENERAL));
+      frame->setMetadata((metadata));
+      frames.insert(make_pair(DUMMY_CTRL_EOP_PIN, frame));
+  }
+
   // if sieve is disabled for atleast one connection - send additional EOP
   // frames - extra EOP frames downstream shouldn't matter
   if (mIsSieveDisabledForAny)
@@ -1586,6 +1607,13 @@ bool Module::handleStop()
   {
     return true;
   }
+  if (myNature == CONTROL)
+  {
+      mRunning = false;
+      term();
+      return true;
+  }
+  // handle SOURCE, TRANSFORM, SINK below
   mStopCount++;
   if (myNature != SOURCE && mStopCount != mForwardPins)
   {

base/src/PipeLine.cpp

@@ -37,8 +37,8 @@ bool PipeLine::addControlModule(boost::shared_ptr<AbsControlModule> cModule)
 	for (int i = 0; i < modules.size(); i++)
 	{
 		modules[i]->addControlModule(cModule);
-		cModule->pipelineModules.push_back(modules[i]);
 	}
+	controlModule = cModule;
 	return true;
 }
 
@@ -133,6 +133,10 @@ bool PipeLine::init()
 			return false;
 		}
 	}
+	if (controlModule != nullptr)
+	{
+		controlModule->init();
+	}
 	myStatus = PL_INITED;
 	LOG_TRACE << " Pipeline initialized";
 	return true;
@@ -159,10 +163,11 @@ void PipeLine::run_all_threaded()
 		m.myThread = boost::thread(ref(m));
 		Utils::setModuleThreadName(m.myThread, m.getId());
 	}
-	if ((modules[0]->controlModule) != nullptr)
+	if (controlModule != nullptr)
 	{
-		Module& m = *(modules[0]->controlModule);
+		Module& m = *(controlModule);
 		m.myThread = boost::thread(ref(m));
+		Utils::setModuleThreadName(m.myThread, m.getId());
 	}
 	mPlay = true;
 }
@@ -183,7 +188,7 @@ void PipeLine::pause()
 			i->get()->play(false);
 		}
 	}
-
+	//Note: controlModule should not be paused
 	mPlay = false;
 }
 
@@ -196,7 +201,11 @@ void PipeLine::play()
 			i->get()->play(true);
 		}
 	}
-
+	// Control module should continue running anyways
+	if (controlModule != nullptr)
+	{
+		controlModule->play(true);
+	}
 	mPlay = true;
 }
 
@@ -215,6 +224,7 @@ void PipeLine::step()
 			i->get()->queueStep();
 		}
 	}
+	// should controlModule step ?
 }
 
 void PipeLine::stop()
@@ -232,6 +242,10 @@ void PipeLine::stop()
 			i->get()->stop();
 		}
 	}
+	if (controlModule != nullptr)
+	{
+		controlModule->stop();
+	}
 }
 
 void PipeLine::wait_for_all(bool ignoreStatus)
@@ -247,10 +261,15 @@ void PipeLine::wait_for_all(bool ignoreStatus)
 		Module& m = *(i->get());
 		m.myThread.join();
 	}
+
+	if (controlModule != nullptr)
+	{
+		controlModule->myThread.join();
+	}
 }
 
 
-void PipeLine::interrup_wait_for_all()
+void PipeLine::interrupt_wait_for_all()
 {
 	if (myStatus > PL_STOPPING)
 	{
@@ -269,6 +288,12 @@ void PipeLine::interrup_wait_for_all()
 		Module& m = *(i->get());
 		m.myThread.join();
 	}
+
+	if (controlModule != nullptr)
+	{
+		controlModule->myThread.interrupt();
+		controlModule->myThread.join();
+	}
 	myStatus = PL_STOPPED;
 }
 
@@ -280,12 +305,18 @@ const char * PipeLine::getStatus()
 	return StatusNames[myStatus];
 }
 
-void PipeLine::flushAllQueues() {
+void PipeLine::flushAllQueues(bool flushControlModuleQ)
+{
+	if (flushControlModuleQ && controlModule != nullptr)
+	{
+		controlModule->flushQue();
+	}
 	for (auto& m : modules)
 	{
 		if (m->myNature == Module::Kind::SOURCE)
 		{
 			m->flushQueRecursive();
 		}
 	}
-}
+}
+

base/src/SimpleControlModule.cpp

@@ -15,6 +15,11 @@ void SimpleControlModule::sendEOPFrame()
 	return Module::sendEoPFrame();
 }
 
+std::string SimpleControlModule::printStatus()
+{
+	return AbsControlModule::printStatus();
+}
+
 // Right Now, Just Logging But Can be used to Do bunch of other things 
 void SimpleControlModule::handleError(const APErrorObject &error)
 {
@@ -26,4 +31,9 @@ void SimpleControlModule::handleError(const APErrorObject &error)
 void SimpleControlModule::handleHealthCallback(const APHealthObject &healthObj)
 {
 	LOG_ERROR << "Health Callback from  module " << healthObj.getModuleId();
+	if (!healthCallbackExtention.empty())
+	{
+		LOG_INFO << "Calling Health Callback Extention...";
+		healthCallbackExtention(&healthObj,1);
+	}
 }