Use monadic seed in monadic scanning combinators.

- The following functions are updated: scanlM, scanlM',
  postscanlM, postscanlM', prescanlM'.
- Update Changelog.md.

Changelog.md

@@ -11,6 +11,8 @@
 * Deprecate `Streamly.Data.Unicode.Stream` in favor of `Streamly.Unicode.Stream`
 * Change the signature of `foldlM'` to make the initial value of the
   accumulator monadic.
+* Change the signature of `scanlM'`, `postscanlM'` to make the initial value of
+  the accumulator monadic.
 * Change the signature of `concatMapWith` to ensure that it can be
   used with a wide variety of combining functions.
 * Exception handling functions `bracket`, `handle`, `finally` now

benchmark/Streamly/Benchmark/Prelude/Serial.hs

@@ -861,7 +861,7 @@ scan n = composeN n $ S.scanl' (+) 0
 
 {-# INLINE scanlM' #-}
 scanlM' :: MonadIO m => Int -> SerialT m Int -> m ()
-scanlM' n = composeN n $ S.scanlM' (\b a -> return $ b + a) 0
+scanlM' n = composeN n $ S.scanlM' (\b a -> return $ b + a) (return 0)
 
 {-# INLINE scanl1' #-}
 scanl1' :: MonadIO m => Int -> SerialT m Int -> m ()
@@ -873,7 +873,7 @@ postscanl' n = composeN n $ S.postscanl' (+) 0
 
 {-# INLINE postscanlM' #-}
 postscanlM' :: MonadIO m => Int -> SerialT m Int -> m ()
-postscanlM' n = composeN n $ S.postscanlM' (\b a -> return $ b + a) 0
+postscanlM' n = composeN n $ S.postscanlM' (\b a -> return $ b + a) (return 0)
 
 {-# INLINE sequence #-}
 sequence ::
@@ -983,7 +983,7 @@ sieveScan =
                 let ps = takeWhile (\p -> p * p <= n) primes
                  in if P.all (\p -> n `mod` p /= 0) ps
                     then (primes ++ [n], Just n)
-                    else (primes, Nothing)) ([2], Just 2)
+                    else (primes, Nothing)) (return ([2], Just 2))
 
 o_n_space_mapping :: Int -> [Benchmark]
 o_n_space_mapping value =

src/Streamly/Internal/Data/Stream/IsStream.hs

@@ -2201,11 +2201,13 @@ scanx = P.scanlx'
 -- complexity when iterated over a stream. We should use StreamK style scanlM'
 -- for linear performance on iteration.
 --
--- | Like 'scanl'' but with a monadic fold function.
+-- | Like 'scanl'' but with a monadic step function and a monadic seed.
 --
--- @since 0.4.0
+-- /Since: 0.4.0/
+--
+-- /Since: 0.8.0 (signature change)/
 {-# INLINE scanlM' #-}
-scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
+scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
 scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
 
 -- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
@@ -2290,11 +2292,13 @@ postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
 postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
 
 -- XXX this needs to be concurrent
--- | Like 'postscanl'' but with a monadic step function.
+-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
 --
--- @since 0.7.0
+-- /Since: 0.7.0/
+--
+-- /Since: 0.8.0 (signature change)/
 {-# INLINE postscanlM' #-}
-postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
+postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
 postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
 
 -- XXX prescanl does not sound very useful, enable only if there is a
@@ -2308,7 +2312,7 @@ prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
 prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
 
 -- XXX this needs to be concurrent
--- | Like postscanl' but with a monadic step function.
+-- | Like prescanl' but with a monadic step function and a monadic seed.
 --
 -- /Internal/
 {-# INLINE prescanlM' #-}
@@ -2605,7 +2609,7 @@ smapM step initial stream =
     --              stream
     let r = concatMap
                 (\s0 ->
-                    postscanlM' (\(s, _) a -> step s a) (s0, undefined) stream
+                    postscanlM' (\(s, _) a -> step s a) (return (s0, undefined)) stream
                 )
                 (yieldM initial)
      in Serial.map snd r

src/Streamly/Internal/Data/Stream/StreamD.hs

@@ -3616,23 +3616,27 @@ scanlx' fstep begin done s =
 ------------------------------------------------------------------------------
 
 {-# INLINE_NORMAL postscanlM' #-}
-postscanlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
+postscanlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
 postscanlM' fstep begin (Stream step state) =
-    begin `seq` Stream step' (state, begin)
+    Stream step' Nothing
   where
     {-# INLINE_LATE step' #-}
-    step' gst (st, acc) = acc `seq` do
+    step' _ Nothing = do
+        !x <- begin
+        return $ Skip (Just (state, x))
+
+    step' gst (Just (st, acc)) =  do
         r <- step (adaptState gst) st
         case r of
             Yield x s -> do
-                y <- fstep acc x
-                y `seq` return (Yield y (s, y))
-            Skip s -> return $ Skip (s, acc)
+                !y <- fstep acc x
+                return $ Yield y (Just (s, y))
+            Skip s -> return $ Skip (Just (s, acc))
             Stop   -> return Stop
 
 {-# INLINE_NORMAL postscanl' #-}
 postscanl' :: Monad m => (a -> b -> a) -> a -> Stream m b -> Stream m a
-postscanl' f = postscanlM' (\a b -> return (f a b))
+postscanl' f seed = postscanlM' (\a b -> return (f a b)) (return seed)
 
 -- We can possibly have the "done" function as a Maybe to provide an option to
 -- emit or not emit the accumulator when the stream stops.
@@ -3660,26 +3664,43 @@ postscanlMAfter' fstep initial done (Stream step1 state1) = do
     step _ Nothing = return Stop
 
 {-# INLINE_NORMAL postscanlM #-}
-postscanlM :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
-postscanlM fstep begin (Stream step state) = Stream step' (state, begin)
+postscanlM :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
+postscanlM fstep begin (Stream step state) = Stream step' Nothing
   where
     {-# INLINE_LATE step' #-}
-    step' gst (st, acc) = do
+    step' _ Nothing = do
+        r <- begin
+        return $ Skip (Just (state, r))
+
+    step' gst (Just (st, acc)) = do
         r <- step (adaptState gst) st
         case r of
             Yield x s -> do
                 y <- fstep acc x
-                return (Yield y (s, y))
-            Skip s -> return $ Skip (s, acc)
+                return (Yield y (Just (s, y)))
+            Skip s -> return $ Skip (Just (s, acc))
             Stop   -> return Stop
 
 {-# INLINE_NORMAL postscanl #-}
 postscanl :: Monad m => (a -> b -> a) -> a -> Stream m b -> Stream m a
-postscanl f = postscanlM (\a b -> return (f a b))
+postscanl f seed = postscanlM (\a b -> return (f a b)) (return seed)
 
 {-# INLINE_NORMAL scanlM' #-}
-scanlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
-scanlM' fstep begin s = begin `seq` (begin `cons` postscanlM' fstep begin s)
+scanlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
+scanlM' fstep begin (Stream step state) = Stream step' Nothing
+  where
+    {-# INLINE_LATE step' #-}
+    step' _ Nothing = do
+        !x <- begin
+        return $ Yield x (Just (state, x))
+    step' gst (Just (st, acc)) =  do
+        r <- step (adaptState gst) st
+        case r of
+            Yield x s -> do
+                !y <- fstep acc x
+                return $ Yield y (Just (s, y))
+            Skip s -> return $ Skip (Just (s, acc))
+            Stop   -> return Stop
 
 {-# INLINE scanlMAfter' #-}
 scanlMAfter' :: Monad m
@@ -3690,15 +3711,28 @@ scanlMAfter' fstep initial done s =
 
 {-# INLINE scanl' #-}
 scanl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
-scanl' f = scanlM' (\a b -> return (f a b))
+scanl' f seed = scanlM' (\a b -> return (f a b)) (return seed)
 
 {-# INLINE_NORMAL scanlM #-}
-scanlM :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
-scanlM fstep begin s = begin `cons` postscanlM fstep begin s
+scanlM :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
+scanlM fstep begin (Stream step state) = Stream step' Nothing
+  where
+    {-# INLINE_LATE step' #-}
+    step' _ Nothing = do
+        x <- begin
+        return $ Yield x (Just (state, x))
+    step' gst (Just (st, acc)) = do
+        r <- step (adaptState gst) st
+        case r of
+            Yield x s -> do
+                y <- fstep acc x
+                return $ Yield y (Just (s, y))
+            Skip s -> return $ Skip (Just (s, acc))
+            Stop   -> return $ Stop
 
 {-# INLINE scanl #-}
 scanl :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
-scanl f = scanlM (\a b -> return (f a b))
+scanl f seed = scanlM (\a b -> return (f a b)) (return seed)
 
 {-# INLINE_NORMAL scanl1M #-}
 scanl1M :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a

test/Streamly/Test/Prelude/Common.hs

@@ -642,8 +642,8 @@ bindAndComposeHierarchyOps ::
 bindAndComposeHierarchyOps desc t1 = do
     let fldldesc = "Bind and compose foldl, " <> desc <> " Stream "
         fldrdesc = "Bind and compose foldr, " <> desc <> " Stream "
-    
-    bindAndComposeHierarchy 
+
+    bindAndComposeHierarchy
         (fldldesc <> "serially") S.serially fldl
     bindAndComposeHierarchy
         (fldrdesc <> "serially") S.serially fldr
@@ -707,21 +707,21 @@ bindAndComposeHierarchyOps desc t1 = do
             mx >>= \x -> my
             >>= \y -> mz
             >>= \z -> return (x + y + z)
-    
-    fldr, fldl :: (IsStream t, Semigroup (t IO Int)) 
+
+    fldr, fldl :: (IsStream t, Semigroup (t IO Int))
                 => [t IO Int] -> t IO Int
     fldr = foldr (<>) nil
     fldl = foldl (<>) nil
 
 -- Nest two lists using different styles of product compositions
-nestTwoStreams 
+nestTwoStreams
     :: (IsStream t, Semigroup (t IO Int), Monad (t IO))
     => String
     -> ([Int] -> [Int])
     -> ([Int] -> [Int])
     -> (t IO Int -> SerialT IO Int)
     -> Spec
-nestTwoStreams desc streamListT listT t = 
+nestTwoStreams desc streamListT listT t =
     it ("Nests two streams using monadic " <> desc <> " composition") $ do
     let s1 = S.concatMapFoldableWith (<>) return [1..4]
         s2 = S.concatMapFoldableWith (<>) return [5..8]
@@ -731,19 +731,19 @@ nestTwoStreams desc streamListT listT t =
                 return $ x + y
     streamListT r `shouldBe` listT [6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12]
 
-nestTwoStreamsApp 
+nestTwoStreamsApp
     :: (IsStream t, Semigroup (t IO Int), Monad (t IO))
     => String
     -> ([Int] -> [Int])
     -> ([Int] -> [Int])
     -> (t IO Int -> SerialT IO Int)
     -> Spec
-nestTwoStreamsApp desc streamListT listT t = 
+nestTwoStreamsApp desc streamListT listT t =
     it ("Nests two streams using applicative " <> desc <> " composition") $ do
     let s1 = S.concatMapFoldableWith (<>) return [1..4]
         s2 = S.concatMapFoldableWith (<>) return [5..8]
         r  = (S.toList . t) $ ((+) <$> s1 <*> s2)
-    streamListT <$> r 
+    streamListT <$> r
         `shouldReturn` listT [6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12]
 
 
@@ -926,7 +926,7 @@ transformCombineOpsCommon constr desc eq t = do
     prop (desc <> " scanl'") $ transform (scanl' (flip const) 0) t
                                        (S.scanl' (flip const) 0)
     prop (desc <> " scanlM'") $ transform (scanl' (flip const) 0) t
-                                       (S.scanlM' (\_ a -> return a) 0)
+                                       (S.scanlM' (\_ a -> return a) (return 0))
     prop (desc <> " scanl") $ transform (scanl' (flip const) 0) t
                                        (S.scanl' (flip const) 0)
     prop (desc <> " scanl1'") $ transform (scanl1 (flip const)) t

test/Streamly/Test/Prelude/Fold.hs

@@ -114,7 +114,7 @@ foldlM'StrictCheck :: IORef Int -> SerialT IO Int -> IO ()
 foldlM'StrictCheck ref = S.foldlM' (\_ _ -> writeIORef ref 1) (return ())
 
 scanlM'StrictCheck :: IORef Int -> SerialT IO Int -> SerialT IO ()
-scanlM'StrictCheck ref = S.scanlM' (\_ _ -> writeIORef ref 1) ()
+scanlM'StrictCheck ref = S.scanlM' (\_ _ -> writeIORef ref 1) (return ())
 
 checkScanlMStrictness :: (IORef Int -> SerialT IO Int -> SerialT IO ()) -> IO ()
 checkScanlMStrictness f = do