Fix/enhance benchmarks for concatPairsWith

* Add a benchmark for mergeByMFused
* Move interleaving benchmarks to WSerial module
* Rename some wserial benchmarks
* Add a Merge module for mergeBy/sorting benchmarks

.hlint.ignore

@@ -42,6 +42,7 @@ benchmark/Streamly/Benchmark/Data/Stream/StreamK.hs
 benchmark/Streamly/Benchmark/Data/Unfold.hs
 benchmark/Streamly/Benchmark/FileSystem/Handle.hs
 benchmark/Streamly/Benchmark/Prelude/Async.hs
+benchmark/Streamly/Benchmark/Prelude/Merge.hs
 benchmark/Streamly/Benchmark/Prelude/Parallel.hs
 benchmark/Streamly/Benchmark/Prelude/Rate.hs
 benchmark/Streamly/Benchmark/Prelude/Serial/Exceptions.hs
@@ -50,4 +51,3 @@ benchmark/Streamly/Benchmark/Prelude/Serial/Split.hs
 benchmark/Streamly/Benchmark/Prelude/Serial/Transformation1.hs
 benchmark/Streamly/Benchmark/Prelude/WAsync.hs
 benchmark/Streamly/Benchmark/Prelude/ZipAsync.hs
-benchmark/Streamly/Benchmark/Prelude/ZipSerial.hs

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

@@ -384,6 +384,14 @@ iterateDropWhileTrue streamLen iterStreamLen =
 zipWith :: Monad m => Stream m Int -> m ()
 zipWith src = drain $ S.zipWith (,) src src
 
