Change the signature of foldlM' to use a monadic seed.

- Also change usages of foldlM' in tests and benchmarks.

.hlint.ignore

@@ -15,7 +15,6 @@ src/Streamly/Internal/Data/Stream/StreamDK/Type.hs
 src/Streamly/Internal/Data/Stream/StreamD.hs
 src/Streamly/Internal/Data/Pipe/Types.hs
 src/Streamly/Internal/Data/Sink.hs
-src/Streamly/Internal/Data/Zipper/Array.hs
 src/Streamly/Internal/Data/Parser/ParserD/Types.hs
 src/Streamly/Internal/Data/Parser/ParserD.hs
 src/Streamly/Internal/Data/Prim/Array/Types.hs

Changelog.md

@@ -1,5 +1,10 @@
 ## Unreleased
 
+### Breaking changes
+
+* Change the signature of `foldlM'` to ensure that the accumulator is
+  monadic.
+
 ### Bug Fixes
 
 * `accept*` and `connect` APIs in `Streamly.Network.Inet.TCP` and the `accept`

benchmark/Streamly/Benchmark/FileIO/Stream.hs

@@ -433,7 +433,7 @@ inspect $ 'chunksOf `hasNoType` ''GroupState
 {-# INLINE chunksOfD #-}
 chunksOfD :: Int -> Handle -> IO Int
 chunksOfD n inh =
-    D.foldlM' (\i _ -> return $ i + 1) 0
+    D.foldlM' (\i _ -> return $ i + 1) (return 0)
         $ D.groupsOf n (AT.writeNUnsafe n)
         $ D.fromStreamK (S.unfold FH.read inh)
 

benchmark/Streamly/Benchmark/Memory/ArrayOps.hs

@@ -173,13 +173,13 @@ toRevList = S.toRevList
 
 foldrMBuild  = S.foldrM  (\x xs -> xs >>= return . (x :)) (return [])
 foldl'Build = S.foldl' (flip (:)) []
-foldlM'Build = S.foldlM' (\xs x -> return $ x : xs) []
+foldlM'Build = S.foldlM' (\xs x -> return $ x : xs) (return [])
 
 foldrMReduce = S.foldrM (\x xs -> xs >>= return . (x +)) (return 0)
 foldl'Reduce = S.foldl' (+) 0
 foldl'ReduceMap = P.fmap (+1) . S.foldl' (+) 0
 foldl1'Reduce = S.foldl1' (+)
-foldlM'Reduce = S.foldlM' (\xs a -> return $ a + xs) 0
+foldlM'Reduce = S.foldlM' (\xs a -> return $ a + xs) (return 0)
 
 last   = S.last
 null   = S.null

benchmark/Streamly/Benchmark/Prelude/Serial.hs

@@ -383,7 +383,7 @@ foldl1'Reduce = S.foldl1' (+)
 
 {-# INLINE foldlM'Reduce #-}
 foldlM'Reduce :: Monad m => SerialT m Int -> m Int
-foldlM'Reduce = S.foldlM' (\xs a -> return $ a + xs) 0
+foldlM'Reduce = S.foldlM' (\xs a -> return $ a + xs) (return 0)
 
 {-# INLINE last #-}
 last :: Monad m => SerialT m Int -> m (Maybe Int)
@@ -539,7 +539,7 @@ foldl'Build = S.foldl' (flip (:)) []
 
 {-# INLINE foldlM'Build #-}
 foldlM'Build :: Monad m => SerialT m Int -> m [Int]
-foldlM'Build = S.foldlM' (\xs x -> return $ x : xs) []
+foldlM'Build = S.foldlM' (\xs x -> return $ x : xs) (return [])
 
 o_n_heap_elimination_foldl :: Int -> [Benchmark]
 o_n_heap_elimination_foldl value =

examples/WordClassifier.hs

@@ -61,7 +61,7 @@ main = do
          & S.map toLower          -- SerialT IO Char
          & S.words FL.toList      -- SerialT IO String
          & S.filter (all isAlpha) -- SerialT IO String
-         & S.foldlM' (flip (Map.alterF alter)) Map.empty -- IO (Map String (IORef Int))
+         & S.foldlM' (flip (Map.alterF alter)) (return Map.empty) -- IO (Map String (IORef Int))
 
     -- Print the top hashmap entries
     counts <-

examples/WordCount.hs

@@ -603,7 +603,7 @@ countArray src = do
 {-# INLINE wc_mwl_parallel #-}
 wc_mwl_parallel :: Handle -> Int -> IO (V.IOVector Int)
 wc_mwl_parallel src n = do
-    counts <- newCounts
+    let counts = newCounts
     S.foldlM' addCounts counts
         $ S.aheadly
         $ S.maxThreads numCapabilities

src/Streamly/Internal/Data/Array.hs

@@ -143,7 +143,7 @@ fromStreamDN limit str = do
     i <-
         D.foldlM'
             (\i x -> i `seq` liftIO $ writeArray marr i x >> return (i + 1))
-            0 $
+            (return 0) $
         D.take limit str
     liftIO $ freezeArray marr 0 i
 

src/Streamly/Internal/Data/Prim/Array.hs

@@ -136,7 +136,7 @@ fromStreamDN limit str = do
     _ <-
         D.foldlM'
             (\i x -> i `seq` liftIO (writePrimArray marr i x) >> return (i + 1))
-            0 $
+            (return 0) $
         D.take limit str
     liftIO $ unsafeFreezePrimArray marr
 

src/Streamly/Internal/Data/SmallArray.hs

@@ -122,7 +122,7 @@ fromStreamDN limit str = do
     i <-
         D.foldlM'
             (\i x -> i `seq` liftIO (writeSmallArray marr i x) >> return (i + 1))
-            0 $
+            (return 0) $
         D.take limit str
     liftIO $ freezeSmallArray marr 0 i
 

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

@@ -480,8 +480,10 @@ foldlx' fstep begin done m =
 
 -- XXX implement in terms of foldlMx'?
 {-# INLINE_NORMAL foldlM' #-}
-foldlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> m b
-foldlM' fstep begin (Stream step state) = go SPEC begin state
+foldlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> m b
+foldlM' fstep mbegin (Stream step state) = do
+    begin <- mbegin
+    go SPEC begin state
   where
     {-# INLINE_LATE go #-}
     go !_ acc st = acc `seq` do
@@ -495,7 +497,7 @@ foldlM' fstep begin (Stream step state) = go SPEC begin state
 
 {-# INLINE foldl' #-}
 foldl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> m b
-foldl' fstep = foldlM' (\b a -> return (fstep b a))
+foldl' fstep begin = foldlM' (\b a -> return (fstep b a)) (return begin)
 
 -- | Convert a list of pure values to a 'Stream'
 {-# INLINE_LATE fromList #-}

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

@@ -433,8 +433,8 @@ foldlMx' step begin done m = go begin m
 
 -- | Like 'foldl'' but with a monadic step function.
 {-# INLINE foldlM' #-}
-foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
-foldlM' step begin = foldlMx' step (return begin) return
+foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> m b
+foldlM' step begin = foldlMx' step begin return
 
 -- | Lazy left fold to a stream.
 {-# INLINE foldlS #-}

src/Streamly/Internal/Memory/Array/Types.hs

@@ -706,7 +706,7 @@ fromStreamDN :: forall m a. (MonadIO m, Storable a)
     => Int -> D.Stream m a -> m (Array a)
 fromStreamDN limit str = do
     arr <- liftIO $ newArray limit
-    end <- D.foldlM' fwrite (aEnd arr) $ D.take limit str
+    end <- D.foldlM' fwrite (return $ aEnd arr) $ D.take limit str
     return $ arr {aEnd = end}
 
     where

src/Streamly/Internal/Memory/ArrayStream.hs

@@ -158,12 +158,12 @@ spliceArraysLenUnsafe :: (MonadIO m, Storable a)
     => Int -> SerialT m (Array a) -> m (Array a)
 spliceArraysLenUnsafe len buffered = do
     arr <- liftIO $ A.newArray len
-    end <- S.foldlM' writeArr (aEnd arr) buffered
+    end <- S.foldlM' writeArr (return $ aEnd arr) buffered
     return $ arr {aEnd = end}
 
     where
 
-    writeArr dst Array{..} =
+    writeArr dst Array{..} = do
         liftIO $ withForeignPtr aStart $ \src -> do
                         let count = aEnd `minusPtr` src
                         A.memcpy (castPtr dst) (castPtr src) count
@@ -181,7 +181,7 @@ _spliceArraysBuffered s = do
 spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)
     => SerialT m (Array a) -> m (Array a)
 spliceArraysRealloced s = do
-    idst <- liftIO $ A.newArray (A.bytesToElemCount (undefined :: a)
+    let idst = liftIO $ A.newArray (A.bytesToElemCount (undefined :: a)
                                 (A.mkChunkSizeKB 4))
 
     arr <- S.foldlM' A.spliceWithDoubling idst s

src/Streamly/Internal/Prelude.hs

@@ -1325,7 +1325,7 @@ foldxM = P.foldlMx'
 --
 -- @since 0.2.0
 {-# INLINE foldlM' #-}
-foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
+foldlM' :: Monad m => (b -> a -> m b) -> m b -> SerialT m a -> m b
 foldlM' step begin m = S.foldlM' step begin $ toStreamS m
 
 ------------------------------------------------------------------------------

test/Main.hs

@@ -706,7 +706,7 @@ checkScanl'Strictness = do
         `shouldReturn` "success"
 
 foldlM'StrictCheck :: IORef Int -> SerialT IO Int -> IO ()
-foldlM'StrictCheck ref = S.foldlM' (\_ _ -> writeIORef ref 1) ()
+foldlM'StrictCheck ref = S.foldlM' (\_ _ -> writeIORef ref 1) (return ())
 
 #ifdef DEVBUILD
 foldxMStrictCheck :: IORef Int -> SerialT IO Int -> IO ()

test/Prop.hs

@@ -374,7 +374,7 @@ concurrentFoldlApplication n =
         -- XXX we should test empty list case as well
         let list = [0..n]
         stream <- run $
-            sourceUnfoldrM1 n |&. S.foldlM' (\xs x -> return (x : xs)) []
+            sourceUnfoldrM1 n |&. S.foldlM' (\xs x -> return (x : xs)) (return [])
         listEquals (==) (reverse stream) list
 
 concurrentFoldrApplication :: Word8 -> Property