Separate "streamly" stream benchmarks from "streamly-core"

harendra-kumar · harendra-kumar · commit 209a4b86923c · 2025-12-01T14:47:41.000+05:30
diff --git a/benchmark/Streamly/Benchmark/Data/Stream/Prelude/Exceptions.hs b/benchmark/Streamly/Benchmark/Data/Stream/Prelude/Exceptions.hs
@@ -22,19 +22,24 @@
 module Main (main) where
 
 import Control.Exception (Exception, throwIO)
-import Stream.Common (drain)
+import Data.HashMap.Strict (HashMap)
+import Data.Proxy (Proxy(..))
+import Stream.Common (drain, benchIOSink)
+import Streamly.Internal.Data.IsMap.HashMap ()
+import Streamly.Internal.Data.Stream (Stream)
+import System.IO (Handle, hClose, hPutChar)
 
 import qualified Data.IORef as Ref
 import qualified Data.Map.Strict as Map
 
-import System.IO (Handle, hClose, hPutChar)
+import qualified Stream.Common as Common
+import qualified Streamly.Internal.Data.Fold as Fold
 import qualified Streamly.FileSystem.Handle as FH
 import qualified Streamly.Internal.FileSystem.Handle as IFH
-import qualified Streamly.Internal.Data.Unfold as IUF
-import qualified Streamly.Internal.Data.Unfold.Prelude as IUF
-
 import qualified Streamly.Internal.Data.Stream as Stream
 import qualified Streamly.Internal.Data.Stream.Prelude as Stream
+import qualified Streamly.Internal.Data.Unfold as IUF
+import qualified Streamly.Internal.Data.Unfold.Prelude as IUF
 
 import Test.Tasty.Bench hiding (env)
 import Prelude hiding (last, length)
@@ -213,11 +218,8 @@ o_1_space_copy_exceptions_toChunks env =
         ]
     ]
 
-moduleName :: String
-moduleName = "Data.Stream.Prelude.Exceptions"
-
-benchmarks :: BenchEnv -> Int -> [Benchmark]
-benchmarks env size =
+excBenchmarks :: BenchEnv -> Int -> [Benchmark]
+excBenchmarks env size =
     [ bgroup (o_1_space_prefix moduleName) $ concat
         [ o_1_space_serial_exceptions size
         , o_1_space_copy_exceptions_readChunks env
@@ -226,7 +228,104 @@ benchmarks env size =
         ]
     ]
 
