@@ -26,41 +26,66 @@ toMicroSecs x = x * 10^(6 :: Int)
2626measureRate' :: IsStream t
2727 => String
2828 -> (t IO Int -> SerialT IO Int)
29− -> Double
29+ -> Int -- buffers
30+ -> Int -- threads
31+ -> Either Double Int -- either rate or count of actions
3032 -> Int
3133 -> (Double, Double)
3234 -> (Double, Double)
3335 -> Spec
34−measureRate' desc t rval consumerDelay producerDelay dur =
35− it (desc <> " rate: " <> show rval
36+measureRate' desc t buffers threads rval consumerDelay producerDelay expectedRange = do
37+
38+ let threadAction =
39+ case rval of
40+ Left r -> S.take (round $ 10 * r) . S.repeatM
41+ Right n -> S.replicateM n
42+
43+ rateDesc = case rval of
44+ Left r -> " rate: " <> show r
45+ Right n -> " count: " <> show n
46+
47+ it (desc <> rateDesc
48+ <> " buffers: " <> show buffers
49+ <> " threads: " <> show threads
3650 <> ", consumer latency: " <> show consumerDelay
3751 <> ", producer latency: " <> show producerDelay)
38− $ durationShouldBe dur $
39− runStream
40− $ (if consumerDelay > 0
41− then S.mapM $ \x ->
42− threadDelay (toMicroSecs consumerDelay) >> return x
43− else id)
44− $ t
45− $ maxBuffer (-1)
46− $ maxThreads (-1)
47− $ avgRate rval
48− $ S.take (round $ rval * 10)
49− $ S.repeatM $ do
50− let (t1, t2) = producerDelay
51− r <- if t1 == t2
52− then return $ round $ toMicroSecs t1
53− else randomRIO ( round $ toMicroSecs t1
54− , round $ toMicroSecs t2)
55− when (r > 0) $ -- do
56− -- t1 <- getTime Monotonic
57− threadDelay r
58− -- t2 <- getTime Monotonic
59− -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
60− -- putStrLn $ "delay took: " <> show delta
61− -- when (delta > 2) $ do
62− -- putStrLn $ "delay took high: " <> show delta
63− return 1
52+ $ durationShouldBe expectedRange $
53+ runStream
54+ $ (if consumerDelay > 0
55+ then S.mapM $ \x ->
56+ threadDelay (toMicroSecs consumerDelay) >> return x
57+ else id)
58+ $ t
59+ $ maxBuffer buffers
60+ $ maxThreads threads
61+ $ (case rval of {Left r -> avgRate r; Right _ -> rate Nothing})
62+ $ threadAction $ do
63+ let (t1, t2) = producerDelay
64+ r <- if t1 == t2
65+ then return $ round $ toMicroSecs t1
66+ else randomRIO ( round $ toMicroSecs t1
67+ , round $ toMicroSecs t2)
68+ when (r > 0) $ -- do
69+ -- t1 <- getTime Monotonic
70+ threadDelay r
71+ -- t2 <- getTime Monotonic
72+ -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
73+ -- putStrLn $ "delay took: " <> show delta
74+ -- when (delta > 2) $ do
75+ -- putStrLn $ "delay took high: " <> show delta
76+ return 1
77+
78+measureRateVariable :: IsStream t
79+ => String
80+ -> (t IO Int -> SerialT IO Int)
81+ -> Double
82+ -> Int
83+ -> (Double, Double)
84+ -> (Double, Double)
85+ -> Spec
86+measureRateVariable desc t rval consumerDelay producerDelay dur =
87+ measureRate' desc t (-1) (-1) (Left rval)
88+ consumerDelay producerDelay dur
6489
6590measureRate :: IsStream t
6691 => String
@@ -72,10 +97,54 @@ measureRate :: IsStream t
7297 -> Spec
7398measureRate desc t rval consumerDelay producerDelay dur =
7499 let d = fromIntegral producerDelay
75− in measureRate' desc t rval consumerDelay (d, d) dur
100+ in measureRateVariable desc t rval consumerDelay (d, d) dur
101+
102+measureThreads :: IsStream t
103+ => String
104+ -> (t IO Int -> SerialT IO Int)
105+ -> Int -- threads
106+ -> Int -- count of actions
107+ -> Spec
108+measureThreads desc t threads count = do
109+ let expectedTime =
110+ if threads < 0
111+ then 1.0
112+ else fromIntegral count / fromIntegral threads
113+ duration = (expectedTime * 0.9, expectedTime * 1.1)
114+ measureRate' desc t (-1) threads (Right count) 0 (1,1) duration
115+
116+measureBuffers :: IsStream t
117+ => String
118+ -> (t IO Int -> SerialT IO Int)
119+ -> Int -- buffers
120+ -> Int -- count of actions
121+ -> Spec
122+measureBuffers desc t buffers count = do
123+ let expectedTime =
124+ if buffers < 0
125+ then 1.0
126+ else fromIntegral count / fromIntegral buffers
127+ duration = (expectedTime * 0.9, expectedTime * 1.1)
128+ measureRate' desc t buffers (-1) (Right count) 0 (1,1) duration
76129
77130main :: IO ()
78131main = hspec $ do
132+
133+ describe "maxBuffers" $ do
134+ measureBuffers "asyncly" asyncly (-1) 5
135+ -- XXX this test fails due to a known issue
136+ -- measureBuffers "maxBuffers" asyncly 1 5
137+ measureBuffers "asyncly" asyncly 5 5
138+
139+ describe "maxThreads" $ do
140+ measureThreads "asyncly" asyncly (-1) 5
141+ measureThreads "asyncly" asyncly 1 5
142+ measureThreads "asyncly" asyncly 5 5
143+
144+ measureThreads "aheadly" aheadly (-1) 5
145+ measureThreads "aheadly" aheadly 1 5
146+ measureThreads "aheadly" aheadly 5 5
147+
79148 let range = (8,12)
80149
81150 -- Note that because after the last yield we don't wait, the last period
@@ -108,7 +177,7 @@ main = hspec $ do
108177 ]
109178 in describe "asyncly no consumer delay and variable producer delay" $
110179 forM_ rates $ \r ->
111− measureRate' "asyncly" asyncly r 0 (0.1, 3) range
180+ measureRateVariable "asyncly" asyncly r 0 (0.1, 3) range
112181
113182 let rates = [1, 10, 100, 1000, 10000
114183#ifndef __GHCJS__