From bcfac88f35f896afd7ba39ebef498c4a9f62d048 Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Fri, 3 Jul 2026 15:22:01 +0100 Subject: [PATCH 01/10] Replace Metadata Hash with a 'Latest Time-stamp' Implement time-stamping feature in plugins. Update time-stamp at locations of structural change to the plugin. Report the latest time-stamp in requestConfiguration(). Remove 'key-hashing' related implementation and tests. Fixes #526 --- cpp/common/include/IpcMessage.h | 32 ------------------- .../include/FrameProcessorPlugin.h | 14 ++++++++ cpp/frameProcessor/src/BloscPlugin.cpp | 2 ++ cpp/frameProcessor/src/FileWriterPlugin.cpp | 3 ++ .../src/FrameProcessorController.cpp | 17 +++++----- cpp/frameProcessor/src/GapFillPlugin.cpp | 2 ++ .../src/KafkaProducerPlugin.cpp | 2 ++ cpp/frameProcessor/src/LiveViewPlugin.cpp | 2 ++ .../src/OffsetAdjustmentPlugin.cpp | 2 ++ .../src/ParameterAdjustmentPlugin.cpp | 2 ++ .../src/ParameterPublishPlugin.cpp | 2 ++ .../src/RawFileWriterPlugin.cpp | 2 ++ cpp/frameReceiver/test/IpcMessageUnitTest.cpp | 27 ---------------- 13 files changed, 41 insertions(+), 68 deletions(-) diff --git a/cpp/common/include/IpcMessage.h b/cpp/common/include/IpcMessage.h index ab67e2101..339b9f183 100644 --- a/cpp/common/include/IpcMessage.h +++ b/cpp/common/include/IpcMessage.h @@ -271,19 +271,6 @@ class IpcMessage { } } - // returns all the key-strings of param_str as a single 'delim' delimited string - std::string get_keys(const std::string& param_str, const char delim = '/') - { - std::string keys_str; - constexpr size_t string_size = 256; - keys_str.reserve(string_size); - if (this->doc_.FindMember(param_str.c_str()) == doc_.MemberEnd()) - return ""; - rapidjson::Value& params = this->doc_[param_str.c_str()]; - get_keys_helper(params, delim, keys_str); - return keys_str; - } - //! Sets the nack message type with a reason parameter void set_nack(const std::string& reason); @@ -475,25 +462,6 @@ class IpcMessage { //! Maps an internal message timestamp representation to an ISO8601 extended format string std::string valid_msg_timestamp(boost::posix_time::ptime msg_timestamp); - // helper function to recursively append key-strings in 'value' to the 'result' variable - static void get_keys_helper(const rapidjson::Value& value, const char delim, std::string& result) - { - if (__builtin_expect(value.IsObject(), 1)) { - for (const auto& member : value.GetObject()) { - // Store the found key - result.append(member.name.GetString()) += delim; - - // Look deeper inside this key's value - get_keys_helper(member.value, delim, result); - } - } else if (value.IsArray()) { - // Arrays don't have keys, but their elements (like child objects) might - for (const auto& item : value.GetArray()) { - get_keys_helper(item, delim, result); - } - } - } - //! Indicates if the message has a params block bool has_params(void) const; diff --git a/cpp/frameProcessor/include/FrameProcessorPlugin.h b/cpp/frameProcessor/include/FrameProcessorPlugin.h index 3c96cc980..13c7ae4d7 100644 --- a/cpp/frameProcessor/include/FrameProcessorPlugin.h +++ b/cpp/frameProcessor/include/FrameProcessorPlugin.h @@ -73,8 +73,20 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO void remove_callback(const std::string& name); void remove_all_callbacks(); void notify_end_of_acquisition(); + static constexpr char METADATA_VERSION[] = "metadata_version"; protected: + /** function to update the config time-stamp! */ + inline void update_config_metadata_version(void) + { + auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::system_clock::now()); + auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); + this->config_metadata_ts_ = val.count(); + } + inline int64_t get_metadata_version() + { + return this->config_metadata_ts_; + } void push(boost::shared_ptr frame); void push(const std::string& plugin_name, boost::shared_ptr frame); @@ -242,6 +254,8 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO boost::mutex mutex_; /** process_frame performance stats */ CallDuration process_duration_; + /** time-stamp to represent the version of config metadata */ + int64_t config_metadata_ts_; }; } /* namespace FrameProcessor */ diff --git a/cpp/frameProcessor/src/BloscPlugin.cpp b/cpp/frameProcessor/src/BloscPlugin.cpp index 7913a8dbb..e1fb32d1b 100644 --- a/cpp/frameProcessor/src/BloscPlugin.cpp +++ b/cpp/frameProcessor/src/BloscPlugin.cpp @@ -104,6 +104,7 @@ BloscPlugin::BloscPlugin() : CONFIG_BLOSC_MODE, PMDD::STRING_T, PMDA::READ_WRITE, { FPB::BLOSC_COMPRESS_MODE_STR, FPB::BLOSC_DECOMPRESS_MODE_STR, FPB::BLOSC_OFF_MODE_STR } ); + update_config_metadata_version(); this->commanded_compression_settings_.blosc_compressor = BLOSC_LZ4_COMPNAME; this->commanded_compression_settings_.shuffle = BLOSC_BITSHUFFLE_STR; @@ -418,6 +419,7 @@ void BloscPlugin::requestConfiguration(OdinData::IpcMessage& reply) this->commanded_compression_settings_.compression_level ); reply.set_param(this->get_name() + '/' + BloscPlugin::CONFIG_BLOSC_MODE, Mode_map::mode_to_str(this->plugin_mode_)); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } int BloscPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index 74a863f7a..565a522ed 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -168,6 +168,7 @@ FileWriterPlugin::FileWriterPlugin() : add_status_param_metadata(prefix + STATUS_LAST_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); add_status_param_metadata(prefix + STATUS_MAX_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); add_status_param_metadata(prefix + STATUS_MEAN_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); + update_config_metadata_version(); this->logger_ = Logger::getLogger("FP.FileWriterPlugin"); LOG4CXX_INFO(logger_, "FileWriterPlugin version " << this->get_version_long() << " loaded"); @@ -366,6 +367,7 @@ void FileWriterPlugin::configure(OdinData::IpcMessage& config, OdinData::IpcMess this->configure_dataset(dataset_name, dsetConfig, reply); } } + update_config_metadata_version(); } // Check to see if we are deleting all datasets @@ -494,6 +496,7 @@ void FileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) } } } + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/FrameProcessorController.cpp b/cpp/frameProcessor/src/FrameProcessorController.cpp index 6a56ed5d9..a1d9c6449 100644 --- a/cpp/frameProcessor/src/FrameProcessorController.cpp +++ b/cpp/frameProcessor/src/FrameProcessorController.cpp @@ -294,11 +294,6 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m sharedMemController_->status(reply); } - // returns all key strings in "params" as a '/' delimited string - std::string param_keys = reply.get_keys("params", '/'); - size_t hash = std::hash {}(param_keys); - reply.set_param(FrameProcessorController::METADATA_HASH, hash); - std::map>::iterator iter; if (metadata) { for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { @@ -541,10 +536,14 @@ void FrameProcessorController::requestConfiguration(OdinData::IpcMessage& reply, iter->second->requestConfiguration(reply); } - // returns all key strings in "params" as a '/' delimited string - std::string param_keys = reply.get_keys("params", '/'); - size_t hash = std::hash {}(param_keys); - reply.set_param(FrameProcessorController::METADATA_HASH, hash); + int64_t latest_metadata_ver = -1; + for (auto iter : plugins_) { + latest_metadata_ver = std::max( + latest_metadata_ver, + reply.get_param(iter.second->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION) + ); + } + reply.set_param(FrameProcessorController::METADATA_HASH, latest_metadata_ver); if (metadata) { for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { diff --git a/cpp/frameProcessor/src/GapFillPlugin.cpp b/cpp/frameProcessor/src/GapFillPlugin.cpp index fe7600344..9be988a01 100644 --- a/cpp/frameProcessor/src/GapFillPlugin.cpp +++ b/cpp/frameProcessor/src/GapFillPlugin.cpp @@ -29,6 +29,7 @@ GapFillPlugin::GapFillPlugin() add_config_param_metadata(GapFillPlugin::CONFIG_CHIP_SIZE, PMDD::INTARR_T, PMDA::READ_WRITE); add_config_param_metadata(GapFillPlugin::CONFIG_GRID_X_GAPS, PMDD::INTARR_T, PMDA::READ_WRITE); add_config_param_metadata(GapFillPlugin::CONFIG_GRID_Y_GAPS, PMDD::INTARR_T, PMDA::READ_WRITE); + update_config_metadata_version(); LOG4CXX_INFO(logger_, "GapFillPlugin version " << this->get_version_long() << " loaded"); } @@ -319,6 +320,7 @@ void GapFillPlugin::requestConfiguration(OdinData::IpcMessage& reply) for (int index = 0; index < gaps_y_.size(); index++) { reply.set_param(get_name() + '/' + CONFIG_GRID_Y_GAPS + "[]", gaps_y_[index]); } + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } } // namespace FrameProcessor diff --git a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp index 289715bae..bb0907788 100644 --- a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp +++ b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp @@ -66,6 +66,7 @@ KafkaProducerPlugin::KafkaProducerPlugin() : add_status_param_metadata(STATUS_FRAMES_SENT, PMDD::UINT_T, PMDA::READ_ONLY, 0); add_status_param_metadata(STATUS_FRAMES_LOST, PMDD::UINT_T, PMDA::READ_ONLY, 0); add_status_param_metadata(STATUS_FRAMES_ACK, PMDD::UINT_T, PMDA::READ_ONLY, 0); + update_config_metadata_version(); this->reset_statistics(); } @@ -126,6 +127,7 @@ void KafkaProducerPlugin::requestConfiguration(OdinData::IpcMessage& reply) reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_PARTITION, this->partition_); reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_DATASET, this->dataset_name_); reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_INCLUDE_PARAMETERS, this->include_parameters_); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/LiveViewPlugin.cpp b/cpp/frameProcessor/src/LiveViewPlugin.cpp index c50a0bd57..9ae0bd6cf 100644 --- a/cpp/frameProcessor/src/LiveViewPlugin.cpp +++ b/cpp/frameProcessor/src/LiveViewPlugin.cpp @@ -40,6 +40,7 @@ LiveViewPlugin::LiveViewPlugin() : add_config_param_metadata(CONFIG_SOCKET_ADDR, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_DATASET_NAME, PMDD::STRINGARR_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_TAGGED_FILTER_NAME, PMDD::STRINGARR_T, PMDA::READ_WRITE); + update_config_metadata_version(); set_frame_freq_config(DEFAULT_FRAME_FREQ); set_per_second_config(DEFAULT_PER_SECOND); @@ -197,6 +198,7 @@ void LiveViewPlugin::requestConfiguration(OdinData::IpcMessage& reply) reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_SOCKET_ADDR, image_view_socket_addr_); reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_PER_SECOND, per_second_); reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_DATASET_NAME, dataset_names_); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp b/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp index 1f2eba10f..0634eef4f 100644 --- a/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp @@ -20,6 +20,7 @@ OffsetAdjustmentPlugin::OffsetAdjustmentPlugin() : logger_ = Logger::getLogger("FP.OffsetAdjustmentPlugin"); LOG4CXX_INFO(logger_, "OffsetAdjustmentPlugin version " << this->get_version_long() << " loaded"); add_config_param_metadata(OFFSET_ADJUSTMENT_CONFIG, PMDD::INT_T, PMDA::READ_WRITE); + update_config_metadata_version(); } /** @@ -75,6 +76,7 @@ void OffsetAdjustmentPlugin::configure(OdinData::IpcMessage& config, OdinData::I void OffsetAdjustmentPlugin::requestConfiguration(OdinData::IpcMessage& reply) { reply.set_param(get_name() + '/' + OFFSET_ADJUSTMENT_CONFIG, offset_adjustment_.load()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } int OffsetAdjustmentPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp index 176a62dbb..8f3bf0c3c 100644 --- a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp @@ -127,6 +127,7 @@ void ParameterAdjustmentPlugin::configure(OdinData::IpcMessage& config, OdinData parameter_adjustments_.clear(); parameter_inputs_.clear(); } + update_config_metadata_version(); } } catch (std::runtime_error& e) { std::stringstream ss; @@ -158,6 +159,7 @@ void ParameterAdjustmentPlugin::requestConfiguration(OdinData::IpcMessage& reply input_iter->second ); } + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } int ParameterAdjustmentPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/ParameterPublishPlugin.cpp b/cpp/frameProcessor/src/ParameterPublishPlugin.cpp index bada4ecd6..acec355d5 100644 --- a/cpp/frameProcessor/src/ParameterPublishPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterPublishPlugin.cpp @@ -24,6 +24,7 @@ ParameterPublishPlugin::ParameterPublishPlugin() : LOG4CXX_INFO(logger_, "ParameterPublishPlugin version " << this->get_version_long() << " loaded"); add_config_param_metadata(CONFIG_ENDPOINT, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(DATA_PARAMETERS, PMDD::STRINGARR_T, PMDA::READ_WRITE); + update_config_metadata_version(); } /** @@ -126,6 +127,7 @@ void ParameterPublishPlugin::requestConfiguration(OdinData::IpcMessage& reply) for (auto& it : this->parameters_) { reply.set_param(parameters_key, it); } + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } /** Bind to endpoint and store for config reporting diff --git a/cpp/frameProcessor/src/RawFileWriterPlugin.cpp b/cpp/frameProcessor/src/RawFileWriterPlugin.cpp index 2c6d9def0..ee6755256 100644 --- a/cpp/frameProcessor/src/RawFileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/RawFileWriterPlugin.cpp @@ -26,6 +26,7 @@ RawFileWriterPlugin::RawFileWriterPlugin() : add_config_param_metadata(CONFIG_FILE_PATH, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_ENABLED, PMDD::BOOL_T, PMDA::READ_WRITE); add_status_param_metadata(STATUS_DROPPED_FRAMES, PMDD::UINT_T, PMDA::READ_ONLY, 0); + update_config_metadata_version(); } void RawFileWriterPlugin::process_frame(boost::shared_ptr frame) @@ -101,6 +102,7 @@ void RawFileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) { reply.set_param(this->get_name() + '/' + RawFileWriterPlugin::CONFIG_FILE_PATH, this->file_path_.string()); reply.set_param(this->get_name() + '/' + RawFileWriterPlugin::CONFIG_ENABLED, this->enabled_); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); } /** Get status of the RawFileWriterPlugin diff --git a/cpp/frameReceiver/test/IpcMessageUnitTest.cpp b/cpp/frameReceiver/test/IpcMessageUnitTest.cpp index 9cfb062b7..555502491 100644 --- a/cpp/frameReceiver/test/IpcMessageUnitTest.cpp +++ b/cpp/frameReceiver/test/IpcMessageUnitTest.cpp @@ -383,31 +383,4 @@ BOOST_AUTO_TEST_CASE(TestIpcMessageCreationSpeed) << rate << " Hz" ); } - -BOOST_AUTO_TEST_CASE(IpcMessageGetKeys) -{ - const char* json_str = "{ \"msg_type\":\"cmd\"," - " \"msg_val\":\"status\"," - " \"timestamp\" : \"2026-05-09T12:10:01.123456\"," - " \"status\" : true," - " \"params\" : {" - " \"param1\" : \"values\"," - " \"param2\" : 274," - " \"info\": {" - " \"phoneNumber\": 2026," - " \"state\": true," - " \"name\":{" - " \"first\": \"odin\"," - " \"last\": \"data\"" - " }," - " \"dob\": 2011" - " }" - " }" - "}"; - OdinData::IpcMessage test_msg(json_str, true); - BOOST_CHECK(test_msg.is_valid() == true); - std::string keys_list = test_msg.get_keys("params"); - BOOST_CHECK(keys_list == "param1/param2/info/phoneNumber/state/name/first/last/dob/"); -} - BOOST_AUTO_TEST_SUITE_END(); From 94da0fe8b0d9ea643658b7129d74a02e5b13ac1b Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Mon, 6 Jul 2026 16:25:07 +0100 Subject: [PATCH 02/10] Implement Timestamp Function for provideStatus --- cpp/frameProcessor/include/FrameProcessorPlugin.h | 15 ++++++++++++++- cpp/frameProcessor/src/BloscPlugin.cpp | 2 +- cpp/frameProcessor/src/FileWriterPlugin.cpp | 6 ++++-- .../src/FrameProcessorController.cpp | 14 ++++++++++++++ cpp/frameProcessor/src/GapFillPlugin.cpp | 2 +- cpp/frameProcessor/src/KafkaProducerPlugin.cpp | 6 ++++-- cpp/frameProcessor/src/LiveViewPlugin.cpp | 2 +- cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp | 2 +- .../src/ParameterAdjustmentPlugin.cpp | 2 +- cpp/frameProcessor/src/ParameterPublishPlugin.cpp | 2 +- cpp/frameProcessor/src/RawFileWriterPlugin.cpp | 6 ++++-- 11 files changed, 46 insertions(+), 13 deletions(-) diff --git a/cpp/frameProcessor/include/FrameProcessorPlugin.h b/cpp/frameProcessor/include/FrameProcessorPlugin.h index 13c7ae4d7..eb5b9f802 100644 --- a/cpp/frameProcessor/include/FrameProcessorPlugin.h +++ b/cpp/frameProcessor/include/FrameProcessorPlugin.h @@ -83,10 +83,21 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); this->config_metadata_ts_ = val.count(); } - inline int64_t get_metadata_version() + /** function to update the status time-stamp! */ + inline void update_status_metadata_version(void) + { + auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::system_clock::now()); + auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); + this->status_metadata_ts_ = val.count(); + } + inline int64_t get_config_metadata_version() { return this->config_metadata_ts_; } + inline int64_t get_status_metadata_version() + { + return this->status_metadata_ts_; + } void push(boost::shared_ptr frame); void push(const std::string& plugin_name, boost::shared_ptr frame); @@ -256,6 +267,8 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO CallDuration process_duration_; /** time-stamp to represent the version of config metadata */ int64_t config_metadata_ts_; + /** time-stamp to represent the version of config metadata */ + int64_t status_metadata_ts_; }; } /* namespace FrameProcessor */ diff --git a/cpp/frameProcessor/src/BloscPlugin.cpp b/cpp/frameProcessor/src/BloscPlugin.cpp index e1fb32d1b..282b96e39 100644 --- a/cpp/frameProcessor/src/BloscPlugin.cpp +++ b/cpp/frameProcessor/src/BloscPlugin.cpp @@ -419,7 +419,7 @@ void BloscPlugin::requestConfiguration(OdinData::IpcMessage& reply) this->commanded_compression_settings_.compression_level ); reply.set_param(this->get_name() + '/' + BloscPlugin::CONFIG_BLOSC_MODE, Mode_map::mode_to_str(this->plugin_mode_)); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } int BloscPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index 565a522ed..b87b2ae76 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -143,6 +143,7 @@ FileWriterPlugin::FileWriterPlugin() : FileWriterPlugin::CLOSE_TIMEOUT_PERIOD, PMDD::UINT_T, PMDA::READ_WRITE, 0, PMD::MAX_UNSET ); add_config_param_metadata(FileWriterPlugin::START_CLOSE_TIMEOUT, PMDD::BOOL_T, PMDA::READ_WRITE); + update_config_metadata_version(); add_status_param_metadata(STATUS_WRITING, PMDD::BOOL_T, PMDA::READ_ONLY); add_status_param_metadata(STATUS_FRAMES_MAX, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); @@ -168,7 +169,7 @@ FileWriterPlugin::FileWriterPlugin() : add_status_param_metadata(prefix + STATUS_LAST_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); add_status_param_metadata(prefix + STATUS_MAX_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); add_status_param_metadata(prefix + STATUS_MEAN_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); - update_config_metadata_version(); + update_status_metadata_version(); this->logger_ = Logger::getLogger("FP.FileWriterPlugin"); LOG4CXX_INFO(logger_, "FileWriterPlugin version " << this->get_version_long() << " loaded"); @@ -496,7 +497,7 @@ void FileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) } } } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** @@ -893,6 +894,7 @@ void FileWriterPlugin::status(OdinData::IpcMessage& status) status.set_param(prefix + STATUS_RANK, (int)this->concurrent_rank_); status.set_param(prefix + STATUS_TIMEOUT_ACTIVE, this->timeout_active_); add_file_writing_stats(status); + status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_status_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/FrameProcessorController.cpp b/cpp/frameProcessor/src/FrameProcessorController.cpp index a1d9c6449..80a85e35e 100644 --- a/cpp/frameProcessor/src/FrameProcessorController.cpp +++ b/cpp/frameProcessor/src/FrameProcessorController.cpp @@ -316,6 +316,20 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m std::vector plugin_warnings = iter->second->get_warnings(); warning_messages.insert(warning_messages.end(), plugin_warnings.begin(), plugin_warnings.end()); } + + int64_t latest_metadata_ver = -1; + for (auto iter : plugins_) { + try { + latest_metadata_ver = std::max( + latest_metadata_ver, + reply.get_param(iter.second->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION) + ); + } catch (...) { + // pass + } + } + reply.set_param(FrameProcessorController::METADATA_HASH, latest_metadata_ver); + std::vector::iterator error_iter; for (error_iter = error_messages.begin(); error_iter != error_messages.end(); ++error_iter) { reply.set_param("error[]", *error_iter); diff --git a/cpp/frameProcessor/src/GapFillPlugin.cpp b/cpp/frameProcessor/src/GapFillPlugin.cpp index 9be988a01..bb8d53585 100644 --- a/cpp/frameProcessor/src/GapFillPlugin.cpp +++ b/cpp/frameProcessor/src/GapFillPlugin.cpp @@ -320,7 +320,7 @@ void GapFillPlugin::requestConfiguration(OdinData::IpcMessage& reply) for (int index = 0; index < gaps_y_.size(); index++) { reply.set_param(get_name() + '/' + CONFIG_GRID_Y_GAPS + "[]", gaps_y_[index]); } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } } // namespace FrameProcessor diff --git a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp index bb0907788..3339c7838 100644 --- a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp +++ b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp @@ -62,11 +62,12 @@ KafkaProducerPlugin::KafkaProducerPlugin() : add_config_param_metadata(CONFIG_PARTITION, PMDD::INT_T, PMDA::READ_WRITE, -1); add_config_param_metadata(CONFIG_DATASET, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_INCLUDE_PARAMETERS, PMDD::BOOL_T, PMDA::READ_WRITE); + update_config_metadata_version(); add_status_param_metadata(STATUS_FRAMES_SENT, PMDD::UINT_T, PMDA::READ_ONLY, 0); add_status_param_metadata(STATUS_FRAMES_LOST, PMDD::UINT_T, PMDA::READ_ONLY, 0); add_status_param_metadata(STATUS_FRAMES_ACK, PMDD::UINT_T, PMDA::READ_ONLY, 0); - update_config_metadata_version(); + update_status_metadata_version(); this->reset_statistics(); } @@ -127,7 +128,7 @@ void KafkaProducerPlugin::requestConfiguration(OdinData::IpcMessage& reply) reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_PARTITION, this->partition_); reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_DATASET, this->dataset_name_); reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_INCLUDE_PARAMETERS, this->include_parameters_); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** @@ -145,6 +146,7 @@ void KafkaProducerPlugin::status(OdinData::IpcMessage& status) status.set_param(get_name() + '/' + STATUS_FRAMES_LOST, frames_lost_); /* Number of acknowledged frames */ status.set_param(get_name() + '/' + STATUS_FRAMES_ACK, frames_ack_); + status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_status_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/LiveViewPlugin.cpp b/cpp/frameProcessor/src/LiveViewPlugin.cpp index 9ae0bd6cf..b2f74deeb 100644 --- a/cpp/frameProcessor/src/LiveViewPlugin.cpp +++ b/cpp/frameProcessor/src/LiveViewPlugin.cpp @@ -198,7 +198,7 @@ void LiveViewPlugin::requestConfiguration(OdinData::IpcMessage& reply) reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_SOCKET_ADDR, image_view_socket_addr_); reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_PER_SECOND, per_second_); reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_DATASET_NAME, dataset_names_); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp b/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp index 0634eef4f..025488366 100644 --- a/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp @@ -76,7 +76,7 @@ void OffsetAdjustmentPlugin::configure(OdinData::IpcMessage& config, OdinData::I void OffsetAdjustmentPlugin::requestConfiguration(OdinData::IpcMessage& reply) { reply.set_param(get_name() + '/' + OFFSET_ADJUSTMENT_CONFIG, offset_adjustment_.load()); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } int OffsetAdjustmentPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp index 8f3bf0c3c..1153762a2 100644 --- a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp @@ -159,7 +159,7 @@ void ParameterAdjustmentPlugin::requestConfiguration(OdinData::IpcMessage& reply input_iter->second ); } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } int ParameterAdjustmentPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/ParameterPublishPlugin.cpp b/cpp/frameProcessor/src/ParameterPublishPlugin.cpp index acec355d5..2dfb479e4 100644 --- a/cpp/frameProcessor/src/ParameterPublishPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterPublishPlugin.cpp @@ -127,7 +127,7 @@ void ParameterPublishPlugin::requestConfiguration(OdinData::IpcMessage& reply) for (auto& it : this->parameters_) { reply.set_param(parameters_key, it); } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** Bind to endpoint and store for config reporting diff --git a/cpp/frameProcessor/src/RawFileWriterPlugin.cpp b/cpp/frameProcessor/src/RawFileWriterPlugin.cpp index ee6755256..f8ea3ad3f 100644 --- a/cpp/frameProcessor/src/RawFileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/RawFileWriterPlugin.cpp @@ -25,8 +25,9 @@ RawFileWriterPlugin::RawFileWriterPlugin() : LOG4CXX_TRACE(logger_, "RawFileWriterPlugin constructor."); add_config_param_metadata(CONFIG_FILE_PATH, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_ENABLED, PMDD::BOOL_T, PMDA::READ_WRITE); - add_status_param_metadata(STATUS_DROPPED_FRAMES, PMDD::UINT_T, PMDA::READ_ONLY, 0); update_config_metadata_version(); + add_status_param_metadata(STATUS_DROPPED_FRAMES, PMDD::UINT_T, PMDA::READ_ONLY, 0); + update_status_metadata_version(); } void RawFileWriterPlugin::process_frame(boost::shared_ptr frame) @@ -102,7 +103,7 @@ void RawFileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) { reply.set_param(this->get_name() + '/' + RawFileWriterPlugin::CONFIG_FILE_PATH, this->file_path_.string()); reply.set_param(this->get_name() + '/' + RawFileWriterPlugin::CONFIG_ENABLED, this->enabled_); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_metadata_version()); + reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** Get status of the RawFileWriterPlugin @@ -112,6 +113,7 @@ void RawFileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) void RawFileWriterPlugin::status(OdinData::IpcMessage& status) { status.set_param(this->get_name() + '/' + RawFileWriterPlugin::STATUS_DROPPED_FRAMES, this->dropped_frames_); + status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_status_metadata_version()); } int RawFileWriterPlugin::get_version_major() From 34f9085193e55517a33b5d6484c453e8a8671ba6 Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Mon, 13 Jul 2026 12:31:38 +0000 Subject: [PATCH 03/10] Restructure Timestamp Implementation Call the update_*_metadata_ts() methods from the parent class - FrameProcessorPlugin's constructor. Add public methods to return the raw timestamp value of the plugin. Remove redundant set_param() calls and extra loops. --- .../include/FrameProcessorController.h | 4 +-- .../include/FrameProcessorPlugin.h | 24 +++++++------- cpp/frameProcessor/src/BloscPlugin.cpp | 2 -- cpp/frameProcessor/src/FileWriterPlugin.cpp | 5 +-- .../src/FrameProcessorController.cpp | 31 +++++-------------- .../src/FrameProcessorPlugin.cpp | 2 ++ cpp/frameProcessor/src/GapFillPlugin.cpp | 2 -- .../src/KafkaProducerPlugin.cpp | 5 +-- cpp/frameProcessor/src/LiveViewPlugin.cpp | 2 -- .../src/OffsetAdjustmentPlugin.cpp | 2 -- .../src/ParameterAdjustmentPlugin.cpp | 1 - .../src/ParameterPublishPlugin.cpp | 2 -- .../src/RawFileWriterPlugin.cpp | 4 --- 13 files changed, 26 insertions(+), 60 deletions(-) diff --git a/cpp/frameProcessor/include/FrameProcessorController.h b/cpp/frameProcessor/include/FrameProcessorController.h index c0f156c41..8cdd28c55 100644 --- a/cpp/frameProcessor/include/FrameProcessorController.h +++ b/cpp/frameProcessor/include/FrameProcessorController.h @@ -77,8 +77,8 @@ class FrameProcessorController : public IFrameCallback, /** Configuration constant for executing setup of shared memory interface **/ static const std::string CONFIG_FR_SETUP; - /** current metadata hash-value **/ - static const std::string METADATA_HASH; + /** latest metadata timestamp **/ + static const std::string METADATA_TS_KEY; /** Configuration constant for control socket endpoint **/ static const std::string CONFIG_CTRL_ENDPOINT; diff --git a/cpp/frameProcessor/include/FrameProcessorPlugin.h b/cpp/frameProcessor/include/FrameProcessorPlugin.h index eb5b9f802..b3a38b2e6 100644 --- a/cpp/frameProcessor/include/FrameProcessorPlugin.h +++ b/cpp/frameProcessor/include/FrameProcessorPlugin.h @@ -73,31 +73,31 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO void remove_callback(const std::string& name); void remove_all_callbacks(); void notify_end_of_acquisition(); - static constexpr char METADATA_VERSION[] = "metadata_version"; + /** return the latest ts of config/status metadata for this plugin instance */ + inline int64_t get_config_metadata_ts() const + { + return this->config_metadata_ts_; + } + inline int64_t get_status_metadata_ts() const + { + return this->status_metadata_ts_; + } protected: /** function to update the config time-stamp! */ inline void update_config_metadata_version(void) { - auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::system_clock::now()); + auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); this->config_metadata_ts_ = val.count(); } /** function to update the status time-stamp! */ inline void update_status_metadata_version(void) { - auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::system_clock::now()); + auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); this->status_metadata_ts_ = val.count(); } - inline int64_t get_config_metadata_version() - { - return this->config_metadata_ts_; - } - inline int64_t get_status_metadata_version() - { - return this->status_metadata_ts_; - } void push(boost::shared_ptr frame); void push(const std::string& plugin_name, boost::shared_ptr frame); @@ -267,7 +267,7 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO CallDuration process_duration_; /** time-stamp to represent the version of config metadata */ int64_t config_metadata_ts_; - /** time-stamp to represent the version of config metadata */ + /** time-stamp to represent the version of status metadata */ int64_t status_metadata_ts_; }; diff --git a/cpp/frameProcessor/src/BloscPlugin.cpp b/cpp/frameProcessor/src/BloscPlugin.cpp index 282b96e39..7913a8dbb 100644 --- a/cpp/frameProcessor/src/BloscPlugin.cpp +++ b/cpp/frameProcessor/src/BloscPlugin.cpp @@ -104,7 +104,6 @@ BloscPlugin::BloscPlugin() : CONFIG_BLOSC_MODE, PMDD::STRING_T, PMDA::READ_WRITE, { FPB::BLOSC_COMPRESS_MODE_STR, FPB::BLOSC_DECOMPRESS_MODE_STR, FPB::BLOSC_OFF_MODE_STR } ); - update_config_metadata_version(); this->commanded_compression_settings_.blosc_compressor = BLOSC_LZ4_COMPNAME; this->commanded_compression_settings_.shuffle = BLOSC_BITSHUFFLE_STR; @@ -419,7 +418,6 @@ void BloscPlugin::requestConfiguration(OdinData::IpcMessage& reply) this->commanded_compression_settings_.compression_level ); reply.set_param(this->get_name() + '/' + BloscPlugin::CONFIG_BLOSC_MODE, Mode_map::mode_to_str(this->plugin_mode_)); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } int BloscPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index b87b2ae76..8fdcdb587 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -143,7 +143,6 @@ FileWriterPlugin::FileWriterPlugin() : FileWriterPlugin::CLOSE_TIMEOUT_PERIOD, PMDD::UINT_T, PMDA::READ_WRITE, 0, PMD::MAX_UNSET ); add_config_param_metadata(FileWriterPlugin::START_CLOSE_TIMEOUT, PMDD::BOOL_T, PMDA::READ_WRITE); - update_config_metadata_version(); add_status_param_metadata(STATUS_WRITING, PMDD::BOOL_T, PMDA::READ_ONLY); add_status_param_metadata(STATUS_FRAMES_MAX, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); @@ -497,7 +496,6 @@ void FileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) } } } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** @@ -882,7 +880,7 @@ void FileWriterPlugin::delete_datasets() void FileWriterPlugin::status(OdinData::IpcMessage& status) { // Record the plugin's status items - std::string prefix = get_name() + '/'; + const std::string prefix = get_name() + '/'; status.set_param(prefix + STATUS_WRITING, this->writing_); status.set_param(prefix + STATUS_FRAMES_MAX, (int)this->current_acquisition_->frames_to_write_); status.set_param(prefix + STATUS_FRAMES_WRITTEN, (int)this->current_acquisition_->frames_written_); @@ -894,7 +892,6 @@ void FileWriterPlugin::status(OdinData::IpcMessage& status) status.set_param(prefix + STATUS_RANK, (int)this->concurrent_rank_); status.set_param(prefix + STATUS_TIMEOUT_ACTIVE, this->timeout_active_); add_file_writing_stats(status); - status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_status_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/FrameProcessorController.cpp b/cpp/frameProcessor/src/FrameProcessorController.cpp index 80a85e35e..5fc979954 100644 --- a/cpp/frameProcessor/src/FrameProcessorController.cpp +++ b/cpp/frameProcessor/src/FrameProcessorController.cpp @@ -43,7 +43,7 @@ const std::string FrameProcessorController::CONFIG_VALUE = "value"; const std::string FrameProcessorController::COMMAND_KEY = "command"; const std::string FrameProcessorController::SUPPORTED_KEY = "supported"; -const std::string FrameProcessorController::METADATA_HASH = "metadata_hash"; +const std::string FrameProcessorController::METADATA_TS_KEY = "metadata_ts"; const int FrameProcessorController::META_TX_HWM = 10000; @@ -302,11 +302,14 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m } } + int64_t latest_metadata_ver = -1; // Loop over plugins, list names and request status from each for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { reply.set_param("plugins/names[]", iter->first); // Request status for the plugin iter->second->status(reply); + // Check for the latest timestamp + latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_status_metadata_ts()); // Add performance statistics iter->second->add_performance_stats(reply); // Read error level @@ -316,19 +319,7 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m std::vector plugin_warnings = iter->second->get_warnings(); warning_messages.insert(warning_messages.end(), plugin_warnings.begin(), plugin_warnings.end()); } - - int64_t latest_metadata_ver = -1; - for (auto iter : plugins_) { - try { - latest_metadata_ver = std::max( - latest_metadata_ver, - reply.get_param(iter.second->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION) - ); - } catch (...) { - // pass - } - } - reply.set_param(FrameProcessorController::METADATA_HASH, latest_metadata_ver); + reply.set_param(FrameProcessorController::METADATA_TS_KEY, latest_metadata_ver); std::vector::iterator error_iter; for (error_iter = error_messages.begin(); error_iter != error_messages.end(); ++error_iter) { @@ -544,20 +535,14 @@ void FrameProcessorController::requestConfiguration(OdinData::IpcMessage& reply, reply.set_param(fr_cnxn_str + FrameProcessorController::CONFIG_FR_RELEASE, frReleaseEndpoint_); // Loop over plugins and request current configuration from each + int64_t latest_metadata_ver = -1; std::map>::iterator iter; for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { reply.set_param("plugins/names[]", iter->first); iter->second->requestConfiguration(reply); + latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_config_metadata_ts()); } - - int64_t latest_metadata_ver = -1; - for (auto iter : plugins_) { - latest_metadata_ver = std::max( - latest_metadata_ver, - reply.get_param(iter.second->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION) - ); - } - reply.set_param(FrameProcessorController::METADATA_HASH, latest_metadata_ver); + reply.set_param(FrameProcessorController::METADATA_TS_KEY, latest_metadata_ver); if (metadata) { for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { diff --git a/cpp/frameProcessor/src/FrameProcessorPlugin.cpp b/cpp/frameProcessor/src/FrameProcessorPlugin.cpp index 097f3f157..416020791 100644 --- a/cpp/frameProcessor/src/FrameProcessorPlugin.cpp +++ b/cpp/frameProcessor/src/FrameProcessorPlugin.cpp @@ -20,6 +20,8 @@ FrameProcessorPlugin::FrameProcessorPlugin() : { OdinData::configure_logging_mdc(OdinData::app_path.c_str()); logger_ = log4cxx::Logger::getLogger("FP.FrameProcessorPlugin"); + update_config_metadata_version(); + update_status_metadata_version(); } /** diff --git a/cpp/frameProcessor/src/GapFillPlugin.cpp b/cpp/frameProcessor/src/GapFillPlugin.cpp index bb8d53585..fe7600344 100644 --- a/cpp/frameProcessor/src/GapFillPlugin.cpp +++ b/cpp/frameProcessor/src/GapFillPlugin.cpp @@ -29,7 +29,6 @@ GapFillPlugin::GapFillPlugin() add_config_param_metadata(GapFillPlugin::CONFIG_CHIP_SIZE, PMDD::INTARR_T, PMDA::READ_WRITE); add_config_param_metadata(GapFillPlugin::CONFIG_GRID_X_GAPS, PMDD::INTARR_T, PMDA::READ_WRITE); add_config_param_metadata(GapFillPlugin::CONFIG_GRID_Y_GAPS, PMDD::INTARR_T, PMDA::READ_WRITE); - update_config_metadata_version(); LOG4CXX_INFO(logger_, "GapFillPlugin version " << this->get_version_long() << " loaded"); } @@ -320,7 +319,6 @@ void GapFillPlugin::requestConfiguration(OdinData::IpcMessage& reply) for (int index = 0; index < gaps_y_.size(); index++) { reply.set_param(get_name() + '/' + CONFIG_GRID_Y_GAPS + "[]", gaps_y_[index]); } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } } // namespace FrameProcessor diff --git a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp index 3339c7838..4e78bc0fa 100644 --- a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp +++ b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp @@ -62,12 +62,10 @@ KafkaProducerPlugin::KafkaProducerPlugin() : add_config_param_metadata(CONFIG_PARTITION, PMDD::INT_T, PMDA::READ_WRITE, -1); add_config_param_metadata(CONFIG_DATASET, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_INCLUDE_PARAMETERS, PMDD::BOOL_T, PMDA::READ_WRITE); - update_config_metadata_version(); add_status_param_metadata(STATUS_FRAMES_SENT, PMDD::UINT_T, PMDA::READ_ONLY, 0); add_status_param_metadata(STATUS_FRAMES_LOST, PMDD::UINT_T, PMDA::READ_ONLY, 0); add_status_param_metadata(STATUS_FRAMES_ACK, PMDD::UINT_T, PMDA::READ_ONLY, 0); - update_status_metadata_version(); this->reset_statistics(); } @@ -128,7 +126,6 @@ void KafkaProducerPlugin::requestConfiguration(OdinData::IpcMessage& reply) reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_PARTITION, this->partition_); reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_DATASET, this->dataset_name_); reply.set_param(get_name() + '/' + KafkaProducerPlugin::CONFIG_INCLUDE_PARAMETERS, this->include_parameters_); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** @@ -146,7 +143,7 @@ void KafkaProducerPlugin::status(OdinData::IpcMessage& status) status.set_param(get_name() + '/' + STATUS_FRAMES_LOST, frames_lost_); /* Number of acknowledged frames */ status.set_param(get_name() + '/' + STATUS_FRAMES_ACK, frames_ack_); - status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_status_metadata_version()); + status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_TS_KEY, get_status_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/LiveViewPlugin.cpp b/cpp/frameProcessor/src/LiveViewPlugin.cpp index b2f74deeb..c50a0bd57 100644 --- a/cpp/frameProcessor/src/LiveViewPlugin.cpp +++ b/cpp/frameProcessor/src/LiveViewPlugin.cpp @@ -40,7 +40,6 @@ LiveViewPlugin::LiveViewPlugin() : add_config_param_metadata(CONFIG_SOCKET_ADDR, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_DATASET_NAME, PMDD::STRINGARR_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_TAGGED_FILTER_NAME, PMDD::STRINGARR_T, PMDA::READ_WRITE); - update_config_metadata_version(); set_frame_freq_config(DEFAULT_FRAME_FREQ); set_per_second_config(DEFAULT_PER_SECOND); @@ -198,7 +197,6 @@ void LiveViewPlugin::requestConfiguration(OdinData::IpcMessage& reply) reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_SOCKET_ADDR, image_view_socket_addr_); reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_PER_SECOND, per_second_); reply.set_param(get_name() + '/' + LiveViewPlugin::CONFIG_DATASET_NAME, dataset_names_); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** diff --git a/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp b/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp index 025488366..1f2eba10f 100644 --- a/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/OffsetAdjustmentPlugin.cpp @@ -20,7 +20,6 @@ OffsetAdjustmentPlugin::OffsetAdjustmentPlugin() : logger_ = Logger::getLogger("FP.OffsetAdjustmentPlugin"); LOG4CXX_INFO(logger_, "OffsetAdjustmentPlugin version " << this->get_version_long() << " loaded"); add_config_param_metadata(OFFSET_ADJUSTMENT_CONFIG, PMDD::INT_T, PMDA::READ_WRITE); - update_config_metadata_version(); } /** @@ -76,7 +75,6 @@ void OffsetAdjustmentPlugin::configure(OdinData::IpcMessage& config, OdinData::I void OffsetAdjustmentPlugin::requestConfiguration(OdinData::IpcMessage& reply) { reply.set_param(get_name() + '/' + OFFSET_ADJUSTMENT_CONFIG, offset_adjustment_.load()); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } int OffsetAdjustmentPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp index 1153762a2..7fe742360 100644 --- a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp @@ -159,7 +159,6 @@ void ParameterAdjustmentPlugin::requestConfiguration(OdinData::IpcMessage& reply input_iter->second ); } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } int ParameterAdjustmentPlugin::get_version_major() diff --git a/cpp/frameProcessor/src/ParameterPublishPlugin.cpp b/cpp/frameProcessor/src/ParameterPublishPlugin.cpp index 2dfb479e4..bada4ecd6 100644 --- a/cpp/frameProcessor/src/ParameterPublishPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterPublishPlugin.cpp @@ -24,7 +24,6 @@ ParameterPublishPlugin::ParameterPublishPlugin() : LOG4CXX_INFO(logger_, "ParameterPublishPlugin version " << this->get_version_long() << " loaded"); add_config_param_metadata(CONFIG_ENDPOINT, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(DATA_PARAMETERS, PMDD::STRINGARR_T, PMDA::READ_WRITE); - update_config_metadata_version(); } /** @@ -127,7 +126,6 @@ void ParameterPublishPlugin::requestConfiguration(OdinData::IpcMessage& reply) for (auto& it : this->parameters_) { reply.set_param(parameters_key, it); } - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** Bind to endpoint and store for config reporting diff --git a/cpp/frameProcessor/src/RawFileWriterPlugin.cpp b/cpp/frameProcessor/src/RawFileWriterPlugin.cpp index f8ea3ad3f..2c6d9def0 100644 --- a/cpp/frameProcessor/src/RawFileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/RawFileWriterPlugin.cpp @@ -25,9 +25,7 @@ RawFileWriterPlugin::RawFileWriterPlugin() : LOG4CXX_TRACE(logger_, "RawFileWriterPlugin constructor."); add_config_param_metadata(CONFIG_FILE_PATH, PMDD::STRING_T, PMDA::READ_WRITE); add_config_param_metadata(CONFIG_ENABLED, PMDD::BOOL_T, PMDA::READ_WRITE); - update_config_metadata_version(); add_status_param_metadata(STATUS_DROPPED_FRAMES, PMDD::UINT_T, PMDA::READ_ONLY, 0); - update_status_metadata_version(); } void RawFileWriterPlugin::process_frame(boost::shared_ptr frame) @@ -103,7 +101,6 @@ void RawFileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) { reply.set_param(this->get_name() + '/' + RawFileWriterPlugin::CONFIG_FILE_PATH, this->file_path_.string()); reply.set_param(this->get_name() + '/' + RawFileWriterPlugin::CONFIG_ENABLED, this->enabled_); - reply.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_config_metadata_version()); } /** Get status of the RawFileWriterPlugin @@ -113,7 +110,6 @@ void RawFileWriterPlugin::requestConfiguration(OdinData::IpcMessage& reply) void RawFileWriterPlugin::status(OdinData::IpcMessage& status) { status.set_param(this->get_name() + '/' + RawFileWriterPlugin::STATUS_DROPPED_FRAMES, this->dropped_frames_); - status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_VERSION, get_status_metadata_version()); } int RawFileWriterPlugin::get_version_major() From 93caa1573d552d21338a589ff69f3fcd4c789f32 Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Mon, 13 Jul 2026 12:35:49 +0000 Subject: [PATCH 04/10] Delete redundant call in kafkaPlugin. --- cpp/frameProcessor/src/KafkaProducerPlugin.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp index 4e78bc0fa..289715bae 100644 --- a/cpp/frameProcessor/src/KafkaProducerPlugin.cpp +++ b/cpp/frameProcessor/src/KafkaProducerPlugin.cpp @@ -143,7 +143,6 @@ void KafkaProducerPlugin::status(OdinData::IpcMessage& status) status.set_param(get_name() + '/' + STATUS_FRAMES_LOST, frames_lost_); /* Number of acknowledged frames */ status.set_param(get_name() + '/' + STATUS_FRAMES_ACK, frames_ack_); - status.set_param(this->get_name() + '/' + FrameProcessorPlugin::METADATA_TS_KEY, get_status_metadata_version()); } /** From 79a45b981544295cca2c1601246ab05857a35aa0 Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Tue, 14 Jul 2026 15:58:27 +0000 Subject: [PATCH 05/10] Minor Fixes --- cpp/frameProcessor/src/FileWriterPlugin.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index 8fdcdb587..6a2578e7f 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -168,7 +168,6 @@ FileWriterPlugin::FileWriterPlugin() : add_status_param_metadata(prefix + STATUS_LAST_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); add_status_param_metadata(prefix + STATUS_MAX_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); add_status_param_metadata(prefix + STATUS_MEAN_CLOSE, PMDD::UINT_T, PMDA::READ_ONLY, 0, PMD::MAX_UNSET); - update_status_metadata_version(); this->logger_ = Logger::getLogger("FP.FileWriterPlugin"); LOG4CXX_INFO(logger_, "FileWriterPlugin version " << this->get_version_long() << " loaded"); @@ -373,6 +372,7 @@ void FileWriterPlugin::configure(OdinData::IpcMessage& config, OdinData::IpcMess // Check to see if we are deleting all datasets if (config.has_param(FileWriterPlugin::CONFIG_DELETE_DATASETS)) { this->delete_datasets(); + update_config_metadata_version(); } // Check to see if we are being told how many frames to write From 8262ea3303b9784e54d2ff37cc3b609a4bc82f1c Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Wed, 15 Jul 2026 12:54:26 +0000 Subject: [PATCH 06/10] Unit Test for FileWriter and ParameterAdjustment Add unit test to validate that the metadata timestamp for FileWriter and ParameterAdjustment are updated When the triggering configuration is passed to it. --- .../include/FrameProcessorPlugin.h | 16 ++++++++-------- cpp/frameProcessor/src/FileWriterPlugin.cpp | 1 + cpp/frameProcessor/test/FileWriterPluginTest.cpp | 14 ++++++++++++++ .../test/ParameterAdjustmentPluginTest.cpp | 12 ++++++++++++ 4 files changed, 35 insertions(+), 8 deletions(-) diff --git a/cpp/frameProcessor/include/FrameProcessorPlugin.h b/cpp/frameProcessor/include/FrameProcessorPlugin.h index b3a38b2e6..c25d933ed 100644 --- a/cpp/frameProcessor/include/FrameProcessorPlugin.h +++ b/cpp/frameProcessor/include/FrameProcessorPlugin.h @@ -74,28 +74,28 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO void remove_all_callbacks(); void notify_end_of_acquisition(); /** return the latest ts of config/status metadata for this plugin instance */ - inline int64_t get_config_metadata_ts() const + __attribute__((always_inline)) int64_t get_config_metadata_ts() const { return this->config_metadata_ts_; } - inline int64_t get_status_metadata_ts() const + __attribute__((always_inline)) int64_t get_status_metadata_ts() const { return this->status_metadata_ts_; } protected: /** function to update the config time-stamp! */ - inline void update_config_metadata_version(void) + __attribute__((always_inline)) void update_config_metadata_version(void) { - auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); - auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); + auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); + auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); this->config_metadata_ts_ = val.count(); } /** function to update the status time-stamp! */ - inline void update_status_metadata_version(void) + __attribute__((always_inline)) void update_status_metadata_version(void) { - auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); - auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); + auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); + auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); this->status_metadata_ts_ = val.count(); } void push(boost::shared_ptr frame); diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index 6a2578e7f..020fd2d33 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -350,6 +350,7 @@ void FileWriterPlugin::configure(OdinData::IpcMessage& config, OdinData::IpcMess LOG4CXX_INFO(logger_, "Dataset name " << dataset_name << " found, creating..."); // If we can retrieve a single string parameter then we are being asked to create // a new dataset. Only create it if it doesn't already exist. + std::cout << "Creating new dataset\n"; create_new_dataset(dataset_name); } catch (OdinData::IpcMessageException& e) { // The object passed to us is a dataset description so pass to the configure_dataset method. diff --git a/cpp/frameProcessor/test/FileWriterPluginTest.cpp b/cpp/frameProcessor/test/FileWriterPluginTest.cpp index 1c8154b22..963e0d4f9 100644 --- a/cpp/frameProcessor/test/FileWriterPluginTest.cpp +++ b/cpp/frameProcessor/test/FileWriterPluginTest.cpp @@ -151,4 +151,18 @@ BOOST_AUTO_TEST_CASE(FileWriterPluginZeroFrames) } } +BOOST_AUTO_TEST_CASE(FileWriterPluginMetadataTimestampUpdate) +{ + FrameProcessor::FileWriterPlugin fwp; + int64_t first_ts = fwp.get_config_metadata_ts(); + OdinData::IpcMessage cfg; + OdinData::IpcMessage reply; + fwp.set_name("hdf"); + cfg.set_param(FrameProcessor::FileWriterPlugin::CONFIG_DATASET, "new_dataset"); + fwp.configure(cfg, reply); + int64_t second_ts = fwp.get_config_metadata_ts(); + std::cout << "The first is: " << first_ts << " the second is: " << second_ts << '\n'; + BOOST_CHECK(second_ts > first_ts); +} + BOOST_AUTO_TEST_SUITE_END(); diff --git a/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp b/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp index 4e53409cc..04fa5cd15 100644 --- a/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp +++ b/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp @@ -106,4 +106,16 @@ BOOST_AUTO_TEST_CASE(AdjustExistingParameter) BOOST_CHECK_EQUAL(14, frame->get_meta_data().get_parameter("UID")); } +BOOST_AUTO_TEST_CASE(ValidateMetadataTimestampUpdate) +{ + FrameProcessor::ParameterAdjustmentPlugin plugin; + OdinData::IpcMessage reply; + OdinData::IpcMessage cfg; + int64_t first_ts = plugin.get_config_metadata_ts(); + cfg.set_param(FrameProcessor::PARAMETER_NAME_CONFIG + "/UID/" + FrameProcessor::PARAMETER_ADJUSTMENT_CONFIG, 4); + plugin.configure(cfg, reply); + int64_t second_ts = plugin.get_config_metadata_ts(); + BOOST_CHECK(second_ts > first_ts); +} + BOOST_AUTO_TEST_SUITE_END(); // ParameterAdjustmentPluginUnitTest From 741028d083ac61368f280c08c9580b199af6e482 Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Wed, 15 Jul 2026 18:04:29 +0000 Subject: [PATCH 07/10] Minor Fix --- cpp/frameProcessor/src/FileWriterPlugin.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index 020fd2d33..6a2578e7f 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -350,7 +350,6 @@ void FileWriterPlugin::configure(OdinData::IpcMessage& config, OdinData::IpcMess LOG4CXX_INFO(logger_, "Dataset name " << dataset_name << " found, creating..."); // If we can retrieve a single string parameter then we are being asked to create // a new dataset. Only create it if it doesn't already exist. - std::cout << "Creating new dataset\n"; create_new_dataset(dataset_name); } catch (OdinData::IpcMessageException& e) { // The object passed to us is a dataset description so pass to the configure_dataset method. From e5e254da5ea77f029943de003cb9aee1bd122355 Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Thu, 16 Jul 2026 12:00:04 +0000 Subject: [PATCH 08/10] Minor Fix Remove errant std::cout statement. --- cpp/frameProcessor/test/FileWriterPluginTest.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/cpp/frameProcessor/test/FileWriterPluginTest.cpp b/cpp/frameProcessor/test/FileWriterPluginTest.cpp index 963e0d4f9..fb1a10f75 100644 --- a/cpp/frameProcessor/test/FileWriterPluginTest.cpp +++ b/cpp/frameProcessor/test/FileWriterPluginTest.cpp @@ -161,7 +161,6 @@ BOOST_AUTO_TEST_CASE(FileWriterPluginMetadataTimestampUpdate) cfg.set_param(FrameProcessor::FileWriterPlugin::CONFIG_DATASET, "new_dataset"); fwp.configure(cfg, reply); int64_t second_ts = fwp.get_config_metadata_ts(); - std::cout << "The first is: " << first_ts << " the second is: " << second_ts << '\n'; BOOST_CHECK(second_ts > first_ts); } From 7b93502ffe2898ca6da01784de7f2c2576a6dd7a Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Thu, 16 Jul 2026 16:20:45 +0000 Subject: [PATCH 09/10] Minor Fix: Update Methods and Variable Names. --- .../include/FrameProcessorPlugin.h | 24 +++++++++---------- cpp/frameProcessor/src/FileWriterPlugin.cpp | 4 ++-- .../src/FrameProcessorController.cpp | 4 ++-- .../src/FrameProcessorPlugin.cpp | 4 ++-- .../src/ParameterAdjustmentPlugin.cpp | 2 +- .../test/FileWriterPluginTest.cpp | 4 ++-- .../test/ParameterAdjustmentPluginTest.cpp | 4 ++-- 7 files changed, 23 insertions(+), 23 deletions(-) diff --git a/cpp/frameProcessor/include/FrameProcessorPlugin.h b/cpp/frameProcessor/include/FrameProcessorPlugin.h index c25d933ed..f228faf1f 100644 --- a/cpp/frameProcessor/include/FrameProcessorPlugin.h +++ b/cpp/frameProcessor/include/FrameProcessorPlugin.h @@ -74,29 +74,29 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO void remove_all_callbacks(); void notify_end_of_acquisition(); /** return the latest ts of config/status metadata for this plugin instance */ - __attribute__((always_inline)) int64_t get_config_metadata_ts() const + __attribute__((always_inline)) int64_t get_config_ts() const { - return this->config_metadata_ts_; + return this->config_ts_; } - __attribute__((always_inline)) int64_t get_status_metadata_ts() const + __attribute__((always_inline)) int64_t get_status_ts() const { - return this->status_metadata_ts_; + return this->status_ts_; } protected: /** function to update the config time-stamp! */ - __attribute__((always_inline)) void update_config_metadata_version(void) + __attribute__((always_inline)) void update_config_ts(void) { auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); - this->config_metadata_ts_ = val.count(); + this->config_ts_ = val.count(); } /** function to update the status time-stamp! */ - __attribute__((always_inline)) void update_status_metadata_version(void) + __attribute__((always_inline)) void update_status_ts(void) { auto time_stamp_ms = std::chrono::time_point_cast(std::chrono::steady_clock::now()); auto val = std::chrono::duration_cast(time_stamp_ms.time_since_epoch()); - this->status_metadata_ts_ = val.count(); + this->status_ts_ = val.count(); } void push(boost::shared_ptr frame); void push(const std::string& plugin_name, boost::shared_ptr frame); @@ -265,10 +265,10 @@ class FrameProcessorPlugin : public IFrameCallback, public OdinData::IVersionedO boost::mutex mutex_; /** process_frame performance stats */ CallDuration process_duration_; - /** time-stamp to represent the version of config metadata */ - int64_t config_metadata_ts_; - /** time-stamp to represent the version of status metadata */ - int64_t status_metadata_ts_; + /** time-stamp to represent the version of config structure */ + int64_t config_ts_; + /** time-stamp to represent the version of status structure */ + int64_t status_ts_; }; } /* namespace FrameProcessor */ diff --git a/cpp/frameProcessor/src/FileWriterPlugin.cpp b/cpp/frameProcessor/src/FileWriterPlugin.cpp index 6a2578e7f..e322dcc7d 100644 --- a/cpp/frameProcessor/src/FileWriterPlugin.cpp +++ b/cpp/frameProcessor/src/FileWriterPlugin.cpp @@ -366,13 +366,13 @@ void FileWriterPlugin::configure(OdinData::IpcMessage& config, OdinData::IpcMess this->configure_dataset(dataset_name, dsetConfig, reply); } } - update_config_metadata_version(); + update_config_ts(); } // Check to see if we are deleting all datasets if (config.has_param(FileWriterPlugin::CONFIG_DELETE_DATASETS)) { this->delete_datasets(); - update_config_metadata_version(); + update_config_ts(); } // Check to see if we are being told how many frames to write diff --git a/cpp/frameProcessor/src/FrameProcessorController.cpp b/cpp/frameProcessor/src/FrameProcessorController.cpp index 5fc979954..4f3c49640 100644 --- a/cpp/frameProcessor/src/FrameProcessorController.cpp +++ b/cpp/frameProcessor/src/FrameProcessorController.cpp @@ -309,7 +309,7 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m // Request status for the plugin iter->second->status(reply); // Check for the latest timestamp - latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_status_metadata_ts()); + latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_status_ts()); // Add performance statistics iter->second->add_performance_stats(reply); // Read error level @@ -540,7 +540,7 @@ void FrameProcessorController::requestConfiguration(OdinData::IpcMessage& reply, for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { reply.set_param("plugins/names[]", iter->first); iter->second->requestConfiguration(reply); - latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_config_metadata_ts()); + latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_config_ts()); } reply.set_param(FrameProcessorController::METADATA_TS_KEY, latest_metadata_ver); diff --git a/cpp/frameProcessor/src/FrameProcessorPlugin.cpp b/cpp/frameProcessor/src/FrameProcessorPlugin.cpp index 416020791..fbec49b61 100644 --- a/cpp/frameProcessor/src/FrameProcessorPlugin.cpp +++ b/cpp/frameProcessor/src/FrameProcessorPlugin.cpp @@ -20,8 +20,8 @@ FrameProcessorPlugin::FrameProcessorPlugin() : { OdinData::configure_logging_mdc(OdinData::app_path.c_str()); logger_ = log4cxx::Logger::getLogger("FP.FrameProcessorPlugin"); - update_config_metadata_version(); - update_status_metadata_version(); + update_config_ts(); + update_status_ts(); } /** diff --git a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp index 7fe742360..bf9d7c275 100644 --- a/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp +++ b/cpp/frameProcessor/src/ParameterAdjustmentPlugin.cpp @@ -127,7 +127,7 @@ void ParameterAdjustmentPlugin::configure(OdinData::IpcMessage& config, OdinData parameter_adjustments_.clear(); parameter_inputs_.clear(); } - update_config_metadata_version(); + update_config_ts(); } } catch (std::runtime_error& e) { std::stringstream ss; diff --git a/cpp/frameProcessor/test/FileWriterPluginTest.cpp b/cpp/frameProcessor/test/FileWriterPluginTest.cpp index fb1a10f75..1b85cad30 100644 --- a/cpp/frameProcessor/test/FileWriterPluginTest.cpp +++ b/cpp/frameProcessor/test/FileWriterPluginTest.cpp @@ -154,13 +154,13 @@ BOOST_AUTO_TEST_CASE(FileWriterPluginZeroFrames) BOOST_AUTO_TEST_CASE(FileWriterPluginMetadataTimestampUpdate) { FrameProcessor::FileWriterPlugin fwp; - int64_t first_ts = fwp.get_config_metadata_ts(); + int64_t first_ts = fwp.get_config_ts(); OdinData::IpcMessage cfg; OdinData::IpcMessage reply; fwp.set_name("hdf"); cfg.set_param(FrameProcessor::FileWriterPlugin::CONFIG_DATASET, "new_dataset"); fwp.configure(cfg, reply); - int64_t second_ts = fwp.get_config_metadata_ts(); + int64_t second_ts = fwp.get_config_ts(); BOOST_CHECK(second_ts > first_ts); } diff --git a/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp b/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp index 04fa5cd15..7c12375ff 100644 --- a/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp +++ b/cpp/frameProcessor/test/ParameterAdjustmentPluginTest.cpp @@ -111,10 +111,10 @@ BOOST_AUTO_TEST_CASE(ValidateMetadataTimestampUpdate) FrameProcessor::ParameterAdjustmentPlugin plugin; OdinData::IpcMessage reply; OdinData::IpcMessage cfg; - int64_t first_ts = plugin.get_config_metadata_ts(); + int64_t first_ts = plugin.get_config_ts(); cfg.set_param(FrameProcessor::PARAMETER_NAME_CONFIG + "/UID/" + FrameProcessor::PARAMETER_ADJUSTMENT_CONFIG, 4); plugin.configure(cfg, reply); - int64_t second_ts = plugin.get_config_metadata_ts(); + int64_t second_ts = plugin.get_config_ts(); BOOST_CHECK(second_ts > first_ts); } From 51fb2b8c2cc6b0e67c00e1e7c24875ec39d52b9f Mon Sep 17 00:00:00 2001 From: Ohisemega Date: Fri, 17 Jul 2026 08:01:25 +0000 Subject: [PATCH 10/10] Rename Variables Add seperate keys for the timestamp of status and config. Rename local variables. --- .../include/FrameProcessorController.h | 5 +++-- .../src/FrameProcessorController.cpp | 15 ++++++++------- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/cpp/frameProcessor/include/FrameProcessorController.h b/cpp/frameProcessor/include/FrameProcessorController.h index 8cdd28c55..fbfddf583 100644 --- a/cpp/frameProcessor/include/FrameProcessorController.h +++ b/cpp/frameProcessor/include/FrameProcessorController.h @@ -77,8 +77,9 @@ class FrameProcessorController : public IFrameCallback, /** Configuration constant for executing setup of shared memory interface **/ static const std::string CONFIG_FR_SETUP; - /** latest metadata timestamp **/ - static const std::string METADATA_TS_KEY; + /** key-strings for latest config and status timestamp **/ + static const std::string CONFIG_TS_KEY; + static const std::string STATUS_TS_KEY; /** Configuration constant for control socket endpoint **/ static const std::string CONFIG_CTRL_ENDPOINT; diff --git a/cpp/frameProcessor/src/FrameProcessorController.cpp b/cpp/frameProcessor/src/FrameProcessorController.cpp index 4f3c49640..1bd2cfe1a 100644 --- a/cpp/frameProcessor/src/FrameProcessorController.cpp +++ b/cpp/frameProcessor/src/FrameProcessorController.cpp @@ -43,7 +43,8 @@ const std::string FrameProcessorController::CONFIG_VALUE = "value"; const std::string FrameProcessorController::COMMAND_KEY = "command"; const std::string FrameProcessorController::SUPPORTED_KEY = "supported"; -const std::string FrameProcessorController::METADATA_TS_KEY = "metadata_ts"; +const std::string FrameProcessorController::CONFIG_TS_KEY = "config_ts"; +const std::string FrameProcessorController::STATUS_TS_KEY = "status_ts"; const int FrameProcessorController::META_TX_HWM = 10000; @@ -302,14 +303,14 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m } } - int64_t latest_metadata_ver = -1; + int64_t latest_ts = -1; // Loop over plugins, list names and request status from each for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { reply.set_param("plugins/names[]", iter->first); // Request status for the plugin iter->second->status(reply); // Check for the latest timestamp - latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_status_ts()); + latest_ts = std::max(latest_ts, iter->second->get_status_ts()); // Add performance statistics iter->second->add_performance_stats(reply); // Read error level @@ -319,7 +320,7 @@ void FrameProcessorController::provideStatus(OdinData::IpcMessage& reply, bool m std::vector plugin_warnings = iter->second->get_warnings(); warning_messages.insert(warning_messages.end(), plugin_warnings.begin(), plugin_warnings.end()); } - reply.set_param(FrameProcessorController::METADATA_TS_KEY, latest_metadata_ver); + reply.set_param(FrameProcessorController::STATUS_TS_KEY, latest_ts); std::vector::iterator error_iter; for (error_iter = error_messages.begin(); error_iter != error_messages.end(); ++error_iter) { @@ -535,14 +536,14 @@ void FrameProcessorController::requestConfiguration(OdinData::IpcMessage& reply, reply.set_param(fr_cnxn_str + FrameProcessorController::CONFIG_FR_RELEASE, frReleaseEndpoint_); // Loop over plugins and request current configuration from each - int64_t latest_metadata_ver = -1; + int64_t latest_ts = -1; std::map>::iterator iter; for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) { reply.set_param("plugins/names[]", iter->first); iter->second->requestConfiguration(reply); - latest_metadata_ver = std::max(latest_metadata_ver, iter->second->get_config_ts()); + latest_ts = std::max(latest_ts, iter->second->get_config_ts()); } - reply.set_param(FrameProcessorController::METADATA_TS_KEY, latest_metadata_ver); + reply.set_param(FrameProcessorController::CONFIG_TS_KEY, latest_ts); if (metadata) { for (iter = plugins_.begin(); iter != plugins_.end(); ++iter) {