+{-# INLINE sortByK #-}
+sortByK :: (a -> a -> Ordering) -> Stream m a -> Stream m a
+sortByK f = S.concatPairsWith (S.mergeBy f) S.fromPure
+
+{-# INLINE sortBy #-}
+sortBy :: Monad m => Stream m Int -> m ()
+sortBy = drain . sortByK compare
+
 -------------------------------------------------------------------------------
 -- Mixed Composition
 -------------------------------------------------------------------------------
@@ -700,6 +708,7 @@ o_1_space_concat streamLen =
             (concatMapBySerial streamLen2 streamLen2)
         , benchIOSrc1 "concatMapBy serial (1 of n)"
             (concatMapBySerial 1 streamLen)
+        , benchFold "sortBy" sortBy (unfoldrM streamLen)
         ]
     where
     streamLen2 = round (P.fromIntegral streamLen**(1/2::P.Double)) -- double nested loop

benchmark/Streamly/Benchmark/Prelude/Merge.hs

@@ -0,0 +1,270 @@
+-- |
+-- Module      : Main
+-- Copyright   : (c) 2018 Composewell Technologies
+--
+-- License     : BSD3
+-- Maintainer  : streamly@composewell.com
+
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE RankNTypes #-}
+
+#ifdef __HADDOCK_VERSION__
+#undef INSPECTION
+#endif
+
+#ifdef INSPECTION
+{-# LANGUAGE TemplateHaskell #-}
+{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
+#endif
+
+import Streamly.Internal.Data.Stream.IsStream (SerialT)
+import qualified Streamly.Internal.Data.Stream.IsStream as Stream
+-- import qualified Streamly.Internal.Data.Unfold as Unfold
+
+import Streamly.Benchmark.Common
+import Streamly.Benchmark.Prelude
+
+import Gauge
+
+#ifdef INSPECTION
+import GHC.Types (SPEC(..))
+import Test.Inspection
+
+import qualified Streamly.Internal.Data.Stream.StreamD as D
+#endif
+
+moduleName :: String
+moduleName = "Prelude.Merge"
+
+-------------------------------------------------------------------------------
+-- Merging
+-------------------------------------------------------------------------------
+
+{-# INLINE mergeWith #-}
+mergeWith ::
+    (  (Int -> Int -> Ordering)
+    -> SerialT IO Int
+    -> SerialT IO Int
+    -> SerialT IO Int
+    )
+    -> (Int -> Int -> Ordering)
+    -> Int -> Int -> IO ()
+mergeWith g cmp count n =
+    Stream.drain
+        $ g
+            cmp
+            (sourceUnfoldrM count n)
+            (sourceUnfoldrM count (n + 1))
+
+{-# INLINE mergeWithM #-}
+mergeWithM ::
+    (  (Int -> Int -> IO Ordering)
+    -> SerialT IO Int
+    -> SerialT IO Int
+    -> SerialT IO Int
+    )
+    -> (Int -> Int -> Ordering)
+    -> Int -> Int -> IO ()
+mergeWithM g cmp count n =
+    Stream.drain
+        $ g
+            (\a b -> return $ cmp a b)
+            (sourceUnfoldrM count n)
+            (sourceUnfoldrM count (n + 1))
+
+{-# INLINE mergeBy #-}
+mergeBy :: (Int -> Int -> Ordering) -> Int -> Int -> IO ()
+mergeBy = mergeWith Stream.mergeBy
+
+{-# INLINE mergeByM #-}
+mergeByM :: (Int -> Int -> Ordering) -> Int -> Int -> IO ()
+mergeByM = mergeWithM Stream.mergeByM
+
+{-# INLINE mergeByMFused #-}
+mergeByMFused :: (Int -> Int -> Ordering) -> Int -> Int -> IO ()
+mergeByMFused = mergeWithM Stream.mergeByMFused
+
+#ifdef INSPECTION
+inspect $ hasNoTypeClasses 'mergeBy
+inspect $ 'mergeBy `hasNoType` ''SPEC
+inspect $ 'mergeBy `hasNoType` ''D.Step
+
+inspect $ hasNoTypeClasses 'mergeByM
+inspect $ 'mergeByM `hasNoType` ''SPEC
+inspect $ 'mergeByM `hasNoType` ''D.Step
+
+{-# INLINE _mergeByMFusedCheck #-}
+_mergeByMFusedCheck :: IO ()
+_mergeByMFusedCheck = mergeWithM Stream.mergeByMFused compare 0 0
+
+inspect $ hasNoTypeClasses '_mergeByMFusedCheck
+inspect $ '_mergeByMFusedCheck `hasNoType` ''SPEC
+inspect $ '_mergeByMFusedCheck `hasNoType` ''D.Step
+#endif
+
+{-# INLINE concatPairsWithMergeBy #-}
+concatPairsWithMergeBy :: (Int -> Int -> Ordering) -> Int -> Int -> Int -> IO ()
+concatPairsWithMergeBy cmp = concatPairsWith (Stream.mergeBy cmp)
+
+{-# INLINE concatPairsWithMergeByFused #-}
+concatPairsWithMergeByFused ::
+    (Int -> Int -> Ordering) -> Int -> Int -> Int -> IO ()
+concatPairsWithMergeByFused cmp =
+    concatPairsWith (Stream.mergeByMFused (\x y -> return $ cmp x y))
+
+-------------------------------------------------------------------------------
+-- Interleaving
+-------------------------------------------------------------------------------
+
+o_1_space_joining :: Int -> [Benchmark]
+o_1_space_joining value =
+    [ bgroup "joining (2 of n/2)"
+        [ benchIOSrc1
+            "mergeBy compare"
+            (mergeBy compare (value `div` 2))
+        , benchIOSrc1
+            "mergeByM compare"
+            (mergeByM compare (value `div` 2))
+        , benchIOSrc1
+            "mergeByMFused compare"
+            (mergeByMFused compare (value `div` 2))
+
+        , benchIOSrc1
+            "mergeBy (flip compare)"
+            (mergeBy (flip compare) (value `div` 2))
+        , benchIOSrc1
+            "mergeByM (flip compare)"
+            (mergeByM (flip compare) (value `div` 2))
+        , benchIOSrc1
+            "mergeByMFused (flip compare)"
+            (mergeByMFused (flip compare) (value `div` 2))
+
+        , benchIOSrc1
+            "concatPairsWithMergeBy compare"
+            (concatPairsWithMergeBy compare 2 (value `div` 2))
+        , benchIOSrc1
+            "concatPairsWithMergeBy (flip compare)"
+            (concatPairsWithMergeBy (flip compare) 2 (value `div` 2))
+
+        , benchIOSrc1
+            "concatPairsWithMergeByFused compare"
+            (concatPairsWithMergeByFused compare 2 (value `div` 2))
+        , benchIOSrc1
+            "concatPairsWithMergeByFused (flip compare)"
+            (concatPairsWithMergeByFused (flip compare) 2 (value `div` 2))
+        ]
+    ]
+
+-------------------------------------------------------------------------------
+-- Concat
+-------------------------------------------------------------------------------
+
+{-
+{-# INLINE sourceUnfoldrMUF #-}
+-- (count, value)
+sourceUnfoldrMUF :: Monad m => Int -> UF.Unfold m (Int, Int) Int
+sourceUnfoldrMUF count = UF.unfoldrM step
+    where
+    step (cnt, start) =
+        return $
+            if cnt > start + count
+            then Nothing
+            else Just (cnt, (cnt + 1, start))
+
+{-# INLINE unfoldManyMergeBy #-}
+unfoldManyMergeBy :: Int -> Int -> Int -> IO ()
+unfoldManyMergeBy outer inner n =
+    S.drain $ (Internal.unfoldManyMergeBy compare)
+        -- (UF.lmap return (UF.replicateM inner))
+        (UF.lmap (\x -> (x,x)) (sourceUnfoldrMUF inner))
+        (sourceUnfoldrM outer n)
+
+#ifdef INSPECTION
+inspect $ hasNoTypeClasses 'unfoldManyMergeBy
+-- inspect $ 'unfoldManyMergeBy `hasNoType` ''SPEC
+-- inspect $ 'unfoldManyMergeBy `hasNoType`
+--      ''D.ConcatUnfoldMergeState
+#endif
+-}
+
+{-# INLINE sortBy #-}
+sortBy :: (Int -> Int -> Ordering) -> SerialT IO Int -> IO ()
+sortBy cmp = Stream.drain . Stream.sortBy cmp
+
+-- For fair comparison with concatPairs, removed sorted segmentation
+{-# INLINE listSortBy #-}
+listSortBy :: (a -> a -> Ordering) -> [a] -> [a]
+listSortBy cmp = mergeAll . sequences
+  where
+    sequences = fmap (: [])
+
+    mergeAll [x] = x
+    mergeAll xs  = mergeAll (mergePairs xs)
+
+    mergePairs (a:b:xs) = let !x = merge a b
+                          in x : mergePairs xs
+    mergePairs xs       = xs
+
+    merge as@(a:as') bs@(b:bs')
+      | a `cmp` b == GT = b:merge as  bs'
+      | otherwise       = a:merge as' bs
+    merge [] bs         = bs
+    merge as []         = as
+
+o_n_heap_concat :: Int -> [Benchmark]
+o_n_heap_concat value =
+    [ bgroup "concatPairsWith"
+        [ benchIOSrc1
+            "concatPairsWithMergeBy compare (n of 1)"
+            (concatPairsWithMergeBy compare value 1)
+        , benchIOSrc1
+            "concatPairsWithMergeBy compare (sqrtVal of sqrtVal)"
+            (concatPairsWithMergeBy compare sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithMergeBy (flip compare) (n of 1)"
+            (concatPairsWithMergeBy (flip compare) value 1)
+        , benchIOSrc1
+            "concatPairsWithMergeBy (flip compare) (sqrtVal of sqrtVal)"
+            (concatPairsWithMergeBy (flip compare) sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithMergeByFused compare (n of 1)"
+            (concatPairsWithMergeByFused compare value 1)
+        , benchIOSrc1
+            "concatPairsWithMergeByFused compare (sqrtVal of sqrtVal)"
+            (concatPairsWithMergeByFused compare sqrtVal sqrtVal)
+        ]
+    , bgroup "sorting"
+        [ benchIOSink value "sortBy compare" (sortBy compare)
+        , benchIOSink value "sortBy (flip compare)" (sortBy (flip compare))
+        , bench "sortByLists compare"
+            $ nf (\x -> listSortBy compare [1..x]) value
+        , bench "sortByLists (flip compare)"
+            $ nf (\x -> listSortBy (flip compare) [1..x]) value
+        ]
+    ]
+
+    where
+
+    sqrtVal = round $ sqrt (fromIntegral value :: Double)
+
+-------------------------------------------------------------------------------
+-- Main
+-------------------------------------------------------------------------------
+
+-- In addition to gauge options, the number of elements in the stream can be
+-- passed using the --stream-size option.
+--
+main :: IO ()
+main = runWithCLIOpts defaultStreamSize allBenchmarks
+
+    where
+
+    allBenchmarks size =
+        [ bgroup (o_1_space_prefix moduleName) (o_1_space_joining size)
+        , bgroup (o_n_heap_prefix moduleName) (o_n_heap_concat size)
+        ]

benchmark/Streamly/Benchmark/Prelude/Serial/Nested.hs

@@ -125,38 +125,6 @@ inspect $ 'append2 `hasNoType` ''SPEC
 inspect $ 'append2 `hasNoType` ''D.AppendState
 #endif
 
--------------------------------------------------------------------------------
--- Merging
--------------------------------------------------------------------------------
-
-{-# INLINE mergeBy #-}
-mergeBy :: Int -> Int -> IO ()
-mergeBy count n =
-    S.drain $
-    S.mergeBy
-        compare
-        (sourceUnfoldrM count n)
-        (sourceUnfoldrM count (n + 1))
-
-{-# INLINE mergeByM #-}
-mergeByM :: Int -> Int -> IO ()
-mergeByM count n =
-    S.drain $
-    S.mergeByM
-        (\a b -> return $ compare a b)
-        (sourceUnfoldrM count n)
-        (sourceUnfoldrM count (n + 1))
-
-#ifdef INSPECTION
-inspect $ hasNoTypeClasses 'mergeBy
-inspect $ 'mergeBy `hasNoType` ''SPEC
-inspect $ 'mergeBy `hasNoType` ''D.Step
-
-inspect $ hasNoTypeClasses 'mergeByM
-inspect $ 'mergeByM `hasNoType` ''SPEC
-inspect $ 'mergeByM `hasNoType` ''D.Step
-#endif
-
 o_1_space_joining :: Int -> [Benchmark]
 o_1_space_joining value =
     [ bgroup "joining"
@@ -164,8 +132,6 @@ o_1_space_joining value =
         , benchIOSrc1 "append (2,x/2)" (append2 (value `div` 2))
         , benchIOSrc1 "serial (2,2,x/4)" (serial4 (value `div` 4))
         , benchIOSrc1 "append (2,2,x/4)" (append4 (value `div` 4))
-        , benchIOSrc1 "mergeBy (2,x/2)" (mergeBy (value `div` 2))
-        , benchIOSrc1 "mergeByM (2,x/2)" (mergeByM (value `div` 2))
         ]
     ]
 
@@ -263,30 +229,14 @@ inspect $ 'concatMapWithAppend `hasNoType` ''SPEC
 
 -- concatPairWith
 
+{-# INLINE concatPairWithSerial #-}
+concatPairWithSerial :: Int -> Int -> Int -> IO ()
+concatPairWithSerial = concatPairsWith Internal.serial
+
 {-# INLINE concatPairWithAppend #-}
 concatPairWithAppend :: Int -> Int -> Int -> IO ()
 concatPairWithAppend = concatPairsWith Internal.append
 
-{-# INLINE concatPairWithInterleave #-}
-concatPairWithInterleave :: Int -> Int -> Int -> IO ()
-concatPairWithInterleave = concatPairsWith Internal.interleave
-
-{-# INLINE concatPairWithInterleaveSuffix #-}
-concatPairWithInterleaveSuffix :: Int -> Int -> Int -> IO ()
-concatPairWithInterleaveSuffix = concatPairsWith Internal.interleaveSuffix
-
-{-# INLINE concatPairWithInterleaveInfix #-}
-concatPairWithInterleaveInfix :: Int -> Int -> Int -> IO ()
-concatPairWithInterleaveInfix = concatPairsWith Internal.interleaveInfix
-
-{-# INLINE concatPairWithInterleaveMin #-}
-concatPairWithInterleaveMin :: Int -> Int -> Int -> IO ()
-concatPairWithInterleaveMin = concatPairsWith Internal.interleaveMin
-
-{-# INLINE concatPairWithRoundrobin #-}
-concatPairWithRoundrobin :: Int -> Int -> Int -> IO ()
-concatPairWithRoundrobin = concatPairsWith Internal.roundrobin
-
 -- unfoldMany
 
 -- unfoldMany replicate/unfoldrM
@@ -356,23 +306,18 @@ o_1_space_concat value = sqrtVal `seq`
 
         -------------------concatPairsWith-----------------
 
-        , benchIOSrc1 "concatPairWithAppend"
+        -- Use large number of streams to check scalability
-            (concatPairWithAppend 2 (value `div` 2))
+        , benchIOSrc1 "concatPairWithSerial (n of 1)"
-        , benchIOSrc1 "concatPairWithInterleave"
+            (concatPairWithSerial value 1)
-        (concatPairWithInterleave 2 (value `div` 2))
+        , benchIOSrc1 "concatPairWithSerial (sqrtVal of sqrtVal)"
-
+            (concatPairWithSerial sqrtVal sqrtVal)
-        , benchIOSrc1 "concatPairWithInterleaveSuffix"
+        , benchIOSrc1 "concatPairWithSerial (2 of n/2)"
-        (concatPairWithInterleaveSuffix 2 (value `div` 2))
+            (concatPairWithSerial 2 (value `div` 2))
-
-        , benchIOSrc1 "concatPairWithInterleaveInfix"
-        (concatPairWithInterleaveInfix 2 (value `div` 2))
-
-        , benchIOSrc1 "concatPairWithInterleaveMin"
-        (concatPairWithInterleaveMin 2 (value `div` 2))
-
-        , benchIOSrc1 "concatPairWithRoundrobin"
-        (concatPairWithRoundrobin 2 (value `div` 2))
 
+        , benchIOSrc1 "concatPairWithAppend (n of 1)"
+            (concatPairWithAppend value 1)
+        , benchIOSrc1 "concatPairWithAppend (sqrtVal of sqrtVal)"
+            (concatPairWithAppend sqrtVal sqrtVal)
         ]
     ]
 

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation2.hs

@@ -20,7 +20,6 @@ import Control.Monad.IO.Class (MonadIO(..))
 import Data.Monoid (Sum(..))
 import GHC.Generics (Generic)
 
-import qualified Data.List as List
 import qualified Streamly.Internal.Data.Fold as FL
 import qualified Streamly.Internal.Data.Stream.IsStream as Internal
 import qualified Streamly.Prelude  as S
@@ -175,19 +174,13 @@ reverse n = composeN n S.reverse
 reverse' :: MonadIO m => Int -> SerialT m Int -> m ()
 reverse' n = composeN n Internal.reverse'
 
-{-# INLINE sortBy #-}
-sortBy :: MonadIO m => Int -> SerialT m Int -> m ()
-sortBy n = composeN n (Internal.sortBy compare)
-
 o_n_heap_buffering :: Int -> [Benchmark]
 o_n_heap_buffering value =
     [ bgroup "buffered"
         [
-        -- Reversing/sorting a stream
+        -- Reversing a stream
           benchIOSink value "reverse" (reverse 1)
         , benchIOSink value "reverse'" (reverse' 1)
-        , benchIOSink value "sortBy" (sortBy 1)
-        , bench "sort Lists" $ nf (\x -> List.sort [1..x]) value
 
         , benchIOSink value "mkAsync" (mkAsync fromSerial)
         ]

benchmark/Streamly/Benchmark/Prelude/WSerial.hs

@@ -88,18 +88,54 @@ inspect $ 'roundRobin2 `hasNoType` ''SPEC
 inspect $ 'roundRobin2 `hasNoType` ''D.InterleaveState
 #endif
 
-o_1_space_joining :: Int -> [Benchmark]
+{-# INLINE sourceUnfoldrMUF #-}
-o_1_space_joining value =
+-- (count, value)
-    [ bgroup "joining"
+sourceUnfoldrMUF :: Monad m => Int -> UF.Unfold m (Int, Int) Int
-        [ benchIOSrc1 "wSerial (2,x/2)" (wSerial2 value)
+sourceUnfoldrMUF count = UF.unfoldrM step
-        , benchIOSrc1 "interleave (2,x/2)" (interleave2 value)
+    where
-        , benchIOSrc1 "roundRobin (2,x/2)" (roundRobin2 value)
+    step (cnt, start) =
-        ]
+        return $
-    ]
+            if cnt > start + count
+            then Nothing
+            else Just (cnt, (cnt + 1, start))
 
--------------------------------------------------------------------------------
+{-# INLINE unfoldManyInterleave #-}
--- Concat
+unfoldManyInterleave :: Int -> Int -> Int -> IO ()
--------------------------------------------------------------------------------
+unfoldManyInterleave outer inner n =
+    S.drain $ Internal.unfoldManyInterleave
+        -- (UF.lmap return (UF.replicateM inner))
+        (UF.lmap (\x -> (x,x)) (sourceUnfoldrMUF inner))
+        (sourceUnfoldrM outer n)
+
+#ifdef INSPECTION
+inspect $ hasNoTypeClasses 'unfoldManyInterleave
+-- inspect $ 'unfoldManyInterleave `hasNoType` ''SPEC
+-- inspect $ 'unfoldManyInterleave `hasNoType`
+--      ''D.ConcatUnfoldInterleaveState
+#endif
+
+{-# INLINE unfoldManyRoundRobin #-}
+unfoldManyRoundRobin :: Int -> Int -> Int -> IO ()
+unfoldManyRoundRobin outer inner n =
+    S.drain $ Internal.unfoldManyRoundRobin
+        -- (UF.lmap return (UF.replicateM inner))
+        (UF.lmap (\x -> (x,x)) (sourceUnfoldrMUF inner))
+        (sourceUnfoldrM outer n)
+
+#ifdef INSPECTION
+inspect $ hasNoTypeClasses 'unfoldManyRoundRobin
+-- inspect $ 'unfoldManyRoundRobin `hasNoType` ''SPEC
+-- inspect $ 'unfoldManyRoundRobin `hasNoType`
+--      ''D.ConcatUnfoldInterleaveState
+#endif
+
+{-# INLINE concatPairsWithWSerial #-}
+concatPairsWithWSerial :: Int -> Int -> Int -> IO ()
+concatPairsWithWSerial = concatPairsWith Internal.wSerial
+
+{-# INLINE concatPairsWithRoundrobin #-}
+concatPairsWithRoundrobin :: Int -> Int -> Int -> IO ()
+concatPairsWithRoundrobin = concatPairsWith Internal.roundrobin
 
 {-# INLINE concatMapWithWSerial #-}
 concatMapWithWSerial :: Int -> Int -> Int -> IO ()
@@ -110,86 +146,189 @@ inspect $ hasNoTypeClasses 'concatMapWithWSerial
 inspect $ 'concatMapWithWSerial `hasNoType` ''SPEC
 #endif
 
-o_1_space_concat :: Int -> [Benchmark]
+o_1_space_joining :: Int -> [Benchmark]
-o_1_space_concat value =
+o_1_space_joining value =
-    [ bgroup "concat"
+    [ bgroup "joining (2 of n/2)"
-        [ benchIOSrc1
+        [ benchIOSrc1 "wSerial" (wSerial2 value)
-              "concatMapWithWSerial (2,x/2)"
+        , benchIOSrc1 "interleave" (interleave2 value)
+        , benchIOSrc1 "roundRobin" (roundRobin2 value)
+        , benchIOSrc1
+              "concatMapWithWSerial"
               (concatMapWithWSerial 2 (value `div` 2))
         , benchIOSrc1
-              "concatMapWithWSerial (x/2,2)"
+              "concatMapWithInterleave"
-              (concatMapWithWSerial (value `div` 2) 2)
+              (concatStreamsWith Internal.interleave 2 (value `div` 2))
+        , benchIOSrc1
+              "concatMapWithRoundrobin"
+              (concatStreamsWith Internal.roundrobin 2 (value `div` 2))
+        , benchIOSrc1
+              "unfoldManyInterleave"
+              (unfoldManyInterleave 2 (value `div` 2))
+        , benchIOSrc1
+            "concatPairsWithWSerial"
+            (concatPairsWithWSerial 2 (value `div` 2))
+        , benchIOSrc1
+            "concatPairsWithRoundrobin"
+            (concatPairsWithRoundrobin 2 (value `div` 2))
         ]
     ]
 
-{-# INLINE unfoldManyInterleaveRepl4xN #-}
+-------------------------------------------------------------------------------
-unfoldManyInterleaveRepl4xN :: Int -> Int -> IO ()
+-- Concat
-unfoldManyInterleaveRepl4xN value n =
+-------------------------------------------------------------------------------
-    S.drain $ Internal.unfoldManyInterleave
-        (UF.lmap return (UF.replicateM 4))
-        (sourceUnfoldrM (value `div` 4) n)
 
-#ifdef INSPECTION
+o_1_space_concat :: Int -> [Benchmark]
-inspect $ hasNoTypeClasses 'unfoldManyInterleaveRepl4xN
+o_1_space_concat value =
--- inspect $ 'unfoldManyInterleaveRepl4xN `hasNoType` ''SPEC
+    [ bgroup "concatMapWith"
--- inspect $ 'unfoldManyInterleaveRepl4xN `hasNoType`
+        [ benchIOSrc1
---      ''D.ConcatUnfoldInterleaveState
+            "concatMapWithWSerial (n of 1)"
-#endif
+            (concatStreamsWith wSerial value 1)
+        , benchIOSrc1
+            "concatMapWithWSerial (sqrtVal of sqrtVal)"
+            (concatStreamsWith wSerial sqrtVal sqrtVal)
+        -- concatMapWith using StreamD versions of interleave operations are
+        -- all quadratic, we just measure the sqrtVal benchmark for comparison.
+        {-
+        , benchIOSrc1
+            "concatMapWithInterleave (n of 1)"
+            (concatStreamsWith Internal.interleave value 1)
+        -}
+        , benchIOSrc1
+            "concatMapWithInterleave (sqrtVal of sqrtVal)"
+            (concatStreamsWith Internal.interleave sqrtVal sqrtVal)
+        , benchIOSrc1
+            "concatMapWithRoundrobin (sqrtVal of sqrtVal)"
+            (concatStreamsWith Internal.roundrobin sqrtVal sqrtVal)
+        ]
+    ]
 
-{-# INLINE unfoldManyRoundRobinRepl4xN #-}
+    where
-unfoldManyRoundRobinRepl4xN :: Int -> Int -> IO ()
-unfoldManyRoundRobinRepl4xN value n =
-    S.drain $ Internal.unfoldManyRoundRobin
-        (UF.lmap return (UF.replicateM 4))
-        (sourceUnfoldrM (value `div` 4) n)
 
-#ifdef INSPECTION
+    sqrtVal = round $ sqrt (fromIntegral value :: Double)
-inspect $ hasNoTypeClasses 'unfoldManyRoundRobinRepl4xN
+
--- inspect $ 'unfoldManyRoundRobinRepl4xN `hasNoType` ''SPEC
+{-# INLINE concatPairsWithInterleave #-}
--- inspect $ 'unfoldManyRoundRobinRepl4xN `hasNoType`
+concatPairsWithInterleave :: Int -> Int -> Int -> IO ()
---      ''D.ConcatUnfoldInterleaveState
+concatPairsWithInterleave = concatPairsWith Internal.interleave
-#endif
+
+{-# INLINE concatPairsWithInterleaveSuffix #-}
+concatPairsWithInterleaveSuffix :: Int -> Int -> Int -> IO ()
+concatPairsWithInterleaveSuffix = concatPairsWith Internal.interleaveSuffix
+
+{-# INLINE concatPairsWithInterleaveInfix #-}
+concatPairsWithInterleaveInfix :: Int -> Int -> Int -> IO ()
+concatPairsWithInterleaveInfix = concatPairsWith Internal.interleaveInfix
+
+{-# INLINE concatPairsWithInterleaveMin #-}
+concatPairsWithInterleaveMin :: Int -> Int -> Int -> IO ()
+concatPairsWithInterleaveMin = concatPairsWith Internal.interleaveMin
 
 o_n_heap_concat :: Int -> [Benchmark]
 o_n_heap_concat value =
-    [ bgroup "concat"
+    [ bgroup "concatPairsWith"
         [
-          -- interleave x/4 streams of 4 elements each. Needs to buffer
-          -- proportional to x/4. This is different from WSerial because
-          -- WSerial expands slowly because of binary interleave behavior and
-          -- this expands immediately because of Nary interleave behavior.
           benchIOSrc1
-              "unfoldManyInterleaveRepl (x/4,4)"
+              "unfoldManyInterleave (n of 1)"
-              (unfoldManyInterleaveRepl4xN value)
+              (unfoldManyInterleave value 1)
         , benchIOSrc1
-              "unfoldManyRoundRobinRepl (x/4,4)"
+              "unfoldManyInterleave (sqrtVal of sqrtVal)"
-              (unfoldManyRoundRobinRepl4xN value)
+              (unfoldManyInterleave sqrtVal sqrtVal)
+
+        , benchIOSrc1
+              "unfoldManyRoundRobin (n of 1)"
+              (unfoldManyRoundRobin value 1)
+        , benchIOSrc1
+              "unfoldManyRoundRobin (sqrtVal of sqrtVal)"
+              (unfoldManyRoundRobin sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithWSerial (n of 1)"
+            (concatPairsWithWSerial value 1)
+        , benchIOSrc1
+            "concatPairsWithWSerial (sqrtVal of sqrtVal)"
+            (concatPairsWithWSerial sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithInterleave (n of 1)"
+            (concatPairsWithInterleave value 1)
+        , benchIOSrc1
+            "concatPairsWithInterleave (sqrtVal of sqrtVal)"
+            (concatPairsWithInterleave sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithInterleaveSuffix (n of 1)"
+            (concatPairsWithInterleaveSuffix value 1)
+        , benchIOSrc1
+            "concatPairsWithInterleaveSuffix (sqrtVal of sqrtVal)"
+            (concatPairsWithInterleaveSuffix sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithInterleaveInfix (n of 1)"
+            (concatPairsWithInterleaveInfix value 1)
+        , benchIOSrc1
+            "concatPairsWithInterleaveInfix (sqrtVal of sqrtVal)"
+            (concatPairsWithInterleaveInfix sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithInterleaveMin (n of 1)"
+            (concatPairsWithInterleaveMin value 1)
+        , benchIOSrc1
+            "concatPairsWithInterleaveMin (sqrtVal of sqrtVal)"
+            (concatPairsWithInterleaveMin sqrtVal sqrtVal)
+
+        , benchIOSrc1
+            "concatPairsWithRoundrobin (n of 1)"
+            (concatPairsWithRoundrobin value 1)
+        , benchIOSrc1
+            "concatPairsWithRoundrobin (sqrtVal of sqrtVal)"
+            (concatPairsWithRoundrobin sqrtVal sqrtVal)
         ]
     ]
 
+    where
+
+    sqrtVal = round $ sqrt (fromIntegral value :: Double)
+
 -------------------------------------------------------------------------------
 -- Monad
 -------------------------------------------------------------------------------
 
 o_1_space_outerProduct :: Int -> [Benchmark]
 o_1_space_outerProduct value =
-    [ bgroup "monad-outer-product"
+    [ bgroup "outer-product"
-        [ benchIO "toNullAp" $ toNullAp value fromWSerial
+        [ benchIO
-        , benchIO "toNullM" $ toNullM value fromWSerial
+            "drain applicative (+) (sqrtVal of sqrtVal)"
-        , benchIO "toNullM3" $ toNullM3 value fromWSerial
+            (toNullAp value fromWSerial)
-        , benchIO "filterAllOutM" $ filterAllOutM value fromWSerial
+        , benchIO
-        , benchIO "filterAllInM" $ filterAllInM value fromWSerial
+            "drain monad (+) (sqrtVal of sqrtVal)"
-        , benchIO "filterSome" $ filterSome value fromWSerial
+            (toNullM value fromWSerial)
-        , benchIO "breakAfterSome" $ breakAfterSome value fromWSerial
+        , benchIO
+            "drain monad (+) (cbrtVal of cbrtVal of cbrtVal)"
+            (toNullM3 value fromWSerial)
+        , benchIO
+            "filterAllOut monad (+) (sqrtVal of sqrtVal)"
+            (filterAllOutM value fromWSerial)
+        , benchIO
+            "filterAllIn monad (+) (sqrtVal of sqrtVal)"
+            (filterAllInM value fromWSerial)
+        , benchIO
+            "filterSome monad (+) (sqrtVal of sqrtVal)"
+            (filterSome value fromWSerial)
+        , benchIO
+            "breakAfterSome monad (+) (sqrtVal of sqrtVal)"
+            (breakAfterSome value fromWSerial)
         ]
     ]
 
 o_n_space_outerProduct :: Int -> [Benchmark]
 o_n_space_outerProduct value =
     [ bgroup
-        "monad-outer-product"
+        "outer-product"
-        [ benchIO "toList" $ toListM value fromWSerial
+        [ benchIO
-        , benchIO "toListSome" $ toListSome value fromWSerial
+            "toList monad (+) (sqrtVal of sqrtVal)"
+            (toListM value fromWSerial)
+        , benchIO
+            "toListSome monad (+) (sqrtVal of sqrtVal)"
+            (toListSome value fromWSerial)
         ]
     ]
 

benchmark/Streamly/Benchmark/Prelude/ZipSerial.hs

@@ -84,11 +84,28 @@ inspect $ 'zipWithM `hasNoType` ''SPEC
 inspect $ 'zipWithM `hasNoType` ''D.Step
 #endif
 
+{-
+{-# INLINE concatMapWithZip #-}
+concatMapWithZip :: Int -> Int -> Int -> IO ()
+concatMapWithZip = concatStreamsWith (S.zipWith (+))
+-}
+
+{-# INLINE concatPairsWithZip #-}
+concatPairsWithZip :: Int -> Int -> Int -> IO ()
+concatPairsWithZip = concatPairsWith (S.zipWith (+))
+
 o_1_space_joining :: Int -> [Benchmark]
 o_1_space_joining value =
     [ bgroup "joining"
-        [ benchIOSrc1 "zip (2,x/2)" (zipWith (value `div` 2))
+        [ benchIOSrc1 "zip (2 of n/2)" (zipWith (value `div` 2))
-        , benchIOSrc1 "zipM (2,x/2)" (zipWithM (value `div` 2))
+        , benchIOSrc1 "zipM (2 of n/2)" (zipWithM (value `div` 2))
+        {-
+        -- Not correct because of nil stream at end issue.
+        , benchIOSrc1 "concatMapWithZip (+) (2 of n/2)"
+            (concatMapWithZip 2 (value `div` 2))
+        -}
+        , benchIOSrc1 "concatPairsWithZip (+) (2 of n/2)"
+            (concatPairsWithZip 2 (value `div` 2))
         ]
     ]
 
@@ -103,11 +120,37 @@ o_1_space_mapping value =
         ]
     ]
 
+-------------------------------------------------------------------------------
+-- Concat
+-------------------------------------------------------------------------------
+
+o_n_heap_concat :: Int -> [Benchmark]
+o_n_heap_concat value =
+    [ bgroup "concatPairsWith"
+        [ {- -- This fails with stack overflow.
+            benchIOSrc1 "concatMapWithZip (n of 1)"
+            (concatMapWithZip value 1)
+        -- Not correct because of nil stream at end issue.
+        , benchIOSrc1 "concatMapWithZip (sqrtVal of sqrtVal)"
+            (concatMapWithZip sqrtVal sqrtVal)
+        -}
+          benchIOSrc1 "concatPairsWithZip (n of 1)"
+            (concatPairsWithZip value 1)
+        , benchIOSrc1 "concatPairsWithZip (sqrtVal of sqrtVal)"
+            (concatPairsWithZip sqrtVal sqrtVal)
+        ]
+    ]
+
+    where
+
+    sqrtVal = round $ sqrt (fromIntegral value :: Double)
+
 -------------------------------------------------------------------------------
 -- Monad outer product
 -------------------------------------------------------------------------------
 
 {-
+-- Not correct because of nil stream at end issue.
 o_1_space_outerProduct :: Int -> [Benchmark]
 o_1_space_outerProduct value =
     [ bgroup "monad-outer-product"
@@ -133,6 +176,7 @@ main = runWithCLIOpts defaultStreamSize allBenchmarks
         [ bgroup (o_1_space_prefix moduleName) $ concat
             [ o_1_space_joining size
             , o_1_space_mapping size
+            , o_n_heap_concat size
             -- XXX need to fix, timing in ns
             -- , o_1_space_outerProduct size
             ]

benchmark/lib/Streamly/Benchmark/Prelude.hs

@@ -346,7 +346,7 @@ sourceConcatMapId value n =
 
 {-# INLINE concatStreamsWith #-}
 concatStreamsWith
-    :: (forall c. S.SerialT IO c -> S.SerialT IO c -> S.SerialT IO c)
+    :: (S.SerialT IO Int -> S.SerialT IO Int -> S.SerialT IO Int)
     -> Int
     -> Int
     -> Int
@@ -358,7 +358,7 @@ concatStreamsWith op outer inner n =
 
 {-# INLINE concatPairsWith #-}
 concatPairsWith
-    :: (forall c. S.SerialT IO c -> S.SerialT IO c -> S.SerialT IO c)
+    :: (S.SerialT IO Int -> S.SerialT IO Int -> S.SerialT IO Int)
     -> Int
     -> Int
     -> Int

benchmark/streamly-benchmarks.cabal

@@ -210,6 +210,16 @@ benchmark Prelude.WSerial
   else
     buildable: True
 
+benchmark Prelude.Merge
+  import: bench-options
+  type: exitcode-stdio-1.0
+  hs-source-dirs: Streamly/Benchmark/Prelude
+  main-is: Merge.hs
+  if impl(ghcjs)
+    buildable: False
+  else
+    buildable: True
+
 benchmark Prelude.ZipSerial
   import: bench-options
   type: exitcode-stdio-1.0

hie.yaml

@@ -26,6 +26,8 @@ cradle:
               component: "bench:Prelude.Ahead"
             - path: "./benchmark/Streamly/Benchmark/Prelude/Async.hs"
               component: "bench:Prelude.Async"
+            - path: "./benchmark/Streamly/Benchmark/Prelude/Merge.hs"
+              component: "bench:Prelude.Merge"
             - path: "./benchmark/Streamly/Benchmark/Prelude/Parallel.hs"
               component: "bench:Prelude.Parallel"
             - path: "./benchmark/Streamly/Benchmark/Prelude/Serial.hs"
@@ -44,6 +46,10 @@ cradle:
               component: "bench:Prelude.Serial"
             - path: "./benchmark/Streamly/Benchmark/Prelude/Serial/Transformation3.hs"
               component: "bench:Prelude.Serial"
+            - path: "./benchmark/Streamly/Benchmark/Prelude/WSerial.hs"
+              component: "bench:Prelude.WSerial"
+            - path: "./benchmark/Streamly/Benchmark/Prelude/ZipSerial.hs"
+              component: "bench:Prelude.ZipSerial"
             - path: "./benchmark/Streamly/Benchmark/Unicode/Stream.hs"
               component: "bench:Unicode.Stream"
             - path: "./benchmark/lib/"