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 [
@@ -1109,7 +1109,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

@@ -116,7 +116,6 @@ This construct uses the `log_entry` singleton template to register type informat
 The diagrams below illustrate the high-level class structure of logging framework components.
 
 The [ara::log][1] implementation conforms to Adaptive AUTOSAR specification R1903.
-![alt text][package-ara-log]
 
 ![alt text][package-datarouter]
 
@@ -135,21 +134,22 @@ The activity diagram below depicts the first-run process:
 ### ara::log implementation
 
 The Adaptive AUTOSAR logging interface implementation follows standard specifications. The system creates LogStream objects dynamically to enable isolated collection of log message items and atomic message commits on stream flush.
-![alt text][seq-ara-log]
 ![alt text][log-filtering-client-end]
 
 ### Ring buffer and linear allocator buffer
 
 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
 
@@ -171,9 +171,7 @@ The datarouter requires two configuration files:
 
 [context-ecu]: uml/context-ecu.png "Context: logging framework in xPAD ECU (hPAD example)"
 [context-highlevel]: uml/context-highlevel.png "Implementation details: general approach"
-[seq-ara-log]: uml/seq-ara-log.png "ara::log call conversion to libtracing"
 [seq-trace]: uml/seq-trace.png "Activity diagram for tracing functionality"
-[package-ara-log]: uml/package-ara-log.png "Package contents for ara::log"
 [package-datarouter]: uml/package-datarouter.png "Package contents for datarouter"
 [log-filtering-client-end]: uml/dlt_message_filtering_frontend.png "DLT log filtering in the frontend (client side)"
 [log-filtering-datarouter]: uml/dlt_message_filtering_backend.png "DLT log filtering in the backend (Datarouter)"

score/datarouter/doc/design/uml/activity_watchdog_session_lifecycle.puml

@@ -0,0 +1,41 @@
+@startuml activity_watchdog_lifecycle
+
+title Per-Session Acquire Watchdog Lifecycle
+
+start
+
+:mp_worker tick fires;
+
+if (Session has acquire_in_flight?) then (yes)
+  if (now() > acquire_deadline?) then (yes)
+    :Clear acquire_in_flight;
+    :Increment acquire_miss_count;
+    if (acquire_miss_count >= max_misses?) then (yes)
+      :Tear down session;
+      stop
+    else (no)
+      :Allow next tick to\nretry acquire;
+    endif
+  else (no)
+    :Wait for response\nor deadline expiry;
+  endif
+else (no)
+  :connection.Notify(kAcquireRequest);
+  :Set acquire_in_flight = true;
+  :Set acquire_deadline =\nnow() + watchdog_config.deadline;
+endif
+
+:Wait for next tick;
+
+if (kAcquireResponse received?) then (yes)
+  :Clear acquire_in_flight;
+  :Reset acquire_deadline;
+  :Reset acquire_miss_count = 0;
+  :Process acquired buffer;
+else (no)
+  :Continue waiting;
+endif
+
+stop
+
+@enduml

score/datarouter/doc/design/uml/client_session_interaction_sequence.puml

@@ -0,0 +1,171 @@
+@startuml sequence_notify_acquire_request
+
+title IServerConnection::Notify() + Watchdog\n(Non-blocking MsgDeliverEvent)
+
+skinparam sequenceMessageAlign center
+skinparam responseMessageBelowArrow true
+
+box "Client A (healthy)" #LightBlue
+    participant "User Threads\n(Client A)" as UTA
+    participant "SharedMemoryWriter\n(Client A)" as SHMA
+    participant "QnxDispatchEngine\n(Client A)" as AE
+end box
+
+box "Client X (unhealthy)" #PeachPuff
+    participant "QnxDispatchEngine\n(Client X)" as XE
+end box
+
+box "Datarouter Process" #LightGreen
+    participant "QnxDispatchEngine\n(server engine)" as SE
+    participant "mp_worker thread" as MW
+    participant "Watchdog Timer" as WD
+    participant "DLT output" as DLT
+end box
+
+note over SE
+  Datarouter sends kAcquireRequest
+  via **IServerConnection::Notify()** on the already-
+  established server connection for each client.
+
+  Notify() calls **MsgDeliverEvent()** which delivers
+  a QNX pulse to the client — **fire-and-forget,
+  returns immediately, never blocks**.
+end note
+
+== Initialization ==
+
+UTA -> AE : DatarouterMessageClientImpl::Run()
+AE -> SE : kConnect (appid, uid, pid)\n[Send() — one-time, acceptable]
+
+SE -> SE : connect_callback()\nStore IServerConnection\nfor this client
+
+note over SE
+  Datarouter now holds the
+  IServerConnection reference
+  for each connected client.
+  This is the handle for Notify().
+end note
+
+MW -> MW : Create session + start watchdog\n(per-client deadline timer)
+
+== Steady-State: Acquire via Notify ==
+
+MW -> MW : RunWorkerThread()\ntick() fires for Client A
+
+MW -> SE : **connection_A.Notify(kAcquireRequest)**
+
+SE -> SE : QnxDispatchServer::\nServerConnection::Notify()
+
+note over SE #LightGreen
+  Internally:
+  1. Copy message to notify_pool_ slot
+  2. Push to send_queue_
+  3. **MsgDeliverEvent(rcvid, &pulse)** - Returns immediately and never blocks.
+end note
+
+SE -->> AE : QNX pulse delivered\n(async, non-blocking)
+
+AE -> AE : dispatch_block() receives pulse\n→ readv() → ProcessInput(NOTIFY)\n→ notify_callback_()
+
+note over AE
+  Client's NotifyCallback
+  (new, registered at Start()):
+  validates message, calls
+  ReadAcquire(), sends response.
+end note
+
+AE -> SHMA : ReadAcquire()
+SHMA -> SHMA : alternating_reader_.Switch()\nBuffer A → B
+
+AE -> SE : **Send(kAcquireResponse + result)**\n[Client → Datarouter: Send is safe,\nDatarouter is always ready to receive]
+
+SE -> MW : MessageCallback()\n→ OnAcquireResponse()
+MW -> MW : session.on_acquire_response(acq)
+
+MW -> SHMA : Wait for writers to release\nRead acquired Buffer A
+MW -> DLT : Forward log records ✓
+
+MW -> WD : Reset watchdog timer for Client A
+
+== Client X is unhealthy ==
+
+note over XE #PeachPuff
+  X's engine thread is unhealthy (probable causes):
+  • crash leading to suspended thread
+  • Priority inversion
+  • CPU starvation
+end note
+
+MW -> MW : tick() fires for Client X
+MW -> SE : **connection_XYZ.Notify(kAcquireRequest)**
+
+SE -> SE : MsgDeliverEvent()\n→ **returns immediately!**
+
+note over SE #LightGreen
+  Pulse is queued in X's
+  QNX receive channel.
+  Datarouter does NOT block.
+  Server engine thread is free.
+end note
+
+note over XE #PeachPuff
+  X never processes the pulse.
+  No response comes back.
+  **But Datarouter is NOT blocked.**
+end note
+
+== Meanwhile: Client A continues unaffected ==
+
+MW -> MW : tick() fires for Client A
+MW -> SE : **connection_A.Notify(kAcquireRequest)**
+SE -->> AE : pulse delivered
+AE -> SHMA : ReadAcquire()
+AE -> SE : Send(kAcquireResponse)
+MW -> DLT : Forward Client A's logs
+
+note over DLT #LightGreen
+  **Client A is unaffected**
+  by X being unhealthy.
+end note
+
+== Watchdog: Detect unresponsive X ==
+
+WD -> WD : Client X deadline expired\n(no kAcquireResponse within T seconds)
+
+WD -> MW : Notify: X unresponsive
+
+MW -> MW : Increment X miss counter
+
+alt X misses < max_allowed_misses
+    MW -> MW : Retry: enqueue next tick\n(X gets another chance)
+    MW -> SE : connection_XYZ.Notify(kAcquireRequest)\n[still non-blocking, still safe]
+else X misses >= max_allowed_misses
+    MW -> MW : **Tear down X session**
+
+    note over MW #LightYellow
+      Session teardown:
+      1. Mark session for delete
+      2. Attempt post-mortem read
+         of X's current buffer
+         (best-effort, no IPC needed)
+      3. Remove from pid_session_map_
+      4. Log diagnostic event
+    end note
+
+    MW -> SHMA : Post-mortem: read whatever\nis in X's last buffer\n(shmem is still mapped)
+    MW -> DLT : Forward recovered records\n(if any valid ones found)
+
+    MW -> MW : Erase session from map\nClose server connection
+    destroy XE
+end
+
+== Subsequent X reconnect (if X recovers) ==
+
+note over MW
+  If X was merely slow (not dead),
+  it may reconnect later with a new
+  kConnect. Datarouter creates a
+  fresh session — clean slate.
+end note
+
+@enduml