Merge pull request #6361 from IntersectMBO/fmaste/trace-forward

mgmeier · web-flow · commit a5e309dca245 · 2025-11-24T17:51:17.000Z
trace-forward | sink write and read optimizations
diff --git a/bench/tx-generator/src/Cardano/Benchmarking/Tracer.hs b/bench/tx-generator/src/Cardano/Benchmarking/Tracer.hs
@@ -172,7 +172,7 @@ initialTraceConfig = TraceConfig {
           , setMaxDetail TracerNameConnect
           , setMaxDetail TracerNameSubmit
           ]
-    , tcForwarder = Just defaultForwarder {tofConnQueueSize = 2048, tofDisconnQueueSize = 4096}
+    , tcForwarder = Just defaultForwarder {tofQueueSize = 4096}
     , tcNodeName = Nothing
     , tcPeerFrequency = Just 2000 -- Every 2 seconds
     , tcResourceFrequency = Just 1000 -- Every second
diff --git a/cardano-tracer/demo/multi/forwarder.hs b/cardano-tracer/demo/multi/forwarder.hs
@@ -6,6 +6,6 @@ import           System.Environment (getArgs)
 
 main :: IO ()
 main = getArgs >>= \case
-  [localSock, "Initiator"] -> launchForwardersSimple Initiator localSock 1000 2000
-  [localSock, "Responder"] -> launchForwardersSimple Responder localSock 1000 2000
+  [localSock, "Initiator"] -> launchForwardersSimple Initiator localSock 2000
+  [localSock, "Responder"] -> launchForwardersSimple Responder localSock 2000
   _ -> putStrLn "Usage: ./demo-forwarder /path/to/local/sock Initiator|Responder"
diff --git a/cardano-tracer/demo/ssh/forwarder.hs b/cardano-tracer/demo/ssh/forwarder.hs
@@ -21,8 +21,8 @@ main = getArgs >>=
              , tsWorkDir      = Identity "."
              }
     in case mode of
-         "Initiator" -> launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 1000 2000
-         "Responder" -> launchForwardersSimple ts Responder (Net.LocalPipe localSock) 1000 2000
+         "Initiator" -> launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 2000
+         "Responder" -> launchForwardersSimple ts Responder (Net.LocalPipe localSock) 2000
          _ -> err
   _ -> err
  where err = error "Usage: ./demo-forwarder /path/to/local/sock Initiator|Responder"
diff --git a/cardano-tracer/test/Cardano/Tracer/Test/DataPoint/Tests.hs b/cardano-tracer/test/Cardano/Tracer/Test/DataPoint/Tests.hs
@@ -41,7 +41,7 @@ propDataPoint ts@TestSetup{..} rootDir localSock = do
   savedDPValues :: TVar DataPointValues <- newTVarIO []
   withAsync (doRunCardanoTracer config (Just $ rootDir <> "/../state") stderrShowTracer stopProtocols dpRequestors) \_ -> do
     sleep 1.0
-    withAsync (launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 1000 10000) \_ -> do
+    withAsync (launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 10000) \_ -> do
       sleep 1.5
       -- We know that there is one single "node" only (and one single requestor too).
       -- requestors ((_, dpRequestor):_) <- M.toList <$> readTVarIO dpRequestors
diff --git a/cardano-tracer/test/Cardano/Tracer/Test/Forwarder.hs b/cardano-tracer/test/Cardano/Tracer/Test/Forwarder.hs
@@ -91,20 +91,18 @@ launchForwardersSimple
   -> ForwardersMode
   -> HowToConnect
   -> Word
-  -> Word
   -> IO ()
