Merge pull request #5460 from input-output-hk/jutaro/deadlock-stdout-thread-tx-generator

Fix deadlock issue with stdout tracer in tx-generator

Author: mgmeier

bench/tx-generator/src/Cardano/Benchmarking/GeneratorTx.hs

@@ -1,5 +1,4 @@
 {-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE UndecidableInstances #-}
@@ -21,14 +20,15 @@ import           Cardano.Prelude
 import           Prelude (String)
 
 import qualified Control.Concurrent.STM as STM
-import           "contra-tracer" Control.Tracer (Tracer, traceWith)
 import qualified Data.Time.Clock as Clock
 
 import qualified Data.List.NonEmpty as NE
 import           Data.Text (pack)
 import           Network.Socket (AddrInfo (..), AddrInfoFlag (..), Family (..), SocketType (Stream),
                    addrFamily, addrFlags, addrSocketType, defaultHints, getAddrInfo)
 
+import           Cardano.Logging
+
 import           Cardano.Node.Configuration.NodeAddress
 
 import           Cardano.Api hiding (txFee)
@@ -45,7 +45,7 @@ import           Cardano.TxGenerator.Types (NumberOfTxs, TPSRate, TxGenError (..
 
 type AsyncBenchmarkControl = (Async (), [Async ()], IO SubmissionSummary, IO ())
 
-waitBenchmark :: Tracer IO (TraceBenchTxSubmit TxId) -> AsyncBenchmarkControl -> ExceptT TxGenError IO ()
+waitBenchmark :: Trace IO (TraceBenchTxSubmit TxId) -> AsyncBenchmarkControl -> ExceptT TxGenError IO ()
 waitBenchmark traceSubmit (feeder, workers, mkSummary, _) = liftIO $ do
   mapM_ waitCatch (feeder : workers)
   traceWith traceSubmit . TraceBenchTxSubSummary =<< mkSummary
@@ -67,7 +67,7 @@ lookupNodeAddress node = do
     }
 
 handleTxSubmissionClientError ::
-     Tracer IO (TraceBenchTxSubmit TxId)
+     Trace IO (TraceBenchTxSubmit TxId)
   -> Network.Socket.AddrInfo
   -> ReportRef
   -> SubmissionErrorPolicy
@@ -91,8 +91,8 @@ handleTxSubmissionClientError
       , show err]
 
 walletBenchmark :: forall era. IsShelleyBasedEra era
-  => Tracer IO (TraceBenchTxSubmit TxId)
-  -> Tracer IO NodeToNodeSubmissionTrace
+  => Trace IO (TraceBenchTxSubmit TxId)
+  -> Trace IO NodeToNodeSubmissionTrace
   -> ConnectClient
   -> String
   -> NonEmpty NodeIPv4Address

bench/tx-generator/src/Cardano/Benchmarking/GeneratorTx/NodeToNode.hs

@@ -14,6 +14,8 @@ module Cardano.Benchmarking.GeneratorTx.NodeToNode
 import           Cardano.Prelude (forever, liftIO)
 import           Prelude
 
+import           "contra-tracer" Control.Tracer (Tracer (..))
+
 import           Codec.Serialise (DeserialiseFailure)
 import           Control.Concurrent.Class.MonadSTM.Strict (newTVarIO)
 import           Control.Monad.Class.MonadTimer (MonadTimer, threadDelay)
@@ -23,7 +25,6 @@ import           Data.Proxy (Proxy (..))
 import           Network.Socket (AddrInfo (..))
 import           System.Random (newStdGen)
 
-import           "contra-tracer" Control.Tracer (Tracer, nullTracer)
 import           Ouroboros.Consensus.Block.Abstract
 import           Ouroboros.Consensus.Byron.Ledger.Mempool (GenTx)
 import qualified Ouroboros.Consensus.Cardano as Consensus (CardanoBlock)
@@ -79,7 +80,7 @@ benchmarkConnectTxSubmit ioManager handshakeTracer submissionTracer codecConfig
   NtN.connectTo
     (socketSnocket ioManager)
     NetworkConnectTracers {
-        nctMuxTracer       = nullTracer,
+        nctMuxTracer       = mempty,
         nctHandshakeTracer = handshakeTracer
       }
     peerMultiplex
@@ -117,12 +118,12 @@ benchmarkConnectTxSubmit ioManager handshakeTracer submissionTracer codecConfig
         NtN.NodeToNodeProtocols
           { NtN.chainSyncProtocol = InitiatorProtocolOnly $
                                       MuxPeer
-                                        nullTracer
+                                        mempty
                                         (cChainSyncCodec myCodecs)
                                         chainSyncPeerNull
           , NtN.blockFetchProtocol = InitiatorProtocolOnly $
                                        MuxPeer
-                                         nullTracer
+                                         mempty
                                          (cBlockFetchCodec myCodecs)
                                          (blockFetchClientPeer blockFetchClientNull)
           , NtN.keepAliveProtocol = InitiatorProtocolOnly $
@@ -135,7 +136,7 @@ benchmarkConnectTxSubmit ioManager handshakeTracer submissionTracer codecConfig
                                            (txSubmissionClientPeer myTxSubClient)
           , NtN.peerSharingProtocol = InitiatorProtocolOnly $
                                          MuxPeer
-                                           nullTracer
+                                           mempty
                                            (cPeerSharingCodec myCodecs)
                                            (peerSharingClientPeer peerSharingClientNull)
           } )
