CW_27 export

score/datarouter/BUILD

@@ -208,8 +208,8 @@ cc_library(
         ":datarouter_types",
         ":dltprotocol",
         ":logparser_interface",
+        "//score/mw/log",
         "//score/mw/log/detail/data_router/shared_memory:reader",
-        "@score_baselibs//score/mw/log",
         "@score_baselibs//score/mw/log/configuration:nvconfig",
         "@score_baselibs//score/static_reflection_with_serialization/serialization",
     ],
@@ -383,7 +383,7 @@ cc_library(
     strip_include_prefix = "src/applications",
     visibility = ["//score/datarouter/test:__subpackages__"],
     deps = [
-        "@score_baselibs//score/mw/log",
+        "//score/mw/log",
     ],
 )
 
@@ -400,7 +400,7 @@ cc_library(
         "@score_logging//score/datarouter:__subpackages__",
     ],
     deps = [
-        "@score_baselibs//score/mw/log",
+        "//score/mw/log",
     ],
 )
 
@@ -420,10 +420,10 @@ cc_library(
         ":logparser_factory_interface",
         ":message_passing_server",
         ":unixdomain_server",
+        "//score/mw/log",
         "//score/mw/log/detail/data_router/shared_memory:reader",
         "@score_baselibs//score/concurrency:synchronized",
         "@score_baselibs//score/language/futurecpp",
-        "@score_baselibs//score/mw/log",
     ],
 )
 
@@ -444,8 +444,8 @@ cc_library(
         ":logparser_factory_interface",
         ":message_passing_server",
         ":unixdomain_mock",
+        "//score/mw/log",
         "@score_baselibs//score/concurrency:synchronized",
-        "@score_baselibs//score/mw/log",
     ],
 )
 
@@ -467,8 +467,8 @@ cc_library(
         ":logparser_testing",
         ":message_passing_server",
         ":unixdomain_mock",
+        "//score/mw/log",
         "@score_baselibs//score/concurrency:synchronized",
-        "@score_baselibs//score/mw/log",
     ],
 )
 
