Add scanl1, scanl1M, scanl1', scanl1M', scanl, scanlM to StreamD.

benchmark/Linear.hs

@@ -127,6 +127,7 @@ main =
         ]
       , bgroup "transformation"
         [ benchIOSink "scan" (Ops.scan 1)
+        , benchIOSink "scanl1" (Ops.scanl1 1)
         , benchIOSink "map" (Ops.map 1)
         , benchIOSink "fmap" (Ops.fmap 1)
         , benchIOSink "mapM" (Ops.mapM serially 1)
@@ -140,6 +141,7 @@ main =
         ]
       , bgroup "transformationX4"
         [ benchIOSink "scan" (Ops.scan 4)
+        , benchIOSink "scanl1" (Ops.scanl1 4)
         , benchIOSink "map" (Ops.map 4)
         , benchIOSink "fmap" (Ops.fmap 4)
         , benchIOSink "mapM" (Ops.mapM serially 4)
@@ -194,11 +196,13 @@ main =
       , benchIOSink "filter-drop" (Ops.filterDrop 4)
       , benchIOSink "filter-take" (Ops.filterTake 4)
       , benchIOSink "filter-scan" (Ops.filterScan 4)
+      , benchIOSink "filter-scanl1" (Ops.filterScanl1 4)
       , benchIOSink "filter-map"  (Ops.filterMap 4)
       ]
     , bgroup "iterated"
       [ benchIOSrc serially "mapM"           Ops.iterateMapM
       , benchIOSrc serially "scan(1/100)"    Ops.iterateScan
+      , benchIOSrc serially "scanl1(1/100)"  Ops.iterateScanl1
       , benchIOSrc serially "filterEven"     Ops.iterateFilterEven
       , benchIOSrc serially "takeAll"        Ops.iterateTakeAll
       , benchIOSrc serially "dropOne"        Ops.iterateDropOne

benchmark/LinearOps.hs

@@ -254,6 +254,7 @@ composeN' n f =
         _ -> undefined
 
 {-# INLINE scan #-}
+{-# INLINE scanl1 #-}
 {-# INLINE map #-}
 {-# INLINE fmap #-}
 {-# INLINE mapMaybe #-}
@@ -271,7 +272,7 @@ composeN' n f =
 {-# INLINE dropWhileFalse #-}
 {-# INLINE findIndices #-}
 {-# INLINE elemIndices #-}
-scan, map, fmap, mapMaybe, filterEven, filterAllOut,
+scan, scanl1, map, fmap, mapMaybe, filterEven, filterAllOut,
     filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropOne,
     dropAll, dropWhileTrue, dropWhileMTrue, dropWhileFalse,
     findIndices, elemIndices
@@ -290,6 +291,7 @@ sequence :: (S.IsStream t, S.MonadAsync m)
     => (t m Int -> S.SerialT m Int) -> t m (m Int) -> m ()
 
 scan          n = composeN n $ S.scanl' (+) 0
+scanl1        n = composeN n $ S.scanl1' (+)
 fmap          n = composeN n $ Prelude.fmap (+1)
 map           n = composeN n $ S.map (+1)
 mapM t        n = composeN' n $ t . S.mapM return
@@ -332,18 +334,21 @@ iterateSource g i n = f i (sourceUnfoldrMN iterStreamLen n)
 
 {-# INLINE iterateMapM #-}
 {-# INLINE iterateScan #-}
+{-# INLINE iterateScanl1 #-}
 {-# INLINE iterateFilterEven #-}
 {-# INLINE iterateTakeAll #-}
 {-# INLINE iterateDropOne #-}
 {-# INLINE iterateDropWhileFalse #-}
 {-# INLINE iterateDropWhileTrue #-}
-iterateMapM, iterateScan, iterateFilterEven, iterateTakeAll, iterateDropOne,
-    iterateDropWhileFalse, iterateDropWhileTrue
+iterateMapM, iterateScan, iterateScanl1, iterateFilterEven, iterateTakeAll,
+    iterateDropOne, iterateDropWhileFalse, iterateDropWhileTrue
     :: S.MonadAsync m
     => Int -> Stream m Int
 
 -- this is quadratic
 iterateScan            = iterateSource (S.scanl' (+) 0) (maxIters `div` 10)
+-- so is this
+iterateScanl1          = iterateSource (S.scanl1' (+)) (maxIters `div` 10)
 
 iterateMapM            = iterateSource (S.mapM return) maxIters
 iterateFilterEven      = iterateSource (S.filter even) maxIters
@@ -412,9 +417,10 @@ cmpBy src = S.cmpBy P.compare src src
 {-# INLINE filterDrop #-}
 {-# INLINE filterTake #-}
 {-# INLINE filterScan #-}
+{-# INLINE filterScanl1 #-}
 {-# INLINE filterMap #-}
 scanMap, dropMap, dropScan, takeDrop, takeScan, takeMap, filterDrop,
-    filterTake, filterScan, filterMap
+    filterTake, filterScan, filterScanl1, filterMap
     :: Monad m => Int -> Stream m Int -> m ()
 
 scanMap    n = composeN n $ S.map (subtract 1) . S.scanl' (+) 0
@@ -426,6 +432,7 @@ takeMap    n = composeN n $ S.map (subtract 1) . S.take maxValue
 filterDrop n = composeN n $ S.drop 1 . S.filter (<= maxValue)
 filterTake n = composeN n $ S.take maxValue . S.filter (<= maxValue)
 filterScan n = composeN n $ S.scanl' (+) 0 . S.filter (<= maxBound)
+filterScanl1 n = composeN n $ S.scanl1' (+) . S.filter (<= maxBound)
 filterMap  n = composeN n $ S.map (subtract 1) . S.filter (<= maxValue)
 
 -------------------------------------------------------------------------------

src/Streamly/Prelude.hs

@@ -169,6 +169,12 @@ module Streamly.Prelude
     -- transformation like map can be expressed in terms of this.
     , scanl'
     , scanlM'
+    , scanlM
+    , scanl
+    , scanl1M'
+    , scanl1'
+    , scanl1M
+    , scanl1
     , scanx
 
     -- ** Mapping
@@ -243,7 +249,7 @@ import Prelude
                foldl, mapM, mapM_, sequence, all, any, sum, product, elem,
                notElem, maximum, minimum, head, last, tail, length, null,
                reverse, iterate, init, and, or, lookup, foldr1, (!!),
-               splitAt)
+               splitAt, scanl, scanl1)
 import qualified Prelude
 import qualified System.IO as IO
 
@@ -851,6 +857,43 @@ scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
 scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
 scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
 
+-- | Like 'scanlM'' but with a non-strict accumulator.
+--
+{-# INLINE scanlM #-}
+scanlM :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
+scanlM step begin m = fromStreamD $ D.scanlM step begin $ toStreamD m
+
+-- | Like 'scanl'' but with a non-strict accumulator.
+{-# INLINE scanl #-}
+scanl :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
+scanl step z m = fromStreamD $ D.scanl step z $ toStreamD m
+
+-- | Left scan with a monadic step function. Uses the first element of
+-- the stream as the starting value. Returns an empty stream if the
+-- given stream is empty.
+--
+{-# INLINE scanl1M #-}
+scanl1M :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
+scanl1M step m = fromStreamD $ D.scanl1M step $ toStreamD m
+
+-- | Like 'scanl1M' but with a strict accumulator.
+--
+{-# INLINE scanl1M' #-}
+scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
+scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
+
+-- | Left scan, uses the first element of the stream as the starting
+-- value. Returns an empty stream if the given stream is empty.
+{-# INLINE scanl1 #-}
+scanl1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
+scanl1 step m = fromStreamD $ D.scanl1 step $ toStreamD m
+
+-- | Like 'scanl1'' but with a strict accumulator.
+--
+{-# INLINE scanl1' #-}
+scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
+scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
+
 ------------------------------------------------------------------------------
 -- Transformation by Filtering
 ------------------------------------------------------------------------------

src/Streamly/Streams/StreamD.hs

@@ -103,6 +103,12 @@ module Streamly.Streams.StreamD
     -- ** By folding (scans)
     , scanlM'
     , scanl'
+    , scanlM
+    , scanl
+    , scanl1M'
+    , scanl1'
+    , scanl1M
+    , scanl1
 
     -- * Filtering
     , filter
@@ -148,7 +154,7 @@ import Prelude
        hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,
                takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,
                notElem, null, head, tail, zipWith, lookup, foldr1, sequence,
-               (!!))
+               (!!), scanl, scanl1)
 
 import Streamly.SVar (MonadAsync, State(..), defState, rstState)
 
@@ -630,6 +636,20 @@ postscanlM' fstep begin (Stream step state) =
             Skip s -> return $ Skip (s, acc)
             Stop   -> 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)
+  where
+    {-# INLINE_LATE step' #-}
+    step' gst (st, acc) = do
+        r <- step (rstState gst) st
+        case r of
+            Yield x s -> do
+                y <- fstep acc x
+                return (Yield y (s, y))
+            Skip s -> return $ Skip (s, acc)
+            Stop   -> return Stop
+
 {-# 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)
@@ -638,6 +658,64 @@ scanlM' fstep begin s = begin `seq` (begin `cons` postscanlM' fstep begin s)
 scanl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
 scanl' f = scanlM' (\a b -> return (f a b))
 
+{-# 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
+
+{-# INLINE scanl #-}
+scanl :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
+scanl f = scanlM (\a b -> return (f a b))
+
+{-# INLINE_NORMAL scanl1M #-}
+scanl1M :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a
+scanl1M fstep (Stream step state) = Stream step' (state, Nothing)
+  where
+    {-# INLINE_LATE step' #-}
+    step' gst (st, Nothing) = do
+        r <- step (rstState gst) st
+        case r of
+            Yield x s -> return $ Yield x (s, Just x)
+            Skip s -> return $ Skip (s, Nothing)
+            Stop   -> return Stop
+
+    step' gst (st, Just acc) = do
+        r <- step (rstState gst) st
+        case r of
+            Yield y s -> do
+                z <- fstep acc y
+                return $ Yield z (s, Just z)
+            Skip s -> return $ Skip (s, Just acc)
+            Stop   -> return Stop
+
+{-# INLINE scanl1 #-}
+scanl1 :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a
+scanl1 f = scanl1M (\x y -> return (f x y))
+
+{-# INLINE_NORMAL scanl1M' #-}
+scanl1M' :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a
+scanl1M' fstep (Stream step state) = Stream step' (state, Nothing)
+  where
+    {-# INLINE_LATE step' #-}
+    step' gst (st, Nothing) = do
+        r <- step (rstState gst) st
+        case r of
+            Yield x s -> x `seq` return $ Yield x (s, Just x)
+            Skip s -> return $ Skip (s, Nothing)
+            Stop   -> return Stop
+
+    step' gst (st, Just acc) = acc `seq` do
+        r <- step (rstState gst) st
+        case r of
+            Yield y s -> do
+                z <- fstep acc y
+                z `seq` return $ Yield z (s, Just z)
+            Skip s -> return $ Skip (s, Just acc)
+            Stop   -> return Stop
+
+{-# INLINE scanl1' #-}
+scanl1' :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a
+scanl1' f = scanl1M' (\x y -> return (f x y))
+
 -------------------------------------------------------------------------------
 -- Filtering
 -------------------------------------------------------------------------------

test/Prop.hs

@@ -464,10 +464,23 @@ transformCombineOpsCommon constr desc eq t = do
     prop (desc <> " mapM (+1)") $
         transform (fmap (+1)) t (S.mapM (\x -> return (x + 1)))
 
-    prop (desc <> " scan") $ transform (scanl' (flip const) 0) t
+    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)
+    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)
+    prop (desc <> " scanl1") $ transform (scanl1 (flip const)) t
+                                         (S.scanl1 (flip const))
+    prop (desc <> " scanl1M") $ transform (scanl1 (flip const)) t
+                                          (S.scanl1M (\_ a -> return a))
+    prop (desc <> " scanl1'") $ transform (scanl1 (flip const)) t
+                                         (S.scanl1' (flip const))
+    prop (desc <> " scanl1M'") $ transform (scanl1 (flip const)) t
+                                          (S.scanl1M' (\_ a -> return a))
+
     prop (desc <> " reverse") $ transform reverse t S.reverse
 
     prop (desc <> " intersperseM") $