@@ -152,14 +153,14 @@ benchmarkConnectTxSubmit ioManager handshakeTracer submissionTracer codecConfig
     keepAliveRng <- newStdGen
     peerGSVMap <- liftIO . newTVarIO $ Map.singleton them defaultGSV
     runPeerWithLimits
-      nullTracer
+      mempty
       (cKeepAliveCodec myCodecs)
       (byteLimitsKeepAlive (const 0)) -- TODO: Real Bytelimits, see #1727
       timeLimitsKeepAlive
       channel
       $ keepAliveClientPeer
       $ keepAliveClient
-          nullTracer
+          mempty
           keepAliveRng
           (continueForever (Proxy :: Proxy IO)) them peerGSVMap
           (KeepAliveInterval 10)

bench/tx-generator/src/Cardano/Benchmarking/GeneratorTx/SubmissionClient.hs

@@ -9,7 +9,6 @@
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NoMonomorphismRestriction #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
@@ -39,8 +38,6 @@ import qualified Data.Text as T
 import Data.Type.Equality               (type (~))
 #endif
 
-import           "contra-tracer" Control.Tracer (Tracer, traceWith)
-
 import           Cardano.Tracing.OrphanInstances.Byron ()
 import           Cardano.Tracing.OrphanInstances.Common ()
 import           Cardano.Tracing.OrphanInstances.Consensus ()
@@ -65,6 +62,8 @@ import           Ouroboros.Network.Protocol.TxSubmission2.Type (BlockingReplyLis
 import           Cardano.Api
 import           Cardano.Api.Shelley (fromShelleyTxId, toConsensusGenTx)
 
+import           Cardano.Logging
+
 import           Cardano.Benchmarking.LogTypes
 import           Cardano.Benchmarking.Types
 type CardanoBlock    = Consensus.CardanoBlock  StandardCrypto
@@ -100,8 +99,8 @@ txSubmissionClient
      , IsShelleyBasedEra era
      , tx      ~ Tx era
      )
-  => Tracer m NodeToNodeSubmissionTrace
-  -> Tracer m (TraceBenchTxSubmit TxId)
+  => Trace m NodeToNodeSubmissionTrace
+  -> Trace m (TraceBenchTxSubmit TxId)
   -> TxSource era
   -> EndOfProtocolCallback m
   -> TxSubmissionClient (GenTxId CardanoBlock) (GenTx CardanoBlock) m ()

bench/tx-generator/src/Cardano/Benchmarking/LogTypes.hs

@@ -5,7 +5,6 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE PackageImports #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 {-# OPTIONS_GHC -Wno-all-missed-specialisations #-}
@@ -32,10 +31,10 @@ import           GHC.Generics
 import           Cardano.Api
 import qualified Codec.CBOR.Term as CBOR
 
-import           "contra-tracer" Control.Tracer
-
 import           Network.Mux (WithMuxBearer (..))
 
+import           Cardano.Logging
+
 import           Cardano.Tracing.OrphanInstances.Byron ()
 import           Cardano.Tracing.OrphanInstances.Common ()
 import           Cardano.Tracing.OrphanInstances.Consensus ()
@@ -57,10 +56,10 @@ import           Cardano.TxGenerator.Types (TPSRate)
 
 data BenchTracers =
   BenchTracers
