diff --git a/benchmark/Streamly/Benchmark/Data/Stream/Expand.hs b/benchmark/Streamly/Benchmark/Data/Stream/Expand.hs
index 11ee4658e4..357bd4b8d5 100644
--- a/benchmark/Streamly/Benchmark/Data/Stream/Expand.hs
+++ b/benchmark/Streamly/Benchmark/Data/Stream/Expand.hs
@@ -205,6 +205,21 @@ inspect $ hasNoTypeClasses 'bfsUnfoldEach
 --      ''S.ConcatUnfoldInterleaveState
 #endif
 
+{-# INLINE altBfsUnfoldEach #-}
+altBfsUnfoldEach :: Int -> Int -> Int -> IO ()
+altBfsUnfoldEach outer inner n =
+    S.drain $ S.altBfsUnfoldEach
+        -- (UF.lmap return (UF.replicateM inner))
+        (UF.lmap (\x -> (x,x)) (sourceUnfoldrMUF inner))
+        (sourceUnfoldrM outer n)
+
+#ifdef INSPECTION
+inspect $ hasNoTypeClasses 'altBfsUnfoldEach
+-- inspect $ 'altBfsUnfoldEach `hasNoType` ''SPEC
+-- inspect $ 'altBfsUnfoldEach `hasNoType`
+--      ''S.ConcatUnfoldInterleaveState
+#endif
+
 {-# INLINE unfoldSched #-}
 unfoldSched :: Int -> Int -> Int -> IO ()
 unfoldSched outer inner n =
@@ -245,6 +260,7 @@ o_1_space_joining value =
 
         -- join 2 streams using n-ary ops
         , benchIOSrc1 "bfsUnfoldEach" (bfsUnfoldEach 2 (value `div` 2))
+        , benchIOSrc1 "altBfsUnfoldEach" (altBfsUnfoldEach 2 (value `div` 2))
         , benchIOSrc1 "unfoldSched" (unfoldSched 2 (value `div` 2))
         , benchIOSrc1 "concatMap" (concatMap 2 (value `div` 2))
         ]
@@ -432,6 +448,13 @@ o_n_heap_concat value = sqrtVal `seq`
               "bfsUnfoldEach (sqrtVal of sqrtVal)"
               (bfsUnfoldEach sqrtVal sqrtVal)
 
+        , benchIOSrc1
+              "altBfsUnfoldEach (n of 1)"
+              (altBfsUnfoldEach value 1)
+        , benchIOSrc1
+              "altBfsUnfoldEach (sqrtVal of sqrtVal)"
+              (altBfsUnfoldEach sqrtVal sqrtVal)
+
         , benchIOSrc1
               "unfoldSched (n of 1)"
               (unfoldSched value 1)
@@ -461,6 +484,39 @@ cross2 linearCount start = drain $
 
     nestedCount2 = round (fromIntegral linearCount**(1/2::Double))
 
+{-# INLINE crossApply #-}
+crossApply :: MonadAsync m => Int -> Int -> m ()
+crossApply linearCount start = drain $
+    Stream.crossApply
+        ((+) <$> sourceUnfoldrM nestedCount2 start)
+        (sourceUnfoldrM nestedCount2 start)
+
+    where
+
+    nestedCount2 = round (fromIntegral linearCount**(1/2::Double))
+
+{-# INLINE crossApplyFst #-}
+crossApplyFst :: MonadAsync m => Int -> Int -> m ()
+crossApplyFst linearCount start = drain $
+    Stream.crossApplyFst
+        (sourceUnfoldrM nestedCount2 start)
+        (sourceUnfoldrM nestedCount2 start)
+
+    where
+
+    nestedCount2 = round (fromIntegral linearCount**(1/2::Double))
+
+{-# INLINE crossApplySnd #-}
+crossApplySnd :: MonadAsync m => Int -> Int -> m ()
+crossApplySnd linearCount start = drain $
+    Stream.crossApplySnd
+        (sourceUnfoldrM nestedCount2 start)
+        (sourceUnfoldrM nestedCount2 start)
+
+    where
+
+    nestedCount2 = round (fromIntegral linearCount**(1/2::Double))
+
 o_1_space_applicative :: Int -> [Benchmark]
 o_1_space_applicative value =
     [ bgroup "Applicative"
@@ -468,6 +524,9 @@ o_1_space_applicative value =
         , benchIO "(<*)" $ apDiscardSnd value
         , benchIO "(<*>)" $ toNullAp value
         , benchIO "liftA2" $ apLiftA2 value
+        , benchIO "crossApply" $ crossApply value
+        , benchIO "crossApplyFst" $ crossApplyFst value
+        , benchIO "crossApplySnd" $ crossApplySnd value
         , benchIO "pureDrain2" $ toNullApPure value
         , benchIO "pureCross2" $ cross2 value
         ]
diff --git a/benchmark/Streamly/Benchmark/Data/Stream/Generate.hs b/benchmark/Streamly/Benchmark/Data/Stream/Generate.hs
index ff68396978..1ed0f96a6c 100644
--- a/benchmark/Streamly/Benchmark/Data/Stream/Generate.hs
+++ b/benchmark/Streamly/Benchmark/Data/Stream/Generate.hs
@@ -39,6 +39,16 @@ import Prelude hiding (repeat, replicate, iterate)
 sourceFromList :: Monad m => Int -> Int -> Stream m Int
 sourceFromList value n = Stream.fromList [n..n+value]
 
+-- | 'fromTuple' yields two elements per tuple. To emit and drain ~value
+-- elements we generate value/2 tuples and reduce each tuple's 'fromTuple'
+-- stream with a light 'sum' fold (avoiding a heavy, non-fusible 'concatMap'
+-- that would mask the cost of 'fromTuple').
+{-# INLINE sourceFromTuple #-}
+sourceFromTuple :: Monad m => Int -> Int -> Stream m Int
+sourceFromTuple value n =
+    Stream.mapM (Stream.fold Fold.sum . Stream.fromTuple)
+        $ Stream.fromList (fmap (\i -> (i, i)) [n .. n + value `div` 2])
+
 {-# INLINE sourceFromListM #-}
 sourceFromListM :: MonadAsync m => Int -> Int -> Stream m Int
 sourceFromListM value n = fromListM (fmap return [n..n+value])
@@ -168,6 +178,7 @@ o_1_space_generation value =
         , benchIOSrc "fracFromThenTo" (sourceFracFromThenTo value)
         , benchIOSrc "fracFromTo" (sourceFracFromTo value)
         , benchIOSrc "fromList" (sourceFromList value)
+        , benchIOSrc "fromTuple" (sourceFromTuple value)
         , benchIOSrc "fromListM" (sourceFromListM value)
         , benchPureSrc "IsList.fromList" (sourceIsList value)
         , benchPureSrc "IsString.fromString" (sourceIsString value)
diff --git a/benchmark/Streamly/Benchmark/Data/Unfold.hs b/benchmark/Streamly/Benchmark/Data/Unfold.hs
index dd2be32984..ba02f82ae1 100644
--- a/benchmark/Streamly/Benchmark/Data/Unfold.hs
+++ b/benchmark/Streamly/Benchmark/Data/Unfold.hs
@@ -111,23 +111,23 @@ drainProductDefault to = drainProduct src src
 -------------------------------------------------------------------------------
 
 {-# INLINE lmap #-}
-lmap :: Monad m => Int -> Int -> m ()
+lmap :: Int -> Int -> IO ()
 lmap size start =
     drainTransformationDefault (size + start) (UF.lmap (+ 1)) start
 
 {-# INLINE lmapM #-}
-lmapM :: Monad m => Int -> Int -> m ()
+lmapM :: Int -> Int -> IO ()
 lmapM size start =
     drainTransformationDefault (size + start) (UF.lmapM (return . (+) 1)) start
 
 {-# INLINE both #-}
-both :: Monad m => Int -> Int -> m ()
+both :: Int -> Int -> IO ()
 both size start =
     drainTransformationDefault (size + start) (UF.supply start) ()
 
 
 {-# INLINE first #-}
-first :: Monad m => Int -> Int -> m ()
+first :: Int -> Int -> IO ()
 first size start =
     drainTransformation
         (UF.take size UF.enumerateFromThenIntegral)
@@ -135,7 +135,7 @@ first size start =
         1
 
 {-# INLINE second #-}
-second :: Monad m => Int -> Int -> m ()
+second :: Int -> Int -> IO ()
 second size start =
     drainTransformation
         (UF.take size UF.enumerateFromThenIntegral)
@@ -143,17 +143,17 @@ second size start =
         start
 
 {-# INLINE discardFirst #-}
-discardFirst :: Monad m => Int -> Int -> m ()
+discardFirst :: Int -> Int -> IO ()
 discardFirst size start =
     drainTransformationDefault (size + start) UF.discardFirst (start, start)
 
 {-# INLINE discardSecond #-}
-discardSecond :: Monad m => Int -> Int -> m ()
+discardSecond :: Int -> Int -> IO ()
 discardSecond size start =
     drainTransformationDefault (size + start) UF.discardSecond (start, start)
 
 {-# INLINE swap #-}
-swap :: Monad m => Int -> Int -> m ()
+swap :: Int -> Int -> IO ()
 swap size start =
     drainTransformation
         (UF.take size UF.enumerateFromThenIntegral)
@@ -172,39 +172,60 @@ fromStream size start =
 -- XXX INVESTIGATE: Although the performance of this should be equivalant to
 -- fromStream, this is considerably worse. More than 4x worse.
 {-# INLINE fromStreamK #-}
-fromStreamK :: Monad m => Int -> Int -> m ()
+fromStreamK :: Int -> Int -> IO ()
 fromStreamK size start = drainGeneration UF.fromStreamK (K.replicate size start)
 
 {-# INLINE fromStreamD #-}
-fromStreamD :: Monad m => Int -> Int -> m ()
+fromStreamD :: Int -> Int -> IO ()
 fromStreamD size start =
     drainGeneration UF.fromStreamD (S.replicate size start)
 
-{-# INLINE _nilM #-}
-_nilM :: Monad m => Int -> Int -> m ()
-_nilM _ start = drainGeneration (UF.nilM return) start
+-- 'nilM' runs its action on the seed but yields no output, so unfold it over an
+-- outer source of value seeds to run it ~value times.
+{-# INLINE nilM #-}
+nilM :: Int -> Int -> IO ()
+nilM value start =
+    drainGeneration (UF.unfoldEach (UF.nilM return) (source (start + value))) start
 
 {-# INLINE consM #-}
-consM :: Monad m => Int -> Int -> m ()
+consM :: Int -> Int -> IO ()
 consM size start =
     drainTransformationDefault (size + start) (UF.consM return) start
 
-{-# INLINE _functionM #-}
-_functionM :: Monad m => Int -> Int -> m ()
-_functionM _ start = drainGeneration (UF.functionM return) start
+-- 'functionM', 'function', 'identity' and 'fromEffect' generate a single
+-- element per seed, so to process ~value elements we unfold them over an outer
+-- source of value seeds.
+{-# INLINE functionM #-}
+functionM :: Int -> Int -> IO ()
+functionM value start =
+    drainGeneration
+        (UF.unfoldEach (UF.functionM return) (source (start + value))) start
 
-{-# INLINE _function #-}
-_function :: Monad m => Int -> Int -> m ()
-_function _ start = drainGeneration (UF.function id) start
+{-# INLINE function #-}
+function :: Int -> Int -> IO ()
+function value start =
+    drainGeneration
+        (UF.unfoldEach (UF.function id) (source (start + value))) start
 
-{-# INLINE _identity #-}
-_identity :: Monad m => Int -> Int -> m ()
-_identity _ start = drainGeneration UF.identity start
+{-# INLINE identity #-}
+identity :: Int -> Int -> IO ()
+identity value start =
+    drainGeneration (UF.unfoldEach UF.identity (source (start + value))) start
 
-{-# INLINE _const #-}
-_const :: Monad m => Int -> Int -> m ()
-_const size start =
-    drainGeneration (UF.take size (UF.fromEffect (return start))) undefined
+{-# INLINE fromEffect #-}
+fromEffect :: Int -> Int -> IO ()
+fromEffect value start =
+    drainGeneration
+        (UF.unfoldEach (UF.fromEffect (return start)) (source (start + value)))
+        start
+
+-- 'fromTuple' generates two elements per seed, so unfold it over value/2 tuples
+-- to emit and drain ~value elements.
+{-# INLINE fromTuple #-}
+fromTuple :: Int -> Int -> IO ()
+fromTuple value start =
+    let outer = UF.map (\i -> (i, i)) (source (start + value `div` 2))
+     in drainGeneration (UF.unfoldEach UF.fromTuple outer) start
 
 {-# INLINE sourceUnfoldrM #-}
 sourceUnfoldrM :: Monad m => Int -> Int -> Unfold m Int Int
@@ -219,15 +240,15 @@ sourceUnfoldrM size start = UF.unfoldrM step
               else Nothing
 
 {-# INLINE unfoldrM #-}
-unfoldrM :: Monad m => Int -> Int -> m ()
+unfoldrM :: Int -> Int -> IO ()
 unfoldrM size start = drainGeneration (sourceUnfoldrM size start) start
 
 {-# INLINE fromList #-}
-fromList :: Monad m => Int -> Int -> m ()
+fromList :: Int -> Int -> IO ()
 fromList size start = drainGeneration UF.fromList [start .. start + size]
 
 {-# INLINE fromListM #-}
-fromListM :: Monad m => Int -> Int -> m ()
+fromListM :: Int -> Int -> IO ()
 fromListM size start =
     drainGeneration UF.fromListM (Prelude.map return [start .. start + size])
 
@@ -240,30 +261,30 @@ _fromProducer :: Int -> Int -> m ()
 _fromProducer = undefined
 
 {-# INLINE replicateM #-}
-replicateM :: Monad m => Int -> Int -> m ()
+replicateM :: Int -> Int -> IO ()
 replicateM size start = drainGeneration UF.replicateM (size, return start)
 
 {-# INLINE repeatM #-}
-repeatM :: Monad m => Int -> Int -> m ()
+repeatM :: Int -> Int -> IO ()
 repeatM size start = drainGeneration (UF.take size UF.repeatM) (return start)
 
 {-# INLINE iterateM #-}
-iterateM :: Monad m => Int -> Int -> m ()
+iterateM :: Int -> Int -> IO ()
 iterateM size start =
     drainGeneration (UF.take size (UF.iterateM return)) (return start)
 
 {-# INLINE fromIndicesM #-}
-fromIndicesM :: Monad m => Int -> Int -> m ()
+fromIndicesM :: Int -> Int -> IO ()
 fromIndicesM size start =
     drainGeneration (UF.take size (UF.fromIndicesM return)) start
 
 {-# INLINE enumerateFromThenIntegral #-}
-enumerateFromThenIntegral :: Monad m => Int -> Int -> m ()
+enumerateFromThenIntegral :: Int -> Int -> IO ()
 enumerateFromThenIntegral size start =
     drainGeneration (UF.take size UF.enumerateFromThenIntegral) (start, 1)
 
 {-# INLINE enumerateFromToIntegral #-}
-enumerateFromToIntegral :: Monad m => Int -> Int -> m ()
+enumerateFromToIntegral :: Int -> Int -> IO ()
 enumerateFromToIntegral size start =
     drainGeneration
     ( UF.supplySecond
@@ -272,21 +293,21 @@ enumerateFromToIntegral size start =
     ) start
 
 {-# INLINE enumerateFromIntegral #-}
-enumerateFromIntegral :: Monad m => Int -> Int -> m ()
+enumerateFromIntegral :: Int -> Int -> IO ()
 enumerateFromIntegral size start =
     drainGeneration (UF.take size UF.enumerateFromIntegral) start
 
 {-# INLINE enumerateFromStepNum #-}
-enumerateFromStepNum :: Monad m => Int -> Int -> m ()
+enumerateFromStepNum :: Int -> Int -> IO ()
 enumerateFromStepNum size start =
     drainGeneration (UF.take size (UF.enumerateFromThenNum)) (start, 1)
 
 {-# INLINE enumerateFromNum #-}
-enumerateFromNum :: Monad m => Int -> Int -> m ()
+enumerateFromNum :: Int -> Int -> IO ()
 enumerateFromNum size start = drainGeneration (UF.take size UF.enumerateFromNum) start
 
 {-# INLINE enumerateFromToFractional #-}
-enumerateFromToFractional :: Monad m => Int -> Int -> m ()
+enumerateFromToFractional :: Int -> Int -> IO ()
 enumerateFromToFractional size start =
     let intToDouble x = (fromInteger (fromIntegral x)) :: Double
      in drainGeneration
@@ -301,21 +322,21 @@ enumerateFromToFractional size start =
 -------------------------------------------------------------------------------
 
 {-# INLINE postscan #-}
-postscan :: Monad m => Int -> Int -> m ()
+postscan :: Int -> Int -> IO ()
 postscan size start =
     drainTransformationDefault (size + start) (UF.postscanl Scanl.sum) start
 
 {-# INLINE map #-}
-map :: Monad m => Int -> Int -> m ()
+map :: Int -> Int -> IO ()
 map size start = drainTransformationDefault (size + start) (UF.map (+1)) start
 
 {-# INLINE mapM #-}
-mapM :: Monad m => Int -> Int -> m ()
+mapM :: Int -> Int -> IO ()
 mapM size start =
     drainTransformationDefault (size + start) (UF.mapM (return . (+) 1)) start
 
 {-# INLINE mapM2 #-}
-mapM2 :: Monad m => Int -> Int -> m ()
+mapM2 :: Int -> Int -> IO ()
 mapM2 size start =
     drainTransformationDefault
         size
@@ -327,7 +348,7 @@ mapM2 size start =
 -------------------------------------------------------------------------------
 
 {-# INLINE takeWhileM #-}
-takeWhileM :: Monad m => Int -> Int -> m ()
+takeWhileM :: Int -> Int -> IO ()
 takeWhileM size start =
     drainTransformationDefault
         size
@@ -335,7 +356,7 @@ takeWhileM size start =
         start
 
 {-# INLINE takeWhile #-}
-takeWhile :: Monad m => Int -> Int -> m ()
+takeWhile :: Int -> Int -> IO ()
 takeWhile size start =
     drainTransformationDefault
         size
@@ -343,34 +364,38 @@ takeWhile size start =
         start
 
 {-# INLINE take #-}
-take :: Monad m => Int -> Int -> m ()
+take :: Int -> Int -> IO ()
 take size start = drainTransformationDefault (size + start) (UF.take size) start
 
 {-# INLINE filter #-}
-filter :: Monad m => Int -> Int -> m ()
+filter :: Int -> Int -> IO ()
 filter size start =
     drainTransformationDefault (size + start) (UF.filter (\_ -> True)) start
 
 {-# INLINE filterM #-}
-filterM :: Monad m => Int -> Int -> m ()
+filterM :: Int -> Int -> IO ()
 filterM size start =
     drainTransformationDefault
         (size + start)
         (UF.filterM (\_ -> (return True)))
         start
 
-{-# INLINE _dropOne #-}
-_dropOne :: Monad m => Int -> Int -> m ()
-_dropOne size start =
-    drainTransformationDefault (size + start) (UF.drop 1) start
+-- Dropping one element from a large stream is dominated by generation, so
+-- instead exercise 'drop' ~value/2 times: generate value/2 two-element streams
+-- with 'fromTuple', 'drop' the first element of each, and flatten the rest.
+{-# INLINE dropOne #-}
+dropOne :: Int -> Int -> IO ()
+dropOne value start =
+    let outer = UF.map (\i -> (i, i)) (source (start + value `div` 2))
+     in drainGeneration (UF.unfoldEach (UF.drop 1 UF.fromTuple) outer) start
 
 {-# INLINE dropAll #-}
-dropAll :: Monad m => Int -> Int -> m ()
+dropAll :: Int -> Int -> IO ()
 dropAll size start =
     drainTransformationDefault (size + start) (UF.drop (size + 1)) start
 
 {-# INLINE dropWhileTrue #-}
-dropWhileTrue :: Monad m => Int -> Int -> m ()
+dropWhileTrue :: Int -> Int -> IO ()
 dropWhileTrue size start =
     drainTransformationDefault
         (size + start)
@@ -378,7 +403,7 @@ dropWhileTrue size start =
         start
 
 {-# INLINE dropWhileFalse #-}
-dropWhileFalse :: Monad m => Int -> Int -> m ()
+dropWhileFalse :: Int -> Int -> IO ()
 dropWhileFalse size start =
     drainTransformationDefault
         (size + start)
@@ -386,7 +411,7 @@ dropWhileFalse size start =
         start
 
 {-# INLINE dropWhileMTrue #-}
-dropWhileMTrue :: Monad m => Int -> Int -> m ()
+dropWhileMTrue :: Int -> Int -> IO ()
 dropWhileMTrue size start =
     drainTransformationDefault
         size
@@ -394,7 +419,7 @@ dropWhileMTrue size start =
         start
 
 {-# INLINE dropWhileMFalse #-}
-dropWhileMFalse :: Monad m => Int -> Int -> m ()
+dropWhileMFalse :: Int -> Int -> IO ()
 dropWhileMFalse size start =
     drainTransformationDefault
         size
@@ -406,12 +431,12 @@ dropWhileMFalse size start =
 -------------------------------------------------------------------------------
 
 {-# INLINE zipWith #-}
-zipWith :: Monad m => Int -> Int -> m ()
+zipWith :: Int -> Int -> IO ()
 zipWith size start =
     drainProductDefault (size + start) (UF.zipWith (+)) start
 
 {-# INLINE zipWithM #-}
-zipWithM :: Monad m => Int -> Int -> m ()
+zipWithM :: Int -> Int -> IO ()
 zipWithM size start =
     drainProductDefault
         (size + start)
@@ -419,10 +444,15 @@ zipWithM size start =
         start
 
 {-# INLINE teeZipWith #-}
-teeZipWith :: Monad m => Int -> Int -> m ()
+teeZipWith :: Int -> Int -> IO ()
 teeZipWith size start =
     drainProductDefault (size + start) (UF.zipWith (+)) start
 
+{-# INLINE interleave #-}
+interleave :: Int -> Int -> IO ()
+interleave size start =
+    drainProductDefault (size + start) UF.interleave (start, start)
+
 -------------------------------------------------------------------------------
 -- Applicative
 -------------------------------------------------------------------------------
@@ -431,20 +461,40 @@ nthRoot :: Double -> Int -> Int
 nthRoot n value = round (fromIntegral value**(1/n))
 
 {-# INLINE toNullAp #-}
-toNullAp :: Monad m => Int -> Int -> m ()
+toNullAp :: Int -> Int -> IO ()
 toNullAp value start =
     let end = start + nthRoot 2 value
         s = source end
     -- in UF.fold ((+) <$> s <*> s) FL.drain start
     in UF.fold FL.drain (((+) `fmap` s) `UF.crossApply` s) start
 
-{-# INLINE _apDiscardFst #-}
-_apDiscardFst :: Int -> Int -> m ()
-_apDiscardFst = undefined
+{-# INLINE crossApplyFst #-}
+crossApplyFst :: Int -> Int -> IO ()
+crossApplyFst value start =
+    let end = start + nthRoot 2 value
+        s = source end
+    in UF.fold FL.drain (s `UF.crossApplyFst` s) start
+
+{-# INLINE crossApplySnd #-}
+crossApplySnd :: Int -> Int -> IO ()
+crossApplySnd value start =
+    let end = start + nthRoot 2 value
+        s = source end
+    in UF.fold FL.drain (s `UF.crossApplySnd` s) start
+
+{-# INLINE cross #-}
+cross :: Int -> Int -> IO ()
+cross value start =
+    let end = start + nthRoot 2 value
+        s = source end
+    in UF.fold FL.drain (s `UF.cross` s) start
 
-{-# INLINE _apDiscardSnd #-}
-_apDiscardSnd :: Int -> Int -> m ()
-_apDiscardSnd = undefined
+{-# INLINE fairCross #-}
+fairCross :: Int -> Int -> IO ()
+fairCross value start =
+    let end = start + nthRoot 2 value
+        s = source end
+    in UF.fold FL.drain (s `UF.fairCross` s) start
 
 -------------------------------------------------------------------------------
 -- Monad
@@ -453,7 +503,7 @@ _apDiscardSnd = undefined
 -- XXX to keep the benchmarks same as Stream we should use sourceUnfoldrM in
 -- all of these, and other benchmarks too.
 {-# INLINE concatMapM #-}
-concatMapM :: Monad m => Int -> Int -> Int -> m ()
+concatMapM :: Int -> Int -> Int -> IO ()
 concatMapM inner outer start =
     drainGeneration (UF.concatMapM unfoldInGen unfoldOut) start
 
@@ -463,7 +513,7 @@ concatMapM inner outer start =
     unfoldOut = UF.supplySecond (start + outer) UF.enumerateFromToIntegral
 
 {-# INLINE toNull #-}
-toNull :: Monad m => Int -> Int -> m ()
+toNull :: Int -> Int -> IO ()
 toNull value start =
     let end = start + nthRoot 2 value
         src = source end
@@ -480,7 +530,7 @@ toNull value start =
 
 
 {-# INLINE toNull3 #-}
-toNull3 :: Monad m => Int -> Int -> m ()
+toNull3 :: Int -> Int -> IO ()
 toNull3 value start =
     let end = start + nthRoot 3 value
         src = source end
@@ -498,7 +548,7 @@ toNull3 value start =
      in UF.fold FL.drain u start
 
 {-# INLINE toList #-}
-toList :: Monad m => Int -> Int -> m [Int]
+toList :: Int -> Int -> IO [Int]
 toList value start = do
     let end = start + nthRoot 2 value
         src = source end
@@ -514,7 +564,7 @@ toList value start = do
      in UF.fold FL.toList u start
 
 {-# INLINE toListSome #-}
-toListSome :: Monad m => Int -> Int -> m [Int]
+toListSome :: Int -> Int -> IO [Int]
 toListSome value start = do
     let end = start + nthRoot 2 value
         src = source end
@@ -530,7 +580,7 @@ toListSome value start = do
      in UF.fold FL.toList (UF.take 1000 u) start
 
 {-# INLINE filterAllOut #-}
-filterAllOut :: Monad m => Int -> Int -> m ()
+filterAllOut :: Int -> Int -> IO ()
 filterAllOut value start = do
     let end = start + nthRoot 2 value
         src = source end
@@ -548,7 +598,7 @@ filterAllOut value start = do
      in UF.fold FL.drain u start
 
 {-# INLINE filterAllIn #-}
-filterAllIn :: Monad m => Int -> Int -> m ()
+filterAllIn :: Int -> Int -> IO ()
 filterAllIn value start = do
     let end = start + nthRoot 2 value
         src = source end
@@ -566,7 +616,7 @@ filterAllIn value start = do
      in UF.fold FL.drain u start
 
 {-# INLINE filterSome #-}
-filterSome :: Monad m => Int -> Int -> m ()
+filterSome :: Int -> Int -> IO ()
 filterSome value start = do
     let end = start + nthRoot 2 value
         src = source end
@@ -608,13 +658,114 @@ breakAfterSome value start =
 -------------------------------------------------------------------------------
 
 {-# INLINE unfoldEach #-}
-unfoldEach :: Monad m => Int -> Int -> Int -> m ()
+unfoldEach :: Int -> Int -> Int -> IO ()
 unfoldEach inner outer start = do
     UF.fold
         FL.drain
         (UF.unfoldEach (sourceUnfoldrM inner start) (sourceUnfoldrM outer start))
         start
 
+{-# INLINE unfoldEachInterleave #-}
+unfoldEachInterleave :: Int -> Int -> Int -> IO ()
+unfoldEachInterleave inner outer start = do
+    UF.fold
+        FL.drain
+        (UF.unfoldEachInterleave
+            (sourceUnfoldrM inner start) (sourceUnfoldrM outer start))
+        start
+
+-------------------------------------------------------------------------------
+-- Inspection
+-------------------------------------------------------------------------------
+
+#ifdef INSPECTION
+-- All benchmarks must fully fuse: no stream constructors (the 'Yield', 'Skip'
+-- and 'Stop' of the 'Step' type) should remain in the optimized core.
+
+-- input
+inspect $ 'lmap `hasNoType` ''S.Step
+inspect $ 'lmapM `hasNoType` ''S.Step
+inspect $ 'both `hasNoType` ''S.Step
+inspect $ 'first `hasNoType` ''S.Step
+inspect $ 'second `hasNoType` ''S.Step
+inspect $ 'discardFirst `hasNoType` ''S.Step
+inspect $ 'discardSecond `hasNoType` ''S.Step
+inspect $ 'swap `hasNoType` ''S.Step
+
+-- generation
+-- 'fromStream', 'fromStreamD' and 'consM' wrap an opaque stream/cons cell, so
+-- the 'Step' is not eliminated.
+-- inspect $ 'fromStream `hasNoType` ''S.Step
+-- inspect $ 'fromStreamD `hasNoType` ''S.Step
+-- inspect $ 'consM `hasNoType` ''S.Step
+inspect $ 'fromStreamK `hasNoType` ''S.Step
+inspect $ 'nilM `hasNoType` ''S.Step
+inspect $ 'functionM `hasNoType` ''S.Step
+inspect $ 'function `hasNoType` ''S.Step
+inspect $ 'identity `hasNoType` ''S.Step
+inspect $ 'fromEffect `hasNoType` ''S.Step
+inspect $ 'fromTuple `hasNoType` ''S.Step
+inspect $ 'unfoldrM `hasNoType` ''S.Step
+inspect $ 'fromList `hasNoType` ''S.Step
+inspect $ 'fromListM `hasNoType` ''S.Step
+inspect $ 'replicateM `hasNoType` ''S.Step
+inspect $ 'repeatM `hasNoType` ''S.Step
+inspect $ 'iterateM `hasNoType` ''S.Step
+inspect $ 'fromIndicesM `hasNoType` ''S.Step
+inspect $ 'enumerateFromThenIntegral `hasNoType` ''S.Step
+inspect $ 'enumerateFromToIntegral `hasNoType` ''S.Step
+inspect $ 'enumerateFromIntegral `hasNoType` ''S.Step
+inspect $ 'enumerateFromStepNum `hasNoType` ''S.Step
+inspect $ 'enumerateFromNum `hasNoType` ''S.Step
+inspect $ 'enumerateFromToFractional `hasNoType` ''S.Step
+
+-- transformation
+inspect $ 'map `hasNoType` ''S.Step
+inspect $ 'mapM `hasNoType` ''S.Step
+inspect $ 'mapM2 `hasNoType` ''S.Step
+inspect $ 'postscan `hasNoType` ''S.Step
+
+-- filtering
+inspect $ 'takeWhileM `hasNoType` ''S.Step
+inspect $ 'takeWhile `hasNoType` ''S.Step
+inspect $ 'take `hasNoType` ''S.Step
+inspect $ 'filter `hasNoType` ''S.Step
+inspect $ 'filterM `hasNoType` ''S.Step
+inspect $ 'dropOne `hasNoType` ''S.Step
+inspect $ 'dropAll `hasNoType` ''S.Step
+inspect $ 'dropWhileTrue `hasNoType` ''S.Step
+inspect $ 'dropWhileFalse `hasNoType` ''S.Step
+inspect $ 'dropWhileMTrue `hasNoType` ''S.Step
+inspect $ 'dropWhileMFalse `hasNoType` ''S.Step
+
+-- zip
+inspect $ 'zipWith `hasNoType` ''S.Step
+inspect $ 'zipWithM `hasNoType` ''S.Step
+inspect $ 'teeZipWith `hasNoType` ''S.Step
+inspect $ 'interleave `hasNoType` ''S.Step
+
+-- nested
+inspect $ 'toNullAp `hasNoType` ''S.Step
+inspect $ 'crossApplyFst `hasNoType` ''S.Step
+inspect $ 'crossApplySnd `hasNoType` ''S.Step
+inspect $ 'cross `hasNoType` ''S.Step
+inspect $ 'fairCross `hasNoType` ''S.Step
+inspect $ 'unfoldEach `hasNoType` ''S.Step
+-- The 'bind'-based benchmarks use the Unfold monad ('UF.bind'), which is a
+-- concatMap and does not fuse, so the 'Step' constructors remain. The same is
+-- true for 'concatMapM' and 'unfoldEachInterleave'.
+-- inspect $ 'concatMapM `hasNoType` ''S.Step
+-- inspect $ 'toNull `hasNoType` ''S.Step
+-- inspect $ 'toNull3 `hasNoType` ''S.Step
+-- inspect $ 'toList `hasNoType` ''S.Step
+-- inspect $ 'toListSome `hasNoType` ''S.Step
+-- inspect $ 'filterAllOut `hasNoType` ''S.Step
+-- inspect $ 'filterAllIn `hasNoType` ''S.Step
+-- inspect $ 'filterSome `hasNoType` ''S.Step
+-- inspect $ 'breakAfterSome `hasNoType` ''S.Step
+-- inspect $ 'unfoldEachInterleave `hasNoType` ''S.Step
+#endif
+
 -------------------------------------------------------------------------------
 -- Benchmarks
 -------------------------------------------------------------------------------
@@ -644,19 +795,16 @@ o_1_space_generation size =
           [ benchIO "fromStream" $ fromStream size
           , benchIO "fromStreamK" $ fromStreamK size
           , benchIO "fromStreamD" $ fromStreamD size
-          -- Very small benchmarks, reporting in ns
-          -- , benchIO "nilM" $ nilM size
+          , benchIO "nilM" $ nilM size
           , benchIO "consM" $ consM size
-          -- , benchIO "functionM" $ functionM size
-          -- , benchIO "function" $ function size
-          -- , benchIO "identity" $ identity size
-          -- , benchIO "const" $ const size
+          , benchIO "functionM" $ functionM size
+          , benchIO "function" $ function size
+          , benchIO "identity" $ identity size
+          , benchIO "fromEffect" $ fromEffect size
+          , benchIO "fromTuple" $ fromTuple size
           , benchIO "unfoldrM" $ unfoldrM size
           , benchIO "fromList" $ fromList size
           , benchIO "fromListM" $ fromListM size
-          -- Unimplemented
-          -- , benchIO "fromSVar" $ fromSVar size
-          -- , benchIO "fromProducer" $ fromProducer size
           , benchIO "replicateM" $ replicateM size
           , benchIO "repeatM" $ repeatM size
           , benchIO "iterateM" $ iterateM size
@@ -690,8 +838,7 @@ o_1_space_filtering size =
           , benchIO "take" $ take size
           , benchIO "filter" $ filter size
           , benchIO "filterM" $ filterM size
-          -- Very small benchmark, reporting in ns
-          -- , benchIO "dropOne" $ dropOne size
+          , benchIO "dropOne" $ dropOne size
           , benchIO "dropAll" $ dropAll size
           , benchIO "dropWhileTrue" $ dropWhileTrue size
           , benchIO "dropWhileFalse" $ dropWhileFalse size
@@ -707,6 +854,7 @@ o_1_space_zip size =
           [ benchIO "zipWithM" $ zipWithM size
           , benchIO "zipWith" $ zipWith size
           , benchIO "teeZipWith" $ teeZipWith size
+          , benchIO "interleave" $ interleave size
           ]
     ]
 
@@ -724,9 +872,10 @@ o_1_space_nested env size =
     [ bgroup
           "nested"
           [ benchIO "crossApply outer=inner=(sqrt Max)" $ toNullAp size
-          -- Unimplemented
-          -- , benchIO "apDiscardFst" $ apDiscardFst size
-          -- , benchIO "apDiscardSnd" $ apDiscardSnd size
+          , benchIO "crossApplyFst outer=inner=(sqrt Max)" $ crossApplyFst size
+          , benchIO "crossApplySnd outer=inner=(sqrt Max)" $ crossApplySnd size
+          , benchIO "cross outer=inner=(sqrt Max)" $ cross size
+          , benchIO "fairCross outer=inner=(sqrt Max)" $ fairCross size
 
           , benchIO "concatMapM outer=inner=(sqrt Max)" $ concatMapM sqrtVal sqrtVal
           , benchIO "bind2" $ toNull size
@@ -739,6 +888,12 @@ o_1_space_nested env size =
           , benchIO "unfoldEach inner=outer=(sqrt Max)" $ unfoldEach sqrtVal sqrtVal
           , benchIO "unfoldEach inner=1 outer=Max" $ unfoldEach 1 size
           , benchIO "unfoldEach inner=Max outer=1" $ unfoldEach size 1
+          , benchIO "unfoldEachInterleave inner=outer=(sqrt Max)"
+            $ unfoldEachInterleave sqrtVal sqrtVal
+          , benchIO "unfoldEachInterleave inner=1 outer=Max"
+            $ unfoldEachInterleave 1 size
+          , benchIO "unfoldEachInterleave inner=Max outer=1"
+            $ unfoldEachInterleave size 1
           , mkBench "foldMany (Fold.takeEndBy_ (== lf) Fold.drain)" env
             $ \inh _ -> foldManySepBy inh
           ]
diff --git a/benchmark/streamly-benchmarks.cabal b/benchmark/streamly-benchmarks.cabal
index 794591a942..6fb72e4c8f 100644
--- a/benchmark/streamly-benchmarks.cabal
+++ b/benchmark/streamly-benchmarks.cabal
@@ -448,7 +448,7 @@ benchmark Data.Stream
     if flag(dev)
       ghc-options: +RTS -M1000M -RTS
     else
-      ghc-options: +RTS -M500M -RTS
+      ghc-options: +RTS -M800M -RTS
 
 benchmark Data.Stream.Adaptive
   import: bench-options-threaded
@@ -596,6 +596,8 @@ benchmark Data.Unfold
     buildable: False
   else
     buildable: True
+  if flag(limit-build-mem)
+    ghc-options: +RTS -M1000M -RTS
 
 benchmark Data.Unfold.Prelude
   import: bench-options
diff --git a/core/streamly-core.cabal b/core/streamly-core.cabal
index e0b0c3a84b..7616b79ec4 100644
--- a/core/streamly-core.cabal
+++ b/core/streamly-core.cabal
@@ -364,6 +364,8 @@ library
 
                      -- streamly-core-streams
                      , Streamly.Internal.Data.StreamK
+                     -- Shared step-function combinators used by Stream and Unfold
+                     , Streamly.Internal.Data.Producer
                      -- StreamD depends on streamly-array-types
                      , Streamly.Internal.Data.Stream
 
@@ -519,8 +521,6 @@ library
                     , Streamly.Internal.Data.Stream.Transformer
                     , Streamly.Internal.Data.Stream.Type
 
-                    , Streamly.Internal.Data.Producer
-
                     , Streamly.Internal.Data.StreamK.Type
                     , Streamly.Internal.Data.StreamK.Transformer
 
diff --git a/test/Streamly/Test/Data/Unfold.hs b/test/Streamly/Test/Data/Unfold.hs
index 969b7c7020..3d48f9dd98 100644
--- a/test/Streamly/Test/Data/Unfold.hs
+++ b/test/Streamly/Test/Data/Unfold.hs
@@ -63,6 +63,19 @@ testUnfold unf seed lst = runIdentity action
 testUnfoldD :: Unfold Identity a Int -> a -> [Int] -> Bool
 testUnfoldD = testUnfold
 
+-- | Like 'testUnfold' but compares the outputs as multisets (order
+-- independent). Useful for combinators like 'UF.fairCross' and
+-- 'UF.unfoldEachInterleave' that produce the same elements as their
+-- non-interleaving counterparts but in a different order.
+testUnfoldSorted :: Ord b => Unfold Identity a b -> a -> [b] -> Bool
+testUnfoldSorted unf seed lst = runIdentity action
+
+    where
+
+    action = do
+        x <- S.fold Fold.toList $ S.unfold unf seed
+        return $ List.sort x == List.sort lst
+
 -------------------------------------------------------------------------------
 -- Operations on input
 -------------------------------------------------------------------------------
@@ -577,6 +590,32 @@ outerProduct =
 
     crossProduct u1 u2 = UF.cross (UF.lmap fst u1) (UF.lmap snd u2)
 
+crossApply :: Bool
+crossApply =
+    let unf1 = UF.enumerateFromToIntegral
+        unf2 = UF.enumerateFromToIntegral
+        unf = UF.crossApply
+                  (UF.map (+) (UF.lmap fst unf1))
+                  (UF.lmap snd unf2)
+        lst = [a + b :: Int | a <- [0 .. 10], b <- [0 .. 20]]
+     in testUnfold unf (((0, 10), (0, 20)) :: ((Int, Int), (Int, Int))) lst
+
+crossApplyFst :: Bool
+crossApplyFst =
+    let unf1 = UF.enumerateFromToIntegral
+        unf2 = UF.enumerateFromToIntegral
+        unf = UF.crossApplyFst (UF.lmap fst unf1) (UF.lmap snd unf2)
+        lst = [a :: Int | a <- [0 .. 10], _ <- [0 .. 20 :: Int]]
+     in testUnfold unf (((0, 10), (0, 20)) :: ((Int, Int), (Int, Int))) lst
+
+crossApplySnd :: Bool
+crossApplySnd =
+    let unf1 = UF.enumerateFromToIntegral
+        unf2 = UF.enumerateFromToIntegral
+        unf = UF.crossApplySnd (UF.lmap fst unf1) (UF.lmap snd unf2)
+        lst = [b :: Int | _ <- [0 .. 10 :: Int], b <- [0 .. 20]]
+     in testUnfold unf (((0, 10), (0, 20)) :: ((Int, Int), (Int, Int))) lst
+
 concatMapM :: Bool
 concatMapM =
     let inner b =
@@ -586,6 +625,34 @@ concatMapM =
         list = List.concatMap (replicate 10) [1 .. 10]
      in testUnfoldMD unf (1, 10) 0 110 list
 
+fromTuple :: Bool
+fromTuple = testUnfold UF.fromTuple ((1, 2) :: (Int, Int)) [1, 2]
+
+interleave :: Bool
+interleave =
+    let unf = UF.interleave UF.fromList UF.fromList
+     in testUnfold unf ([1, 3, 5], [2, 4, 6]) ([1 .. 6] :: [Int])
+
+-- | 'fairCross' yields the same pairs as 'cross' (outerProduct) but in a
+-- breadth-first order, so compare as multisets.
+fairCross :: Bool
+fairCross =
+    let unf1 = UF.enumerateFromToIntegral
+        unf2 = UF.enumerateFromToIntegral
+        unf = UF.fairCross (UF.lmap fst unf1) (UF.lmap snd unf2)
+        lst = [(a, b) :: (Int, Int) | a <- [0 .. 10], b <- [0 .. 20]]
+     in testUnfoldSorted unf (((0, 10), (0, 20)) :: ((Int, Int), (Int, Int))) lst
+
+-- | 'unfoldEachInterleave' yields the same elements as 'unfoldEach' (concat)
+-- but interleaved breadth-first, so compare as multisets.
+unfoldEachInterleave :: Bool
+unfoldEachInterleave =
+    let unfIn = UF.replicateM
+        unfOut = UF.map ((10,) . return) UF.enumerateFromToIntegral
+        unf = UF.unfoldEachInterleave unfIn unfOut
+        lst = Prelude.concat $ Prelude.map (Prelude.replicate 10) [1 .. 10]
+     in testUnfoldSorted unf (1, 10) (lst :: [Int])
+
 -------------------------------------------------------------------------------
 -- Test groups
 -------------------------------------------------------------------------------
@@ -621,6 +688,7 @@ testGeneration =
             prop "unfoldrM" unfoldrM
             -- prop "fromList" fromList
             prop "fromListM" fromListM
+            prop "fromTuple" fromTuple
             -- prop "fromSVar" fromSVar
             -- prop "fromProducer" fromProducer
             prop "replicateM" replicateM
@@ -693,9 +761,12 @@ testCombination =
             prop "concat" concat
             prop "concatMapM" concatMapM
             prop "outerProduct" outerProduct
-            -- prop "ap" ap
-            -- prop "apDiscardFst" apDiscardFst
-            -- prop "apDiscardSnd" apDiscardSnd
+            prop "crossApply" crossApply
+            prop "crossApplyFst" crossApplyFst
+            prop "crossApplySnd" crossApplySnd
+            prop "fairCross" fairCross
+            prop "interleave" interleave
+            prop "unfoldEachInterleave" unfoldEachInterleave
 
 -------------------------------------------------------------------------------
 -- Main