diff --git a/src/datadog/impl/core/context.cpp b/src/datadog/impl/core/context.cpp index cd16055c..575fb939 100644 --- a/src/datadog/impl/core/context.cpp +++ b/src/datadog/impl/core/context.cpp @@ -12,6 +12,7 @@ #include "datadog/uuid.hpp" #include "datadog/impl/assert.hpp" +#include "datadog/impl/core/message_bus.hpp" #include "datadog/impl/core/version.hpp" namespace datadog::impl { @@ -187,11 +188,24 @@ CoreContext CoreContextProvider::Get() const { } void CoreContextProvider::Update(const std::function& callback) { - // Acquire an exclusive write lock, then allow the callback to mutate the context - std::unique_lock lock(_mutex); - callback(_context); + // Acquire an exclusive write lock, mutate the context, and capture a snapshot. The + // lock is released before Send() to minimize contention on the message bus. + CoreContext snapshot = [&] { + std::unique_lock lock(_mutex); + callback(_context); + return _context; + }(); + + // CoreContext has been mutated (and we have a valid snapshot): notify all registered + // message-handlers of the change, so that Features can perform work (on the messaging + // thread) in response to the updated context + if (_message_bus) { + _message_bus->Send(ContextChangedMessage{std::move(snapshot)}); + } } +void CoreContextProvider::SetMessageBus(MessageBus* bus) { _message_bus = bus; } + const HttpContext& CoreContextProvider::GetHttpContext() const { std::shared_lock lock(_mutex); return *_context.http; diff --git a/src/datadog/impl/core/context.hpp b/src/datadog/impl/core/context.hpp index d66c6407..8ac38ac1 100644 --- a/src/datadog/impl/core/context.hpp +++ b/src/datadog/impl/core/context.hpp @@ -24,6 +24,8 @@ struct DeviceInfo; namespace datadog::impl { +class MessageBus; + /** * SDK configuration details that influence how HTTP requests are built. Immutable for * the lifetime of the Core. @@ -159,6 +161,7 @@ class CoreContextProvider { private: CoreContext _context; mutable std::shared_mutex _mutex; + MessageBus* _message_bus{nullptr}; public: explicit CoreContextProvider(const CoreContext& context); @@ -171,6 +174,10 @@ class CoreContextProvider { /** * Mutates the current CoreContext value, by invoking the provided callback after * obtaining exclusive write access. + * + * After mutation, dispatches a `ContextChangedMessage` on the `MessageBus`, carrying + * an immutable snapshot of the updated context to any Features that have registered + * an interest in receiving notifications of context changes. */ void Update(const std::function& callback); @@ -178,6 +185,14 @@ class CoreContextProvider { * Synchronously returns an immutable reference to the HttpContext value. */ const HttpContext& GetHttpContext() const; + + /** + * Registers the `MessageBus` that `Update()` will dispatch `ContextChangedMessage` + * values to. Pass `nullptr` to detach the bus (e.g. during shutdown). Called by + * `Core::Start()` before the context thread launches; no synchronization is needed + * because the context thread is not yet running at that point. + */ + void SetMessageBus(MessageBus* bus); }; } // namespace datadog::impl diff --git a/src/datadog/impl/core/core.cpp b/src/datadog/impl/core/core.cpp index ac976702..13957fe9 100644 --- a/src/datadog/impl/core/core.cpp +++ b/src/datadog/impl/core/core.cpp @@ -13,6 +13,7 @@ #include "datadog/impl/assert.hpp" #include "datadog/impl/core/context_thread.hpp" +#include "datadog/impl/core/messaging_thread.hpp" #include "datadog/impl/core/storage_thread.hpp" #include "datadog/impl/core/types.hpp" #include "datadog/impl/core/version.hpp" @@ -266,6 +267,7 @@ bool Core::RegisterFeature(const std::shared_ptr& impl) { std::move(*event_read_directory), std::move(upload_state) ); + _diagnostic_logger.Debug( "Feature registered", {{"feature", name}, {"feature_id", static_cast(id)}} @@ -320,8 +322,29 @@ bool Core::Start() { std::ref(_features) ); - // Initialize a thread-safe queue for functions that will execute on the context - // thread + // Iterate through all registered features and call MakeMessageHandler(), allowing + // features to register their intent to be notified when messages are sent on the + // message bus + std::vector> handlers; + handlers.reserve(_features.size()); + for (const auto& f : _features) { + if (auto h = f.impl->MakeMessageHandler()) { + handlers.push_back(std::move(*h)); + } + } + + // Create the message bus, which will queue all core-to-feature and feature-to-feature + // messages that need to be handled by interested features + DATADOG_ASSERT(!_message_bus, "_message_bus already exists on Start()"); + _message_bus = std::make_unique(std::move(handlers)); + + // Install the MessageBus into the CoreContextProvider so that it can send + // ContextChangedMessage in response to updates: context thread doesn't exist yet, so + // no synchronization is required here + _context_provider->SetMessageBus(_message_bus.get()); + + // Initialize a thread-safe queue for feature-submitted functions that will execute on + // the context thread DATADOG_ASSERT(!_context_queue, "_context_queue already exists on Start()"); _context_queue = std::make_unique>>(); @@ -383,6 +406,18 @@ bool Core::Start() { ) ); } + + // Start the messaging thread only after all features have completed OnCoreStarted(). + // This ensures that any ContextChangedMessages dispatched during startup (e.g. from + // a feature calling UpdateContext in its Start()) are not delivered to a feature's + // handler before that feature's own OnCoreStarted() has run. Messages enqueued + // during the loop above are buffered in the queue and will be drained once this + // thread starts. + DATADOG_ASSERT(!_message_bus_thread, "_message_bus_thread already exists on Start()"); + _message_bus_thread = std::thread( + MessagingThreadMain, std::ref(_diagnostic_logger), std::ref(*_message_bus) + ); + return true; } @@ -422,14 +457,31 @@ void Core::Stop() { } _context_thread.reset(); + // Stop the messaging thread after the context thread exits: the context thread is + // the primary producer of messages (via Update()), so draining it first prevents new + // messages from arriving after we signal the messaging thread to stop. After joining, + // detach the bus from the context provider so any stray Update() calls (which should + // not happen at this point) do not reference the destroyed bus. + DATADOG_ASSERT(_message_bus, "_message_bus is invalid on Stop"); + DATADOG_ASSERT( + _message_bus_thread && _message_bus_thread->joinable(), + "_message_bus_thread is non-joinable on Stop" + ); + _message_bus->Stop(); + _diagnostic_logger.Debug("Joining on messaging thread"); + _message_bus_thread->join(); + _message_bus_thread.reset(); + _context_provider->SetMessageBus(nullptr); + // Notify each registered feature that the core has stopped, now that no more // context-thread functions enqueued by those features may be running for (const auto& feature : _features) { feature.impl->OnCoreStopping(); } - // Destroy the context queue, now that no more features exist + // Destroy the context queue and message bus, now that no more features exist _context_queue.reset(); + _message_bus.reset(); // Stop all queue processing, then block until the consumer thread drains the queue // and exits, at which point it's safe to release the queue diff --git a/src/datadog/impl/core/core.hpp b/src/datadog/impl/core/core.hpp index 94e33ca0..80a06eab 100644 --- a/src/datadog/impl/core/core.hpp +++ b/src/datadog/impl/core/core.hpp @@ -20,6 +20,7 @@ #include "datadog/impl/core/block.hpp" #include "datadog/impl/core/context.hpp" #include "datadog/impl/core/feature.hpp" +#include "datadog/impl/core/message_bus.hpp" #include "datadog/impl/core/queue.hpp" #include "datadog/impl/core/storage_queue.hpp" #include "datadog/impl/core/storage_write.hpp" @@ -389,6 +390,9 @@ class Core { std::unique_ptr>> _context_queue; std::optional _context_thread; + std::unique_ptr _message_bus; + std::optional _message_bus_thread; + std::unique_ptr _upload_scheduler; std::optional _upload_thread; diff --git a/src/datadog/impl/core/feature.hpp b/src/datadog/impl/core/feature.hpp index 6ee1d4e3..58385529 100644 --- a/src/datadog/impl/core/feature.hpp +++ b/src/datadog/impl/core/feature.hpp @@ -18,6 +18,7 @@ #include "datadog/impl/core/block.hpp" #include "datadog/impl/core/context.hpp" #include "datadog/impl/core/feature_id.hpp" +#include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/feature_scope.hpp" #include "datadog/impl/core/tlv.hpp" #include "datadog/impl/platform/filesystem.hpp" @@ -136,6 +137,23 @@ class Feature : public std::enable_shared_from_this { const HttpContext& context, class BatchReader& reader ) = 0; + /** + * Called by `Core::Start()` each time the SDK starts up. If the feature wants to + * react to messages dispatched on the `MessageBus`, it should override this method + * and return a handler function: typically a lambda that captures `weak_from_this()` + * so that message delivery is silently skipped if the feature has already been + * destroyed. + * + * Implementations must be idempotent, as this function called on every `Start()`. + * + * The default implementation returns `std::nullopt`, meaning the feature opts out of + * message handling entirely. + */ + virtual std::optional> + MakeMessageHandler() { + return std::nullopt; + } + protected: std::optional _scope; }; diff --git a/tests/impl/core/core_test.cpp b/tests/impl/core/core_test.cpp index 110529f5..1634866b 100644 --- a/tests/impl/core/core_test.cpp +++ b/tests/impl/core/core_test.cpp @@ -6,8 +6,11 @@ #include "datadog/impl/core/core.hpp" +#include #include +#include "datadog/impl/core/feature_message.hpp" + #include "mock/clock.hpp" #include "mock/feature.hpp" #include "mock/filesystem.hpp" @@ -288,3 +291,58 @@ TEST_CASE("Core Lifecycle", "[unit]") { REQUIRE(feature->num_stop_calls == 1); } } + +/** + * Feature that counts received ContextChangedMessages via MakeMessageHandler(), and + * triggers a context update (causing one such message to be dispatched) each time the + * core starts. + */ +class MessageHandlerFeature : public MockFeature { + public: + std::atomic messages_received{0}; + + MessageHandlerFeature() + : MockFeature(CreateFeatureId("MHFT"), "message_handler_test") {} + + std::optional> + MakeMessageHandler() override { + auto weak = weak_from_this(); + return [weak](const FeatureMessage&) { + if (auto self = std::static_pointer_cast(weak.lock())) { + self->messages_received.fetch_add(1); + } + }; + } + + void Start() override { + MockFeature::Start(); + // Trigger a context update so that a ContextChangedMessage is dispatched to + // all registered handlers on each start + _scope->UpdateContext([](CoreContext& ctx) { ctx.rum.emplace(); }); + } +}; + +TEST_CASE("Core Messaging", "[unit]") { + SECTION( + "M deliver messages to handler after restart W core is stopped and " + "started again" + ) { + impl::Core core = _make_core(); + REQUIRE(core.Init()); + + auto feature = std::make_shared(); + REQUIRE(core.RegisterFeature(feature)); + + // First run: start triggers one context update, stop drains both the context + // thread and the messaging thread before returning + REQUIRE(core.Start()); + core.Stop(); + REQUIRE(feature->messages_received.load() == 1); + + // Second run: the handler must still be registered after the restart; the + // same context update should deliver a second message + REQUIRE(core.Start()); + core.Stop(); + REQUIRE(feature->messages_received.load() == 2); + } +} diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp index 26b6248b..a4615457 100644 --- a/tests/impl/core/message_bus_test.cpp +++ b/tests/impl/core/message_bus_test.cpp @@ -11,6 +11,8 @@ #include #include +#include "datadog/impl/core/feature.hpp" +#include "datadog/impl/core/feature_id.hpp" #include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/messaging_thread.hpp" @@ -45,6 +47,171 @@ static uint64_t ReadSentinel(const ContextChangedMessage& msg) { return value; } +/** + * Minimal Feature subclass used to test MakeMessageHandler() wiring. Writes delivered + * ContextChangedMessage values into an external vector whose lifetime exceeds the + * feature's, which allows test assertions even when the feature is destroyed early. + */ +struct MessageBusTestFeature final : public Feature { + std::vector& received; + + explicit MessageBusTestFeature(std::vector& out) + : received(out) {} + + FeatureId GetId() const override { return CreateFeatureId("MBUS"); } + std::string_view GetName() const override { return "message_bus_test"; } + std::optional UploadThread_PrepareReport( + const HttpContext&, class BatchReader& + ) override { + return std::nullopt; + } + + std::optional> + MakeMessageHandler() override { + // Capture both the weak_ptr (for lifetime gating) and a direct reference to the + // output vector (which outlives the Feature in tests). Since `received` is a + // reference member, `&out` in the lambda binds to the external vector, not to + // this object, so the reference remains valid after the Feature is destroyed. + auto weak = weak_from_this(); + std::vector& out = received; + return [weak, &out](const FeatureMessage& msg) { + if (!weak.lock()) { + return; + } + out.push_back(std::get(msg)); + }; + } +}; + +TEST_CASE("MessageBus — integration", "[unit][core]") { + DiagnosticMessageBuffer diagnostics; + DiagnosticLogger logger = diagnostics.CreateTestLogger(); + + SECTION( + "M deliver ContextChangedMessage W MakeMessageHandler() wired through " + "CoreContextProvider::Update()" + ) { + // Given a feature registered via MakeMessageHandler() + std::vector received; + auto feature = std::make_shared(received); + auto handler = feature->MakeMessageHandler(); + REQUIRE(handler.has_value()); + + std::vector> handlers; + handlers.push_back(std::move(*handler)); + + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); + + CoreContextProvider context_provider(MOCK_CONTEXT); + context_provider.SetMessageBus(&bus); + + // When Update() is called + context_provider.Update([](CoreContext& ctx) { ctx.rum.emplace(); }); + + // Then the handler receives the message + bus.Stop(); + t.join(); + context_provider.SetMessageBus(nullptr); + + REQUIRE(received.size() == 1); + } + + SECTION("M deliver correct context snapshot W context mutated in Update()") { + // Given a bus with a plain lambda handler + std::vector received; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage& msg) { + received.push_back(std::get(msg)); + }); + + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); + + CoreContextProvider context_provider(MOCK_CONTEXT); + context_provider.SetMessageBus(&bus); + + // When Update() writes a known sentinel value into the RUM session_id + constexpr uint64_t sentinel = 0xc0ffee; + context_provider.Update([=](CoreContext& ctx) { + ctx.rum.emplace(); + std::memcpy(ctx.rum->session_id.bytes.data(), &sentinel, sizeof(sentinel)); + }); + + bus.Stop(); + t.join(); + context_provider.SetMessageBus(nullptr); + + // Then the delivered snapshot carries that exact value + REQUIRE(received.size() == 1); + REQUIRE(received[0].context.rum.has_value()); + uint64_t got = 0; + std::memcpy(&got, received[0].context.rum->session_id.bytes.data(), sizeof(got)); + REQUIRE(got == sentinel); + } + + SECTION( + "M skip delivery gracefully W feature is destroyed before message is handled" + ) { + // Given a feature whose handler captures weak_from_this(), and a second handler + // that counts deliveries to confirm the bus continues to function + std::vector received; + auto feature = std::make_shared(received); + auto handler = feature->MakeMessageHandler(); + REQUIRE(handler.has_value()); + + std::atomic other_count{0}; + std::vector> handlers; + handlers.push_back(std::move(*handler)); + handlers.push_back([&](const FeatureMessage&) { other_count.fetch_add(1); }); + + // Enqueue a message while the feature is still alive, then destroy it before the + // messaging thread has a chance to drain the queue. Running the thread function + // synchronously (after Stop()) guarantees the feature is gone when the handler + // runs. + MessageBus bus(std::move(handlers)); + REQUIRE(bus.Send(MakeMessage(0x1))); + + feature.reset(); // weak_from_this() in handler now expires + + bus.Stop(); + MessagingThreadMain(logger, bus); // drain synchronously; no background thread + + // The TestFeature handler silently dropped the message; the other handler ran + REQUIRE(received.empty()); + REQUIRE(other_count.load() == 1); + } + + SECTION( + "M deliver all messages enqueued before Stop() W shutdown via " + "CoreContextProvider::Update()" + ) { + // Given a bus wired to a context provider with a counting handler + std::atomic delivered{0}; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage&) { delivered.fetch_add(1); }); + + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); + + CoreContextProvider context_provider(MOCK_CONTEXT); + context_provider.SetMessageBus(&bus); + + // When multiple Update() calls are made before the bus is stopped + const int num_updates = 20; + for (int i = 0; i < num_updates; i++) { + context_provider.Update([](CoreContext& ctx) { ctx.rum.emplace(); }); + } + + bus.Stop(); + t.join(); + context_provider.SetMessageBus(nullptr); + + // Then all messages were delivered before the join returned + REQUIRE(delivered.load() == num_updates); + } +} + TEST_CASE("MessageBus", "[unit][core]") { DiagnosticMessageBuffer diagnostics; DiagnosticLogger logger = diagnostics.CreateTestLogger();