Add SharedMemoryPlugin Implementation

cpp/frameProcessor/include/CMakeLists.txt

@@ -24,6 +24,7 @@ SET(HEADERS DataBlock.h
             KafkaProducerPlugin.h
             ParamMetadata.h
             RawFileWriterPlugin.h
-            ParameterPublishPlugin.h)
+            ParameterPublishPlugin.h
+            SharedMemoryPlugin.h)
 
 INSTALL(FILES ${HEADERS} DESTINATION include/frameProcessor)

cpp/frameProcessor/include/FrameProcessorController.h

@@ -16,8 +16,6 @@
 #include "IpcChannel.h"
 #include "IpcReactor.h"
 #include "OdinDataDefaults.h"
-#include "SharedBufferManager.h"
-#include "SharedMemoryController.h"
 #include "logging.h"
 
 namespace FrameProcessor {
@@ -68,15 +66,6 @@ class FrameProcessorController : public IFrameCallback,
     /** Configuration constant to set the debug level of the frame processor **/
     static const std::string CONFIG_DEBUG;
 
-    /** Configuration constant for name of shared memory storage **/
-    static const std::string CONFIG_FR_SHARED_MEMORY;
-    /** Configuration constant for connection string for frame release **/
-    static const std::string CONFIG_FR_RELEASE;
-    /** Configuration constant for connection string for frame ready **/
-    static const std::string CONFIG_FR_READY;
-    /** Configuration constant for executing setup of shared memory interface **/
-    static const std::string CONFIG_FR_SETUP;
-
     /** key-strings for latest config and status timestamp **/
     static const std::string CONFIG_TS_KEY;
     static const std::string STATUS_TS_KEY;
@@ -122,8 +111,6 @@ class FrameProcessorController : public IFrameCallback,
     /** Configuration constant for the meta TX channel high water mark **/
     static const int META_TX_HWM;
 
-    void setupFrameReceiverInterface(const std::string& frPublisherString, const std::string& frSubscriberString);
-    void closeFrameReceiverInterface();
     void setupControlInterface(const std::string& ctrlEndpointString);
     void closeControlInterface();
     void setupMetaRxInterface();
@@ -136,8 +123,6 @@ class FrameProcessorController : public IFrameCallback,
 
     /** Pointer to the logging facility */
     log4cxx::LoggerPtr logger_;
-    /** Pointer to the shared memory controller instance for this process */
-    boost::shared_ptr<SharedMemoryController> sharedMemController_;
     /** Map of plugins loaded, indexed by plugin index */
     std::map<std::string, boost::shared_ptr<FrameProcessorPlugin>> plugins_;
     /** Map of stored configuration objects */

cpp/frameProcessor/include/FrameProcessorPlugin.h

@@ -43,6 +43,7 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO
     virtual bool reset_statistics();
     std::vector<std::string> get_errors();
     std::vector<std::string> get_warnings();
+    void inject_EOA();
     virtual void configure(OdinData::IpcMessage& config, OdinData::IpcMessage& reply);
     virtual void requestConfiguration(OdinData::IpcMessage& reply);
 

cpp/frameProcessor/include/SharedMemoryController.h

@@ -36,20 +36,30 @@ namespace FrameProcessor {
  * shared memory location is available for re-use.
  */
 class SharedMemoryController {
+    typedef boost::function<void(boost::shared_ptr<Frame> frame)> TProcess_frame_cb;
+
 public:
     SharedMemoryController(
         boost::shared_ptr<OdinData::IpcReactor> reactor,
         const std::string& rxEndPoint,
         const std::string& txEndPoint
     );
     virtual ~SharedMemoryController();
-    void setSharedBufferManager(const std::string& shared_buffer_name);
+    void setSharedBufferManager(std::string& shared_buffer_name);
     void requestSharedBufferConfig(const bool deferred = false);
-    void registerCallback(const std::string& name, boost::shared_ptr<IFrameCallback> cb);
-    void removeCallback(const std::string& name);
     void handleRxChannel();
-    void status(OdinData::IpcMessage& status);
-    void injectEOA();
+    const std::string& getName() const
+    {
+        return shbName_;
+    }
+    bool isConfigured() const
+    {
+        return sharedBufferConfigured_;
+    }
+    void inject_process_frame_cb(TProcess_frame_cb callback)
+    {
+        callback_ = callback;
+    }
 
 private:
     /** Pointer to logger */
@@ -64,11 +74,15 @@ class SharedMemoryController {
     OdinData::IpcChannel rxChannel_;
     /** IpcChannel for sending notifications of frame release */
     OdinData::IpcChannel txChannel_;
+    /** Shared buffer name */
+    std::string shbName_;
     /** Shared buffer configured status flag */
     bool sharedBufferConfigured_;
     /** Shared buffer config request deferred flag */
     bool sharedBufferConfigRequestDeferred_;
 
+    TProcess_frame_cb callback_;
+
     /** Name of class used in status messages */
     static const std::string SHARED_MEMORY_CONTROLLER_NAME;
 };

cpp/frameProcessor/include/SharedMemoryPlugin.h

@@ -0,0 +1,59 @@
+/*
+ * SharedMemoryPlugin.h
+ *
+ *  Created on: 08 Sep. 2026
+ *      Author: Famous Alele
+ */
+
+#ifndef SHAREDMEMORYPLUGIN_H_
+#define SHAREDMEMORYPLUGIN_H_
+#include <log4cxx/logger.h>
+
+using namespace log4cxx;
+
+#include "ClassLoader.h"
+#include "FrameProcessorPlugin.h"
+#include "IpcReactor.h"
+#include "SharedMemoryController.h"
+
+namespace FrameProcessor {
+class SharedMemoryPlugin : public FrameProcessorPlugin {
+public:
+    SharedMemoryPlugin();
+    ~SharedMemoryPlugin();
+    void start_reactor();
+    void process_frame(boost::shared_ptr<Frame> frame);
+    void configure(OdinData::IpcMessage& config, OdinData::IpcMessage& reply);
+    void requestConfiguration(OdinData::IpcMessage& reply);
+    void status(OdinData::IpcMessage& reply);
+    int get_version_major() override;
+    int get_version_minor() override;
+    int get_version_patch() override;
+    std::string get_version_short() override;
+    std::string get_version_long() override;
+
+    const static std::string CONFIG_FR_RELEASE;
+    const static std::string CONFIG_FR_READY;
+    const static std::string STATUS_SHB_NAME;
+    const static std::string STATUS_SHB_CONFIGURED;
+
+private:
+    std::string frReleaseEndpoint_;
+    std::string frReadyEndpoint_;
+    boost::shared_ptr<OdinData::IpcReactor> reactor_;
+
+    /** IpcReactor thread */
+    boost::thread m_thread_;
+    /** The shared memory controller object */
+    boost::shared_ptr<SharedMemoryController> shmctrlr_handle_;
+    /** Pointer to logger */
+    LoggerPtr logger_;
+
+    static void dummy_timer()
+    {
+    }
+    void setupFrameReceiverInterface(const std::string&, const std::string&);
+};
+}
+
+#endif // end SHAREDMEMORYPLUGIN_H_

cpp/frameProcessor/src/CMakeLists.txt

@@ -105,6 +105,11 @@ add_library(ParameterPublishPlugin SHARED ParameterPublishPlugin.cpp ParameterPu
 target_link_libraries(ParameterPublishPlugin ${LIB_PROCESSOR} ${Boost_LIBRARIES} ${LOG4CXX_LIBRARIES} ${ZEROMQ_LIBRARIES} ${COMMON_LIBRARY})
 install(TARGETS ParameterPublishPlugin DESTINATION lib)
 
+# Add library for SharedMemory plugin
+add_library(SharedMemoryPlugin SHARED SharedMemoryPlugin.cpp SharedMemoryController.cpp SharedMemoryPluginLib.cpp)
+target_link_libraries(SharedMemoryPlugin ${LIB_PROCESSOR} ${Boost_LIBRARIES} ${LOG4CXX_LIBRARIES} ${ZEROMQ_LIBRARIES} ${COMMON_LIBRARY})
+install(TARGETS SharedMemoryPlugin DESTINATION lib)
+
 # Add test and project source files to executable
 if ( ${CMAKE_SYSTEM_NAME} MATCHES Linux )
   # librt required for timing functions

cpp/frameProcessor/src/FrameProcessorController.cpp

@@ -19,10 +19,6 @@ const std::string FrameProcessorController::META_RX_INTERFACE = "inproc://meta_r
 const std::string FrameProcessorController::CONFIG_EOA = "inject_eoa";
 const std::string FrameProcessorController::CONFIG_DEBUG = "debug_level";
 
-const std::string FrameProcessorController::CONFIG_FR_RELEASE = "fr_release_cnxn";
-const std::string FrameProcessorController::CONFIG_FR_READY = "fr_ready_cnxn";
-const std::string FrameProcessorController::CONFIG_FR_SETUP = "fr_setup";
-
 const std::string FrameProcessorController::CONFIG_CTRL_ENDPOINT = "ctrl_endpoint";
 const std::string FrameProcessorController::CONFIG_META_ENDPOINT = "meta_endpoint";
 
@@ -290,11 +286,6 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m
     // Error messages
     std::vector<std::string> error_messages, warning_messages;
 
-    // Request status information from the shared memory controller
-    if (sharedMemController_) {
-        sharedMemController_->status(reply);
-    }
-
     std::map<std::string, boost::shared_ptr<FrameProcessorPlugin>>::iterator iter;
     if (metadata) {
         for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) {
@@ -403,8 +394,9 @@ void FrameProcessorController::configure(OdinData::IpcMessage& config, OdinData:
     // Check for a request to inject an End Of Acquisition object
     if (config.has_param(FrameProcessorController::CONFIG_EOA)) {
         LOG4CXX_DEBUG_LEVEL(1, logger_, "Injecting End Of Acquisition object into plugin chain");
-        if (sharedMemController_) {
-            sharedMemController_->injectEOA();
+        std::string plugin_name = config.get_param<std::string>(FrameProcessorController::CONFIG_EOA);
+        if (plugins_.count(plugin_name)) {
+            this->plugins_[plugin_name]->inject_EOA();
         }
     }
 
@@ -432,19 +424,6 @@ void FrameProcessorController::configure(OdinData::IpcMessage& config, OdinData:
         this->configurePlugin(pluginConfig, reply);
     }
 
-    // Check if we are being passed the shared memory configuration
-    if (config.has_param(FrameProcessorController::CONFIG_FR_SETUP)) {
-        OdinData::IpcMessage frConfig(
-            config.get_param<const rapidjson::Value&>(FrameProcessorController::CONFIG_FR_SETUP)
-        );
-        if (frConfig.has_param(FrameProcessorController::CONFIG_FR_RELEASE)
-            && frConfig.has_param(FrameProcessorController::CONFIG_FR_READY)) {
-            std::string pubString = frConfig.get_param<std::string>(FrameProcessorController::CONFIG_FR_RELEASE);
-            std::string subString = frConfig.get_param<std::string>(FrameProcessorController::CONFIG_FR_READY);
-            this->setupFrameReceiverInterface(pubString, subString);
-        }
-    }
-
     // Check if we are being asked to store a configuration object
     if (config.has_param(FrameProcessorController::CONFIG_STORE)) {
         OdinData::IpcMessage storeConfig(
@@ -531,9 +510,6 @@ void FrameProcessorController::requestConfiguration(OdinData::IpcMessage& reply,
     // Add local configuration parameter values to the reply
     reply.set_param(FrameProcessorController::CONFIG_CTRL_ENDPOINT, ctrlChannelEndpoint_);
     reply.set_param(FrameProcessorController::CONFIG_META_ENDPOINT, metaTxChannelEndpoint_);
-    std::string fr_cnxn_str = FrameProcessorController::CONFIG_FR_SETUP + '/';
-    reply.set_param(fr_cnxn_str + FrameProcessorController::CONFIG_FR_READY, frReadyEndpoint_);
-    reply.set_param(fr_cnxn_str + FrameProcessorController::CONFIG_FR_RELEASE, frReleaseEndpoint_);
 
     // Loop over plugins and request current configuration from each
     int64_t latest_ts = -1;
@@ -775,24 +751,8 @@ void FrameProcessorController::loadPlugin(const std::string& index, const std::s
 void FrameProcessorController::connectPlugin(const std::string& index, const std::string& connectTo)
 {
     // Check that the plugin is loaded
-    if (plugins_.count(index) > 0) {
-        // Check for the shared memory connection
-        if (connectTo == "frame_receiver") {
-            if (sharedMemController_) {
-                sharedMemController_->registerCallback(index, plugins_[index]);
-            } else {
-                LOG4CXX_ERROR(
-                    logger_, "Cannot connect " << index << " to frame_receiver, frame_receiver is not configured"
-                );
-                std::stringstream is;
-                is << "Cannot connect " << index << " to frame_receiver, frame_receiver is not configured";
-                throw std::runtime_error(is.str().c_str());
-            }
-        } else {
-            if (plugins_.count(connectTo) > 0) {
-                plugins_[connectTo]->register_callback(index, plugins_[index]);
-            }
-        }
+    if ((plugins_.count(index) > 0) & (plugins_.count(connectTo) > 0)) {
+        plugins_[connectTo]->register_callback(index, plugins_[index]);
     } else {
         LOG4CXX_ERROR(logger_, "Cannot connect plugin with index = " << index << ", plugin isn't loaded");
         std::stringstream is;
@@ -809,15 +769,8 @@ void FrameProcessorController::connectPlugin(const std::string& index, const std
 void FrameProcessorController::disconnectPlugin(const std::string& index, const std::string& disconnectFrom)
 {
     // Check that the plugin is loaded
-    if (plugins_.count(index) > 0) {
-        // Check for the shared memory connection
-        if (disconnectFrom == "frame_receiver") {
-            sharedMemController_->removeCallback(index);
-        } else {
-            if (plugins_.count(disconnectFrom) > 0) {
-                plugins_[disconnectFrom]->remove_callback(index);
-            }
-        }
+    if ((plugins_.count(index) > 0) & ((plugins_.count(disconnectFrom) > 0))) {
+        plugins_[disconnectFrom]->remove_callback(index);
     } else {
         LOG4CXX_ERROR(logger_, "Cannot disconnect plugin with index = " << index << ", plugin isn't loaded");
         std::stringstream is;
@@ -880,8 +833,6 @@ void FrameProcessorController::shutdown()
 
         // Close control IPC channel
         closeControlInterface();
-        // Close FrameReceiver interface IPC channels
-        closeFrameReceiverInterface();
 
         // Destroy any allocated DataBlocks
         LOG4CXX_DEBUG_LEVEL(1, logger_, "Tearing down DataBlockPool");
@@ -901,64 +852,6 @@ void FrameProcessorController::waitForShutdown()
     exitCondition_.wait(lock);
 }
 
-/** Set up the frame receiver interface.
- *
- * This method creates new SharedMemoryController and SharedMemoryParser objects,
- * which manage the receipt of frame ready notifications and construction of
- * Frame objects from shared memory.
- * Pointers to the two objects are kept by this class.
- *
- * \param[in] sharedMemName - Name of the shared memory block opened by the frame receiver.
- * \param[in] frPublisherString - Endpoint for sending frame release notifications.
- * \param[in] frSubscriberString - Endpoint for receiving frame ready notifications.
- */
-void FrameProcessorController::setupFrameReceiverInterface(
-    const std::string& frPublisherString,
-    const std::string& frSubscriberString
-)
-{
-    LOG4CXX_DEBUG_LEVEL(
-        1, logger_, "Shared Memory Config: Publisher=" << frPublisherString << " Subscriber=" << frSubscriberString
-    );
-
-    // Only reconstruct the shared memory controller if it has never been created or either
-    // of the endpoints has been changed
-    if (!sharedMemController_ || frPublisherString != frReleaseEndpoint_ || frSubscriberString != frReadyEndpoint_) {
-        try {
-            // Release the current shared memory controller if one exists
-            if (sharedMemController_) {
-                sharedMemController_.reset();
-            }
-            // Create the new shared memory controller and give it the parser and publisher
-            sharedMemController_ = boost::shared_ptr<SharedMemoryController>(
-                new SharedMemoryController(reactor_, frSubscriberString, frPublisherString)
-            );
-            frReadyEndpoint_ = frSubscriberString;
-            frReleaseEndpoint_ = frPublisherString;
-
-        } catch (const boost::interprocess::interprocess_exception& e) {
-            LOG4CXX_ERROR(logger_, "Unable to access shared memory: " << e.what());
-        }
-    } else {
-        LOG4CXX_ERROR(logger_, "*** Not updating shared memory, endpoints were not changed");
-    }
-}
-
-/** Close the frame receiver interface.
- */
-void FrameProcessorController::closeFrameReceiverInterface()
-{
-    LOG4CXX_DEBUG_LEVEL(1, logger_, "Closing FrameReceiver interface.");
-    try {
-        // Release the current shared memory controller if one exists
-        if (sharedMemController_) {
-            sharedMemController_.reset();
-        }
-    } catch (const boost::interprocess::interprocess_exception& e) {
-        LOG4CXX_ERROR(logger_, "Error occurred when closing FrameReceiver interface: " << e.what());
-    }
-}
-
 /** Set up the control interface.
  *
  * This method binds the control IpcChannel to the provided endpoint,

cpp/frameProcessor/src/FrameProcessorPlugin.cpp

@@ -472,4 +472,17 @@ void FrameProcessorPlugin::process_end_of_acquisition()
 {
 }
 
+/** Method to inject End Of Frame sentinel
+ * down the pipeline. Typically, the FrameProcessorCOntroller
+ * sends this message to the SharedMemoryPlugin
+ */
+void FrameProcessorPlugin::inject_EOA()
+{
+    // Create the EOA frame object
+    boost::shared_ptr<FrameProcessor::EndOfAcquisitionFrame> eoa
+        = boost::make_shared<FrameProcessor::EndOfAcquisitionFrame>();
+    this->process_end_of_acquisition();
+    this->push(eoa);
+}
+
 } /* namespace FrameProcessor */