Adaptive.hs

-- |
-- Module      : Main
-- Copyright   : (c) 2018 Composewell Technologies
-- License     : BSD3
-- Maintainer  : streamly@composewell.com

import Control.Concurrent (threadDelay)
import Control.Monad (when)
import Control.Monad.IO.Class (MonadIO(liftIO))
import Streamly.Data.Stream.Prelude (Stream, Config)
import System.Random (randomRIO)

import qualified Streamly.Data.Fold as Fold
import qualified Streamly.Data.Stream.Prelude as Stream

import Test.Tasty.Bench

-- Note that we should also compare the cpuTime especially when threaded
-- runtime is used with this benchmark because thread scheduling is not
-- predictable and can add non-deterministic delay to the total time measured.
--
-- Also, the worker dispatch depends on the worker dispatch latency which is
-- set to fixed 200 us. We need to keep that in mind when designing tests.

moduleName :: String
moduleName = "Data.Stream.Adaptive"

value :: Int
value = 1000

produce :: MonadIO m => (Int, Int) -> m Int
produce range = do
    r <- randomRIO range
    when (r /= 0) $ liftIO $ threadDelay r
    return r

consume :: MonadIO m => (Int, Int) -> Int -> m Int
consume crange x = liftIO $ do
    d <- randomRIO crange
    when (d /= 0) $ threadDelay d
    return x

{-# INLINE serialSrc #-}
serialSrc :: (Int, Int) -> Stream IO Int
serialSrc range = Stream.replicateM value $ produce range

{-# INLINE parSrc #-}
parSrc :: (Config -> Config) -> (Int, Int) -> Stream IO Int
parSrc cfg range = Stream.parReplicateM cfg value $ produce range

{-# INLINE serialRun #-}
serialRun :: (Int, Int) -> (Int, Int) -> IO ()
serialRun srange crange =
    Stream.fold Fold.drain
        $ Stream.mapM (consume crange) (serialSrc srange)

{-# INLINE parRun #-}
parRun :: (Config -> Config) -> (Int, Int) -> (Int, Int) -> IO ()
parRun cfg srange crange =
    Stream.fold Fold.drain
        $ Stream.parMapM cfg (consume crange) (parSrc cfg srange)

low, medium, high :: Int
low = 10
medium = 20
high = 30

{-# INLINE noDelaySerial #-}
noDelaySerial :: IO ()
noDelaySerial = serialRun (0,0) (0,0)

{-# INLINE noDelay #-}
noDelay :: (Config -> Config) -> IO ()
noDelay cfg = parRun cfg (0,0) (0,0)

{-# INLINE alwaysConstSlowSerial #-}
alwaysConstSlowSerial :: IO ()
alwaysConstSlowSerial = serialRun (0,0) (medium,medium)

{-# INLINE alwaysConstSlow #-}
alwaysConstSlow :: (Config -> Config) -> IO ()
alwaysConstSlow cfg = parRun cfg (low,low) (medium,medium)

{-# INLINE alwaysConstFast #-}
alwaysConstFast :: (Config -> Config) -> IO ()
alwaysConstFast cfg = parRun cfg (high,high) (medium,medium)

{-# INLINE alwaysVarSlow #-}
alwaysVarSlow :: (Config -> Config) -> IO ()
alwaysVarSlow cfg = parRun cfg (low,low) (low,high)

{-# INLINE alwaysVarFast #-}
alwaysVarFast :: (Config -> Config) -> IO ()
alwaysVarFast cfg = parRun cfg (high,high) (low,high)

-- XXX add variable producer tests as well

{-# INLINE runVarSometimesFast #-}
runVarSometimesFast :: (Config -> Config) -> IO ()
runVarSometimesFast cfg = parRun cfg (medium,medium) (low,high)

{-# INLINE randomVar #-}
randomVar :: (Config -> Config) -> IO ()
randomVar cfg = parRun cfg (low,high) (low,high)

main :: IO ()
main =
  defaultMain [bgroup moduleName allBenchmarks]

  where

  allBenchmarks =
    [
      bgroup "serialConstantSlowConsumer"
      [ bench "serial" $ nfIO alwaysConstSlowSerial
      ]
    , bgroup "default"
      [ bench "serial"      $ nfIO noDelaySerial
      , bench "async"       $ nfIO $ noDelay id
      , bench "ordered"     $ nfIO $ noDelay (Stream.ordered True)
      , bench "interleaved" $ nfIO $ noDelay (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ noDelay (Stream.eager True)
      ]
    , bgroup "constantSlowConsumer"
      [ bench "async"       $ nfIO $ alwaysConstSlow id
      , bench "ordered"     $ nfIO $ alwaysConstSlow (Stream.ordered True)
      , bench "interleaved" $ nfIO $ alwaysConstSlow (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ alwaysConstSlow (Stream.eager True)
      ]
   ,  bgroup "constantFastConsumer"
      [ bench "async"       $ nfIO $ alwaysConstFast id
      , bench "ordered"     $ nfIO $ alwaysConstFast (Stream.ordered True)
      , bench "interleaved" $ nfIO $ alwaysConstFast (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ alwaysConstFast (Stream.eager True)
      ]
   ,  bgroup "variableSlowConsumer"
      [ bench "async"       $ nfIO $ alwaysVarSlow id
      , bench "ordered"     $ nfIO $ alwaysVarSlow (Stream.ordered True)
      , bench "interleaved" $ nfIO $ alwaysVarSlow (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ alwaysVarSlow (Stream.eager True)
      ]
   ,  bgroup "variableFastConsumer"
      [ bench "async"       $ nfIO $ alwaysVarFast id
      , bench "ordered"     $ nfIO $ alwaysVarFast (Stream.ordered True)
      , bench "interleaved" $ nfIO $ alwaysVarFast (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ alwaysVarFast (Stream.eager True)
      ]
   ,  bgroup "variableSometimesFastConsumer"
      [ bench "async"       $ nfIO $ runVarSometimesFast id
      , bench "ordered"     $ nfIO $ runVarSometimesFast (Stream.ordered True)
      , bench "interleaved" $ nfIO $ runVarSometimesFast (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ runVarSometimesFast (Stream.eager True)
      ]
   ,  bgroup "variableFullOverlap"
      [ bench "async"       $ nfIO $ randomVar id
      , bench "ordered"     $ nfIO $ randomVar (Stream.ordered True)
      , bench "interleaved" $ nfIO $ randomVar (Stream.interleaved True)
      , bench "parallel"    $ nfIO $ randomVar (Stream.eager True)
      ]
   ]