Merge #3496

3496: trace-forward: new API for acceptor's part. r=denisshevchenko a=denisshevchenko

This PR changes API of the acceptor's part only, it doesn't touch the forwarder's part (used by `trace-dispatcher`):

1. `IO`-action returning `DataPointAsker` is provided. It allows avoiding `unsafePerformIO` in the acceptor application.
2. Peer error handlers were added. It allows the acceptor application to know if the connection with the node was dropped.

Additionally:

1. `ekg-forward` dependency is updated.
2. `DataPoint` protocol is fixed: now there is a check if the reply with `DataPoint`s is empty.

**Please note that `cardano-tracer` service, the next part of the new tracing infrastructure, depends on this PR. After it will be merged, the PR for `cardano-tracer` will be opened.**

Co-authored-by: Denis Shevchenko <[email protected]>

Author: iohk-bors[bot]

cabal.project

@@ -279,8 +279,8 @@ source-repository-package
 source-repository-package
   type: git
   location: https://github.com/input-output-hk/ekg-forward
-  tag: 2adc8b698443bb10154304b24f6c1d6913bb65b9
-  --sha256: 0cyixq3jmq43zs1yzrycqw1klyjy0zxf1vifknnr1k9d6sc3zf6b
+  tag: 297cd9db5074339a2fb2e5ae7d0780debb670c63
+  --sha256: 1zcwry3y5rmd9lgxy89wsb3k4kpffqji35dc7ghzbz603y1gy24g
 
 source-repository-package
   type: git

trace-forward/src/Trace/Forward/Run/DataPoint/Acceptor.hs

@@ -8,6 +8,8 @@ module Trace.Forward.Run.DataPoint.Acceptor
   ) where
 
 import qualified Codec.Serialise as CBOR