-launchForwardersSimple ts mode howToConnect connSize disconnSize = withIOManager \iomgr ->
-  runInLoop (launchForwardersSimple' ts iomgr mode howToConnect connSize disconnSize) (Just Minimum) howToConnect 1
+launchForwardersSimple ts mode howToConnect queueSize = withIOManager \iomgr ->
+  runInLoop (launchForwardersSimple' ts iomgr mode howToConnect queueSize) (Just Minimum) howToConnect 1
 
 launchForwardersSimple'
   :: TestSetup Identity
   -> IOManager
   -> ForwardersMode
   -> HowToConnect
   -> Word
-  -> Word
   -> IO ()
-launchForwardersSimple' ts iomgr mode howToConnect connSize disconnSize =
+launchForwardersSimple' ts iomgr mode howToConnect queueSize =
   case (howToConnect, mode) of
     (Net.RemoteSocket (Text.unpack -> host) (show -> port), Initiator) -> do
       result <- try @IOException do
@@ -177,8 +175,7 @@ launchForwardersSimple' ts iomgr mode howToConnect connSize disconnSize =
   tfConfig =
     TOF.ForwarderConfiguration
       { TOF.forwarderTracer = nullTracer -- contramap show stdoutTracer
-      , TOF.disconnectedQueueSize = disconnSize
-      , TOF.connectedQueueSize = connSize
+      , TOF.queueSize = queueSize
       }
 
   dpfConfig :: DPF.ForwarderConfiguration
diff --git a/cardano-tracer/test/Cardano/Tracer/Test/Logs/Tests.hs b/cardano-tracer/test/Cardano/Tracer/Test/Logs/Tests.hs
@@ -51,7 +51,7 @@ propLogs ts@TestSetup{..} format logRotLimitBytes logRotMaxAgeMinutes rootDir lo
   withAsync (doRunCardanoTracer (acceptConfig rootDir) (Just $ rootDir <> "/../state") stderrShowTracer stopProtocols dpRequestors) \async1 -> do
     link async1
     sleep 1.0
-    withAsync (launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 1000 10000) \async2 -> do
+    withAsync (launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 10000) \async2 -> do
       link async2
       sleep 8.0 -- Wait till some rotation is done.
       applyBrake stopProtocols
@@ -92,10 +92,10 @@ propMultiInit ts@TestSetup{..} format rootDir howToConnect1 howToConnect2 = do
   withAsync (doRunCardanoTracer initConfig (Just $ rootDir <> "/../state") stderrShowTracer stopProtocols dpRequestors) \async1 -> do
     link async1
     sleep 1.0
-    withAsync (launchForwardersSimple ts Responder howToConnect1 1000 10000) \async2 -> do
+    withAsync (launchForwardersSimple ts Responder howToConnect1 10000) \async2 -> do
       link async2
       sleep 1.0
-      withAsync (launchForwardersSimple ts Responder howToConnect2 1000 10000) \async3 -> do
+      withAsync (launchForwardersSimple ts Responder howToConnect2 10000) \async3 -> do
         link async3
         sleep 5.0 -- Wait till some work is done.
         applyBrake stopProtocols
@@ -132,12 +132,12 @@ propMultiResp ts@TestSetup{..} format rootDir howToConnect = do
   withAsync (doRunCardanoTracer respConfig (Just $ rootDir <> "/../state") stderrShowTracer stopProtocols dpRequestors) \async1 -> do
     link async1
     sleep 1.0
-    -- withAsync (launchForwardersSimple ts Initiator howToConnect 1000 10000) \async2 -> do
-    withAsync (launchForwardersSimple ts Initiator howToConnect 1000 10000) \async2 -> do
+    -- withAsync (launchForwardersSimple ts Initiator howToConnect 10000) \async2 -> do
+    withAsync (launchForwardersSimple ts Initiator howToConnect 10000) \async2 -> do
       link async2
       sleep 1.0
-      -- withAsync (launchForwardersSimple ts Initiator howToConnect 1000 10000) \async3 -> do
-      withAsync (launchForwardersSimple ts Initiator howToConnect 1000 10000) \async3 -> do
+      -- withAsync (launchForwardersSimple ts Initiator howToConnect 10000) \async3 -> do
+      withAsync (launchForwardersSimple ts Initiator howToConnect 10000) \async3 -> do
         link async3
         sleep 5.0 -- Wait till some work is done.
         applyBrake stopProtocols
diff --git a/cardano-tracer/test/Cardano/Tracer/Test/Queue/Tests.hs b/cardano-tracer/test/Cardano/Tracer/Test/Queue/Tests.hs
@@ -37,8 +37,8 @@ propQueue ts rootDir localSocket = do
   -- Run the forwarder only. It imitates the case when the acceptor is
   -- misconfigured and cannot be launched, so the connection cannot be established.
   -- In this case, the forwarder should collect trace items in its internal
-  -- "flexible queue" and periodically flush them to stdout.
-  withAsyncBound (launchForwardersSimple ts Responder (Net.LocalPipe localSocket) connSize disconnSize) . const $
+  -- fixed-capacity queue and periodically flush them to stdout.
+  withAsyncBound (launchForwardersSimple ts Responder (Net.LocalPipe localSocket) queueSize) . const $
     -- Wait till the queue will be redirected to stdout.
     sleep 7.0
   -- Return the normal stdout.
@@ -48,8 +48,8 @@ propQueue ts rootDir localSocket = do
   content <- TIO.readFile tmpPath
   removeFile tmpPath
   let flushedTraceObjectsNum = T.count "TraceObject" content
-  return $ flushedTraceObjectsNum === fromIntegral disconnSize
+  return $ flushedTraceObjectsNum === fromIntegral queueSize
+
+queueSize :: Word
+queueSize = 100
 
-connSize, disconnSize :: Word
-connSize = 50
-disconnSize = 100
diff --git a/cardano-tracer/test/Cardano/Tracer/Test/Restart/Tests.hs b/cardano-tracer/test/Cardano/Tracer/Test/Restart/Tests.hs
@@ -41,7 +41,7 @@ propNetworkForwarder ts rootDir localSock = do
   brake <- initProtocolsBrake
   dpRequestors <- initDataPointRequestors
   propNetwork' ts rootDir
-    ( launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 1000 10000
+    ( launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 10000
     , doRunCardanoTracer config (Just $ rootDir <> "/../state") stderrShowTracer brake dpRequestors
     )
 
diff --git a/trace-dispatcher/CHANGELOG.md b/trace-dispatcher/CHANGELOG.md
@@ -1,5 +1,9 @@
 # Revision history for trace-dispatcher
 
+## NEXT -- Nov 2025
+
+* Replaced both `tofDisconnQueueSize` and `tofConnQueueSize` with `tofQueueSize` (See #6361 for details). 
+
 ## 2.10.0 -- July, 2025
 * Forwarding protocol supports connections over TCP socket, in addition to Unix domain sockets.
 * Failure to initialise the `PrometheusSimple` backend is now lenient - i.e., won't result in an exception being propagated.
diff --git a/trace-dispatcher/src/Cardano/Logging/Types.hs b/trace-dispatcher/src/Cardano/Logging/Types.hs
@@ -487,50 +487,59 @@ instance AE.FromJSON Verbosity where
                                     <> "Unknown Verbosity: " <> show other
 
 data TraceOptionForwarder = TraceOptionForwarder {
-    tofConnQueueSize       :: Word
-  , tofDisconnQueueSize    :: Word
+    tofQueueSize           :: Word
   , tofVerbosity           :: Verbosity
   , tofMaxReconnectDelay   :: Word
 } deriving stock (Eq, Ord, Show, Generic)
 
--- A word regarding queue sizes:
+-- A word regarding queue size:
+--
 -- In case of a missing forwarding service consumer, traces messages will be
--- buffered. This mitigates short forwarding interruptions, or delays at startup time.
+-- buffered. This mitigates short forwarding interruptions, or delays at startup
+-- time.
 --
--- The queue capacity should thus correlate to the expected log lines per second given
--- a particular tracing configuration - to avoid unnecessarily increasing memory footprint.
+-- The queue capacity should thus correlate to the expected log lines per second
+-- given a particular tracing configuration - to avoid unnecessarily increasing
+-- memory footprint.
 --
 -- The default values here are chosen to accomodate verbose tracing output
--- (i.e., buffering 1min worth of trace data given ~32 messages per second). A config
--- that results in less than 5 msgs per second should also provide TraceOptionForwarder
--- queue size values considerably lower. The `disconnQueueSize` is the hard limit in that case.
+-- (i.e., buffering 1min worth of trace data given ~32 messages per second). A
+-- config that results in less than 5 msgs per second should also provide
+-- `TraceOptionForwarder` a queue size value considerably lower.
 --
--- The queue sizes tie in with the max number of trace objects cardano-tracer requests periodically,
--- the default for that being 100. Here, the basic queue can hold enough traces for 10 subsequent polls
--- by cardano-tracer.
+-- The queue size ties in with the max number of trace objects cardano-tracer
+-- requests periodically, the default for that being 100. Here, the queue can
+-- hold enough traces for 10 subsequent polls by cardano-tracer.
 instance AE.FromJSON TraceOptionForwarder where
-    parseJSON (AE.Object obj) =
-      TraceOptionForwarder
-        <$> obj AE..:? "connQueueSize"      AE..!= 1024
-        <*> obj AE..:? "disconnQueueSize"   AE..!= 2048
-        <*> obj AE..:? "verbosity"          AE..!= Minimum
-        <*> obj AE..:? "maxReconnectDelay"  AE..!= 60
+    parseJSON (AE.Object obj) = do
+      -- Field "queueSize" is the new field that replaces and unifies
+      -- both "connQueueSize" and "disconnQueueSize".
+      maybeQueueSize <- obj AE..:? "queueSize"
+      queueSize <- case maybeQueueSize of
+                     -- If the new field was provided we use it.
+                     (Just qs) -> return qs
+                     -- Else we look for the deprectaed fields.
+                     Nothing   -> do
+                       -- We keep the same default values.
+                       connQueueSize    <- obj AE..:? "connQueueSize"    AE..!= 1024
+                       disconnQueueSize <- obj AE..:? "disconnQueueSize" AE..!= 2048
+                       return $ max connQueueSize disconnQueueSize
+      verbosity         <- obj AE..:? "verbosity"         AE..!= Minimum
+      maxReconnectDelay <- obj AE..:? "maxReconnectDelay" AE..!= 60
+      return $ TraceOptionForwarder queueSize verbosity maxReconnectDelay
     parseJSON _ = mempty
 
-
 instance AE.ToJSON TraceOptionForwarder where
   toJSON TraceOptionForwarder{..} = AE.object
     [
-      "connQueueSize"     AE..= tofConnQueueSize,
-      "disconnQueueSize"  AE..= tofDisconnQueueSize,
+      "queueSize"         AE..= tofQueueSize,
       "verbosity"         AE..= tofVerbosity,
       "maxReconnectDelay" AE..= tofMaxReconnectDelay
     ]
 
 defaultForwarder :: TraceOptionForwarder
 defaultForwarder = TraceOptionForwarder {
-    tofConnQueueSize       = 1024
-  , tofDisconnQueueSize    = 2048
+    tofQueueSize           = 2048
   , tofVerbosity           = Minimum
   , tofMaxReconnectDelay   = 60
 }
diff --git a/trace-dispatcher/test/Cardano/Logging/Test/Unit/Configuration.hs b/trace-dispatcher/test/Cardano/Logging/Test/Unit/Configuration.hs
@@ -52,8 +52,7 @@ config1 = TraceConfig {
           , (["tracer3","TestMessage"], [ConfSeverity (SeverityF (Just Info))])
           ]
     , tcForwarder = Just TraceOptionForwarder {
-        tofConnQueueSize = 100
-      , tofDisconnQueueSize = 1000
+        tofQueueSize = 1000
       , tofVerbosity = Minimum
       , tofMaxReconnectDelay = 60
       }
@@ -72,8 +71,7 @@ config2 = TraceConfig {
         , (["tracer3","TestMessage"], [ConfSeverity (SeverityF (Just Warning))])
         ]
     , tcForwarder = Just TraceOptionForwarder {
-        tofConnQueueSize = 100
-      , tofDisconnQueueSize = 1000
+        tofQueueSize = 1000
       , tofVerbosity = Minimum
       , tofMaxReconnectDelay = 60
       }
diff --git a/trace-dispatcher/test/Cardano/Logging/Test/Unit/Documentation.hs b/trace-dispatcher/test/Cardano/Logging/Test/Unit/Documentation.hs
@@ -35,8 +35,7 @@ config1 = TraceConfig {
           , (["node1"], [ConfSeverity (SeverityF (Just Warning)),  ConfBackend [Stdout MachineFormat]])
           ]
     , tcForwarder = Just TraceOptionForwarder {
-        tofConnQueueSize = 100
-      , tofDisconnQueueSize = 1000
+        tofQueueSize = 1000
       , tofVerbosity = Minimum
       , tofMaxReconnectDelay = 60
       }
diff --git a/trace-forward/CHANGELOG.md b/trace-forward/CHANGELOG.md
@@ -1,6 +1,11 @@
 # ChangeLog
 
+## NEXT - Nov 2025
+
+* Replaced both `disconnectedQueueSize` and `connectedQueueSize` with `queueSize` (See #6361 for details).
+
 ## 2.3.1 - Oct 2025
+
 * Updated to `ekg-forward-1.0`, `ouroboros-network-framework-0.19.2` and `typed-protocols-1.0`.
 
 ## 2.3.0 - Jul 2025
diff --git a/trace-forward/src/Trace/Forward/Configuration/TraceObject.hs b/trace-forward/src/Trace/Forward/Configuration/TraceObject.hs
@@ -28,14 +28,11 @@ data AcceptorConfiguration lo = AcceptorConfiguration
 data ForwarderConfiguration lo = ForwarderConfiguration
   { -- | The tracer that will be used by the forwarder in its network layer.
     forwarderTracer :: !(Tracer IO (TraceSendRecv (TraceObjectForward lo)))
-    -- | The big size of internal queue for tracing items. We use it in
-    --   the beginning of the session, to avoid queue overflow, because
-    --   initially there is no connection with acceptor yet, and the
-    --   number of tracing items after node's start may be very big.
-  , disconnectedQueueSize :: !Word
-    -- | The small size of internal queue for tracing items. We use it
-    --   after the big queue is empty, which means that acceptor is connected
-    --   and tracing items are already forwarded to it. We switch to small
-    --   queue to reduce memory usage in the node.
-  , connectedQueueSize :: !Word
+    -- | The size of the internal queue for tracing items.
+    --   Use a size suitable for the beginning of the session, to avoid queue
+    --   overflows, because initially there is no connection with acceptor yet,
+    --   and the number of tracing items after the node starts may be very big.
+    --   At the same time choose a number that reduces memory usage in the node.
+  , queueSize :: !Word
   }
+
diff --git a/trace-forward/src/Trace/Forward/Forwarding.hs b/trace-forward/src/Trace/Forward/Forwarding.hs
@@ -111,9 +111,8 @@ initForwardingDelayed iomgr config magic ekgStore tracerSocketMode = liftIO $ do
       Nothing -> EKGF.LocalPipe ""
       Just (LocalPipe str, _mode) -> EKGF.LocalPipe str
       Just (RemoteSocket host port, _mode) -> EKGF.RemoteSocket host port
-  connSize = tofConnQueueSize config
-  disconnSize = tofDisconnQueueSize config
-  verbosity = tofVerbosity config
+  queueSize         = tofQueueSize         config
+  verbosity         = tofVerbosity         config
   maxReconnectDelay = tofMaxReconnectDelay config
 
   ekgConfig :: EKGF.ForwarderConfiguration
@@ -129,9 +128,8 @@ initForwardingDelayed iomgr config magic ekgStore tracerSocketMode = liftIO $ do
   tfConfig :: TF.ForwarderConfiguration TraceObject
   tfConfig =
     TF.ForwarderConfiguration
-      { TF.forwarderTracer       = mkTracer verbosity
-      , TF.disconnectedQueueSize = disconnSize
-      , TF.connectedQueueSize    = connSize
+      { TF.forwarderTracer = mkTracer verbosity
+      , TF.queueSize       = queueSize
       }
 
   dpfConfig :: DPF.ForwarderConfiguration
diff --git a/trace-forward/src/Trace/Forward/Utils/ForwardSink.hs b/trace-forward/src/Trace/Forward/Utils/ForwardSink.hs
@@ -1,17 +1,11 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE GADTs #-}
-
 module Trace.Forward.Utils.ForwardSink
   ( ForwardSink (..)
   ) where
 
-import           Control.Concurrent.STM.TBQueue
-import           Control.Concurrent.STM.TVar
+import Control.Concurrent.STM.TBQueue
 
 data ForwardSink lo = ForwardSink
-  { forwardQueue     :: !(TVar (TBQueue lo))
-  , disconnectedSize :: !Word
-  , connectedSize    :: !Word
-  , wasUsed          :: !(TVar Bool)
+  { forwardQueue     :: !(TBQueue lo)
   , overflowCallback :: !([lo] -> IO ())
   }
+
diff --git a/trace-forward/src/Trace/Forward/Utils/TraceObject.hs b/trace-forward/src/Trace/Forward/Utils/TraceObject.hs

Original file line number	Diff line number	Diff line change
`@@ -172,7 +172,7 @@ initialTraceConfig = TraceConfig {`
`172`	`172`	`, setMaxDetail TracerNameConnect`
`173`	`173`	`, setMaxDetail TracerNameSubmit`
`174`	`174`	`]`
`175`		`- , tcForwarder = Just defaultForwarder {tofConnQueueSize = 2048, tofDisconnQueueSize = 4096}`
	`175`	`+ , tcForwarder = Just defaultForwarder {tofQueueSize = 4096}`
`176`	`176`	`, tcNodeName = Nothing`
`177`	`177`	`, tcPeerFrequency = Just 2000 -- Every 2 seconds`
`178`	`178`	`, tcResourceFrequency = Just 1000 -- Every second`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ propNetworkForwarder ts rootDir localSock = do`
`41`	`41`	`brake <- initProtocolsBrake`
`42`	`42`	`dpRequestors <- initDataPointRequestors`
`43`	`43`	`propNetwork' ts rootDir`
`44`		`- ( launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 1000 10000`
	`44`	`+ ( launchForwardersSimple ts Initiator (Net.LocalPipe localSock) 10000`
`45`	`45`	`, doRunCardanoTracer config (Just $ rootDir <> "/../state") stderrShowTracer brake dpRequestors`
`46`	`46`	`)`
`47`	`47`