From b793af5607ad993304f84a1ac13bcfa5f2cdb0d7 Mon Sep 17 00:00:00 2001 From: Michael Jackson Date: Wed, 5 Aug 2026 11:05:22 -0400 Subject: [PATCH 01/12] ENH: Add named MessageHandler send methods and ThrottledMessageHandler First step of the messaging API overhaul. Adds the new API alongside the existing operator() overloads so the tree keeps building; call sites are migrated separately. * Fold ProgressMessage into Message as an int32 progress field, defaulting to -1 for messages that carry no percent. ProgressMessage becomes an alias so existing aggregate initializations keep compiling. * Add sendMessage/sendInfoMessage/sendDebugMessage/sendWarningMessage/ sendErrorMessage to MessageHandler. * Report progress through two intent-bearing calls rather than one generic one, so a filter records what it meant to display. sendProgressCount() renders "label: 35/100" for loops whose counts mean something to a user; sendProgressPercent() renders "label: 33.33%" for loops whose bounds are too large for counts to be readable. Overloading on numeric type was rejected because an integer literal would silently pick a form, hiding the intent. sendProgressMessage() remains for pre-rendered text. * The sender renders the display text, since "35/100" contains no percent to append. Message::progress carries only the number a progress bar needs, which is the pair PipelineFilter and DREAM3DNX already consume. The bar value clamps to [0, 100] and a zero denominator yields 0; the rendered text is not clamped, so an overshooting loop honestly reports 150.00% while the bar stays pinned at 100. * Add ThrottledMessageHandler, which rate-limits progress and status messages using a per-instance thread that opens an atomic_bool gate once per interval. The loop body only reads the flag, measured at 0.25 ns per iteration against 14.7 ns for a steady_clock::now() call. Written as a relaxed load followed by an exchange, because a bare exchange every iteration is an unconditional read-modify-write on a shared cache line and costs 2.5x more. Messages are formatted only when one is due, so throttled loops do not allocate. Exposes updateCount/updatePercent and incrementCount/incrementPercent mirroring the two display intents. The class is not internally thread-safe; callers serialize it. * Drop the ProgressMessage downcast in PipelineFilter::notifyFilterMessage and read the progress field directly. * Forward warning and error message text instead of discarding it. Only the fault state was previously sent, so the reason for a fault was lost. * Remove three PipelineObserver connections to Pipeline-level filter signals that have no emitter, and the handlers they targeted. Only the per-filter connections fire: Pipeline::onNotify re-wraps into a PipelineNodeMessage and emits on the generic signal, never the typed ones. Verified by running a pipeline before and after. Also removes the misaligned argument binding on the progress signal, which read filterIndex as progress. * Remove PipelineFilterMessage, which was never constructed. * Update the Python bindings for the merged Message type and add snake_case send methods. IFilter.ProgressMessage is retained as a static factory. * Add 26 tests covering the send methods, both progress display forms and their decimal handling, percent clamping, zero denominators, the throttle gate, cross-thread sender election, counter accumulation, prompt destruction, and message routing. Signed-off-by: Michael Jackson --- CMakeLists.txt | 4 +- .../SimplnxCore/wrapping/python/simplnxpy.cpp | 32 +- src/nxrunner/src/CliObserver.cpp | 31 -- src/nxrunner/src/CliObserver.hpp | 6 - src/simplnx/Filter/IFilter.cpp | 37 +++ src/simplnx/Filter/IFilter.hpp | 104 +++++- .../Messaging/PipelineFilterMessage.cpp | 47 --- .../Messaging/PipelineFilterMessage.hpp | 42 --- src/simplnx/Pipeline/PipelineFilter.cpp | 29 +- .../Utilities/ThrottledMessageHandler.cpp | 105 +++++++ .../Utilities/ThrottledMessageHandler.hpp | 140 +++++++++ test/CMakeLists.txt | 3 + test/MessageHandlerTest.cpp | 179 +++++++++++ test/PipelineFilterMessagingTest.cpp | 108 +++++++ test/ThrottledMessageHandlerTest.cpp | 295 ++++++++++++++++++ 15 files changed, 1003 insertions(+), 159 deletions(-) delete mode 100644 src/simplnx/Pipeline/Messaging/PipelineFilterMessage.cpp delete mode 100644 src/simplnx/Pipeline/Messaging/PipelineFilterMessage.hpp create mode 100644 src/simplnx/Utilities/ThrottledMessageHandler.cpp create mode 100644 src/simplnx/Utilities/ThrottledMessageHandler.hpp create mode 100644 test/MessageHandlerTest.cpp create mode 100644 test/PipelineFilterMessagingTest.cpp create mode 100644 test/ThrottledMessageHandlerTest.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 10680552e8..dc03a5cf92 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -539,7 +539,6 @@ set(SIMPLNX_HDRS ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/NodeRemovedMessage.hpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/NodeStatusMessage.hpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/OutputRenamedMessage.hpp - ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/PipelineFilterMessage.hpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/PipelineNodeMessage.hpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/PipelineNodeObserver.hpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/RenamedMessage.hpp @@ -568,6 +567,7 @@ set(SIMPLNX_HDRS ${SIMPLNX_SOURCE_DIR}/Utilities/MaskCompareUtilities.hpp ${SIMPLNX_SOURCE_DIR}/Utilities/MemoryUtilities.hpp ${SIMPLNX_SOURCE_DIR}/Utilities/MessageHelper.hpp + ${SIMPLNX_SOURCE_DIR}/Utilities/ThrottledMessageHandler.hpp ${SIMPLNX_SOURCE_DIR}/Utilities/StringUtilities.hpp ${SIMPLNX_SOURCE_DIR}/Utilities/StringInterpretationUtilities.hpp ${SIMPLNX_SOURCE_DIR}/Utilities/IntersectionUtilities.hpp @@ -761,7 +761,6 @@ set(SIMPLNX_SRCS ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/NodeRemovedMessage.cpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/NodeStatusMessage.cpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/OutputRenamedMessage.cpp - ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/PipelineFilterMessage.cpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/PipelineNodeMessage.cpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/PipelineNodeObserver.cpp ${SIMPLNX_SOURCE_DIR}/Pipeline/Messaging/RenamedMessage.cpp @@ -783,6 +782,7 @@ set(SIMPLNX_SRCS ${SIMPLNX_SOURCE_DIR}/Utilities/MaskCompareUtilities.cpp ${SIMPLNX_SOURCE_DIR}/Utilities/MemoryUtilities.cpp ${SIMPLNX_SOURCE_DIR}/Utilities/MessageHelper.cpp + ${SIMPLNX_SOURCE_DIR}/Utilities/ThrottledMessageHandler.cpp ${SIMPLNX_SOURCE_DIR}/Utilities/IParallelAlgorithm.cpp ${SIMPLNX_SOURCE_DIR}/Utilities/ParallelDataAlgorithm.cpp ${SIMPLNX_SOURCE_DIR}/Utilities/ParallelData2DAlgorithm.cpp diff --git a/src/Plugins/SimplnxCore/wrapping/python/simplnxpy.cpp b/src/Plugins/SimplnxCore/wrapping/python/simplnxpy.cpp index ab82dc4cbd..b6c4ddad76 100644 --- a/src/Plugins/SimplnxCore/wrapping/python/simplnxpy.cpp +++ b/src/Plugins/SimplnxCore/wrapping/python/simplnxpy.cpp @@ -1482,24 +1482,34 @@ PYBIND11_MODULE(simplnx, mod) filterMessage.def(py::init<>()); filterMessage.def(py::init()); + filterMessage.def(py::init([](IFilter::Message::Type type, std::string message, int32 progress) { return IFilter::Message{type, std::move(message), progress}; }), "type"_a, "message"_a, + "progress"_a); filterMessage.def_readwrite("type", &IFilter::Message::type); filterMessage.def_readwrite("message", &IFilter::Message::message); + filterMessage.def_readwrite("progress", &IFilter::Message::progress); - py::class_ progressMessage(filter, "ProgressMessage"); - progressMessage.def(py::init([](std::string message, int32 progress) { - IFilter::ProgressMessage progressMessage_; - progressMessage_.type = IFilter::Message::Type::Progress; - progressMessage_.message = std::move(message); - progressMessage_.progress = progress; - return progressMessage_; - }), - "message"_a, "progress"_a); - progressMessage.def_readwrite("progress", &IFilter::ProgressMessage::progress); + // Progress information now lives on Message itself, so ProgressMessage is no longer a distinct + // type. This factory keeps the IFilter.ProgressMessage(message, progress) call shape working. + filter.def_static( + "ProgressMessage", [](std::string message, int32 progress) { return IFilter::Message{IFilter::Message::Type::Progress, std::move(message), progress}; }, "message"_a, "progress"_a); py::class_ messageHandler(filter, "MessageHandler"); messageHandler.def(py::init<>()); messageHandler.def_readwrite("callback", &IFilter::MessageHandler::m_Callback); - messageHandler.def("__call__", [](const IFilter::MessageHandler& self, const IFilter::Message& message) { self(message); }); + messageHandler.def("send_message", [](const IFilter::MessageHandler& self, const IFilter::Message& message) { self.sendMessage(message); }, "message"_a); + messageHandler.def("send_info_message", [](const IFilter::MessageHandler& self, std::string message) { self.sendInfoMessage(std::move(message)); }, "message"_a); + messageHandler.def("send_debug_message", [](const IFilter::MessageHandler& self, std::string message) { self.sendDebugMessage(std::move(message)); }, "message"_a); + messageHandler.def("send_warning_message", [](const IFilter::MessageHandler& self, std::string message) { self.sendWarningMessage(std::move(message)); }, "message"_a); + messageHandler.def("send_error_message", [](const IFilter::MessageHandler& self, std::string message) { self.sendErrorMessage(std::move(message)); }, "message"_a); + messageHandler.def( + "send_progress_message", [](const IFilter::MessageHandler& self, std::string message, int32 percent) { self.sendProgressMessage(std::move(message), percent); }, "message"_a, "percent"_a); + messageHandler.def( + "send_progress_count", [](const IFilter::MessageHandler& self, std::string label, usize current, usize max) { self.sendProgressCount(std::move(label), current, max); }, "label"_a, "current"_a, + "max"_a); + messageHandler.def( + "send_progress_percent", [](const IFilter::MessageHandler& self, std::string label, usize current, usize max, int32 decimals) { self.sendProgressPercent(std::move(label), current, max, decimals); }, + "label"_a, "current"_a, "max"_a, "decimals"_a = 2); + messageHandler.def("__call__", [](const IFilter::MessageHandler& self, const IFilter::Message& message) { self.sendMessage(message); }); py::class_ preflightValue(filter, "PreflightValue"); preflightValue.def(py::init<>()); diff --git a/src/nxrunner/src/CliObserver.cpp b/src/nxrunner/src/CliObserver.cpp index 89f9540f92..3f8ba69938 100644 --- a/src/nxrunner/src/CliObserver.cpp +++ b/src/nxrunner/src/CliObserver.cpp @@ -16,9 +16,6 @@ PipelineObserver::PipelineObserver(Pipeline* pipeline) { startObservingNode(pipeline); pipeline->getCancelledSignal().connect([this](void) { onCancelled(); }); - pipeline->getFilterProgressSignal().connect([this](AbstractPipelineNode* node, int32_t progress, int32_t max, const std::string& msg) { onFilterProgress(node, progress, max, msg); }); - pipeline->getFilterRunStateSignal().connect([this](AbstractPipelineNode* node, int32_t index, RunState state) { onRunStateChanged(node, state); }); - pipeline->getFilterUpdateSignal().connect([this](AbstractPipelineNode* node, int32 index, const std::string& msg) { onFilterUpdate(node, msg); }); pipeline->getPipelineFaultSignal().connect([this](AbstractPipelineNode* node, FaultState state) { onFaultStateChanged(node, state); }); } if(pipeline == nullptr) @@ -83,34 +80,6 @@ void PipelineObserver::onCancelled() const std::cout << timestamp() << " Pipeline has been cancelled" << std::endl; } -void PipelineObserver::onFilterProgress(AbstractPipelineNode* node, int32 progress, int32 maxProgress, const std::string& msg) const -{ - std::cout << fmt::format("{} ({} / {}): {}", node->getName(), progress, maxProgress, msg) << std::endl; -} - -void PipelineObserver::onRunStateChanged(AbstractPipelineNode* node, RunState state) const -{ - switch(state) - { - case RunState::Executing: - std::cout << timestamp() << fmt::format(" {} has begun executing", node->getName()) << std::endl; - break; - case RunState::Preflighting: - std::cout << timestamp() << fmt::format(" {} has begun preflighting", node->getName()) << std::endl; - break; - case RunState::Idle: - std::cout << timestamp() << fmt::format(" {} has completed", node->getName()) << std::endl; - break; - case RunState::Queued: - break; - } -} - -void PipelineObserver::onFilterUpdate(AbstractPipelineNode* node, const std::string& msg) const -{ - std::cout << fmt::format("{}: {}", node->getName(), msg) << std::endl; -} - void PipelineObserver::onFaultStateChanged(AbstractPipelineNode* node, FaultState state) const { switch(state) diff --git a/src/nxrunner/src/CliObserver.hpp b/src/nxrunner/src/CliObserver.hpp index 21e1f57b56..9543e2cca8 100644 --- a/src/nxrunner/src/CliObserver.hpp +++ b/src/nxrunner/src/CliObserver.hpp @@ -27,12 +27,6 @@ class PipelineObserver : public PipelineNodeObserver void onCancelled() const; - void onFilterProgress(AbstractPipelineNode* node, int32 progress, int32 maxProgress, const std::string& msg) const; - - void onRunStateChanged(AbstractPipelineNode* node, RunState state) const; - - void onFilterUpdate(AbstractPipelineNode* node, const std::string& msg) const; - void onFaultStateChanged(AbstractPipelineNode* node, FaultState state) const; private: diff --git a/src/simplnx/Filter/IFilter.cpp b/src/simplnx/Filter/IFilter.cpp index b06536e26b..8aad9212ad 100644 --- a/src/simplnx/Filter/IFilter.cpp +++ b/src/simplnx/Filter/IFilter.cpp @@ -7,11 +7,48 @@ #include #include +#include #include #include using namespace nx::core; +namespace +{ +/** + * @brief Computes the integer percent used to drive a progress bar, clamped to [0, 100]. A zero + * denominator yields 0 rather than dividing by zero. + * @param current + * @param max + * @return + */ +int32 CalculateProgressBarValue(usize current, usize max) +{ + if(max == 0) + { + return 0; + } + auto percent = static_cast(static_cast(current) / static_cast(max) * 100.0); + return std::clamp(percent, 0, 100); +} +} // namespace + +void IFilter::MessageHandler::sendProgressMessage(std::string message, int32 percent) const +{ + sendMessage(Message{Message::Type::Progress, std::move(message), std::clamp(percent, 0, 100)}); +} + +void IFilter::MessageHandler::sendProgressCount(std::string label, usize current, usize max) const +{ + sendMessage(Message{Message::Type::Progress, fmt::format("{}: {}/{}", label, current, max), CalculateProgressBarValue(current, max)}); +} + +void IFilter::MessageHandler::sendProgressPercent(std::string label, usize current, usize max, int32 decimals) const +{ + const float64 percent = (max == 0) ? 0.0 : (static_cast(current) / static_cast(max) * 100.0); + sendMessage(Message{Message::Type::Progress, fmt::format("{}: {:.{}f}%", label, percent, decimals), CalculateProgressBarValue(current, max)}); +} + namespace { template diff --git a/src/simplnx/Filter/IFilter.hpp b/src/simplnx/Filter/IFilter.hpp index 23e1745036..713997245d 100644 --- a/src/simplnx/Filter/IFilter.hpp +++ b/src/simplnx/Filter/IFilter.hpp @@ -50,15 +50,21 @@ class SIMPLNX_EXPORT IFilter Type type = Type::Info; std::string message; + + /** + * @brief Percent complete, clamped to [0, 100], for driving a progress bar. A value of -1 means + * this message carries no progress value. The human-readable form of the progress lives in + * `message`, which is already rendered by the sender; this field exists only for consumers that + * need a number. + */ + int32 progress = -1; }; /** - * @brief Extends Message to include progress information. + * @brief Retained as an alias so existing aggregate initializations keep compiling. Progress + * information now lives on Message itself. */ - struct ProgressMessage : public Message - { - int32 progress = 0; - }; + using ProgressMessage = Message; /** * @brief Handler for processing filter messages during execution. @@ -68,10 +74,10 @@ class SIMPLNX_EXPORT IFilter using Callback = std::function; /** - * @brief Invokes the callback with a message. + * @brief Sends a message. * @param message The message to send */ - void operator()(const Message& message) const + void sendMessage(const Message& message) const { if(m_Callback) { @@ -79,6 +85,90 @@ class SIMPLNX_EXPORT IFilter } } + /** + * @brief Sends a message of the given type. + * @param type The message type + * @param message The message text + */ + void sendMessage(Message::Type type, std::string message) const + { + sendMessage(Message{type, std::move(message)}); + } + + /** + * @brief Sends an informational message. + * @param message The message text + */ + void sendInfoMessage(std::string message) const + { + sendMessage(Message{Message::Type::Info, std::move(message)}); + } + + /** + * @brief Sends a debug message. + * @param message The message text + */ + void sendDebugMessage(std::string message) const + { + sendMessage(Message{Message::Type::Debug, std::move(message)}); + } + + /** + * @brief Sends a warning message. + * @param message The message text + */ + void sendWarningMessage(std::string message) const + { + sendMessage(Message{Message::Type::Warning, std::move(message)}); + } + + /** + * @brief Sends an error message. + * @param message The message text + */ + void sendErrorMessage(std::string message) const + { + sendMessage(Message{Message::Type::Error, std::move(message)}); + } + + /** + * @brief Sends a progress message whose text has already been rendered by the caller. Prefer + * sendProgressCount() or sendProgressPercent(), which record what the sender intended to + * display; reach for this only when neither form fits. + * @param message The fully rendered message text + * @param percent Percent complete for the progress bar, clamped to [0, 100] + */ + void sendProgressMessage(std::string message, int32 percent) const; + + /** + * @brief Sends progress as a count of completed items, rendered as "