@@ -627,8 +627,8 @@ cc_library(
         ":unixdomain_server",
         "//score/datarouter/network:vlan",
         "//score/datarouter/src/persistent_logging/persistent_logging_stub:sysedr_stub",
+        "//score/mw/log",
         "@score_baselibs//score/language/futurecpp",
-        "@score_baselibs//score/mw/log",
         "@score_baselibs//score/mw/log/configuration:nvconfig",
         "@score_baselibs//score/os:socket",
         "@score_baselibs//score/os:stat",
@@ -656,7 +656,7 @@ cc_library(
         ],
         deps = [
             ":dltserver_common",
-            "@score_baselibs//score/mw/log",
+            "//score/mw/log",
             "@score_baselibs//score/mw/log/configuration:nvconfig",
             "@score_baselibs//score/os:socket",
             "@score_baselibs//score/os:stat",
@@ -716,8 +716,8 @@ cc_library(
         ":logchannel_utility",
         ":socketserver_config_helpers",
         "//score/datarouter/src/persistency:interface",
+        "//score/mw/log",
         "@rapidjson",
-        "@score_baselibs//score/mw/log",
     ],
 )
 
@@ -866,7 +866,7 @@ cc_library(
         visibility = ["//score/datarouter/test:__subpackages__"],
         deps = [
             ":datarouter_options",
-            "@score_baselibs//score/mw/log",
+            "//score/mw/log",
         ] + deps,
     )
     for (name, deps, test_only) in [
@@ -901,15 +901,9 @@ cc_binary(
         "@score_baselibs//score/os:errno_logging",
         "@score_baselibs//score/os:pthread",
         "@score_baselibs//score/os/utils:path",
-        "@score_baselibs//score/mw/log:console",
-        "@score_baselibs//score/mw/log:frontend",
-        "//score/mw/log/backend:file",
-        "//score/mw/log/backend:remote",
+        "//score/mw/log",
         # "//third_party/jemalloc", # Ticket-231781
-    ] + select({
-        "@platforms//os:qnx": ["//score/mw/log/backend:slog"],
-        "//conditions:default": [],
-    }),
+    ],
 )
 
 cc_library(
@@ -1109,7 +1103,7 @@ cc_library(
         "@score_baselibs//score/os:stdio",
         "@score_baselibs//score/os/utils:path",
         "@score_baselibs//score/os/utils:thread",
-        "@score_baselibs//score/mw/log",
+        "//score/mw/log",
         "@score_baselibs//score/mw/log/configuration:nvconfig",
         "//score/datarouter/dlt_filetransfer_trigger_lib:filetransfer_message_types",
         "//score/datarouter/src/persistent_logging/persistent_logging_stub:sysedr_stub",

score/datarouter/doc/design/logging_architecture.md

@@ -143,13 +143,15 @@ The Adaptive AUTOSAR logging interface implementation follows standard specifica
 The ring buffer operates in shared memory to minimize copy overhead.
 Shared memory IPC provides optimal speed and flexibility for this implementation.
 
-### Message passing interaction
+### Datarouter-Client Session
 
-The message passing connection transmits initial connection information, notifications, and handles disconnections. The connection lifecycle proceeds as follows:
+The Datarouter-Client Session uses [message_passing](https://github.com/eclipse-score/communication/tree/main/score/message_passing) IPC for the initial connection, buffer acquire requests, notifications, and disconnections. The `DataRouterRecorder` sets up the session when it is created and closes it when it is destroyed. In between, the datarouter keeps one `IServerConnection` handle per client and drives the log acquisition from the server side.
 
-- The `DataRouterRecorder` constructor creates a `DatarouterMessageClient` instance, spawning a thread that attempts server connections every 100ms.
-- Upon successful connection, the client transmits a message containing application information to the server.
-- The `DataRouterRecorder` destructor notifies the thread, causing normal termination.
+The main idea is to keep clients independent, so that one non-responsive client should not stall the datarouter. This matters because the datarouter serves all clients from a single thread, one after another on each periodic tick, so a single blocking call would hold up every other client. To avoid this, the datarouter asks for buffers (to read) using non-blocking QNX pulses (`IServerConnection::Notify()`), which return right away and never wait on the client. So if a client is slow or stuck, its pulse simply stays unanswered while the healthy clients keep getting served. See [session_sequence_diagram](uml/client_session_interaction_sequence.puml) for the full flow.
+
+However, with such a design there are risks of stale sessions. A small per-session watchdog limits how long the datarouter waits for an acquire response. If a client keeps missing its deadline, the watchdog cleans up the stale session, but first it does a best-effort read of the client's last buffer so no logs are lost unnecessarily. See [activity_watchdog_session](uml/activity_watchdog_session_lifecycle.puml).
+
+The setup also handles an early-disconnect race, wherein if a client crashes while the datarouter is still building its session, the connect and disconnect paths work together to throw away the half-built session instead of keeping a dangling connection pointer. The client can then reconnect cleanly afterwards. See [early_disconnect_race](uml/sequence_early_disconnect_race.puml).
 
 ### datarouter
 

score/datarouter/include/daemon/dlt_log_server.h

@@ -187,6 +187,10 @@ class DltLogServer : score::platform::datarouter::DltNonverboseHandlerType::IOut
                                         AssignmentAction assignment_flag) override;
     std::string SetDltOutputEnable(bool enable) override;
 
+    /// Returns the current output-enable state.
+    /// Thread-safe: reads std::atomic<bool> with acquire ordering.
+    bool IsOutputEnabled() const noexcept override final;
+
     // This is used for test purpose only in google tests, to have an access to the private members.
     // Do not use this calls in implementation except unit tests.
     class DltLogServerTest;
@@ -311,8 +315,6 @@ class DltLogServer : score::platform::datarouter::DltNonverboseHandlerType::IOut
 
     std::unique_ptr<ISysedrHandler> sysedr_handler_;
 
-    bool IsOutputEnabled() const noexcept override final;
-
     void SendNonVerbose(const score::mw::log::config::NvMsgDescriptor& desc,
                         uint32_t tmsp,
                         const void* data,

score/datarouter/include/daemon/i_dlt_log_server.h

@@ -60,6 +60,10 @@ class IDltLogServer
                                                 AssignmentAction assignment_flag) = 0;
     virtual std::string SetDltOutputEnable(bool enable) = 0;
 
+    /// Returns the current output-enable state.
+    /// Thread-safe: backed by std::atomic<bool> in the concrete implementation.
+    virtual bool IsOutputEnabled() const noexcept = 0;
+
     virtual ~IDltLogServer() = default;
 };
 

score/datarouter/include/daemon/message_passing_server.h

@@ -18,7 +18,6 @@
 
 #include "score/mw/log/detail/data_router/shared_memory/common.h"
 
-#include "score/message_passing/i_client_factory.h"
 #include "score/message_passing/i_server_connection.h"
 #include "score/message_passing/i_server_factory.h"
 #include "score/mw/log/detail/logging_identifier.h"
@@ -28,13 +27,15 @@
 
 #include "score/concurrency/interruptible_wait.h"
 #include <score/stop_token.hpp>
+#include <chrono>
 #include <condition_variable>
 #include <functional>
 #include <mutex>
 #include <optional>
 #include <queue>
 #include <thread>
 #include <unordered_map>
+#include <unordered_set>
 
 namespace score
 {
@@ -68,25 +69,37 @@ class IMessagePassingServerSessionWrapper
 class MessagePassingServer : public IMessagePassingServerSessionWrapper
 {
   public:
-    class SessionHandle : public daemon::ISessionHandle
+    struct AcquireWatchdogConfig
     {
-      public:
-        SessionHandle(pid_t pid,
-                      MessagePassingServer* server,
-                      score::cpp::pmr::unique_ptr<score::message_passing::IClientConnection> sender)
-            : daemon::ISessionHandle(), sender_(std::move(sender)), pid_(pid), server_(server), sender_state_{}
+        AcquireWatchdogConfig(std::chrono::milliseconds deadline_in = std::chrono::milliseconds{1000},
+                              std::uint32_t max_misses_in = 3U)
+            : deadline(deadline_in), max_misses(max_misses_in)
         {
         }
 
+        std::chrono::milliseconds deadline;
+        std::uint32_t max_misses;
+    };
+    class SessionHandle : public daemon::ISessionHandle
+    {
+      public:
+        SessionHandle(pid_t pid, MessagePassingServer* server) : daemon::ISessionHandle(), pid_(pid), server_(server) {}
+
         bool AcquireRequest() const override;
 
       private:
-        score::cpp::pmr::unique_ptr<score::message_passing::IClientConnection> sender_;
         pid_t pid_;
         MessagePassingServer* server_;
-        mutable std::optional<score::message_passing::IClientConnection::State> sender_state_;
     };
 
+    /// Thread-safety contract:
+    /// - Tick() is called from the worker thread without the server mutex held.
+    /// - OnAcquireResponse() is called from the dispatch thread with the server mutex held.
+    /// - OnClosedByPeer() is called from the worker thread without the server mutex held.
+    /// - IsSourceClosed() is called from the worker thread with the server mutex held.
+    /// Implementations must ensure that shared state accessed by Tick() and by any method
+    /// called under the mutex (OnAcquireResponse, IsSourceClosed) is properly synchronized
+    /// (e.g., via atomics or an internal lock).
     class ISession
     {
       public:
@@ -102,31 +115,39 @@ class MessagePassingServer : public IMessagePassingServerSessionWrapper
                                                 const score::mw::log::detail::ConnectMessageFromClient&,
                                                 score::cpp::pmr::unique_ptr<daemon::ISessionHandle>)>;
 
-    explicit MessagePassingServer(SessionFactory factory,
-                                  std::shared_ptr<score::message_passing::IServerFactory> server_factory = nullptr,
-                                  std::shared_ptr<score::message_passing::IClientFactory> client_factory = nullptr);
+    MessagePassingServer(SessionFactory factory,
+                         std::shared_ptr<score::message_passing::IServerFactory> server_factory = nullptr,
+                         AcquireWatchdogConfig watchdog_config = AcquireWatchdogConfig{});
     ~MessagePassingServer() noexcept;
 
     // for unit test only. to keep rest of functions in private
     class MessagePassingServerForTest;
 
   private:
-    void NotifyAcquireRequestFailed(std::int32_t pid);
+    friend class SessionHandle;
+
+    bool NotifyAcquireRequest(pid_t pid);
 
-    void MessageCallback(const score::cpp::span<const std::uint8_t> message, const pid_t pid);
-    void OnConnectRequest(const score::cpp::span<const std::uint8_t> message, const pid_t pid);
-    void OnAcquireResponse(const score::cpp::span<const std::uint8_t> message, const pid_t pid);
+    void MessageCallback(score::message_passing::IServerConnection& connection, score::cpp::span<const std::uint8_t> message);
+    void OnConnectRequest(score::message_passing::IServerConnection& connection,
+                          const score::cpp::span<const std::uint8_t> message,
+                          pid_t pid);
+    void OnAcquireResponse(score::message_passing::IServerConnection& connection,
+                           const score::cpp::span<const std::uint8_t> message,
+                           pid_t pid);
 
     using TimestampT = std::chrono::steady_clock::time_point;
 
     struct SessionWrapper
     {
-        SessionWrapper(IMessagePassingServerSessionWrapper* server_instance,
-                       pid_t process_id,
-                       std::unique_ptr<ISession> session_instance)
-            : server(server_instance),
-              pid(process_id),
-              session(std::move(session_instance)),
+        SessionWrapper(IMessagePassingServerSessionWrapper* message_passing_server,
+                       pid_t client_pid,
+                       std::unique_ptr<ISession> message_passing_session)
+            : server(message_passing_server),
+              pid(client_pid),
+              session(std::move(message_passing_session)),
+              connection(nullptr),
+              acquire_in_flight(false),
               enqueued(false),
               running(false),
               to_delete(false),
@@ -165,6 +186,11 @@ class MessagePassingServer : public IMessagePassingServerSessionWrapper
         pid_t pid;
         std::unique_ptr<ISession> session;
 
+        score::message_passing::IServerConnection* connection;
+        bool acquire_in_flight;
+        std::optional<TimestampT> acquire_deadline;
+        std::uint32_t acquire_miss_count{0U};
+
         bool enqueued;
         bool running;
         bool to_delete;
@@ -187,13 +213,18 @@ class MessagePassingServer : public IMessagePassingServerSessionWrapper
     score::cpp::jthread worker_thread_;
     std::condition_variable worker_cond_;  // to wake up worker thread
     std::unordered_map<pid_t, SessionWrapper> pid_session_map_;
+    // Tracks client PIDs that disconnected before a session could be created/emplaced.
+    // This closes a race where OnConnectRequest creates a session outside mutex_ while
+    // disconnect_callback may already have run and the connection object may be gone.
+    std::unordered_set<pid_t> disconnected_pids_;
     std::queue<pid_t> work_queue_;
     std::atomic<bool> workers_exit_;
     std::condition_variable server_cond_;  // to wake up server thread
     bool session_finishing_;
 
     std::shared_ptr<score::message_passing::IServerFactory> server_factory_;
-    std::shared_ptr<score::message_passing::IClientFactory> client_factory_;
+
+    AcquireWatchdogConfig watchdog_config_;
 };
 
 }  // namespace internal

score/datarouter/mocks/daemon/dlt_log_server_mock.h

@@ -44,6 +44,7 @@ class DltLogServerMock : public IDltLogServer
                 (score::platform::DltidT, score::platform::DltidT, score::platform::DltidT, AssignmentAction),
                 (override));
     MOCK_METHOD(std::string, SetDltOutputEnable, (bool), (override));
+    MOCK_METHOD(bool, IsOutputEnabled, (), (const, noexcept, override));
 };
 
 }  // namespace mock

score/datarouter/src/daemon/dlt_log_server.cpp

@@ -213,11 +213,13 @@ void DltLogServer::InitLogChannelsDefault(const bool reloading)
 
 void DltLogServer::SetOutputEnabled(const bool enabled)
 {
-    const bool update = (dlt_output_enabled_.load(std::memory_order_acquire) != enabled);
-
-    if (update)
+    bool expected = !enabled;
+    // Atomically flips only if the current value differs — avoids the TOCTOU
+    // window between load() and store() for the flag itself.
+    if (dlt_output_enabled_.compare_exchange_strong(
+            expected, enabled, std::memory_order_acq_rel, std::memory_order_acquire))
     {
-        dlt_output_enabled_.store(enabled, std::memory_order_release);
+        // Entered only by the thread that won the CAS; callback fires exactly once.
         if (enabled_callback_)
         {
             enabled_callback_(enabled);