git.iko.soy/streamly
on 24375dc help sign in
streamly/test/MaxRate.hs
129 lines 5.3 kB history raw
1{-# LANGUAGE FlexibleContexts #-}
3import Streamly
4import qualified Streamly.Prelude as S
5import Control.Concurrent
6import Control.Monad
7import System.Clock
8import Test.Hspec
9import System.Random
11durationShouldBe :: (Double, Double) -> IO () -> Expectation
12durationShouldBe d@(tMin, tMax) action = do
13 t0 <- getTime Monotonic
14 action
15 t1 <- getTime Monotonic
16 let t = fromIntegral (toNanoSecs (t1 - t0)) / 1e9
17 -- tMax = fromNanoSecs (round $ d*10^9*1.2)
18 -- tMin = fromNanoSecs (round $ d*10^9*0.8)
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 -> Double
29 -> Int
30 -> (Double, Double)
31 -> (Double, Double)
32 -> Spec
33measureRate' desc t rval consumerDelay producerDelay dur =
34 it (desc <> " rate: " <> show rval
35 <> ", consumer latency: " <> show consumerDelay
36 <> ", producer latency: " <> show producerDelay)
37 $ durationShouldBe dur $
38 runStream
39 $ (if consumerDelay > 0
40 then S.mapM $ \x ->
41 threadDelay (toMicroSecs consumerDelay) >> return x
42 else id)
43 $ t
44 $ maxBuffer (-1)
45 $ maxThreads (-1)
46 $ avgRate rval
47 $ S.take (round $ rval * 10)
48 $ S.repeatM $ do
49 let (t1, t2) = producerDelay
50 r <- if t1 == t2
51 then return $ round $ toMicroSecs t1
52 else randomRIO ( round $ toMicroSecs t1
53 , round $ toMicroSecs t2)
54 when (r > 0) $ -- do
55 -- t1 <- getTime Monotonic
56 threadDelay r
57 -- t2 <- getTime Monotonic
58 -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
59 -- putStrLn $ "delay took: " <> show delta
60 -- when (delta > 2) $ do
61 -- putStrLn $ "delay took high: " <> show delta
62 return 1
64measureRate :: IsStream t
65 => String
66 -> (t IO Int -> SerialT IO Int)
67 -> Double
68 -> Int
69 -> Int
70 -> (Double, Double)
71 -> Spec
72measureRate desc t rval consumerDelay producerDelay dur =
73 let d = fromIntegral producerDelay
74 in measureRate' desc t rval consumerDelay (d, d) dur
76main :: IO ()
77main = hspec $ do
78 let range = (8,12)
80 -- Note that because after the last yield we don't wait, the last period
81 -- will be effectively shorter. This becomes significant when the rates are
82 -- lower (1 or lower). For rate 1 we lose 1 second in the end and for rate
83 -- 10 0.1 second.
84 let rates = [1, 10, 100, 1000, 10000, 100000, 1000000]
85 in describe "asyncly no consumer delay no producer delay" $
86 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 0 range)
88 -- XXX try staggering the dispatches to achieve higher rates
89 let rates = [1, 10, 100, 1000, 10000, 25000]
90 in describe "asyncly no consumer delay and 1 sec producer delay" $
91 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 1 range)
93 -- At lower rates (1/10) this is likely to vary quite a bit depending on
94 -- the spread of random producer latencies generated.
95 let rates = [1, 10, 100, 1000, 10000, 25000]
96 in describe "asyncly no consumer delay and variable producer delay" $
97 forM_ rates $ \r ->
98 measureRate' "asyncly" asyncly r 0 (0.1, 3) range
100 let rates = [1, 10, 100, 1000, 10000, 100000, 1000000]
101 in describe "wAsyncly no consumer delay no producer delay" $
102 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 0 range)
104 let rates = [1, 10, 100, 1000, 10000, 25000]
105 in describe "wAsyncly no consumer delay and 1 sec producer delay" $
106 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 1 range)
108 let rates = [1, 10, 100, 1000, 10000, 100000, 1000000]
109 in describe "aheadly no consumer delay no producer delay" $
110 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 0 range)
112 -- XXX after the change to stop workers when the heap is clearing
113 -- thi does not work well at a 25000 ops per second, need to fix.
114 let rates = [1, 10, 100, 1000, 10000, 12500]
115 in describe "aheadly no consumer delay and 1 sec producer delay" $
116 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 1 range)
118 describe "asyncly with 1 sec producer delay and some consumer delay" $ do
119 -- ideally it should take 10 x 1 + 1 seconds
120 forM_ [1] (\r -> measureRate "asyncly" asyncly r 1 1 (11, 16))
121 -- ideally it should take 10 x 2 + 1 seconds
122 forM_ [1] (\r -> measureRate "asyncly" asyncly r 2 1 (21, 23))
123 -- ideally it should take 10 x 3 + 1 seconds
124 forM_ [1] (\r -> measureRate "asyncly" asyncly r 3 1 (31, 33))
126 describe "aheadly with 1 sec producer delay and some consumer delay" $ do
127 forM_ [1] (\r -> measureRate "aheadly" aheadly r 1 1 (11, 16))
128 forM_ [1] (\r -> measureRate "aheadly" aheadly r 2 1 (21, 23))
129 forM_ [1] (\r -> measureRate "aheadly" aheadly r 3 1 (31, 33))