git.iko.soy/streamly
on 240a082 help sign in
streamly/test/MaxRate.hs
226 lines 8.1 kB history raw
1import Streamly
2import qualified Streamly.Prelude as S
3import Streamly.Internal.Data.Time.Clock (getTime, Clock(..))
4import Streamly.Internal.Data.Time.Units
5 (NanoSecond64, diffAbsTime64, fromRelTime64)
7import Control.Concurrent
8import Control.Monad
9import System.Random
10import Test.Hspec
12durationShouldBe :: (Double, Double) -> IO () -> Expectation
13durationShouldBe d@(tMin, tMax) action = do
14 t0 <- getTime Monotonic
15 action
16 t1 <- getTime Monotonic
17 let diff = fromRelTime64 (diffAbsTime64 t1 t0) :: NanoSecond64
18 let t = fromIntegral diff / 1e9
19 putStrLn $ "Expected: " <> show d <> " Took: " <> show t
20 (t <= tMax && t >= tMin) `shouldBe` True
22toMicroSecs :: Num a => a -> a
23toMicroSecs x = x * 10^(6 :: Int)
25measureRate' :: IsStream t
26 => String
27 -> (t IO Int -> SerialT IO Int)
28 -> Int -- buffers
29 -> Int -- threads
30 -> Either Double Int -- either rate or count of actions
31 -> Int
32 -> (Double, Double)
33 -> (Double, Double)
34 -> Spec
35measureRate' desc t buffers threads rval consumerDelay producerDelay expectedRange = do
37 let threadAction =
38 case rval of
39 Left r -> S.take (round $ 10 * r) . S.repeatM
40 Right n -> S.replicateM n
42 rateDesc = case rval of
43 Left r -> " rate: " <> show r
44 Right n -> " count: " <> show n
46 it (desc <> rateDesc
47 <> " buffers: " <> show buffers
48 <> " threads: " <> show threads
49 <> ", consumer latency: " <> show consumerDelay
50 <> ", producer latency: " <> show producerDelay)
51 $ durationShouldBe expectedRange $
52 S.drain
53 $ (if consumerDelay > 0
54 then S.mapM $ \x ->
55 threadDelay (toMicroSecs consumerDelay) >> return x
56 else id)
57 $ t
58 $ maxBuffer buffers
59 $ maxThreads threads
60 $ (case rval of {Left r -> avgRate r; Right _ -> rate Nothing})
61 $ threadAction $ do
62 let (t1, t2) = producerDelay
63 r <- if t1 == t2
64 then return $ round $ toMicroSecs t1
65 else randomRIO ( round $ toMicroSecs t1
66 , round $ toMicroSecs t2)
67 when (r > 0) $ -- do
68 -- t1 <- getTime Monotonic
69 threadDelay r
70 -- t2 <- getTime Monotonic
71 -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
72 -- putStrLn $ "delay took: " <> show delta
73 -- when (delta > 2) $ do
74 -- putStrLn $ "delay took high: " <> show delta
75 return 1
77measureRateVariable :: IsStream t
78 => String
79 -> (t IO Int -> SerialT IO Int)
80 -> Double
81 -> Int
82 -> (Double, Double)
83 -> (Double, Double)
84 -> Spec
85measureRateVariable desc t rval consumerDelay producerDelay dur =
86 measureRate' desc t (-1) (-1) (Left rval)
87 consumerDelay producerDelay dur
89measureRate :: IsStream t
90 => String
91 -> (t IO Int -> SerialT IO Int)
92 -> Double
93 -> Int
94 -> Int
95 -> (Double, Double)
96 -> Spec
97measureRate desc t rval consumerDelay producerDelay dur =
98 let d = fromIntegral producerDelay
99 in measureRateVariable desc t rval consumerDelay (d, d) dur
101measureThreads :: IsStream t
102 => String
103 -> (t IO Int -> SerialT IO Int)
104 -> Int -- threads
105 -> Int -- count of actions
106 -> Spec
107measureThreads desc t threads count = do
108 let expectedTime =
109 if threads < 0
110 then 1.0
111 else fromIntegral count / fromIntegral threads
112 duration = (expectedTime * 0.9, expectedTime * 1.1)
113 measureRate' desc t (-1) threads (Right count) 0 (1,1) duration
115measureBuffers :: IsStream t
116 => String
117 -> (t IO Int -> SerialT IO Int)
118 -> Int -- buffers
119 -> Int -- count of actions
120 -> Spec
121measureBuffers desc t buffers count = do
122 let expectedTime =
123 if buffers < 0
124 then 1.0
125 else fromIntegral count / fromIntegral buffers
126 duration = (expectedTime * 0.9, expectedTime * 1.1)
127 measureRate' desc t buffers (-1) (Right count) 0 (1,1) duration
129main :: IO ()
130main = hspec $ do
132 describe "maxBuffers" $ do
133 measureBuffers "asyncly" asyncly (-1) 5
134 -- XXX this test fails due to a known issue
135 -- measureBuffers "maxBuffers" asyncly 1 5
136 measureBuffers "asyncly" asyncly 5 5
138 describe "maxThreads" $ do
139 measureThreads "asyncly" asyncly (-1) 5
140 measureThreads "asyncly" asyncly 1 5
141 measureThreads "asyncly" asyncly 5 5
143 measureThreads "aheadly" aheadly (-1) 5
144 measureThreads "aheadly" aheadly 1 5
145 measureThreads "aheadly" aheadly 5 5
147 let range = (8,12)
149 -- Note that because after the last yield we don't wait, the last period
150 -- will be effectively shorter. This becomes significant when the rates are
151 -- lower (1 or lower). For rate 1 we lose 1 second in the end and for rate
152 -- 10 0.1 second.
153 let rates = [1, 10, 100, 1000, 10000
154#ifndef __GHCJS__
155 , 100000, 1000000
156#endif
157 ]
158 in describe "asyncly no consumer delay no producer delay" $
159 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 0 range)
161 -- XXX try staggering the dispatches to achieve higher rates
162 let rates = [1, 10, 100, 1000
163#ifndef __GHCJS__
164 , 10000, 25000
165#endif
166 ]
167 in describe "asyncly no consumer delay and 1 sec producer delay" $
168 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 1 range)
170 -- At lower rates (1/10) this is likely to vary quite a bit depending on
171 -- the spread of random producer latencies generated.
172 let rates = [1, 10, 100, 1000
173#ifndef __GHCJS__
174 , 10000, 25000
175#endif
176 ]
177 in describe "asyncly no consumer delay and variable producer delay" $
178 forM_ rates $ \r ->
179 measureRateVariable "asyncly" asyncly r 0 (0.1, 3) range
181 let rates = [1, 10, 100, 1000, 10000
182#ifndef __GHCJS__
183 , 100000, 1000000
184#endif
185 ]
186 in describe "wAsyncly no consumer delay no producer delay" $
187 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 0 range)
189 let rates = [1, 10, 100, 1000
190#ifndef __GHCJS__
191 , 10000, 25000
192#endif
193 ]
194 in describe "wAsyncly no consumer delay and 1 sec producer delay" $
195 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 1 range)
197 let rates = [1, 10, 100, 1000, 10000
198#ifndef __GHCJS__
199 , 100000, 1000000
200#endif
201 ]
202 in describe "aheadly no consumer delay no producer delay" $
203 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 0 range)
205 -- XXX after the change to stop workers when the heap is clearing
206 -- thi does not work well at a 25000 ops per second, need to fix.
207 let rates = [1, 10, 100, 1000
208#ifndef __GHCJS__
209 , 10000, 12500
210#endif
211 ]
212 in describe "aheadly no consumer delay and 1 sec producer delay" $
213 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 1 range)
215 describe "asyncly with 1 sec producer delay and some consumer delay" $ do
216 -- ideally it should take 10 x 1 + 1 seconds
217 forM_ [1] (\r -> measureRate "asyncly" asyncly r 1 1 (11, 16))
218 -- ideally it should take 10 x 2 + 1 seconds
219 forM_ [1] (\r -> measureRate "asyncly" asyncly r 2 1 (21, 23))
220 -- ideally it should take 10 x 3 + 1 seconds
221 forM_ [1] (\r -> measureRate "asyncly" asyncly r 3 1 (31, 33))
223 describe "aheadly with 1 sec producer delay and some consumer delay" $ do
224 forM_ [1] (\r -> measureRate "aheadly" aheadly r 1 1 (11, 16))
225 forM_ [1] (\r -> measureRate "aheadly" aheadly r 2 1 (21, 23))
226 forM_ [1] (\r -> measureRate "aheadly" aheadly r 3 1 (31, 33))