+import           Control.Exception (finally)
+import           Control.Monad (unless)
 import           Control.Monad.Extra (ifM)
 import           Control.Monad.STM (atomically, check)
 import           Control.Concurrent.STM.TVar (modifyTVar', readTVar, readTVarIO)
@@ -21,46 +23,52 @@ import qualified Trace.Forward.Protocol.DataPoint.Acceptor as Acceptor
 import qualified Trace.Forward.Protocol.DataPoint.Codec as Acceptor
 import           Trace.Forward.Protocol.DataPoint.Type (DataPointName)
 import           Trace.Forward.Configuration.DataPoint (AcceptorConfiguration (..))
-import           Trace.Forward.Utils.DataPoint (DataPointAsker (..))
+import           Trace.Forward.Utils.DataPoint (DataPointRequestor (..))
 
 acceptDataPointsInit
   :: AcceptorConfiguration
-  -> DataPointAsker
+  -> IO DataPointRequestor
+  -> IO ()
   -> RunMiniProtocol 'InitiatorMode LBS.ByteString IO () Void
-acceptDataPointsInit config dpAsker =
-  InitiatorProtocolOnly $ runPeerWithAsker config dpAsker
+acceptDataPointsInit config mkDPRequestor peerErrorHandler =
+  InitiatorProtocolOnly $ runPeerWithRequestor config mkDPRequestor peerErrorHandler
 
 acceptDataPointsResp
   :: AcceptorConfiguration
-  -> DataPointAsker
+  -> IO DataPointRequestor
+  -> IO ()
   -> RunMiniProtocol 'ResponderMode LBS.ByteString IO Void ()
-acceptDataPointsResp config dpAsker =
-  ResponderProtocolOnly $ runPeerWithAsker config dpAsker
+acceptDataPointsResp config mkDPRequestor peerErrorHandler =
+  ResponderProtocolOnly $ runPeerWithRequestor config mkDPRequestor peerErrorHandler
 
-runPeerWithAsker
+runPeerWithRequestor
   :: AcceptorConfiguration
-  -> DataPointAsker
+  -> IO DataPointRequestor
+  -> IO ()
   -> MuxPeer LBS.ByteString IO ()
-runPeerWithAsker config dpAsker = 
-  MuxPeerRaw $ \channel ->
-      runPeer
-        (acceptorTracer config)
-        (Acceptor.codecDataPointForward CBOR.encode CBOR.decode
-                                        CBOR.encode CBOR.decode)
-        channel
-        (Acceptor.dataPointAcceptorPeer $ acceptorActions config dpAsker [])
+runPeerWithRequestor config mkDPRequestor peerErrorHandler = 
+  MuxPeerRaw $ \channel -> do
+    dpRequestor <- mkDPRequestor
+    runPeer
+      (acceptorTracer config)
+      (Acceptor.codecDataPointForward CBOR.encode CBOR.decode
+                                      CBOR.encode CBOR.decode)
+      channel
+      (Acceptor.dataPointAcceptorPeer $ acceptorActions config dpRequestor [])
+    `finally` peerErrorHandler
 
 acceptorActions
   :: AcceptorConfiguration
-  -> DataPointAsker
+  -> DataPointRequestor
   -> [DataPointName]
   -> Acceptor.DataPointAcceptor IO ()
 acceptorActions config@AcceptorConfiguration{shouldWeStop}
-                dpAsker@DataPointAsker{askDataPoints, dataPointsNames, dataPointsReply}
+                dpRequestor@DataPointRequestor{askDataPoints, dataPointsNames, dataPointsReply}
                 dpNames =
   Acceptor.SendMsgDataPointsRequest dpNames $ \replyWithDataPoints -> do
-    -- Ok, reply with 'DataPoint's is already here, update the asker.
-    atomically $ do
+    -- Ok, reply with 'DataPoint's is already here, update the requestor.
+    unless (null replyWithDataPoints) $ atomically $ do
+      -- Store the reply for acceptor's external context.
       putTMVar dataPointsReply replyWithDataPoints
       -- To prevent new automatic request.
       modifyTVar' askDataPoints $ const False
@@ -72,4 +80,4 @@ acceptorActions config@AcceptorConfiguration{shouldWeStop}
         -- Ok, external context asked for 'DataPoint's, take their names.
         dpNames' <- readTVarIO dataPointsNames
         -- Ask.
-        return $ acceptorActions config dpAsker dpNames'
+        return $ acceptorActions config dpRequestor dpNames'

trace-forward/src/Trace/Forward/Run/TraceObject/Acceptor.hs

@@ -13,7 +13,7 @@ import           Control.Concurrent.Async (race)
 import           Control.Monad.Extra (ifM)
 import           Control.Monad.STM (atomically, check)
 import           Control.Concurrent.STM.TVar (TVar, readTVar, readTVarIO, registerDelay)
-import           Control.Exception (Exception, throwIO)
+import           Control.Exception (Exception, finally, throwIO)
 import qualified Data.ByteString.Lazy as LBS
 import           Data.Typeable (Typeable)
 import           Data.Void (Void)
@@ -34,28 +34,31 @@ acceptTraceObjectsInit
       Typeable lo)
   => AcceptorConfiguration lo -- ^ Acceptor's configuration.
   -> ([lo] -> IO ())          -- ^ The handler for accepted 'TraceObject's.
+  -> IO ()                    -- ^ The handler for exceptions from 'runPeer'.
   -> RunMiniProtocol 'InitiatorMode LBS.ByteString IO () Void
-acceptTraceObjectsInit config loHandler =
-  InitiatorProtocolOnly $ runPeerWithHandler config loHandler
+acceptTraceObjectsInit config loHandler peerErrorHandler =
+  InitiatorProtocolOnly $ runPeerWithHandler config loHandler peerErrorHandler
 
 acceptTraceObjectsResp
   :: (CBOR.Serialise lo,
       ShowProxy lo,
       Typeable lo)
   => AcceptorConfiguration lo -- ^ Acceptor's configuration.
   -> ([lo] -> IO ())          -- ^ The handler for accepted 'TraceObject's.
