Concurrency: update docs, fix dev build, fix rate control, fix/add benchmarks/tests

.github/workflows/haskell.yml

@@ -184,6 +184,7 @@ jobs:
             cabal_version: 3.6.2.0
             cabal_project: cabal.project
             disable_sdist_build: "y"
+            disable_docs: "y"
             ignore_error: false
           - name: 8.6.5-debug-unoptimized
             ghc_version: 8.6.5
@@ -193,6 +194,7 @@ jobs:
             cabal_project: cabal.project
             cabal_build_options: "--flag debug --flag -opt"
             disable_sdist_build: "y"
+            disable_docs: "y"
             ignore_error: false
         # - name: hlint
         #   build: hlint

.github/workflows/packdiff.yml

@@ -11,10 +11,10 @@ jobs:
 
     - name: Download ghc
       run: |
-        GHCUP_VER=0.1.18.0
+        GHCUP_VER=0.1.20.0
         curl -sL -o ./ghcup https://downloads.haskell.org/~ghcup/$GHCUP_VER/x86_64-linux-ghcup-$GHCUP_VER
         chmod +x ./ghcup
-        GHCVER=8.10.7
+        GHCVER=9.8.1
         ./ghcup install ghc $GHCVER
         ./ghcup set ghc $GHCVER
         cabal update

benchmark/Streamly/Benchmark/Data/Stream/ConcurrentCommon.hs

@@ -147,10 +147,10 @@ concatFmapStreamsWith f outer inner n =
         $ Async.parConcat f
         $ fmap (sourceUnfoldrM inner) (sourceUnfoldrM outer n)
 
-o_1_space_concatMap :: Int -> (Config -> Config) -> [Benchmark]
-o_1_space_concatMap value f =
+o_1_space_concatMap :: String -> Int -> (Config -> Config) -> [Benchmark]
+o_1_space_concatMap label value f =
     value2 `seq`
