ConcurrentThreadHeavy.hs

{-# LANGUAGE CPP #-}
{-# LANGUAGE RankNTypes #-}
-- |
-- Module      : Main
-- Copyright   : (c) 2018 Composewell Technologies
--
-- License     : BSD3
-- Maintainer  : streamly@composewell.com

import Control.Concurrent
import Control.Monad (when, replicateM)
import Streamly.Data.Stream.Prelude (Config)

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

import Test.Tasty (localOption)
import Test.Tasty.Bench

-------------------------------------------------------------------------------
-- Append
-------------------------------------------------------------------------------

-- | Run @tcount@ number of actions concurrently using the given concurrency
-- style. Each thread produces a single output after a delay of @d@
-- microseconds.
--
{-# INLINE append #-}
append :: Int -> Int -> Int -> (Config -> Config) -> IO ()
append buflen tcount d modifier =
    let work i = when (d /= 0) (threadDelay d) >> return i
        cfg =
              modifier
            . Stream.maxBuffer buflen
            . Stream.maxThreads (-1)
    in Stream.fold Fold.drain
        $ Stream.parMapM cfg work
        $ Stream.fromList [1..tcount]

-- | Run @threads@ concurrently, each producing streams of @elems@ elements
-- with a delay of @d@ microseconds between successive elements, and merge
-- their outputs in a single output stream. The individual streams are produced
-- serially but merged using the provided concurrency style.
--
{-# INLINE concated #-}
concated
    :: Int
    -> Int
    -> Int
    -> Int
    -> (Config -> Config)
    -> IO ()
concated buflen threads d elems modifier =
    let work i =
            Stream.replicateM i (when (d /= 0) (threadDelay d) >> return i)
        cfg =
              modifier
            . Stream.maxBuffer buflen
            . Stream.maxThreads (-1)
    in Stream.fold Fold.drain
        $ Stream.parConcatMap cfg work
        $ Stream.replicate threads elems

appendGroup :: Int -> Int -> Int -> [Benchmark]
appendGroup buflen threads usec =
    [ -- bench "serial"   $ nfIO $ append buflen threads delay fromSerial
      bench "async"
        $ nfIO $ append buflen threads usec id
    , bench "ordered"
        $ nfIO $ append buflen threads usec (Stream.ordered True)
    , bench "interleaved"
        $ nfIO $ append buflen threads usec (Stream.interleaved True)
    , bench "parallel"
        $ nfIO $ append buflen threads usec (Stream.eager True)
    ]

concatGroup :: Int -> Int -> Int -> Int -> [Benchmark]
concatGroup buflen threads usec n =
    [ -- bench "serial"   $ nfIO $ concated buflen threads usec n serial
      bench "async"
        $ nfIO $ concated buflen threads usec n id
    , bench "ordered"
        $ nfIO $ concated buflen threads usec n (Stream.ordered True)
    , bench "interleaved"
        $ nfIO $ concated buflen threads usec n (Stream.interleaved True)
    , bench "parallel"
        $ nfIO $ concated buflen threads usec n (Stream.eager True)
    ]

main :: IO ()
main =
    defaultMain $ map (localOption (RelStDev 100000))
    [ -- bgroup "append/buf-1-threads-10k-0sec"  (appendGroup 1 10000 0)
    -- , bgroup "append/buf-100-threads-100k-0sec"  (appendGroup 100 100000 0)
      bgroup "streamly/threads10k-delay5s-singleton(buf10k)"  (appendGroup 10000 10000 5000000)
    --  bgroup "concat/buf-1-threads-100k-count-1" (concatGroup 1 100000 0 1)
    --  bgroup "concat/buf-1-threads-1-count-10m" (concatGroup 1 1 0 10000000)
    , bgroup "streamly/threads100-delay0s-stream500k(buf100)"  (concatGroup 100 100 0 500000)

    , bench "forkIO/threads10k-delay5s-singleton" $
        let delay = threadDelay 5000000
            count = 10000 :: Int
            list = [1..count]
            work i = delay >> return i
        in nfIO $ do
            ref <- newEmptyMVar
            mapM_ (\i -> forkIO (work i >>= \j -> putMVar ref j)) list
            replicateM 10000 (takeMVar ref)
   ]