-  { btTxSubmit_   :: Tracer IO (TraceBenchTxSubmit TxId)
-  , btConnect_    :: Tracer IO SendRecvConnect
-  , btSubmission2_:: Tracer IO SendRecvTxSubmission2
-  , btN2N_        :: Tracer IO NodeToNodeSubmissionTrace
+  { btTxSubmit_   :: Trace IO (TraceBenchTxSubmit TxId)
+  , btConnect_    :: Trace IO SendRecvConnect
+  , btSubmission2_:: Trace IO SendRecvTxSubmission2
+  , btN2N_        :: Trace IO NodeToNodeSubmissionTrace
   }
 
 data TraceBenchTxSubmit txid

bench/tx-generator/src/Cardano/Benchmarking/Script.hs

@@ -17,10 +17,9 @@ import           Ouroboros.Network.NodeToClient (IOManager)
 
 import           Cardano.Benchmarking.Script.Action
 import           Cardano.Benchmarking.Script.Aeson (parseScriptFileAeson)
-import           Cardano.Benchmarking.Script.Core (setProtocolParameters, traceTxGeneratorVersion)
+import           Cardano.Benchmarking.Script.Core (setProtocolParameters)
 import           Cardano.Benchmarking.Script.Env
 import           Cardano.Benchmarking.Script.Types
-import           Cardano.Benchmarking.Tracer
 
 type Script = [Action]
 
@@ -37,8 +36,6 @@ runScript script iom = runActionM execScript iom >>= \case
  where
   cleanup s a = void $ runActionMEnv s a iom
   execScript = do
-    liftIO (initTxGenTracers Nothing) >>= setBenchTracers
-    traceTxGeneratorVersion
     setProtocolParameters QueryLocalNode
     forM_ script action
 

bench/tx-generator/src/Cardano/Benchmarking/Script/Core.hs

@@ -18,12 +18,13 @@
 module Cardano.Benchmarking.Script.Core
 where
 
+import           "contra-tracer" Control.Tracer (Tracer (..))
+
 import           Control.Concurrent (threadDelay)
 import           Control.Monad
 import           Control.Monad.IO.Class
 import           Control.Monad.Trans.Except
 import           Control.Monad.Trans.Except.Extra
-import           "contra-tracer" Control.Tracer (nullTracer)
 import           Data.ByteString.Lazy.Char8 as BSL (writeFile)
 import           Data.Ratio ((%))
 
@@ -37,6 +38,8 @@ import           Cardano.Api
 import           Cardano.Api.Shelley (PlutusScriptOrReferenceInput (..), ProtocolParameters,
                    protocolParamMaxTxExUnits, protocolParamPrices)
 
+import           Cardano.Logging hiding(LocalSocket)
+
 import           Cardano.TxGenerator.Fund as Fund
 import qualified Cardano.TxGenerator.FundQueue as FundQueue
 import           Cardano.TxGenerator.Setup.Plutus as Plutus
@@ -141,8 +144,8 @@ getConnectClient = do
   ioManager <- askIOManager
   return $ benchmarkConnectTxSubmit
                        ioManager
-                       (btConnect_ tracers)
-                       nullTracer -- (btSubmission2_ tracers)
+                       (Tracer $ traceWith (btConnect_ tracers))
+                       mempty -- (btSubmission2_ tracers)
                        (protocolToCodecConfig protocol)
                        networkMagic
 waitBenchmark :: String -> ActionM ()

bench/tx-generator/src/Cardano/Benchmarking/Script/Env.hs

@@ -3,7 +3,6 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
@@ -63,14 +62,15 @@ import           Control.Monad.Trans.Class
 import           Control.Monad.Trans.Except
 import           Control.Monad.Trans.RWS.Strict (RWST)
 import qualified Control.Monad.Trans.RWS.Strict as RWS
-import           "contra-tracer" Control.Tracer (traceWith)
 import           Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
 import qualified Data.Text as Text
 import           Prelude
 
 import           Cardano.Api (File (..), SocketPath)
 
+import           Cardano.Logging
+
 import           Cardano.Benchmarking.GeneratorTx
 import qualified Cardano.Benchmarking.LogTypes as Tracer
 import           Cardano.Benchmarking.OuroborosImports (NetworkId, PaymentKey, ShelleyGenesis,

bench/tx-generator/src/Cardano/Benchmarking/Tracer.hs

@@ -8,7 +8,6 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
@@ -25,7 +24,6 @@ module Cardano.Benchmarking.Tracer
   )
 where
 
-import           "contra-tracer" Control.Tracer (Tracer (..), nullTracer)
 import           GHC.Generics
 
 import           Data.Aeson as A
@@ -72,10 +70,10 @@ generatorTracer tracerName mbTrStdout mbTrForward = do
 
 initNullTracers :: BenchTracers
 initNullTracers = BenchTracers
-    { btTxSubmit_    = nullTracer
-    , btConnect_     = nullTracer
-    , btSubmission2_ = nullTracer
-    , btN2N_         = nullTracer
+    { btTxSubmit_    = mempty
+    , btConnect_     = mempty
+    , btSubmission2_ = mempty
+    , btN2N_         = mempty
     }
 
 -- if the first argument isJust, we assume we have a socket path
@@ -87,20 +85,24 @@ initTxGenTracers mbForwarding = do
   confState       <- emptyConfigReflection
 
   let
-    mkTracer :: (LogFormatting a, MetaTrace a) => Text -> IO (Tracer IO a)
-    mkTracer namespace
-      | isPrefixSilent namespace = pure nullTracer
+    mkTracer :: (LogFormatting a, MetaTrace a)
+                  => Text
+                  -> Maybe (Trace IO FormattedMessage)
+                  -> Maybe (Trace IO FormattedMessage)
+                  -> IO (Trace IO a)
+    mkTracer namespace mbStdoutTracer' mbForwardingTracer'
+      | isPrefixSilent namespace = pure mempty
       | otherwise = do
-        tracer <-  generatorTracer namespace mbStdoutTracer mbForwardingTracer
+        tracer <-  generatorTracer namespace mbStdoutTracer' mbForwardingTracer'
         configureTracers confState initialTraceConfig [tracer]
-        pure $ Tracer (traceWith tracer)
+        pure tracer
 
-  benchTracer@(Tracer traceBench) <- mkTracer "Benchmark"
-  n2nSubmitTracer <- mkTracer "SubmitN2N"
-  connectTracer <- mkTracer "Connect"
-  submitTracer <- mkTracer "Submit"
+  benchTracer <- mkTracer "Benchmark" mbStdoutTracer mbForwardingTracer
+  n2nSubmitTracer <- mkTracer "SubmitN2N" mbStdoutTracer mbForwardingTracer
+  connectTracer <- mkTracer "Connect" mbStdoutTracer mbForwardingTracer
+  submitTracer <- mkTracer "Submit" mbStdoutTracer mbForwardingTracer
 
-  traceBench $ TraceTxGeneratorVersion Version.txGeneratorVersion
+  traceWith benchTracer (TraceTxGeneratorVersion Version.txGeneratorVersion)
 
   return $ BenchTracers
     { btTxSubmit_    = benchTracer

cardano-node/src/Cardano/Node/Tracing/Tracers/Peer.hs

@@ -4,14 +4,11 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PackageImports #-}
 
-module Cardano.Node.Tracing.Tracers.Peer where
---   ( PeerT (..)
---   , startPeerTracer
---   , namesForPeers
---   , severityPeers
---   , docPeers
---   , ppPeer
---   ) where
+module Cardano.Node.Tracing.Tracers.Peer
+  ( PeerT (..)
+  , startPeerTracer
+  , ppPeer
+  ) where
 
 import           Cardano.Node.Orphans ()
 

trace-dispatcher/src/Cardano/Logging/Tracer/Standard.hs

@@ -5,15 +5,16 @@ module Cardano.Logging.Tracer.Standard (
     standardTracer
 ) where
 
+import           Control.Concurrent (myThreadId)
 import           Control.Concurrent.Async
 import           Control.Concurrent.Chan.Unagi.Bounded
-import           Control.Exception (BlockedIndefinitelyOnMVar, catch)
 import           Control.Monad (forever, when)
 import           Control.Monad.IO.Class
 import           Data.IORef (IORef, modifyIORef', newIORef, readIORef)
 import           Data.Maybe (isNothing)
 import           Data.Text (Text)
 import qualified Data.Text.IO as TIO
+import           GHC.Conc (labelThread)
 import           System.IO (hFlush, stdout)
 
 import           Cardano.Logging.DocuGenerator
@@ -29,6 +30,8 @@ newtype StandardTracerState =  StandardTracerState {
 emptyStandardTracerState :: StandardTracerState
 emptyStandardTracerState = StandardTracerState Nothing
 
+-- | It is mandatory to construct only one standard tracer in any application!
+-- Throwing away a standard tracer and using a new one will result in an exception
 standardTracer :: forall m. (MonadIO m)
   => m (Trace m FormattedMessage)
 standardTracer = do
@@ -66,9 +69,10 @@ standardTracer = do
 startStdoutThread :: IORef StandardTracerState -> IO ()
 startStdoutThread stateRef = do
     (inChan, outChan) <- newChan 2048
-    as <- async (catch
-      (stdoutThread outChan)
-      (\(_ :: BlockedIndefinitelyOnMVar) -> pure ()))
+    as <- async (do
+                    tid <- myThreadId
+                    labelThread tid "StdoutTrace"
+                    stdoutThread outChan)
     link as
     modifyIORef' stateRef (\ st ->
       st {stRunning = Just (inChan, outChan, as)})