+  -> IO ()                    -- ^ The handler for exceptions from 'runPeer'.
   -> RunMiniProtocol 'ResponderMode LBS.ByteString IO Void ()
-acceptTraceObjectsResp config loHandler =
-  ResponderProtocolOnly $ runPeerWithHandler config loHandler
+acceptTraceObjectsResp config loHandler peerErrorHandler =
+  ResponderProtocolOnly $ runPeerWithHandler config loHandler peerErrorHandler
 
 runPeerWithHandler
   :: (CBOR.Serialise lo,
       ShowProxy lo,
       Typeable lo)
   => AcceptorConfiguration lo
   -> ([lo] -> IO ())
+  -> IO ()
   -> MuxPeer LBS.ByteString IO ()
-runPeerWithHandler config@AcceptorConfiguration{acceptorTracer, shouldWeStop} loHandler =
+runPeerWithHandler config@AcceptorConfiguration{acceptorTracer, shouldWeStop} loHandler peerErrorHandler =
   MuxPeerRaw $ \channel ->
     timeoutWhenStopped
       shouldWeStop
@@ -66,6 +69,7 @@ runPeerWithHandler config@AcceptorConfiguration{acceptorTracer, shouldWeStop} lo
                                             CBOR.encode CBOR.decode)
           channel
           (Acceptor.traceObjectAcceptorPeer $ acceptorActions config loHandler)
+        `finally` peerErrorHandler
 
 acceptorActions
   :: (CBOR.Serialise lo,
@@ -96,7 +100,7 @@ timeoutWhenStopped stopVar delay action =
   either id id <$>
     race action
          ( do atomically (readTVar stopVar >>= check)
-              v <- registerDelay delay    
-              atomically (readTVar v >>= check) 
+              v <- registerDelay delay
+              atomically (readTVar v >>= check)
               throwIO Timeout
          )

trace-forward/src/Trace/Forward/Utils/DataPoint.hs

@@ -5,9 +5,9 @@
 module Trace.Forward.Utils.DataPoint
   ( DataPoint (..)
   , DataPointStore
-  , DataPointAsker (..)
+  , DataPointRequestor (..)
   , initDataPointStore
-  , initDataPointAsker
+  , initDataPointRequestor
   , writeToStore
   , readFromStore
   , askForDataPoints
@@ -72,7 +72,7 @@ readFromStore dpStore =
 -- | Since 'DataPointForward' protocol does not assume the stream of requests/replies,
 --   we use the 'TVar's to provide to acceptor's side an ability to ask 'DataPoint's
 --   explicitly.
-data DataPointAsker = DataPointAsker
+data DataPointRequestor = DataPointRequestor
   { -- | The "ask flag": we use it to notify that we want 'DataPoint's.
     askDataPoints   :: !(TVar Bool)
     -- | The names of 'DataPoint's we need.
@@ -83,18 +83,18 @@ data DataPointAsker = DataPointAsker
   , dataPointsReply :: !(TMVar DataPointValues)
   }
 
-initDataPointAsker :: IO DataPointAsker
-initDataPointAsker = DataPointAsker
+initDataPointRequestor :: IO DataPointRequestor
+initDataPointRequestor = DataPointRequestor
   <$> newTVarIO False
   <*> newTVarIO []
   <*> newEmptyTMVarIO
 
 askForDataPoints
-  :: DataPointAsker
+  :: DataPointRequestor
   -> [DataPointName]
   -> IO DataPointValues
 askForDataPoints _ [] = return []
-askForDataPoints DataPointAsker{askDataPoints, dataPointsNames, dataPointsReply} dpNames = do
+askForDataPoints DataPointRequestor{askDataPoints, dataPointsNames, dataPointsReply} dpNames = do
   atomically $ do
     modifyTVar' dataPointsNames $ const dpNames -- Fill the names of 'DataPoint's we need.
     modifyTVar' askDataPoints $ const True      -- Ask them! The flag for acceptor's part