streamly-0.11.1: test/Streamly/Test/Data/Stream/Rate.hs
-- |
-- Module : Streamly.Test.Data.Stream.Rate
-- Copyright : (c) 2018 Composewell Technologies
-- License : BSD-3-Clause
-- Maintainer : streamly@composewell.com
-- Stability : experimental
-- Portability : GHC
module Streamly.Test.Data.Stream.Rate (main) where
import Data.Int (Int64)
import Streamly.Data.Stream.Prelude (Config)
import Streamly.Internal.Data.Time.Clock (getTime, Clock(..))
import Streamly.Internal.Data.Time.Units
(NanoSecond64, diffAbsTime64, fromRelTime64)
import System.Mem (performMajorGC)
import qualified Streamly.Data.Fold as Fold
import qualified Streamly.Data.Stream as Stream
import qualified Streamly.Internal.Data.Stream.Prelude as Stream
import Control.Concurrent
import Control.Monad
import System.Random
import Test.Hspec
durationShouldBe :: (Double, Double) -> IO () -> Expectation
durationShouldBe d@(tMin, tMax) action = do
t0 <- getTime Monotonic
action
t1 <- getTime Monotonic
let diff = fromRelTime64 (diffAbsTime64 t1 t0) :: NanoSecond64
let t = fromIntegral diff / 1e9
putStrLn $ "Expected: " <> show d <> " Took: " <> show t
(t <= tMax && t >= tMin) `shouldBe` True
toMicroSecs :: Num a => a -> a
toMicroSecs x = x * 10^(6 :: Int)
measureRate' ::
String
-> (Config -> Config)
-> Int -- buffers
-> Int -- threads
-> Either Double Int -- either rate or count of actions
-> Int
-> (Double, Double)
-> (Double, Double)
-> Spec
measureRate'
desc modifier buffers threads rval consumerDelay producerDelay expectedRange = do
it (desc <> rateDesc
<> " buffers: " <> show buffers
<> " threads: " <> show threads
<> ", consumer latency: " <> show consumerDelay
<> ", producer latency: " <> show producerDelay)
runTest
where
-- Keep a minimum of 2 for the very low rate cases, otherwise the
-- timing would be 0 because we will finish as soon as the first result
-- arrives.
yieldCount :: Int
yieldCount = case rval of
Left r -> max 2 (round (10 * r))
Right n -> max 2 n
rateDesc = (case rval of
Left r -> ", rate: " <> show r <> ", count: " <> show yieldCount
Right n -> ", count: " <> show n) <> ","
cfg (_n :: Maybe Int64) =
modifier
. Stream.maxBuffer buffers
-- . Stream.inspect True
. Stream.maxThreads threads
. case rval of
Left r -> Stream.avgRate r
Right _ -> Stream.rate Nothing
-- XXX it comes out less than expected for ordered streams at high
-- rates, need to fix.
-- . Stream.maxYields (Just (fromIntegral yieldCount))
threadAction f =
case rval of
Left _ ->
Stream.take yieldCount
$ Stream.parMapM (cfg (Just (fromIntegral yieldCount))) f
$ Stream.enumerateFrom (1 :: Int)
Right _ ->
Stream.parReplicateM
(cfg (Just (fromIntegral yieldCount))) yieldCount (f 1)
runTest = do
durationShouldBe expectedRange $ do
res <- Stream.fold Fold.length
$ (if consumerDelay > 0
then Stream.mapM $ \x ->
threadDelay (toMicroSecs consumerDelay) >> return x
else id)
$ threadAction $ \_idx -> do
let (t1, t2) = producerDelay
r <- if t1 == t2
then return $ round $ toMicroSecs t1
else randomRIO ( round $ toMicroSecs t1
, round $ toMicroSecs t2)
when (r > 0) $ -- do
-- t1 <- getTime Monotonic
threadDelay r
-- t2 <- getTime Monotonic
-- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
-- putStrLn $ "delay took: " <> show delta
-- when (delta > 2) $ do
-- putStrLn $ "delay took high: " <> show delta
-- putStrLn $ "Done: " ++ show idx
return (1 :: Int)
when (res /= yieldCount) $
error $ "expected yield count: " ++ show yieldCount
++ " actual: " ++ show res
-- To ensure that when we use "inspect" option on the channel, GC
-- occurs and cleans up the channel to print the debug info.
performMajorGC
measureRateVariable ::
String
-> (Config -> Config)
-> Double
-> Int
-> (Double, Double)
-> (Double, Double)
-> Spec
measureRateVariable desc modifier rval =
measureRate' desc modifier (-1) (-1) (Left rval)
measureRate ::
String
-> (Config -> Config)
-> Double
-> Int
-> Int
-> (Double, Double)
-> Spec
measureRate desc modifier rval consumerDelay producerDelay dur =
let d = fromIntegral producerDelay
in measureRateVariable desc modifier rval consumerDelay (d, d) dur
measureThreads ::
String
-> (Config -> Config)
-> Int -- threads
-> Int -- count of actions
-> Spec
measureThreads desc modifier threads count = do
let expectedTime =
if threads < 0
then 1.0
else fromIntegral count / fromIntegral threads
duration = (expectedTime * 0.9, expectedTime * 1.1)
measureRate' desc modifier (-1) threads (Right count) 0 (1,1) duration
measureBuffers ::
String
-> (Config -> Config)
-> Int -- buffers
-> Int -- count of actions
-> Spec
measureBuffers desc modifier buffers count = do
let expectedTime =
if buffers < 0
then 1.0
else fromIntegral count / fromIntegral buffers
duration = (expectedTime * 0.9, expectedTime * 1.1)
measureRate' desc modifier buffers (-1) (Right count) 0 (1,1) duration
moduleName :: String
moduleName = "Data.Stream.Rate"
main :: IO ()
main = hspec $ do
describe moduleName $ do
describe "maxBuffers" $ do
measureBuffers "async" id (-1) 5
-- XXX this test fails due to a known issue
-- measureBuffers "maxBuffers" id 1 5
measureBuffers "async" id 5 5
describe "maxThreads" $ do
measureThreads "async" id (-1) 5
measureThreads "async" id 1 5
measureThreads "async" id 5 5
measureThreads "ordered" (Stream.ordered True) (-1) 5
measureThreads "ordered" (Stream.ordered True) 1 5
measureThreads "ordered" (Stream.ordered True) 5 5
measureThreads "interleaved" (Stream.interleaved True) (-1) 5
measureThreads "interleaved" (Stream.interleaved True) 1 5
measureThreads "interleaved" (Stream.interleaved True) 5 5
describe "max rate possible (count / time)" $ do
measureRate "async" (Stream.rate Nothing) 1000000 0 0 (0, 1e9)
let range = (8,12)
-- Note that because after the last yield we don't wait, the last period
-- will be effectively shorter. This becomes significant when the rates are
-- lower (1 or lower). For rate 1 we lose 1 second in the end and for rate
-- 10 0.1 second.
let rates = [0.1, 1, 10, 100, 1000, 10000
#ifndef __GHCJS__
, 100000, 1000000
#endif
]
in describe "async no consumer delay no producer delay" $
forM_ rates (\r -> measureRate "async" id r 0 0 range)
-- XXX try staggering the dispatches to achieve higher rates
-- Producer delay causes a lot of threads to be created, consuming large
-- amounts of memory at higher rates.
let rates = [0.1, 1, 10, 100
#if !defined(__GHCJS__) && defined USE_LARGE_MEMORY
1000, 10000, 25000
#endif
]
in describe "async no consumer delay and 1 sec producer delay" $
forM_ rates (\r -> measureRate "async" id r 0 1 range)
-- At lower rates (1/10) this is likely to vary quite a bit depending on
-- the spread of random producer latencies generated.
let rates = [1, 10, 100
#if !defined(__GHCJS__) && defined USE_LARGE_MEMORY
, 1000, 10000, 25000
#endif
]
in describe "async, no consumer delay and variable producer delay" $
forM_ rates $ \r ->
measureRateVariable "async" id r 0 (0.1, 3) range
let rates = [0.1, 1, 10, 100, 1000, 10000
#ifndef __GHCJS__
, 100000, 1000000
#endif
]
in describe "interleaved, no consumer delay no producer delay" $
forM_ rates (\r -> measureRate "interleaved" (Stream.interleaved True) r 0 0 range)
let rates = [0.1, 1, 10, 100, 1000
#if !defined(__GHCJS__) && defined USE_LARGE_MEMORY
, 10000, 25000
#endif
]
in describe "interleaved, no consumer delay and 1 sec producer delay" $
forM_ rates (\r -> measureRate "interleaved" (Stream.interleaved True) r 0 1 range)
let rates = [0.1, 1, 10, 100, 1000, 10000
#ifndef __GHCJS__
, 100000, 1000000
#endif
]
in describe "ordered, no consumer delay no producer delay" $
forM_ rates (\r -> measureRate "ordered" (Stream.ordered True) r 0 0 range)
-- XXX after the change to stop workers when the heap is clearing
-- thi does not work well at a 25000 ops per second, need to fix.
let rates = [0.1, 1, 10, 100, 1000
#if !defined(__GHCJS__) && defined USE_LARGE_MEMORY
, 10000, 12500
#endif
]
in describe "ordered, no consumer delay and 1 sec producer delay" $
forM_ rates (\r -> measureRate "ordered" (Stream.ordered True) r 0 1 range)
describe "async, some consumer delay and 1 sec producer delay" $ do
-- ideally it should take 10 x 1 + 1 seconds
forM_ [1] (\r -> measureRate "async" id r 1 1 (11, 16))
-- ideally it should take 10 x 2 + 1 seconds
forM_ [1] (\r -> measureRate "async" id r 2 1 (21, 23))
-- ideally it should take 10 x 3 + 1 seconds
forM_ [1] (\r -> measureRate "async" id r 3 1 (31, 33))
describe "ordered, some consumer delay and 1 sec producer delay" $ do
forM_ [1] (\r -> measureRate "ordered" (Stream.ordered True) r 1 1 (11, 16))
forM_ [1] (\r -> measureRate "ordered" (Stream.ordered True) r 2 1 (21, 23))
forM_ [1] (\r -> measureRate "ordered" (Stream.ordered True) r 3 1 (31, 33))