Skip to content

Commit cdf528a

Browse files
jutarodeepfire
jutaro
authored and
deepfire
committed
trace-dispatcher: Starting forwarding stress test
1 parent 2aed8e1 commit cdf528a

File tree

10 files changed

+309
-14
lines changed

10 files changed

+309
-14
lines changed

‎.gitignore

+2
Original file line numberDiff line numberDiff line change
@@ -63,3 +63,5 @@ logs
6363
/example
6464
/.hlint.yaml
6565
/testnet
66+
67+
.vscode/

‎trace-dispatcher/bench/trace-dispatcher-bench.hs

-1
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@ import Cardano.Logging.Test.Tracer
1010
import Cardano.Logging.Test.Types
1111
import System.Remote.Monitoring (forkServer)
1212

13-
import Debug.Trace
1413

1514
-- Can be run with:
1615
-- cabal bench trace-dispatcher-bench --benchmark-option='-o benchmark-trace.html'

‎trace-dispatcher/src/Cardano/Logging/Tracer/Forward.hs

+4-1
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import Trace.Forward.Utils.TraceObject (ForwardSink, writeToSink)
1515
import Cardano.Logging.DocuGenerator
1616
import Cardano.Logging.Types
1717

18+
1819
---------------------------------------------------------------------------
1920

2021
forwardTracer :: forall m. (MonadIO m)
@@ -34,4 +35,6 @@ forwardTracer forwardSink =
3435
pure ()
3536
output _sink lk (Left c@Document {}) =
3637
docIt Forwarder (lk, Left c)
37-
output _sink LoggingContext {} _ = pure ()
38+
output _sink LoggingContext {} (Right _) = pure ()
39+
-- writeToSink sink lo
40+
output _sink LoggingContext {} _ = pure ()

‎trace-dispatcher/test/Cardano/Logging/ForwardingStressTest/Script.hs