+{-# INLINE pollCounts #-}
+pollCounts :: Stream IO Int -> IO ()
+pollCounts = drain . Stream.parTapCount (const True) f
+
+    where
+
+    f = Stream.drain . Stream.rollingMap2 (-) . Stream.delayPost 1
+
+{-# INLINE takeInterval #-}
+takeInterval :: Double -> Stream IO Int -> IO ()
+takeInterval i = drain . Stream.takeInterval i
+
+-- Inspection testing is disabled for takeInterval
+-- Enable it when looking at it throughly
+#ifdef INSPECTION
+-- inspect $ hasNoType 'takeInterval ''SPEC
+-- inspect $ hasNoTypeClasses 'takeInterval
+-- inspect $ 'takeInterval `hasNoType` ''D.Step
+#endif
+
+{-# INLINE dropInterval #-}
+dropInterval :: Double -> Stream IO Int -> IO ()
+dropInterval i = drain . Stream.dropInterval i
+
+-- Inspection testing is disabled for dropInterval
+-- Enable it when looking at it throughly
+#ifdef INSPECTION
+-- inspect $ hasNoTypeClasses 'dropInterval
+-- inspect $ 'dropInterval `hasNoType` ''D.Step
+#endif
+
+-- XXX Decide on the time interval
+{-# INLINE _intervalsOfSum #-}
+_intervalsOfSum :: Stream.MonadAsync m => Double -> Stream m Int -> m ()
+_intervalsOfSum i = drain . Stream.intervalsOf i Fold.sum
+
+timeBenchmarks :: BenchEnv -> Int -> [Benchmark]
+timeBenchmarks _env size =
+    [ benchIOSink size "parTapCount 1 second" pollCounts
+    , benchIOSink size "takeInterval-all" (takeInterval 10000)
+    , benchIOSink size "dropInterval-all" (dropInterval 10000)
+    ]
+
+-------------------------------------------------------------------------------
+-- Grouping/Splitting
+-------------------------------------------------------------------------------
+
+{-# INLINE classifySessionsOf #-}
+classifySessionsOf :: Stream.MonadAsync m => (Int -> Int) -> Stream m Int -> m ()
+classifySessionsOf getKey =
+      Common.drain
+    . Stream.classifySessionsOf
+        (const (return False)) 3 (Fold.take 10 Fold.sum)
+    . Stream.timestamped
+    . fmap (\x -> (getKey x, x))
+
+{-# INLINE classifySessionsOfHash #-}
+classifySessionsOfHash :: Stream.MonadAsync m =>
+    (Int -> Int) -> Stream m Int -> m ()
+classifySessionsOfHash getKey =
+      Common.drain
+    . Stream.classifySessionsByGeneric
+        (Proxy :: Proxy (HashMap k))
+        1 False (const (return False)) 3 (Fold.take 10 Fold.sum)
+    . Stream.timestamped
+    . fmap (\x -> (getKey x, x))
+
+o_1_space_grouping :: BenchEnv -> Int -> [Benchmark]
+o_1_space_grouping _env value =
+    -- Buffering operations using heap proportional to group/window sizes.
+    [ bgroup "grouping"
+        [ benchIOSink value "classifySessionsOf (10000 buckets)"
+            (classifySessionsOf (getKey 10000))
+        , benchIOSink value "classifySessionsOf (64 buckets)"
+            (classifySessionsOf (getKey 64))
+        , benchIOSink value "classifySessionsOfHash (10000 buckets)"
+            (classifySessionsOfHash (getKey 10000))
+        , benchIOSink value "classifySessionsOfHash (64 buckets)"
+            (classifySessionsOfHash (getKey 64))
+        ]
+    ]
+
+    where
+
+    getKey :: Int -> Int -> Int
+    getKey n = (`mod` n)
+
+moduleName :: String
+moduleName = "Data.Stream.Prelude"
+
 main :: IO ()
 main = do
     env <- mkHandleBenchEnv
-    runWithCLIOpts defaultStreamSize (benchmarks env)
+    runWithCLIOpts defaultStreamSize (allBenchmarks env)
+
+    where
+
+    allBenchmarks env size =
+        excBenchmarks env size
+            ++ timeBenchmarks env size
+            ++ o_1_space_grouping env size
diff --git a/benchmark/Streamly/Benchmark/Data/Stream/Reduce.hs b/benchmark/Streamly/Benchmark/Data/Stream/Reduce.hs
@@ -24,26 +24,14 @@ import qualified Streamly.Internal.Data.Refold.Type as Refold
 import qualified Streamly.Internal.Data.Fold as FL
 import qualified Stream.Common as Common
 
-#ifndef USE_STREAMLY_CORE
-import Data.HashMap.Strict (HashMap)
-import Data.Proxy (Proxy(..))
-import Streamly.Internal.Data.IsMap.HashMap ()
-#endif
-
 import Streamly.Internal.Data.Stream (Stream)
 import qualified Streamly.Internal.Data.Stream as S
-#ifndef USE_STREAMLY_CORE
-import qualified Streamly.Data.Stream.Prelude as S
-import qualified Streamly.Internal.Data.Stream.Prelude as S
-#endif
 
 import Test.Tasty.Bench
 import Streamly.Benchmark.Common
 import Stream.Common
 import Prelude hiding (reverse, tail)
 
-
-
 -- Apply transformation g count times on a stream of length len
 {-# INLINE iterateSource #-}
 iterateSource ::
@@ -146,27 +134,9 @@ o_1_space_grouping value =
         , benchIOSink value "refoldMany" refoldMany
         , benchIOSink value "foldIterateM" foldIterateM
         , benchIOSink value "refoldIterateM" refoldIterateM
-
-#ifndef USE_STREAMLY_CORE
-        , benchIOSink value "classifySessionsOf (10000 buckets)"
-            (classifySessionsOf (getKey 10000))
-        , benchIOSink value "classifySessionsOf (64 buckets)"
-            (classifySessionsOf (getKey 64))
-        , benchIOSink value "classifySessionsOfHash (10000 buckets)"
-            (classifySessionsOfHash (getKey 10000))
-        , benchIOSink value "classifySessionsOfHash (64 buckets)"
-            (classifySessionsOfHash (getKey 64))
-#endif
         ]
     ]
 
-#ifndef USE_STREAMLY_CORE
-    where
-
-    getKey :: Int -> Int -> Int
-    getKey n = (`mod` n)
-#endif
-
 -------------------------------------------------------------------------------
 -- Size conserving transformations (reordering, buffering, etc.)
 -------------------------------------------------------------------------------
@@ -189,32 +159,6 @@ o_n_heap_buffering value =
         ]
     ]
 
--------------------------------------------------------------------------------
--- Grouping/Splitting
--------------------------------------------------------------------------------
-
-#ifndef USE_STREAMLY_CORE
-{-# INLINE classifySessionsOf #-}
-classifySessionsOf :: S.MonadAsync m => (Int -> Int) -> Stream m Int -> m ()
-classifySessionsOf getKey =
-      Common.drain
-    . S.classifySessionsOf
-        (const (return False)) 3 (FL.take 10 FL.sum)
-    . S.timestamped
-    . fmap (\x -> (getKey x, x))
-
-{-# INLINE classifySessionsOfHash #-}
-classifySessionsOfHash :: S.MonadAsync m =>
-    (Int -> Int) -> Stream m Int -> m ()
-classifySessionsOfHash getKey =
-      Common.drain
-    . S.classifySessionsByGeneric
-        (Proxy :: Proxy (HashMap k))
-        1 False (const (return False)) 3 (FL.take 10 FL.sum)
-    . S.timestamped
-    . fmap (\x -> (getKey x, x))
-#endif
-
 -------------------------------------------------------------------------------
 -- Mixed Transformation
 -------------------------------------------------------------------------------
diff --git a/benchmark/Streamly/Benchmark/Data/Stream/Transform.hs b/benchmark/Streamly/Benchmark/Data/Stream/Transform.hs
@@ -24,28 +24,21 @@
 module Stream.Transform (benchmarks) where
 
 import Control.Monad.IO.Class (MonadIO(..))
-
+import Streamly.Internal.Data.Stream (Stream)
 import System.Random (randomRIO)
 
 import qualified Streamly.Internal.Data.Fold as FL
 import qualified Streamly.Internal.Data.Scanl as Scanl
 
 import qualified Stream.Common as Common
 import qualified Streamly.Internal.Data.Unfold as Unfold
-
-import Streamly.Internal.Data.Stream (Stream)
 import qualified Streamly.Internal.Data.Stream as Stream
-#ifndef USE_STREAMLY_CORE
-import qualified Streamly.Internal.Data.Stream.Prelude as Stream
-#endif
 
 import Test.Tasty.Bench
 import Stream.Common hiding (scanl')
 import Streamly.Benchmark.Common
 import Prelude hiding (sequence, mapM)
 
-
-
 -------------------------------------------------------------------------------
 -- Pipelines (stream-to-stream transformations)
 -------------------------------------------------------------------------------
@@ -98,17 +91,6 @@ sequence = Common.drain . Stream.sequence
 tap :: MonadIO m => Int -> Stream m Int -> m ()
 tap n = composeN n $ Stream.tap FL.sum
 
-#ifndef USE_STREAMLY_CORE
-{-# INLINE pollCounts #-}
-pollCounts :: Int -> Stream IO Int -> IO ()
-pollCounts n =
-    composeN n (Stream.parTapCount (const True) f)
-
-    where
-
-    f = Stream.drain . Stream.rollingMap2 (-) . Stream.delayPost 1
-#endif
-
 {-# INLINE _timestamped #-}
 _timestamped :: MonadIO m => Stream m Int -> m ()
 _timestamped = Stream.drain . Stream.timestamped
@@ -142,9 +124,6 @@ o_1_space_mapping value =
               sequence (sourceUnfoldrAction value n)
         , benchIOSink value "mapM" (mapM 1)
         , benchIOSink value "tap" (tap 1)
-#ifndef USE_STREAMLY_CORE
-        , benchIOSink value "parTapCount 1 second" (pollCounts 1)
-#endif
         -- XXX tasty-bench hangs benchmarking this
         -- , benchIOSink value "timestamped" _timestamped
         -- Scanning
@@ -314,31 +293,6 @@ takeWhileTrue value n = composeN n $ Stream.takeWhile (<= (value + 1))
 takeWhileMTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()
 takeWhileMTrue value n = composeN n $ Stream.takeWhileM (return . (<= (value + 1)))
 
-#if !defined(USE_STREAMLY_CORE)
-{-# INLINE takeInterval #-}
-takeInterval :: Double -> Int -> Stream IO Int -> IO ()
-takeInterval i n = composeN n (Stream.takeInterval i)
-
--- Inspection testing is disabled for takeInterval
--- Enable it when looking at it throughly
-#ifdef INSPECTION
--- inspect $ hasNoType 'takeInterval ''SPEC
--- inspect $ hasNoTypeClasses 'takeInterval
--- inspect $ 'takeInterval `hasNoType` ''D.Step
-#endif
-
-{-# INLINE dropInterval #-}
-dropInterval :: Double -> Int -> Stream IO Int -> IO ()
-dropInterval i n = composeN n (Stream.dropInterval i)
-
--- Inspection testing is disabled for dropInterval
--- Enable it when looking at it throughly
-#ifdef INSPECTION
--- inspect $ hasNoTypeClasses 'dropInterval
--- inspect $ 'dropInterval `hasNoType` ''D.Step
-#endif
-#endif
-
 {-# INLINE dropOne #-}
 dropOne :: MonadIO m => Int -> Stream m Int -> m ()
 dropOne n = composeN n $ Stream.drop 1
@@ -359,13 +313,6 @@ dropWhileMTrue value n = composeN n $ Stream.dropWhileM (return . (<= (value + 1
 dropWhileFalse :: MonadIO m => Int -> Int -> Stream m Int -> m ()
 dropWhileFalse value n = composeN n $ Stream.dropWhile (> (value + 1))
 
-#ifndef USE_STREAMLY_CORE
--- XXX Decide on the time interval
-{-# INLINE _intervalsOfSum #-}
-_intervalsOfSum :: Stream.MonadAsync m => Double -> Int -> Stream m Int -> m ()
-_intervalsOfSum i n = composeN n (Stream.intervalsOf i FL.sum)
-#endif
-
 {-# INLINE findIndices #-}
 findIndices :: MonadIO m => Int -> Int -> Stream m Int -> m ()
 findIndices value n = composeN n $ Stream.findIndices (== (value + 1))
@@ -420,10 +367,6 @@ o_1_space_filtering value =
      -- , benchIOSink value "takeWhileM-true" (_takeWhileMTrue value 1)
         , benchIOSink value "drop-one" (dropOne 1)
         , benchIOSink value "drop-all" (dropAll value 1)
-#if !defined(USE_STREAMLY_CORE)
-        , benchIOSink value "takeInterval-all" (takeInterval 10000 1)
-        , benchIOSink value "dropInterval-all" (dropInterval 10000 1)
-#endif
         , benchIOSink value "dropWhile-true" (dropWhileTrue value 1)
      -- , benchIOSink value "dropWhileM-true" (_dropWhileMTrue value 1)
         , benchIOSink
