@@ -7,8 +7,7 @@
77-- Maintainer : streamly@composewell.com
88
99import Control.Concurrent
10−import Control.Monad (when)
11−-- import Data.IORef
10+import Control.Monad (when, replicateM)
1211
1312import Gauge
1413import Streamly
@@ -18,7 +17,10 @@ import qualified Streamly.Prelude as S
1817-- Append
1918-------------------------------------------------------------------------------
2019
21−-- Single work item yielded per thread
20+-- | Run @tcount@ number of actions concurrently using the given concurrency
21+-- style. Each thread produces a single output after a delay of @d@
22+-- microseconds.
23+--
2224{-# INLINE append #-}
2325append :: IsStream t
2426 => Int -> Int -> Int -> (t IO Int -> SerialT IO Int) -> IO ()
@@ -30,11 +32,22 @@ append buflen tcount d t =
3032 $ maxThreads (-1)
3133 $ S.fromFoldableM $ map work [1..tcount]
3234
33−-- Big stream of items yielded per thread
35+-- | Run @threads@ concurrently, each producing streams of @elems@ elements
36+-- with a delay of @d@ microseconds between successive elements, and merge
37+-- their outputs in a single output stream. The individual streams are produced
38+-- serially but merged using the provided concurrency style.
39+--
3440{-# INLINE concated #-}
35−concated :: Int -> Int -> Int -> Int -> (forall a. SerialT IO a -> SerialT IO a -> SerialT IO a) -> IO ()
41+concated
42+ :: Int
43+ -> Int
44+ -> Int
45+ -> Int
46+ -> (forall a. SerialT IO a -> SerialT IO a -> SerialT IO a)
47+ -> IO ()
3648concated buflen threads d elems t =
37− let work = (\i -> (S.replicateM i ((when (d /= 0) (threadDelay d)) >> return i)))
49+ let work = \i -> S.replicateM i
50+ ((when (d /= 0) (threadDelay d)) >> return i)
3851 in S.drain
3952 $ adapt
4053 $ maxThreads (-1)
@@ -72,25 +85,19 @@ main = do
7285
7386 [ -- bgroup "append/buf-1-threads-10k-0sec" (appendGroup 1 10000 0)
7487 -- , bgroup "append/buf-100-threads-100k-0sec" (appendGroup 100 100000 0)
75− bgroup "append/buf-10k-threads-10k-5sec" (appendGroup 10000 10000 5000000)
88+ bgroup "stream1x10k/buf10k-threads10k-5sec" (appendGroup 10000 10000 5000000)
7689 -- bgroup "concat/buf-1-threads-100k-count-1" (concatGroup 1 100000 0 1)
7790 -- bgroup "concat/buf-1-threads-1-count-10m" (concatGroup 1 1 0 10000000)
78− , bgroup "concat/buf-100-threads-100-count-500k" (concatGroup 100 100 0 500000)
79− {-
80− , bgroup "forkIO-5000ms-10k" $
81− let delay = threadDelay 5000000
82− count = 10000 :: Int
83− list = [1..count]
84− work i = delay >> return i
85− in
86− [ bench "discard" $ nfIO $ do
87− mapM_ (\i -> forkIO $ work i >> return ()) list
88− threadDelay 6000000
89− , bench "collect" $ nfIO $ do
90− ref <- newIORef []
91− mapM_ (\i -> forkIO $ work i >>=
92− \j -> atomicModifyIORef ref $ \xs -> (j : xs, ())) list
93− threadDelay 6000000
94− ]
95− -}
91+ , bgroup "streams100x500k/buf100-threads100" (concatGroup 100 100 0 500000)
92+
93+ , bench "forkIO/threads10k-5sec" $
94+ let delay = threadDelay 5000000
95+ count = 10000 :: Int
96+ list = [1..count]
97+ work i = delay >> return i
98+ in nfIO $ do
99+ ref <- newEmptyMVar
100+ mapM_ (\i -> forkIO $ work i >>=
101+ \j -> putMVar ref j) list
102+ replicateM 10000 (takeMVar ref)
96103 ]