+216
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,216 @@
1+
{-# LANGUAGE DeriveGeneric #-}
2+
{-# LANGUAGE FlexibleInstances #-}
3+
{-# LANGUAGE LambdaCase #-}
4+
{-# LANGUAGE RecordWildCards #-}
5+
{-# LANGUAGE ScopedTypeVariables #-}
6+
{-# LANGUAGE StandaloneDeriving #-}
7+
{-# LANGUAGE UndecidableInstances #-}
8+
9+
module Cardano.Logging.ForwardingStressTest.Script
10+
( TestSetup(..)
11+
, simpleTestConfig
12+
, getTestSetup
13+
, runScriptForwarding
14+
) where
15+
16+
import Control.Concurrent (ThreadId, forkFinally, threadDelay)
17+
import Control.Concurrent.MVar
18+
import Control.Exception.Base (SomeException, throw)
19+
import Control.Monad (join, when)
20+
import Data.Functor ((<&>))
21+
import Data.Functor.Identity
22+
import Data.IORef
23+
import Data.List (sort)
24+
import Data.Map (fromList)
25+
import Data.Maybe
26+
import Data.Monoid
27+
import GHC.Generics (Generic)
28+
import Generic.Data (gmappend)
29+
import Options.Applicative
30+
import Ouroboros.Network.Magic (NetworkMagic (..))
31+
32+
import Test.QuickCheck
33+
34+
import Cardano.Logging
35+
import Cardano.Logging.Test.Config ()
36+
import Cardano.Logging.Test.Messages
37+
import Cardano.Logging.Test.Types
38+
39+
40+
import Debug.Trace
41+
42+
43+
data TestSetup a
44+
= TestSetup
45+
{ tsTime :: !(a Double)
46+
, tsThreads :: !(a Int)
47+
, tsMessages :: !(a (Maybe Int))
48+
, tsSocketPath :: !(a FilePath)
49+
, tsNetworkMagic :: !(a NetworkMagic)
50+
} deriving (Generic)
51+
instance Semigroup (TestSetup Last) where
52+
(<>) = gmappend
53+
54+
deriving instance Show (TestSetup Identity)
55+
56+
defaultTestSetup :: TestSetup Last
57+
defaultTestSetup =
58+
TestSetup
59+
{ tsTime = Last $ Just 10.0
60+
, tsThreads = Last $ Just 5
61+
, tsMessages = Last Nothing
62+
, tsSocketPath = Last $ Just "/tmp/tracer.sock"
63+
, tsNetworkMagic = Last $ Just testnetMagic
64+
}
65+
where testnetMagic = NetworkMagic 764824073
66+
67+
parseTestSetup :: Parser (TestSetup Last)
68+
parseTestSetup =
69+
TestSetup
70+
<$> (Last <$> optional (option auto (long "time" <> metavar "SEC")))
71+
<*> (Last <$> optional (option auto (long "threads" <> metavar "THRDS")))
72+
<*> (Last <$> optional (option auto (long "messages" <> metavar "MSGS")))
73+
<*> (Last <$> optional (option auto (long "socket" <> metavar "FILE")))
74+
<*> (Last <$> optional (option (NetworkMagic <$> auto)
75+
(long "network-magic" <> metavar "INT")))
76+
77+
mergeTestSetup :: TestSetup Last -> TestSetup Identity
78+
mergeTestSetup TestSetup{..} =
79+
TestSetup
80+
{ tsTime = get "Missing tsTime" tsTime
81+
, tsThreads = get "Missing tsThreads" tsThreads
82+
, tsMessages = Identity . join $ getLast tsMessages
83+
, tsSocketPath = get "Missing tsSocketPath" tsSocketPath
84+
, tsNetworkMagic = get "Missing tsNetworkMagic" tsNetworkMagic
85+
}
86+
where
87+
get desc = Identity . fromMaybe (error $ "Missing " <> desc) . getLast
88+
89+
getTestSetup :: IO (TestSetup Identity)
90+
getTestSetup =
91+
customExecParser
92+
(prefs showHelpOnEmpty)
93+
(info parseTestSetup mempty)
94+
<&> (defaultTestSetup <>)
95+
<&> mergeTestSetup
96+
97+
-- | configuration for testing
98+
simpleTestConfig :: TraceConfig
99+
simpleTestConfig = emptyTraceConfig {
100+
tcOptions = fromList
101+
[([] :: Namespace,
102+
[ ConfSeverity (SeverityF (Just Debug))
103+
, ConfDetail DNormal
104+
, ConfBackend [Forwarder]
105+
])
106+
]
107+
}
108+
109+
-- | Run scripts in three threads in parallel.
110+
-- The duration of the test is given by time in seconds
111+
runScriptForwarding ::
112+
TestSetup Identity
113+
-> Trace IO FormattedMessage
114+
-> Trace IO FormattedMessage
115+
-> IORef Int
116+
-> Property
117+
runScriptForwarding ts@TestSetup{..} fwdTracer stdoutTracer' accumulationCounter =
118+
trace ("Test setup " ++ show ts) $ do
119+
let generator :: Gen [Script] = vectorOf (runIdentity tsThreads) $
120+
case runIdentity tsMessages of
121+
Nothing -> scale (* 1000) arbitrary
122+
Just numMsg -> Script <$> vectorOf numMsg arbitrary
123+
forAll generator (\ (scripts :: [Script])
124+
-> ioProperty $ do
125+
tr <- mkCardanoTracer
126+
stdoutTracer'
127+
fwdTracer
128+
Nothing
129+
["Test"]
130+
namesForMessage
131+
severityForMessage
132+
privacyForMessage
133+
configureTracers simpleTestConfig docMessage [tr]
134+
let scripts' = map (\ (Script sc) -> Script
135+
$ filter (\(ScriptedMessage _ msg) ->
136+
namesForMessage msg /= ["Message2"]) sc) scripts
137+
scripts'' = map (\ (Script sc) -> Script (sort sc)) scripts'
138+
scripts''' = zipWith (\ (Script sc) ind -> Script (
139+
withMessageIds (runIdentity tsThreads) ind sc)) scripts'' [0..]
140+
scripts'''' = map (\ (Script sc) -> Script
141+
$ map (withTimeFactor (runIdentity tsTime)) sc) scripts'''
142+
143+
144+
-- putStrLn ("runTest " ++ show scripts)
145+
children :: MVar [MVar (Either SomeException ())] <- newMVar []
146+
mapM_ (\sc -> forkChild children (playIt sc tr 0.0)) scripts''''
147+
res <- waitForChildren children []
148+
let resErr = mapMaybe
149+
(\case
150+
Right _ -> Nothing
151+
Left err -> Just err) res
152+
threadDelay 500000 --wait 0,5 seconds
153+
if not (null resErr)
154+
then throw (head resErr)
155+
else -- Oracle
156+
let numMsg = sum (map (\ (Script sc) -> length sc) scripts'''')
157+
in if numMsg > 0 then do
158+
-- TODO mutiple files
159+
contents <- readFile "/tmp/cardano-forwarder-test-logs/tmp-tracersock@0/node.json"
160+
let lineLength = length (lines contents) - 1
161+
putStrLn $ "Line length " ++ show lineLength
162+
putStrLn $ "Msg length " ++ show numMsg
163+
totalNumMsg <- atomicModifyIORef accumulationCounter (\ac ->
164+
let nc = ac + numMsg
165+
in (nc, nc))
166+
pure (totalNumMsg == lineLength)
167+
else do
168+
putStrLn "Empty test"
169+
pure True
170+
171+
)
172+
173+
forkChild :: MVar [MVar (Either SomeException ())] -> IO () -> IO ThreadId
174+
forkChild children io = do
175+
mvar <- newEmptyMVar
176+
childs <- takeMVar children
177+
putMVar children (mvar:childs)
178+
forkFinally io (putMVar mvar)
179+
180+
waitForChildren :: MVar [MVar (Either SomeException ())]
181+
-> [Either SomeException ()]
182+
-> IO [Either SomeException ()]
183+
waitForChildren children accum = do
184+
cs <- takeMVar children
185+
case cs of
186+
[] -> pure accum
187+
m:ms -> do
188+
putMVar children ms
189+
res <- takeMVar m
190+
waitForChildren children (res : accum)
191+
192+
193+
-- | Play the current script in one thread
194+
-- The time is in milliseconds
195+
playIt :: Script -> Trace IO Message -> Double -> IO ()
196+
playIt (Script []) _tr _d = pure ()
197+
playIt (Script (ScriptedMessage d1 m1 : rest)) tr d = do
198+
when (d < d1) $ threadDelay (round ((d1 - d) * 1000000))
199+
-- this is in microseconds
200+
traceWith tr m1
201+
playIt (Script rest) tr d1
202+
203+
-- | Adds a message id to every message.
204+
-- MessageId gives the id to start with.
205+
-- Returns a tuple with the messages with ids and
206+
-- the successor of the last used messageId
207+
withMessageIds :: Int -> MessageID -> [ScriptedMessage] -> [ScriptedMessage]
208+
withMessageIds numThreads mid sMsgs = go mid sMsgs []
209+
where
210+
go _mid' [] acc = reverse acc
211+
go mid' (ScriptedMessage time msg : tl) acc =
212+
go (mid' + numThreads) tl (ScriptedMessage time (setMessageID msg mid') : acc)
213+
214+
withTimeFactor :: Double -> ScriptedMessage -> ScriptedMessage
215+
withTimeFactor factor (ScriptedMessage time msg) =
216+
ScriptedMessage (time * factor) msg

‎trace-dispatcher/test/Cardano/Logging/Test/Oracles.hs

+1-3
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,6 @@ import Cardano.Logging
1414
import Cardano.Logging.Test.Messages
1515
import Cardano.Logging.Test.Types
1616

17-
import Debug.Trace
18-
1917

2018
-- | Checks for every message that it appears or does not appear at the right
2119
-- backend. Tests filtering and routing to backends
@@ -41,7 +39,7 @@ oracleMessages conf ScriptRes {..} =
4139
res = isCorrectStdout && isCorrectForwarder && isCorrectEKG
4240
in case traceMessage isCorrectStdout isCorrectForwarder isCorrectEKG msg of
4341
Nothing -> res
44-
Just str -> trace str res
42+
Just str -> error (str ++ " " ++ show res)
4543
traceMessage :: Bool -> Bool -> Bool -> Message -> Maybe String
4644
traceMessage isCorrectStdout isCorrectForwarder isCorrectEKG msg
4745
| not isCorrectStdout

‎trace-dispatcher/test/Cardano/Logging/Test/Script.hs

-2
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,6 @@ import Cardano.Logging.Test.Messages
2626
import Cardano.Logging.Test.Tracer
2727
import Cardano.Logging.Test.Types
2828

29-
-- import Debug.Trace
30-
3129

3230
-- | Run a script in a single thread and uses the oracle to test for correctness
3331
-- The duration of the test is given by time in seconds

‎trace-dispatcher/test/Cardano/Logging/Test/Types.hs

+1-1
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ instance LogFormatting Message where
3939
]
4040
forMachine _dtal (Message3 mid d) =
4141
mconcat [ "kind" .= String "Message3"
42-
, "mid" .= String (showT mid)
42+
, "mid" .= String ("<" <> showT mid <> ">")
4343
, "workload" .= String (showT d)
4444
]
4545
forHuman (Message1 mid i) =

‎trace-dispatcher/test/cardano-tracer-config.yaml

+9
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
2+
networkMagic: 764824073
3+
network:
4+
tag: AcceptAt
5+
contents: "/tmp/tracer.sock"
6+
logging:
7+
- logRoot: "/tmp/cardano-forwarder-test-logs"
8+
logMode: FileMode
9+
logFormat: ForMachine

‎trace-dispatcher/test/trace-dispatcher-test.hs

+65-5
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,79 @@
1-
1+
{-# LANGUAGE ScopedTypeVariables #-}
22
{-# OPTIONS_GHC -Wno-unused-imports #-}
33

4+
import Control.Concurrent (threadDelay)
5+
import Control.Exception
6+
import Control.Monad (when)
7+
import Data.Functor.Identity
8+
import System.Directory (removeDirectoryRecursive)
9+
import System.PosixCompat.Files (fileExist)
10+
import System.Process (cleanupProcess, spawnProcess)
411
import Test.Tasty
512
import Test.Tasty.QuickCheck
13+
import Data.IORef (IORef, newIORef)
614

15+
import Cardano.Api (SocketPath (..))
716
import Cardano.Logging
17+
import Cardano.Logging.ForwardingStressTest.Script
818
import Cardano.Logging.Test.Oracles
919
import Cardano.Logging.Test.Script
20+
import Ouroboros.Network.NodeToClient (withIOManager)
1021

1122

1223
main :: IO ()
13-
main = defaultMain tests
24+
main = do
25+
ts <- getTestSetup
26+
let logPath = "/tmp/cardano-forwarder-test-logs"
27+
fe <- fileExist logPath
28+
when fe
29+
(removeDirectoryRecursive logPath)
30+
cardanoTracerHdl <- spawnProcess "cardano-tracer"
31+
[ "-c"
32+
, "test/cardano-tracer-config.yaml"]
33+
threadDelay 1000000 --wait 1 seconds
34+
accumulationCounter <- newIORef 0
35+
fwdTracer <- do
36+
-- TODO: check if this is the correct way to use withIOManager
37+
(forwardSink, _dpStore) <- withIOManager $ \iomgr -> do
38+
-- For simplicity, we are always 'Initiator',
39+
-- so 'cardano-tracer' is always a 'Responder'.
40+
let tracerSocketMode = Just (runIdentity (tsSocketPath ts), Initiator)
41+
initForwarding iomgr simpleTestConfig (runIdentity (tsNetworkMagic ts)) Nothing tracerSocketMode
42+
pure (forwardTracer forwardSink)
43+
stdoutTracer' <- standardTracer
44+
45+
defaultMain (allTests ts fwdTracer stdoutTracer' accumulationCounter)
46+
`catch` (\ (e :: SomeException) -> do
47+
cleanupProcess (Nothing, Nothing, Nothing, cardanoTracerHdl)
48+
throwIO e)
49+
50+
51+
allTests ::
52+
TestSetup Identity
53+
-> Trace IO FormattedMessage
54+
-> Trace IO FormattedMessage
55+
-> IORef Int
56+
-> TestTree
57+
allTests ts fwdTracer stdoutTracer' accumulationCounter =
58+
testGroup "Tests"
59+
[ localTests
60+
, forwarderTests ts fwdTracer stdoutTracer' accumulationCounter
61+
]
1462

15-
tests :: TestTree
16-
tests = localOption (QuickCheckTests 10) $ testGroup "trace-dispatcher"
63+
forwarderTests ::
64+
TestSetup Identity
65+
-> Trace IO FormattedMessage
66+
-> Trace IO FormattedMessage
67+
-> IORef Int
68+
-> TestTree
69+
forwarderTests ts fwdTracer stdoutTracer' accumulationCounter =
70+
localOption (QuickCheckTests 10) $ testGroup "trace-forwarder"
71+
[ testProperty "multi-threaded forwarder stress test" $
72+
runScriptForwarding ts fwdTracer stdoutTracer' accumulationCounter
73+
]
74+
75+
localTests :: TestTree
76+
localTests = localOption (QuickCheckTests 10) $ testGroup "trace-dispatcher"
1777
[ testProperty "single-threaded send tests" $
1878
runScriptSimple 1.0 oracleMessages
1979
, testProperty "multi-threaded send tests" $
@@ -22,4 +82,4 @@ tests = localOption (QuickCheckTests 10) $ testGroup "trace-dispatcher"
2282
-- runScriptMultithreadedWithReconfig 1.0 oracleMessages
2383
, testProperty "reconfiguration stress test" $
2484
runScriptMultithreadedWithConstantReconfig 1.0 (\ _ _ -> property True)
25-
]
85+
]

0 commit comments

Comments
 (0)