git.iko.soy/streamly
on 3da7028 help sign in
streamly/test/MaxRate.hs
227 lines 8.1 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE FlexibleContexts #-}
4import Streamly
5import qualified Streamly.Prelude as S
6import Control.Concurrent
7import Control.Monad
8import System.Clock
9import Test.Hspec
10import System.Random
12durationShouldBe :: (Double, Double) -> IO () -> Expectation
13durationShouldBe d@(tMin, tMax) action = do
14 t0 <- getTime Monotonic
15 action
16 t1 <- getTime Monotonic
17 let t = fromIntegral (toNanoSecs (t1 - t0)) / 1e9
18 -- tMax = fromNanoSecs (round $ d*10^9*1.2)
19 -- tMin = fromNanoSecs (round $ d*10^9*0.8)
20 putStrLn $ "Expected: " <> show d <> " Took: " <> show t
21 (t <= tMax && t >= tMin) `shouldBe` True
23toMicroSecs :: Num a => a -> a
24toMicroSecs x = x * 10^(6 :: Int)
26measureRate' :: IsStream t
27 => String
28 -> (t IO Int -> SerialT IO Int)
29 -> Int -- buffers
30 -> Int -- threads
31 -> Either Double Int -- either rate or count of actions
32 -> Int
33 -> (Double, Double)
34 -> (Double, Double)
35 -> Spec
36measureRate' desc t buffers threads rval consumerDelay producerDelay expectedRange = do
38 let threadAction =
39 case rval of
40 Left r -> S.take (round $ 10 * r) . S.repeatM
41 Right n -> S.replicateM n
43 rateDesc = case rval of
44 Left r -> " rate: " <> show r
45 Right n -> " count: " <> show n
47 it (desc <> rateDesc
48 <> " buffers: " <> show buffers
49 <> " threads: " <> show threads
50 <> ", consumer latency: " <> show consumerDelay
51 <> ", producer latency: " <> show producerDelay)
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
78measureRateVariable :: IsStream t
79 => String
80 -> (t IO Int -> SerialT IO Int)
81 -> Double
82 -> Int
83 -> (Double, Double)
84 -> (Double, Double)
85 -> Spec
86measureRateVariable desc t rval consumerDelay producerDelay dur =
87 measureRate' desc t (-1) (-1) (Left rval)
88 consumerDelay producerDelay dur
90measureRate :: IsStream t
91 => String
92 -> (t IO Int -> SerialT IO Int)
93 -> Double
94 -> Int
95 -> Int
96 -> (Double, Double)
97 -> Spec
98measureRate desc t rval consumerDelay producerDelay dur =
99 let d = fromIntegral producerDelay
100 in measureRateVariable desc t rval consumerDelay (d, d) dur
102measureThreads :: IsStream t
103 => String
104 -> (t IO Int -> SerialT IO Int)
105 -> Int -- threads
106 -> Int -- count of actions
107 -> Spec
108measureThreads 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
116measureBuffers :: IsStream t
117 => String
118 -> (t IO Int -> SerialT IO Int)
119 -> Int -- buffers
120 -> Int -- count of actions
121 -> Spec
122measureBuffers 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
130main :: IO ()
131main = hspec $ do
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
139 describe "maxThreads" $ do
140 measureThreads "asyncly" asyncly (-1) 5
141 measureThreads "asyncly" asyncly 1 5
142 measureThreads "asyncly" asyncly 5 5
144 measureThreads "aheadly" aheadly (-1) 5
145 measureThreads "aheadly" aheadly 1 5
146 measureThreads "aheadly" aheadly 5 5
148 let range = (8,12)
150 -- Note that because after the last yield we don't wait, the last period
151 -- will be effectively shorter. This becomes significant when the rates are
152 -- lower (1 or lower). For rate 1 we lose 1 second in the end and for rate
153 -- 10 0.1 second.
154 let rates = [1, 10, 100, 1000, 10000
155#ifndef __GHCJS__
156 , 100000, 1000000
157#endif
158 ]
159 in describe "asyncly no consumer delay no producer delay" $
160 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 0 range)
162 -- XXX try staggering the dispatches to achieve higher rates
163 let rates = [1, 10, 100, 1000
164#ifndef __GHCJS__
165 , 10000, 25000
166#endif
167 ]
168 in describe "asyncly no consumer delay and 1 sec producer delay" $
169 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 1 range)
171 -- At lower rates (1/10) this is likely to vary quite a bit depending on
172 -- the spread of random producer latencies generated.
173 let rates = [1, 10, 100, 1000
174#ifndef __GHCJS__
175 , 10000, 25000
176#endif
177 ]
178 in describe "asyncly no consumer delay and variable producer delay" $
179 forM_ rates $ \r ->
180 measureRateVariable "asyncly" asyncly r 0 (0.1, 3) range
182 let rates = [1, 10, 100, 1000, 10000
183#ifndef __GHCJS__
184 , 100000, 1000000
185#endif
186 ]
187 in describe "wAsyncly no consumer delay no producer delay" $
188 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 0 range)
190 let rates = [1, 10, 100, 1000
191#ifndef __GHCJS__
192 , 10000, 25000
193#endif
194 ]
195 in describe "wAsyncly no consumer delay and 1 sec producer delay" $
196 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 1 range)
198 let rates = [1, 10, 100, 1000, 10000
199#ifndef __GHCJS__
200 , 100000, 1000000
201#endif
202 ]
203 in describe "aheadly no consumer delay no producer delay" $
204 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 0 range)
206 -- XXX after the change to stop workers when the heap is clearing
207 -- thi does not work well at a 25000 ops per second, need to fix.
208 let rates = [1, 10, 100, 1000
209#ifndef __GHCJS__
210 , 10000, 12500
211#endif
212 ]
213 in describe "aheadly no consumer delay and 1 sec producer delay" $
214 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 1 range)
216 describe "asyncly with 1 sec producer delay and some consumer delay" $ do
217 -- ideally it should take 10 x 1 + 1 seconds
218 forM_ [1] (\r -> measureRate "asyncly" asyncly r 1 1 (11, 16))
219 -- ideally it should take 10 x 2 + 1 seconds
220 forM_ [1] (\r -> measureRate "asyncly" asyncly r 2 1 (21, 23))
221 -- ideally it should take 10 x 3 + 1 seconds
222 forM_ [1] (\r -> measureRate "asyncly" asyncly r 3 1 (31, 33))
224 describe "aheadly with 1 sec producer delay and some consumer delay" $ do
225 forM_ [1] (\r -> measureRate "aheadly" aheadly r 1 1 (11, 16))
226 forM_ [1] (\r -> measureRate "aheadly" aheadly r 2 1 (21, 23))
227 forM_ [1] (\r -> measureRate "aheadly" aheadly r 3 1 (31, 33))