-        [ bgroup "concat"
+        [ bgroup ("concat" ++ label)
             [ benchIO "parConcatMap (n of 1)"
                   (concatMapStreamsWith f value 1)
             , benchIO "parConcatMap (sqrt x of sqrt x)"
@@ -198,7 +198,9 @@ allBenchmarks moduleName wide modifier value =
     [ bgroup (o_1_space_prefix moduleName) $ concat
         [ o_1_space_mapping value modifier
         , o_1_space_concatFoldable value modifier
-        , o_1_space_concatMap value modifier
+        , o_1_space_concatMap "" value modifier
+        , o_1_space_concatMap "-maxThreads-1" value (modifier . Async.maxThreads 1)
+        , o_1_space_concatMap "-rate-Nothing" value (modifier . Async.rate Nothing)
         , o_1_space_joining value modifier
         ] ++ if wide then [] else o_1_space_outerProduct value modifier
     , bgroup (o_n_heap_prefix moduleName) $ concat

benchmark/Streamly/Benchmark/Data/Stream/Rate.hs

@@ -0,0 +1,129 @@
+{-# LANGUAGE FlexibleContexts #-}
+
+-- |
+-- Module      : Main
+-- Copyright   : (c) 2018 Composewell Technologies
+--
+-- License     : BSD3
+-- Maintainer  : [email protected]
+
+import Stream.Common (benchIOSrc, sourceUnfoldrM)
+import Streamly.Data.Stream (Stream)
+import Streamly.Internal.Data.Stream.Prelude (MonadAsync, Config)
+
+import qualified Streamly.Data.Stream.Prelude as Stream
+
+import Streamly.Benchmark.Common
+import Test.Tasty.Bench
+
+moduleName :: String
+moduleName = "Data.Stream.Rate"
+
+-------------------------------------------------------------------------------
+-- Average Rate
+-------------------------------------------------------------------------------
+
+{-# INLINE rateNothing #-}
+rateNothing :: MonadAsync m =>
+    ((Config -> Config) -> Stream m Int -> Stream m Int)
+    -> (Config -> Config) -> Int -> Int -> Stream m Int
+rateNothing f cfg value = f (Stream.rate Nothing . cfg) . sourceUnfoldrM value
+
+{-# INLINE avgRate #-}
+avgRate :: MonadAsync m =>
+    ((Config -> Config) -> Stream m Int -> Stream m Int)
+    -> (Config -> Config) -> Int -> Double -> Int -> Stream m Int
+avgRate f cfg value rt = f (Stream.avgRate rt . cfg) . sourceUnfoldrM value
+
+{-
+-- parEval should be maxThreads 1 anyway
+{-# INLINE avgRateThreads1 #-}
+avgRateThreads1 :: MonadAsync m =>
+    ((Config -> Config) -> Stream m Int -> Stream m Int)
+    -> (Config -> Config) -> Int -> Double -> Int -> Stream m Int
+avgRateThreads1 f cfg value rt =
+    f (Stream.maxThreads 1 . Stream.avgRate rt . cfg) . sourceUnfoldrM value
+-}
+
+{-# INLINE minRate #-}
+minRate :: MonadAsync m =>
+    ((Config -> Config) -> Stream m Int -> Stream m Int)
+    -> (Config -> Config) -> Int -> Double -> Int -> Stream m Int
+minRate f cfg value rt = f (Stream.minRate rt . cfg) . sourceUnfoldrM value
+
+{-# INLINE maxRate #-}
+maxRate :: MonadAsync m =>
+    ((Config -> Config) -> Stream m Int -> Stream m Int)
+    -> (Config -> Config) -> Int -> Double -> Int -> Stream m Int
+maxRate f cfg value rt = f (Stream.minRate rt . cfg) . sourceUnfoldrM value
+
+{-# INLINE constRate #-}
+constRate :: MonadAsync m =>
+    ((Config -> Config) -> Stream m Int -> Stream m Int)
+    -> (Config -> Config) -> Int -> Double -> Int -> Stream m Int
+constRate f cfg value rt = f (Stream.constRate rt . cfg) . sourceUnfoldrM value
+
+-- XXX arbitrarily large rate should be the same as rate Nothing
+-- XXX Add tests for multiworker cases as well - parMapM
+o_1_space_async :: Int -> [Benchmark]
+o_1_space_async value =
+    [ bgroup
+          "default/parEval"
+          [ bgroup
+                "avgRate"
+                -- benchIO "unfoldr" $ toNull
+                [ benchIOSrc "baseline" $ sourceUnfoldrM value
+                , benchIOSrc "Nothing" $ rateNothing Stream.parEval id value
+                , benchIOSrc "1M" $ avgRate Stream.parEval id value 1000000
+                , benchIOSrc "3M" $ avgRate Stream.parEval id value 3000000
+                -- , benchIOSrc "10M/maxThreads1" $ avgRateThreads1 Stream.parEval value 10000000
+                , benchIOSrc "10M" $ avgRate Stream.parEval id value 10000000
+                , benchIOSrc "20M" $ avgRate Stream.parEval id value 20000000
+                ]
+          , bgroup
+                "minRate"
+                [ benchIOSrc "1M" $ minRate Stream.parEval id value 1000000
+                , benchIOSrc "10M" $ minRate Stream.parEval id value 10000000
+                , benchIOSrc "20M" $ minRate Stream.parEval id value 20000000
+                ]
+          , bgroup
+                "maxRate"
+                [ -- benchIOSrc "10K" $ maxRate value 10000
+                  benchIOSrc "10M" $ maxRate Stream.parEval id value 10000000
+                ]
+          , bgroup
+                "constRate"
+                [ -- benchIOSrc "10K" $ constRate value 10000
+                  benchIOSrc "1M" $ constRate Stream.parEval id value 1000000
+                , benchIOSrc "10M" $ constRate Stream.parEval id value 10000000
+                ]
+          ]
+    ]
+
+o_1_space_ahead :: Int -> [Benchmark]
+o_1_space_ahead value =
+    [ bgroup
+        "ordered/parEval"
+        [ benchIOSrc "avgRate/1M"
+            $ avgRate Stream.parEval (Stream.ordered True) value 1000000
+        , benchIOSrc "minRate/1M"
+            $ minRate Stream.parEval (Stream.ordered True) value 1000000
+        , benchIOSrc "maxRate/1M"
+            $ maxRate Stream.parEval (Stream.ordered True) value 1000000
+        , benchIOSrc "constRate/1M"
+            $ constRate Stream.parEval (Stream.ordered True) value 1000000
+        ]
+    ]
+
+-------------------------------------------------------------------------------
+-- Main
+-------------------------------------------------------------------------------
+
+main :: IO ()
+main = runWithCLIOpts defaultStreamSize allBenchmarks
+
+    where
+
+    allBenchmarks value =
+        [ bgroup (o_1_space_prefix moduleName)
+        $ concat [o_1_space_async value, o_1_space_ahead value]]

benchmark/lib/Streamly/Benchmark/Prelude.hs

@@ -62,7 +62,7 @@ module Streamly.Benchmark.Prelude
     , transformComposeMapM
     , transformMapM
     , transformTeeMapM
-    , transformZipMapM
+    -- , transformZipMapM
     )
 where
 
@@ -78,7 +78,7 @@ import qualified Data.Foldable as F
 import qualified Data.List as List
 import qualified Streamly.Prelude  as S
 import qualified Streamly.Internal.Data.Stream.IsStream as Internal
-import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
+import qualified Streamly.Internal.Data.Stream.IsStream as IsStream
 import qualified Streamly.Internal.Data.Pipe as Pipe
 import qualified Streamly.Internal.Data.Stream.Serial as Serial
 
@@ -318,9 +318,10 @@ transformTeeMapM t n =
     composeN n $
     t .
     Internal.transform
-        (Pipe.mapM (\x -> return (x + 1)) `Pipe.tee`
+        (Pipe.mapM (\x -> return (x + 1)) `Pipe.teeMerge`
          Pipe.mapM (\x -> return (x + 2)))
 
+{-
 {-# INLINE transformZipMapM #-}
 transformZipMapM ::
        (S.IsStream t, S.MonadAsync m)
@@ -336,6 +337,7 @@ transformZipMapM t n =
              (+)
              (Pipe.mapM (\x -> return (x + 1)))
              (Pipe.mapM (\x -> return (x + 2))))
+-}
 
 -------------------------------------------------------------------------------
 -- Streams of streams