composewell/streamly
1
1
Fork 0
mirror of https://github.com/composewell/streamly.git synced 2024-08-16 23:10:36 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add a finalizer action in Fold type

Browse Source 
This was needed especially for concurrent fold combinators. A fold
combinator that uses concurrent folds needs to wait for the concurrent
folds to finish before it can finish. The finalizing action in folds can
deallocate any resources allocated by the "initial" action and also wait
for folds that it has initialized.

This complicates fold combinators in general. We can potentially
introduce a type for non-failing parsers and support finalization only
in those. The current use cases can be covered by that. Parsers do not
support scanning, which is not required in the use cases where we need
finalization (there is no known use case).
This commit is contained in:
Harendra Kumar 2023-05-23 22:51:13 +05:30
parent 0ed37ff344
commit 32390c592e
36 changed files with 762 additions and 477 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
benchmark/Streamly/Benchmark/Data/Parser.hs
View File

@ -348,7 +348,7 @@ split_ value =
-- PR.dropWhile (<= (value * 1 `div` 4)) *> PR.die "alt"
{-# INLINE takeWhileFail #-}
takeWhileFail :: Monad m => (a -> Bool) -> Fold m a b -> Parser a m b
takeWhileFail predicate (Fold fstep finitial fextract) =
takeWhileFail predicate (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -369,7 +369,7 @@ takeWhileFail predicate (Fold fstep finitial fextract) =
Fold.Done b -> Done 0 b
else return $ Error "fail"
extract s = fmap (Done 0) (fextract s)
extract s = fmap (Done 0) (ffinal s)
{-# INLINE alt2 #-}
alt2 :: Monad m

4
benchmark/Streamly/Benchmark/Data/ParserK.hs
View File

@ -213,7 +213,7 @@ sequence_ value =
{-# INLINE takeWhileFailD #-}
takeWhileFailD :: Monad m => (a -> Bool) -> Fold m a b -> Parser a m b
takeWhileFailD predicate (Fold fstep finitial fextract) =
takeWhileFailD predicate (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -234,7 +234,7 @@ takeWhileFailD predicate (Fold fstep finitial fextract) =
Fold.Done b -> Done 0 b
else return $ Error "fail"
extract s = fmap (Done 0) (fextract s)
extract s = fmap (Done 0) (ffinal s)
{-# INLINE takeWhileFail #-}
takeWhileFail :: CONSTRAINT =>

19
benchmark/Streamly/Benchmark/Unicode/Stream.hs
View File

@ -11,6 +11,11 @@
{-# LANGUAGE CPP #-}
{-# LANGUAGE ScopedTypeVariables #-}
#undef FUSION_CHECK
#ifdef FUSION_CHECK
{-# OPTIONS_GHC -ddump-simpl -ddump-to-file -dsuppress-all #-}
#endif
#ifdef __HADDOCK_VERSION__
#undef INSPECTION
#endif
@ -280,7 +285,8 @@ _copyStreamUtf8'Fold :: Handle -> Handle -> IO ()
_copyStreamUtf8'Fold inh outh =
Stream.fold (Handle.write outh)
$ Unicode.encodeUtf8
$ Stream.foldMany Unicode.writeCharUtf8'
$ Stream.catRights
$ Stream.parseMany Unicode.writeCharUtf8'
$ Stream.unfold Handle.reader inh
{-# NOINLINE _copyStreamUtf8Parser #-}
@ -317,6 +323,7 @@ o_1_space_decode_encode_read env =
main :: IO ()
main = do
#ifndef FUSION_CHECK
env <- mkHandleBenchEnv
defaultMain (allBenchmarks env)
@ -329,3 +336,13 @@ main = do
, o_1_space_decode_encode_read env
]
]
#else
-- Enable FUSION_CHECK macro at the beginning of the file
-- Enable one benchmark below, and run the benchmark
-- Check the .dump-simpl output
env <- mkHandleBenchEnv
let mkHandles (RefHandles {bigInH = inh, outputH = outh}) = Handles inh outh
(Handles inh outh) <- getHandles env mkHandles
copyStreamLatin1' inh outh
return ()
#endif

2
benchmark/lib/Streamly/Benchmark/Common/Handle.hs
View File

@ -30,6 +30,8 @@ module Streamly.Benchmark.Common.Handle
, isSpace
, isSp
, mkHandleBenchEnv
, Handles(..)
, getHandles
)
where

2
core/src/Streamly/Internal/Data/Array.hs
View File

@ -258,7 +258,7 @@ writeLastN ::
(Storable a, Unbox a, MonadIO m) => Int -> Fold m a (Array a)
writeLastN n
| n <= 0 = fmap (const mempty) FL.drain
| otherwise = A.unsafeFreeze <$> Fold step initial done
| otherwise = A.unsafeFreeze <$> Fold step initial done done
where

3
core/src/Streamly/Internal/Data/Array/Type.hs
View File

@ -596,10 +596,11 @@ pinnedWrite = fmap unsafeFreeze MA.pinnedWrite
--
{-# INLINE unsafeMakePure #-}
unsafeMakePure :: Monad m => Fold IO a b -> Fold m a b
unsafeMakePure (Fold step initial extract) =
unsafeMakePure (Fold step initial extract final) =
Fold (\x a -> return $! unsafeInlineIO (step x a))
(return $! unsafePerformIO initial)
(\s -> return $! unsafeInlineIO $ extract s)
(\s -> return $! unsafeInlineIO $ final s)
-- | Convert a pure stream in Identity monad to an immutable array.
--

6
core/src/Streamly/Internal/Data/Fold/Chunked.hs
View File

@ -103,8 +103,8 @@ newtype ChunkFold m a b = ChunkFold (ParserD.Parser (Array a) m b)
{-# INLINE fromFold #-}
fromFold :: forall m a b. (MonadIO m, Unbox a) =>
Fold.Fold m a b -> ChunkFold m a b
fromFold (Fold.Fold fstep finitial fextract) =
ChunkFold (ParserD.Parser step initial (fmap (Done 0) . fextract))
fromFold (Fold.Fold fstep finitial _ ffinal) =
ChunkFold (ParserD.Parser step initial extract)
where
@ -134,6 +134,8 @@ fromFold (Fold.Fold fstep finitial fextract) =
Fold.Partial fs1 ->
goArray SPEC next fs1
extract = fmap (Done 0) . ffinal
-- | Convert an element 'ParserD.Parser' into an array stream fold. If the
-- parser fails the fold would throw an exception.
--

138
core/src/Streamly/Internal/Data/Fold/Combinators.hs
View File

@ -443,8 +443,8 @@ trace f = lmapM (tracing f)
-- /Pre-release/
{-# INLINE transform #-}
transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
Fold step initial extract
transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract ffinal) =
Fold step initial extract final
where
@ -477,10 +477,14 @@ transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
extract (Tuple' _ fs) = fextract fs
final (Tuple' _ fs) = ffinal fs
{-# INLINE scanWith #-}
scanWith :: Monad m => Bool -> Fold m a b -> Fold m b c -> Fold m a c
scanWith isMany (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
scanWith isMany
(Fold stepL initialL extractL finalL)
(Fold stepR initialR extractR finalR) =
Fold step initial extract final
where
@ -494,15 +498,14 @@ scanWith isMany (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Partial sR1 ->
if isMany
then runStep initialL sR1
else Done <$> extractR sR1
else Done <$> finalR sR1
Done bR -> return $ Done bR
Partial sL -> do
!b <- extractL sL
rR <- stepR sR b
return
$ case rR of
Partial sR1 -> Partial (sL, sR1)
Done bR -> Done bR
case rR of
Partial sR1 -> return $ Partial (sL, sR1)
Done bR -> finalL sL >> return (Done bR)
initial = do
r <- initialR
@ -514,6 +517,8 @@ scanWith isMany (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
extract = extractR . snd
final (sL, sR) = finalL sL *> finalR sR
-- | Scan the input of a 'Fold' to change it in a stateful manner using another
-- 'Fold'. The scan stops as soon as the fold terminates.
--
@ -565,7 +570,7 @@ deleteBy eq x0 = fmap extract $ foldl' step (Tuple' False Nothing)
--
{-# INLINE slide2 #-}
slide2 :: Monad m => Fold m (a, Maybe a) b -> Fold m a b
slide2 (Fold step1 initial1 extract1) = Fold step initial extract
slide2 (Fold step1 initial1 extract1 final1) = Fold step initial extract final
where
@ -577,6 +582,8 @@ slide2 (Fold step1 initial1 extract1) = Fold step initial extract
extract (Tuple' _ s) = extract1 s
final (Tuple' _ s) = final1 s
-- | Return the latest unique element using the supplied comparison function.
-- Returns 'Nothing' if the current element is same as the last element
-- otherwise returns 'Just'.
@ -932,7 +939,7 @@ rollingHashFirstN n = take n rollingHash
--
{-# INLINE rollingMapM #-}
rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Fold m a b
rollingMapM f = Fold step initial extract
rollingMapM f = Fold step initial extract extract
where
@ -1173,7 +1180,8 @@ head = one
-- /Pre-release/
{-# INLINE findM #-}
findM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
findM predicate = Fold step (return $ Partial ()) (const $ return Nothing)
findM predicate =
Fold step (return $ Partial ()) extract extract
where
@ -1184,6 +1192,8 @@ findM predicate = Fold step (return $ Partial ()) (const $ return Nothing)
else Partial ()
in f <$> predicate a
extract = const $ return Nothing
-- | Returns the first element that satisfies the given predicate.
--
{-# INLINE find #-}
@ -1426,7 +1436,7 @@ splitAt n fld = splitWith (,) (take n fld)
{-# INLINE takingEndByM #-}
takingEndByM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
takingEndByM p = Fold step initial (return . toMaybe)
takingEndByM p = Fold step initial extract extract
where
@ -1439,6 +1449,8 @@ takingEndByM p = Fold step initial (return . toMaybe)
then Done $ Just a
else Partial $ Just' a
extract = return . toMaybe
-- |
--
-- >>> takingEndBy p = Fold.takingEndByM (return . p)
@ -1449,7 +1461,7 @@ takingEndBy p = takingEndByM (return . p)
{-# INLINE takingEndByM_ #-}
takingEndByM_ :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
takingEndByM_ p = Fold step initial (return . toMaybe)
takingEndByM_ p = Fold step initial extract extract
where
@ -1462,6 +1474,8 @@ takingEndByM_ p = Fold step initial (return . toMaybe)
then Done Nothing
else Partial $ Just' a
extract = return . toMaybe
-- |
--
-- >>> takingEndBy_ p = Fold.takingEndByM_ (return . p)
@ -1472,7 +1486,7 @@ takingEndBy_ p = takingEndByM_ (return . p)
{-# INLINE droppingWhileM #-}
droppingWhileM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
droppingWhileM p = Fold step initial (return . toMaybe)
droppingWhileM p = Fold step initial extract extract
where
@ -1487,6 +1501,8 @@ droppingWhileM p = Fold step initial (return . toMaybe)
else Just' a
step _ a = return $ Partial $ Just' a
extract = return . toMaybe
-- |
-- >>> droppingWhile p = Fold.droppingWhileM (return . p)
--
@ -1515,15 +1531,15 @@ droppingWhile p = droppingWhileM (return . p)
{-# INLINE takeEndBy_ #-}
takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
-- takeEndBy_ predicate = scanMaybe (takingEndBy_ predicate)
takeEndBy_ predicate (Fold fstep finitial fextract) =
Fold step finitial fextract
takeEndBy_ predicate (Fold fstep finitial fextract ffinal) =
Fold step finitial fextract ffinal
where
step s a =
if not (predicate a)
then fstep s a
else Done <$> fextract s
else Done <$> ffinal s
-- Note:
-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
@ -1544,8 +1560,8 @@ takeEndBy_ predicate (Fold fstep finitial fextract) =
{-# INLINE takeEndBy #-}
takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
-- takeEndBy predicate = scanMaybe (takingEndBy predicate)
takeEndBy predicate (Fold fstep finitial fextract) =
Fold step finitial fextract
takeEndBy predicate (Fold fstep finitial fextract ffinal) =
Fold step finitial fextract ffinal
where
@ -1555,7 +1571,7 @@ takeEndBy predicate (Fold fstep finitial fextract) =
then return res
else do
case res of
Partial s1 -> Done <$> fextract s1
Partial s1 -> Done <$> ffinal s1
Done b -> return $ Done b
------------------------------------------------------------------------------
@ -1590,8 +1606,8 @@ takeEndBySeq :: forall m a b. (MonadIO m, Storable a, Unbox a, Enum a, Eq a) =>
Array.Array a
-> Fold m a b
-> Fold m a b
takeEndBySeq patArr (Fold fstep finitial fextract) =
Fold step initial extract
takeEndBySeq patArr (Fold fstep finitial fextract ffinal) =
Fold step initial extract final
where
@ -1604,7 +1620,7 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
| patLen == 0 ->
-- XXX Should we match nothing or everything on empty
-- pattern?
-- Done <$> fextract acc
-- Done <$> ffinal acc
return $ Partial $ SplitOnSeqEmpty acc
| patLen == 1 -> do
pat <- liftIO $ Array.unsafeIndexIO 0 patArr
@ -1655,7 +1671,7 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
case res of
Partial s1
| pat /= x -> return $ Partial $ SplitOnSeqSingle s1 pat
| otherwise -> Done <$> fextract s1
| otherwise -> Done <$> ffinal s1
Done b -> return $ Done b
step (SplitOnSeqWord s idx wrd) x = do
res <- fstep s x
@ -1664,7 +1680,7 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
Partial s1
| idx == maxIndex -> do
if wrd1 .&. wordMask == wordPat
then Done <$> fextract s1
then Done <$> ffinal s1
else return $ Partial $ SplitOnSeqWordLoop s1 wrd1
| otherwise ->
return $ Partial $ SplitOnSeqWord s1 (idx + 1) wrd1
@ -1675,7 +1691,7 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
case res of
Partial s1
| wrd1 .&. wordMask == wordPat ->
Done <$> fextract s1
Done <$> ffinal s1
| otherwise ->
return $ Partial $ SplitOnSeqWordLoop s1 wrd1
Done b -> return $ Done b
@ -1689,7 +1705,7 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
let fld = Ring.unsafeFoldRing (Ring.ringBound rb)
let !ringHash = fld addCksum 0 rb
if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
then Done <$> fextract s1
then Done <$> ffinal s1
else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
else
return $ Partial $ SplitOnSeqKR s1 (idx + 1) rb rh1
@ -1702,11 +1718,11 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
rh1 <- liftIO $ Ring.unsafeInsert rb rh x
let ringHash = deltaCksum cksum old x
if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
then Done <$> fextract s1
then Done <$> ffinal s1
else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
Done b -> return $ Done b
extract state =
extractFunc fex state =
let st =
case state of
SplitOnSeqEmpty s -> s
@ -1715,7 +1731,11 @@ takeEndBySeq patArr (Fold fstep finitial fextract) =
SplitOnSeqWordLoop s _ -> s
SplitOnSeqKR s _ _ _ -> s
SplitOnSeqKRLoop s _ _ _ -> s
in fextract st
in fex st
extract state = extractFunc fextract state
final state = extractFunc ffinal state
-- | Like 'takeEndBySeq' but discards the matched sequence.
--
@ -1726,8 +1746,8 @@ takeEndBySeq_ :: forall m a b. (MonadIO m, Storable a, Unbox a, Enum a, Eq a) =>
Array.Array a
-> Fold m a b
-> Fold m a b
takeEndBySeq_ patArr (Fold fstep finitial fextract) =
Fold step initial extract
takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) =
Fold step initial extract final
where
@ -1740,7 +1760,7 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
| patLen == 0 ->
-- XXX Should we match nothing or everything on empty
-- pattern?
-- Done <$> fextract acc
-- Done <$> ffinal acc
return $ Partial $ SplitOnSeqEmpty acc
| patLen == 1 -> do
pat <- liftIO $ Array.unsafeIndexIO 0 patArr
@ -1797,13 +1817,13 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
case res of
Partial s1 -> return $ Partial $ SplitOnSeqSingle s1 pat
Done b -> return $ Done b
else Done <$> fextract s
else Done <$> ffinal s
step (SplitOnSeqWord s idx wrd) x = do
let wrd1 = addToWord wrd x
if idx == maxIndex
then do
if wrd1 .&. wordMask == wordPat
then Done <$> fextract s
then Done <$> ffinal s
else return $ Partial $ SplitOnSeqWordLoop s wrd1
else return $ Partial $ SplitOnSeqWord s (idx + 1) wrd1
step (SplitOnSeqWordLoop s wrd) x = do
@ -1814,7 +1834,7 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
case res of
Partial s1
| wrd1 .&. wordMask == wordPat ->
Done <$> fextract s1
Done <$> ffinal s1
| otherwise ->
return $ Partial $ SplitOnSeqWordLoop s1 wrd1
Done b -> return $ Done b
@ -1825,7 +1845,7 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
let fld = Ring.unsafeFoldRing (Ring.ringBound rb)
let !ringHash = fld addCksum 0 rb
if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
then Done <$> fextract s
then Done <$> ffinal s
else return $ Partial $ SplitOnSeqKRLoop s ringHash rb rh1
else return $ Partial $ SplitOnSeqKR s (idx + 1) rb rh1
step (SplitOnSeqKRLoop s cksum rb rh) x = do
@ -1836,7 +1856,7 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
rh1 <- liftIO $ Ring.unsafeInsert rb rh x
let ringHash = deltaCksum cksum old x
if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
then Done <$> fextract s1
then Done <$> ffinal s1
else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
Done b -> return $ Done b
@ -1844,10 +1864,10 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
-- terminates early inside extract, we may still have buffered data
-- remaining which will be lost if we do not communicate that to the
-- driver.
extract state = do
extractFunc fex state = do
let consumeWord s n wrd = do
if n == 0
then fextract s
then fex s
else do
let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
r <- fstep s (toEnum $ fromIntegral old)
@ -1857,7 +1877,7 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
let consumeRing s n rb rh =
if n == 0
then fextract s
then fex s
else do
old <- liftIO $ peek rh
let rh1 = Ring.advance rb rh
@ -1867,13 +1887,17 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract) =
Done b -> return b
case state of
SplitOnSeqEmpty s -> fextract s
SplitOnSeqSingle s _ -> fextract s
SplitOnSeqEmpty s -> fex s
SplitOnSeqSingle s _ -> fex s
SplitOnSeqWord s idx wrd -> consumeWord s idx wrd
SplitOnSeqWordLoop s wrd -> consumeWord s patLen wrd
SplitOnSeqKR s idx rb _ -> consumeRing s idx rb (Ring.startOf rb)
SplitOnSeqKRLoop s _ rb rh -> consumeRing s patLen rb rh
extract state = extractFunc fextract state
final state = extractFunc ffinal state
------------------------------------------------------------------------------
-- Distributing
------------------------------------------------------------------------------
@ -2317,8 +2341,8 @@ toStreamRev = fmap StreamD.fromList toListRev
-- /Pre-release/
{-# INLINE unfoldMany #-}
unfoldMany :: Monad m => Unfold m a b -> Fold m b c -> Fold m a c
unfoldMany (Unfold ustep inject) (Fold fstep initial extract) =
Fold consume initial extract
unfoldMany (Unfold ustep inject) (Fold fstep initial extract final) =
Fold consume initial extract final
where
@ -2345,7 +2369,7 @@ bottomBy :: (MonadIO m, Unbox a) =>
(a -> a -> Ordering)
-> Int
-> Fold m a (MutArray a)
bottomBy cmp n = Fold step initial extract
bottomBy cmp n = Fold step initial extract extract
where
@ -2440,8 +2464,8 @@ intersperseWithQuotes
quote
esc
separator
(Fold stepL initialL extractL)
(Fold stepR initialR extractR) = Fold step initial extract
(Fold stepL initialL _ finalL)
(Fold stepR initialR extractR finalR) = Fold step initial extract final
where
@ -2484,11 +2508,13 @@ intersperseWithQuotes
r <- stepL sL a
case r of
Partial s -> return $ Partial (nextState sR s)
Done _ -> error "Collecting fold finished inside quote"
Done _ -> do
_ <- finalR sR
error "Collecting fold finished inside quote"
step (IntersperseQUnquoted sR sL) a
| a == separator = do
b <- extractL sL
b <- finalL sL
collect IntersperseQUnquoted sR b
| a == quote = processQuoted a sL sR IntersperseQQuoted
| otherwise = process a sL sR IntersperseQUnquoted
@ -2506,3 +2532,13 @@ intersperseWithQuotes
error "intersperseWithQuotes: finished inside quote"
extract (IntersperseQQuotedEsc _ _) =
error "intersperseWithQuotes: finished inside quote, at escape char"
final (IntersperseQUnquoted sR sL) = finalL sL *> finalR sR
final (IntersperseQQuoted sR sL) = do
_ <- finalR sR
_ <- finalL sL
error "intersperseWithQuotes: finished inside quote"
final (IntersperseQQuotedEsc sR sL) = do
_ <- finalR sR
_ <- finalL sL
error "intersperseWithQuotes: finished inside quote, at escape char"

101
core/src/Streamly/Internal/Data/Fold/Container.hs
View File

@ -270,14 +270,15 @@ demuxGeneric :: (Monad m, IsMap f, Traversable f) =>
(a -> Key f)
-> (a -> m (Fold m a b))
-> Fold m a (m (f b), Maybe (Key f, b))
demuxGeneric getKey getFold = fmap extract $ foldlM' step initial
demuxGeneric getKey getFold =
Fold (\s a -> Partial <$> step s a) (Partial <$> initial) extract final
where
initial = return $ Tuple' IsMap.mapEmpty Nothing
{-# INLINE runFold #-}
runFold kv (Fold step1 initial1 extract1) (k, a) = do
runFold kv (Fold step1 initial1 extract1 final1) (k, a) = do
res <- initial1
case res of
Partial s -> do
@ -285,7 +286,7 @@ demuxGeneric getKey getFold = fmap extract $ foldlM' step initial
return
$ case res1 of
Partial _ ->
let fld = Fold step1 (return res1) extract1
let fld = Fold step1 (return res1) extract1 final1
in Tuple' (IsMap.mapInsert k fld kv) Nothing
Done b -> Tuple' (IsMap.mapDelete k kv) (Just (k, b))
Done b -> return $ Tuple' kv (Just (k, b))
@ -298,15 +299,25 @@ demuxGeneric getKey getFold = fmap extract $ foldlM' step initial
runFold kv fld (k, a)
Just f -> runFold kv f (k, a)
extract (Tuple' kv x) = (Prelude.mapM f kv, x)
extract (Tuple' kv x) = return (Prelude.mapM f kv, x)
where
f (Fold _ i e) = do
f (Fold _ i e _) = do
r <- i
case r of
Partial s -> e s
Done b -> return b
_ -> error "demuxGeneric: unreachable code"
final (Tuple' kv x) = return (Prelude.mapM f kv, x)
where
f (Fold _ i _ fin) = do
r <- i
case r of
Partial s -> fin s
_ -> error "demuxGeneric: unreachable code"
-- | @demux getKey getFold@: In a key value stream, fold values corresponding
-- to each key using a key specific fold. @getFold@ is invoked to generate a
@ -348,35 +359,36 @@ demuxGenericIO :: (MonadIO m, IsMap f, Traversable f) =>
(a -> Key f)
-> (a -> m (Fold m a b))
-> Fold m a (m (f b), Maybe (Key f, b))
demuxGenericIO getKey getFold = fmap extract $ foldlM' step initial
demuxGenericIO getKey getFold =
Fold (\s a -> Partial <$> step s a) (Partial <$> initial) extract final
where
initial = return $ Tuple' IsMap.mapEmpty Nothing
{-# INLINE initFold #-}
initFold kv (Fold step1 initial1 extract1) (k, a) = do
initFold kv (Fold step1 initial1 extract1 final1) (k, a) = do
res <- initial1
case res of
Partial s -> do
res1 <- step1 s a
case res1 of
Partial _ -> do
let fld = Fold step1 (return res1) extract1
let fld = Fold step1 (return res1) extract1 final1
ref <- liftIO $ newIORef fld
return $ Tuple' (IsMap.mapInsert k ref kv) Nothing
Done b -> return $ Tuple' kv (Just (k, b))
Done b -> return $ Tuple' kv (Just (k, b))
{-# INLINE runFold #-}
runFold kv ref (Fold step1 initial1 extract1) (k, a) = do
runFold kv ref (Fold step1 initial1 extract1 final1) (k, a) = do
res <- initial1
case res of
Partial s -> do
res1 <- step1 s a
case res1 of
Partial _ -> do
let fld = Fold step1 (return res1) extract1
let fld = Fold step1 (return res1) extract1 final1
liftIO $ writeIORef ref fld
return $ Tuple' kv Nothing
Done b ->
@ -394,16 +406,27 @@ demuxGenericIO getKey getFold = fmap extract $ foldlM' step initial
f <- liftIO $ readIORef ref
runFold kv ref f (k, a)
extract (Tuple' kv x) = (Prelude.mapM f kv, x)
extract (Tuple' kv x) = return (Prelude.mapM f kv, x)
where
f ref = do
(Fold _ i e) <- liftIO $ readIORef ref
Fold _ i e _ <- liftIO $ readIORef ref
r <- i
case r of
Partial s -> e s
Done b -> return b
_ -> error "demuxGenericIO: unreachable code"
final (Tuple' kv x) = return (Prelude.mapM f kv, x)
where
f ref = do
Fold _ i _ fin <- liftIO $ readIORef ref
r <- i
case r of
Partial s -> fin s
_ -> error "demuxGenericIO: unreachable code"
-- | This is specialized version of 'demux' that uses mutable IO cells as
-- fold accumulators for better performance.
@ -418,6 +441,13 @@ demuxIO :: (MonadIO m, Ord k) =>
-> Fold m a (m (Map k b), Maybe (k, b))
demuxIO = demuxGenericIO
-- | Fold a key value stream to a key-value Map. If the same key appears
-- multiple times, only the last value is retained.
{-# INLINE kvToMapOverwriteGeneric #-}
kvToMapOverwriteGeneric :: (Monad m, IsMap f) => Fold m (Key f, a) (f a)
kvToMapOverwriteGeneric =
foldl' (\kv (k, v) -> IsMap.mapInsert k v kv) IsMap.mapEmpty
{-# INLINE demuxToContainer #-}
demuxToContainer :: (Monad m, IsMap f, Traversable f) =>
(a -> Key f) -> (a -> m (Fold m a b)) -> Fold m a (f b)
@ -505,8 +535,8 @@ classifyGeneric :: (Monad m, IsMap f, Traversable f, Ord (Key f)) =>
-- for that use case. We return an action because we want it to be lazy so
-- that the downstream consumers can choose to process or discard it.
(a -> Key f) -> Fold m a b -> Fold m a (m (f b), Maybe (Key f, b))
classifyGeneric f (Fold step1 initial1 extract1) =
fmap extract $ foldlM' step initial
classifyGeneric f (Fold step1 initial1 extract1 final1) =
Fold (\s a -> Partial <$> step s a) (Partial <$> initial) extract final
where
@ -543,7 +573,16 @@ classifyGeneric f (Fold step1 initial1 extract1) =
let kv1 = IsMap.mapDelete k kv
in Tuple3' kv1 (Set.insert k set) (Just (k, b))
extract (Tuple3' kv _ x) = (Prelude.mapM extract1 kv, x)
extract (Tuple3' kv _ x) = return (Prelude.mapM extract1 kv, x)
final (Tuple3' kv set x) = return (IsMap.mapTraverseWithKey f1 kv, x)
where
f1 k s = do
if Set.member k set
then extract1 s
else final1 s
-- | Folds the values for each key using the supplied fold. When scanning, as
-- soon as the fold is complete, its result is available in the second
@ -568,8 +607,8 @@ classify = classifyGeneric
{-# INLINE classifyGenericIO #-}
classifyGenericIO :: (MonadIO m, IsMap f, Traversable f, Ord (Key f)) =>
(a -> Key f) -> Fold m a b -> Fold m a (m (f b), Maybe (Key f, b))
classifyGenericIO f (Fold step1 initial1 extract1) =
fmap extract $ foldlM' step initial
classifyGenericIO f (Fold step1 initial1 extract1 final1) =
Fold (\s a -> Partial <$> step s a) (Partial <$> initial) extract final
where
@ -608,8 +647,21 @@ classifyGenericIO f (Fold step1 initial1 extract1) =
in return
$ Tuple3' kv1 (Set.insert k set) (Just (k, b))
extract (Tuple3' kv _ x) =
(Prelude.mapM (\ref -> liftIO (readIORef ref) >>= extract1) kv, x)
extract (Tuple3' kv _ x) = return (Prelude.mapM g kv, x)
where
g ref = liftIO (readIORef ref) >>= extract1
final (Tuple3' kv set x) = return (IsMap.mapTraverseWithKey g kv, x)
where
g k ref = do
s <- liftIO $ readIORef ref
if Set.member k set
then extract1 s
else final1 s
-- | Same as classify except that it uses mutable IORef cells in the
-- Map providing better performance. Be aware that if this is used as a scan,
@ -624,13 +676,6 @@ classifyIO :: (MonadIO m, Ord k) =>
(a -> k) -> Fold m a b -> Fold m a (m (Map k b), Maybe (k, b))
classifyIO = classifyGenericIO
-- | Fold a key value stream to a key-value Map. If the same key appears
-- multiple times, only the last value is retained.
{-# INLINE kvToMapOverwriteGeneric #-}
kvToMapOverwriteGeneric :: (Monad m, IsMap f) => Fold m (Key f, a) (f a)
kvToMapOverwriteGeneric =
foldl' (\kv (k, v) -> IsMap.mapInsert k v kv) IsMap.mapEmpty
{-# INLINE toContainer #-}
toContainer :: (Monad m, IsMap f, Traversable f, Ord (Key f)) =>
(a -> Key f) -> Fold m a b -> Fold m a (f b)

2
core/src/Streamly/Internal/Data/Fold/Step.hs
View File

@ -40,7 +40,7 @@ data Step s b
= Partial !s
| Done !b
-- | 'first' maps over 'Partial' and 'second' maps over 'Done'.
-- | 'first' maps over the fold state and 'second' maps over the fold result.
--
instance Bifunctor Step where
{-# INLINE bimap #-}

326
core/src/Streamly/Internal/Data/Fold/Type.hs
View File

@ -445,7 +445,7 @@ where
#if !MIN_VERSION_base(4,18,0)
import Control.Applicative (liftA2)
#endif
import Control.Monad ((>=>))
import Control.Monad ((>=>), void)
import Data.Bifunctor (Bifunctor(..))
import Data.Either (fromLeft, fromRight, isLeft, isRight)
import Data.Functor.Identity (Identity(..))
@ -471,23 +471,49 @@ import Streamly.Internal.Data.Fold.Step
-- The type @b@ is the accumulator of the writer. That's the reason the
-- default folds in various modules are called "write".
-- | The type @Fold m a b@ having constructor @Fold step initial extract@
-- represents a fold over an input stream of values of type @a@ to a final
-- value of type @b@ in 'Monad' @m@.
-- An alternative to using an "extract" function is to use "Partial s b" style
-- partial value so that we always emit the output value and there is no need
-- to extract. Then extract can be used for cleanup purposes. But in this case
-- in some cases we may need a "Continue" constructor where an output value is
-- not available, this was implicit earlier. Also, "b" should be lazy here so
-- that we do not always compute it even if we do not need it.
--
-- Partial s b --> extract :: s -> b
-- Continue --> extract :: s -> Maybe b
--
-- But keeping 'b' lazy does not let the fold optimize well. It leads to
-- significant regressions in the key-value folds.
--
-- The "final" function complicates combinators that take other folds as
-- argument because we need to call their finalizers at right places. An
-- alternative to reduce this complexity where it is not required is to use a
-- separate type for bracketed folds but then we need to manage the complexity
-- of two different fold types.
-- The "final" function could be (s -> m (Step s b)), like in parsers so that
-- it can be called in a loop to drain the fold.
-- | The type @Fold m a b@ having constructor @Fold step initial extract
-- final@ represents a fold over an input stream of values of type @a@ to a
-- final value of type @b@ in 'Monad' @m@.
--
-- The fold uses an intermediate state @s@ as accumulator, the type @s@ is
-- internal to the specific fold definition. The initial value of the fold
-- state @s@ is returned by @initial@. The @step@ function consumes an input
-- and either returns the final result @b@ if the fold is done or the next
-- intermediate state (see 'Step'). At any point the fold driver can extract
-- the result from the intermediate state using the @extract@ function.
-- the result from the intermediate state using the @extract@ function. The
-- "final" function is used to finalize the fold, the driver can call it
-- whenever it holds a valid fold state which it will not be using anymore. The
-- state should not be used after finalization. Note that if the fold
-- terminates itself we won't have a valid fold state.
--
-- NOTE: The constructor is not yet released, smart constructors are provided
-- to create folds.
--
data Fold m a b =
-- | @Fold @ @ step @ @ initial @ @ extract@
forall s. Fold (s -> a -> m (Step s b)) (m (Step s b)) (s -> m b)
-- | @Fold@ @step@ @initial@ @extract@ @final@
forall s. Fold (s -> a -> m (Step s b)) (m (Step s b)) (s -> m b) (s -> m b)
------------------------------------------------------------------------------
-- Mapping on the output
@ -497,7 +523,8 @@ data Fold m a b =
--
{-# INLINE rmapM #-}
rmapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
rmapM f (Fold step initial extract) = Fold step1 initial1 (extract >=> f)
rmapM f (Fold step initial extract final) =
Fold step1 initial1 (extract >=> f) (final >=> f)
where
@ -528,6 +555,7 @@ foldl' step initial =
(\s a -> return $ Partial $ step s a)
(return (Partial initial))
return
return
-- | Make a fold from a left fold style monadic step function and initial value
-- of the accumulator.
@ -542,7 +570,7 @@ foldl' step initial =
{-# INLINE foldlM' #-}
foldlM' :: Monad m => (b -> a -> m b) -> m b -> Fold m a b
foldlM' step initial =
Fold (\s a -> Partial <$> step s a) (Partial <$> initial) return
Fold (\s a -> Partial <$> step s a) (Partial <$> initial) return return
-- | Make a strict left fold, for non-empty streams, using first element as the
-- starting value. Returns Nothing if the stream is empty.
@ -640,7 +668,11 @@ foldrM' g z =
{-# INLINE foldt' #-}
foldt' :: Monad m => (s -> a -> Step s b) -> Step s b -> (s -> b) -> Fold m a b
foldt' step initial extract =
Fold (\s a -> return $ step s a) (return initial) (return . extract)
Fold
(\s a -> return $ step s a)
(return initial)
(return . extract)
(return . extract)
-- | Make a terminating fold with an effectful step function and initial state,
-- and a state extraction function.
@ -653,7 +685,7 @@ foldt' step initial extract =
--
{-# INLINE foldtM' #-}
foldtM' :: (s -> a -> m (Step s b)) -> m (Step s b) -> (s -> m b) -> Fold m a b
foldtM' = Fold
foldtM' step initial extract = Fold step initial extract extract
------------------------------------------------------------------------------
-- Refold
@ -667,7 +699,7 @@ foldtM' = Fold
-- /Internal/
fromRefold :: Refold m c a b -> c -> Fold m a b
fromRefold (Refold step inject extract) c =
Fold step (inject c) extract
Fold step (inject c) extract extract
------------------------------------------------------------------------------
-- Basic Folds
@ -727,7 +759,8 @@ toStreamK = foldr K.cons K.nil
-- | Maps a function on the output of the fold (the type @b@).
instance Functor m => Functor (Fold m a) where
{-# INLINE fmap #-}
fmap f (Fold step1 initial1 extract) = Fold step initial (fmap2 f extract)
fmap f (Fold step1 initial1 extract final) =
Fold step initial (fmap2 f extract) (fmap2 f final)
where
@ -762,7 +795,7 @@ instance Functor m => Functor (Fold m a) where
--
{-# INLINE fromPure #-}
fromPure :: Applicative m => b -> Fold m a b
fromPure b = Fold undefined (pure $ Done b) pure
fromPure b = Fold undefined (pure $ Done b) pure pure
-- | Make a fold that yields the result of the supplied effectful action
-- without consuming any further input.
@ -771,7 +804,7 @@ fromPure b = Fold undefined (pure $ Done b) pure
--
{-# INLINE fromEffect #-}
fromEffect :: Applicative m => m b -> Fold m a b
fromEffect b = Fold undefined (Done <$> b) pure
fromEffect b = Fold undefined (Done <$> b) pure pure
{-# ANN type SeqFoldState Fuse #-}
data SeqFoldState sl f sr = SeqFoldL !sl | SeqFoldR !f !sr
@ -804,8 +837,10 @@ data SeqFoldState sl f sr = SeqFoldL !sl | SeqFoldR !f !sr
{-# INLINE splitWith #-}
splitWith :: Monad m =>
(a -> b -> c) -> Fold m x a -> Fold m x b -> Fold m x c
splitWith func (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
splitWith func
(Fold stepL initialL _ finalL)
(Fold stepR initialR _ finalR) =
Fold step initial extract final
where
@ -822,13 +857,18 @@ splitWith func (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
step (SeqFoldL st) a = runL (stepL st a)
step (SeqFoldR f st) a = runR (stepR st a) f
extract (SeqFoldR f sR) = fmap f (extractR sR)
extract (SeqFoldL sL) = do
rL <- extractL sL
-- XXX splitWith should not be used for scanning
-- It would rarely make sense and resource cleanup would be expensive.
-- especially when multiple splitWith are chained.
extract _ = error "splitWith: cannot be used for scanning"
final (SeqFoldR f sR) = fmap f (finalR sR)
final (SeqFoldL sL) = do
rL <- finalL sL
res <- initialR
fmap (func rL)
$ case res of
Partial sR -> extractR sR
Partial sR -> finalR sR
Done rR -> return rR
{-# DEPRECATED serialWith "Please use \"splitWith\" instead" #-}
@ -848,8 +888,8 @@ data SeqFoldState_ sl sr = SeqFoldL_ !sl | SeqFoldR_ !sr
--
{-# INLINE split_ #-}
split_ :: Monad m => Fold m x a -> Fold m x b -> Fold m x b
split_ (Fold stepL initialL _) (Fold stepR initialR extractR) =
Fold step initial extract
split_ (Fold stepL initialL _ finalL) (Fold stepR initialR _ finalR) =
Fold step initial extract final
where
@ -872,11 +912,16 @@ split_ (Fold stepL initialL _) (Fold stepR initialR extractR) =
resR <- stepR st a
return $ first SeqFoldR_ resR
extract (SeqFoldR_ sR) = extractR sR
extract (SeqFoldL_ _) = do
-- XXX split_ should not be used for scanning
-- See splitWith for more details.
extract _ = error "split_: cannot be used for scanning"
final (SeqFoldR_ sR) = finalR sR
final (SeqFoldL_ sL) = do
_ <- finalL sL
res <- initialR
case res of
Partial sR -> extractR sR
Partial sR -> finalR sR
Done rR -> return rR
-- | 'Applicative' form of 'splitWith'. Split the input serially over two
@ -924,8 +969,10 @@ data TeeState sL sR bL bR
--
{-# INLINE teeWith #-}
teeWith :: Monad m => (a -> b -> c) -> Fold m x a -> Fold m x b -> Fold m x c
teeWith f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
teeWith f
(Fold stepL initialL extractL finalL)
(Fold stepR initialR extractR finalR) =
Fold step initial extract final
where
@ -952,6 +999,10 @@ teeWith f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
extract (TeeLeft bR sL) = (`f` bR) <$> extractL sL
extract (TeeRight bL sR) = f bL <$> extractR sR
final (TeeBoth sL sR) = f <$> finalL sL <*> finalR sR
final (TeeLeft bR sL) = (`f` bR) <$> finalL sL
final (TeeRight bL sR) = f bL <$> finalR sR
{-# ANN type TeeFstState Fuse #-}
data TeeFstState sL sR b
= TeeFstBoth !sL !sR
@ -964,8 +1015,10 @@ data TeeFstState sL sR b
{-# INLINE teeWithFst #-}
teeWithFst :: Monad m =>
(b -> c -> d) -> Fold m a b -> Fold m a c -> Fold m a d
teeWithFst f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
teeWithFst f
(Fold stepL initialL extractL finalL)
(Fold stepR initialR extractR finalR) =
Fold step initial extract final
where
@ -984,7 +1037,7 @@ teeWithFst f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Done bl -> do
Done . f bl <$>
case resR of
Partial sr -> extractR sr
Partial sr -> finalR sr
Done br -> return br
initial = runBoth initialL initialR
@ -995,6 +1048,9 @@ teeWithFst f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
extract (TeeFstBoth sL sR) = f <$> extractL sL <*> extractR sR
extract (TeeFstLeft bR sL) = (`f` bR) <$> extractL sL
final (TeeFstBoth sL sR) = f <$> finalL sL <*> finalR sR
final (TeeFstLeft bR sL) = (`f` bR) <$> finalL sL
-- | Like 'teeWith' but terminates as soon as any one of the two folds
-- terminates.
--
@ -1003,8 +1059,10 @@ teeWithFst f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
{-# INLINE teeWithMin #-}
teeWithMin :: Monad m =>
(b -> c -> d) -> Fold m a b -> Fold m a c -> Fold m a d
teeWithMin f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
teeWithMin f
(Fold stepL initialL extractL finalL)
(Fold stepR initialR extractR finalR) =
Fold step initial extract final
where
@ -1016,12 +1074,12 @@ teeWithMin f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Partial sl -> do
case resR of
Partial sr -> return $ Partial $ Tuple' sl sr
Done br -> Done . (`f` br) <$> extractL sl
Done br -> Done . (`f` br) <$> finalL sl
Done bl -> do
Done . f bl <$>
case resR of
Partial sr -> extractR sr
Partial sr -> finalR sr
Done br -> return br
initial = runBoth initialL initialR
@ -1030,6 +1088,8 @@ teeWithMin f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
extract (Tuple' sL sR) = f <$> extractL sL <*> extractR sR
final (Tuple' sL sR) = f <$> finalL sL <*> finalR sR
-- | Shortest alternative. Apply both folds in parallel but choose the result
-- from the one which consumed least input i.e. take the shortest succeeding
-- fold.
@ -1041,8 +1101,8 @@ teeWithMin f (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
--
{-# INLINE shortest #-}
shortest :: Monad m => Fold m x a -> Fold m x b -> Fold m x (Either a b)
shortest (Fold stepL initialL extractL) (Fold stepR initialR _) =
Fold step initial extract
shortest (Fold stepL initialL extractL finalL) (Fold stepR initialR _ finalR) =
Fold step initial extract final
where
@ -1050,10 +1110,16 @@ shortest (Fold stepL initialL extractL) (Fold stepR initialR _) =
runBoth actionL actionR = do
resL <- actionL
resR <- actionR
return $
case resL of
Partial sL -> bimap (Tuple' sL) Right resR
Done bL -> Done $ Left bL
case resL of
Partial sL ->
case resR of
Partial sR -> return $ Partial $ Tuple' sL sR
Done bR -> finalL sL >> return (Done (Right bR))
Done bL -> do
case resR of
Partial sR -> void (finalR sR)
Done _ -> return ()
return (Done (Left bL))
initial = runBoth initialL initialR
@ -1061,6 +1127,8 @@ shortest (Fold stepL initialL extractL) (Fold stepR initialR _) =
extract (Tuple' sL _) = Left <$> extractL sL
final (Tuple' sL sR) = Left <$> finalL sL <* finalR sR
{-# ANN type LongestState Fuse #-}
data LongestState sL sR
= LongestBoth !sL !sR
@ -1078,8 +1146,10 @@ data LongestState sL sR
--
{-# INLINE longest #-}
longest :: Monad m => Fold m x a -> Fold m x b -> Fold m x (Either a b)
longest (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
longest
(Fold stepL initialL _ finalL)
(Fold stepR initialR _ finalR) =
Fold step initial extract final
where
@ -1102,14 +1172,18 @@ longest (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
step (LongestLeft sL) a = bimap LongestLeft Left <$> stepL sL a
step (LongestRight sR) a = bimap LongestRight Right <$> stepR sR a
left sL = Left <$> extractL sL
extract (LongestLeft sL) = left sL
extract (LongestRight sR) = Right <$> extractR sR
extract (LongestBoth sL _) = left sL
-- XXX Scan with this may not make sense as we cannot determine the longest
-- until one of them have exhausted.
extract _ = error $ "longest: scan is not allowed as longest cannot be "
++ "determined until one fold has exhausted."
data ConcatMapState m sa a c
= B !sa
| forall s. C (s -> a -> m (Step s c)) !s (s -> m c)
final (LongestLeft sL) = Left <$> finalL sL
final (LongestRight sR) = Right <$> finalR sR
final (LongestBoth sL sR) = Left <$> finalL sL <* finalR sR
data ConcatMapState m sa a b c
= B !sa (sa -> m b)
| forall s. C (s -> a -> m (Step s c)) !s (s -> m c) (s -> m c)
-- | Map a 'Fold' returning function on the result of a 'Fold' and run the
-- returned fold. This operation can be used to express data dependencies
@ -1132,43 +1206,47 @@ data ConcatMapState m sa a c
--
{-# INLINE concatMap #-}
concatMap :: Monad m => (b -> Fold m a c) -> Fold m a b -> Fold m a c
concatMap f (Fold stepa initiala extracta) = Fold stepc initialc extractc
concatMap f (Fold stepa initiala _ finala) =
Fold stepc initialc extractc finalc
where
initialc = do
r <- initiala
case r of
Partial s -> return $ Partial (B s)
Partial s -> return $ Partial (B s finala)
Done b -> initInnerFold (f b)
stepc (B s) a = do
stepc (B s fin) a = do
r <- stepa s a
case r of
Partial s1 -> return $ Partial (B s1)
Partial s1 -> return $ Partial (B s1 fin)
Done b -> initInnerFold (f b)
stepc (C stepInner s extractInner) a = do
stepc (C stepInner s extractInner fin) a = do
r <- stepInner s a
return $ case r of
Partial sc -> Partial (C stepInner sc extractInner)
Partial sc -> Partial (C stepInner sc extractInner fin)
Done c -> Done c
extractc (B s) = do
r <- extracta s
initExtract (f r)
extractc (C _ sInner extractInner) = extractInner sInner
-- XXX Cannot use for scanning
extractc _ = error "concatMap: cannot be used for scanning"
initInnerFold (Fold step i e) = do
initInnerFold (Fold step i e fin) = do
r <- i
return $ case r of
Partial s -> Partial (C step s e)
Partial s -> Partial (C step s e fin)
Done c -> Done c
initExtract (Fold _ i e) = do
initFinalize (Fold _ i _ fin) = do
r <- i
case r of
Partial s -> e s
Partial s -> fin s
Done c -> return c
finalc (B s fin) = do
r <- fin s
initFinalize (f r)
finalc (C _ sInner _ fin) = fin sInner
------------------------------------------------------------------------------
-- Mapping on input
------------------------------------------------------------------------------
@ -1187,7 +1265,7 @@ concatMap f (Fold stepa initiala extracta) = Fold stepc initialc extractc
--
{-# INLINE lmap #-}
lmap :: (a -> b) -> Fold m b r -> Fold m a r
lmap f (Fold step begin done) = Fold step' begin done
lmap f (Fold step begin done final) = Fold step' begin done final
where
step' x a = step x (f a)
@ -1195,7 +1273,7 @@ lmap f (Fold step begin done) = Fold step' begin done
--
{-# INLINE lmapM #-}
lmapM :: Monad m => (a -> m b) -> Fold m b r -> Fold m a r
lmapM f (Fold step begin done) = Fold step' begin done
lmapM f (Fold step begin done final) = Fold step' begin done final
where
step' x a = f a >>= step x
@ -1207,8 +1285,10 @@ lmapM f (Fold step begin done) = Fold step' begin done
-- /Pre-release/
{-# INLINE postscan #-}
postscan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
postscan (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Fold step initial extract
postscan
(Fold stepL initialL extractL finalL)
(Fold stepR initialR extractR finalR) =
Fold step initial extract final
where
@ -1219,30 +1299,32 @@ postscan (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
Done bL -> do
rR <- stepR sR bL
case rR of
Partial sR1 -> Done <$> extractR sR1
Partial sR1 -> Done <$> finalR sR1
Done bR -> return $ Done bR
Partial sL -> do
!b <- extractL sL
rR <- stepR sR b
return
$ case rR of
Partial sR1 -> Partial (sL, sR1)
Done bR -> Done bR
case rR of
Partial sR1 -> return $ Partial (sL, sR1)
Done bR -> finalL sL >> return (Done bR)
initial = do
r <- initialR
rL <- initialL
case r of
Partial sR ->
rR <- initialR
case rR of
Partial sR -> do
rL <- initialL
case rL of
Done _ -> Done <$> extractR sR
Done _ -> Done <$> finalR sR
Partial sL -> return $ Partial (sL, sR)
Done b -> return $ Done b
-- XXX should use Tuple'
step (sL, sR) x = runStep (stepL sL x) sR
extract = extractR . snd
final (sL, sR) = finalL sL *> finalR sR
------------------------------------------------------------------------------
-- Filtering
------------------------------------------------------------------------------
@ -1255,7 +1337,7 @@ postscan (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
--
{-# INLINE_NORMAL catMaybes #-}
catMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
catMaybes (Fold step initial extract) = Fold step1 initial extract
catMaybes (Fold step initial extract final) = Fold step1 initial extract final
where
@ -1295,7 +1377,7 @@ filtering f = foldl' step Nothing
{-# INLINE filter #-}
filter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
-- filter p = scanMaybe (filtering p)
filter f (Fold step begin done) = Fold step' begin done
filter f (Fold step begin extract final) = Fold step' begin extract final
where
step' x a = if f a then step x a else return $ Partial x
@ -1306,7 +1388,7 @@ filter f (Fold step begin done) = Fold step' begin done
--
{-# INLINE filterM #-}
filterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
filterM f (Fold step begin done) = Fold step' begin done
filterM f (Fold step begin extract final) = Fold step' begin extract final
where
step' x a = do
use <- f a
@ -1395,7 +1477,7 @@ dropping n = foldt' step initial extract
{-# INLINE take #-}
take :: Monad m => Int -> Fold m a b -> Fold m a b
-- take n = scanMaybe (taking n)
take n (Fold fstep finitial fextract) = Fold step initial extract
take n (Fold fstep finitial fextract ffinal) = Fold step initial extract final
where
@ -1407,7 +1489,7 @@ take n (Fold fstep finitial fextract) = Fold step initial extract
s1 = Tuple'Fused i1 s
if i1 < n
then return $ Partial s1
else Done <$> fextract s
else Done <$> ffinal s
Done b -> return $ Done b
initial = finitial >>= next (-1)
@ -1416,6 +1498,8 @@ take n (Fold fstep finitial fextract) = Fold step initial extract
extract (Tuple'Fused _ r) = fextract r
final (Tuple'Fused _ r) = ffinal r
------------------------------------------------------------------------------
-- Nesting
------------------------------------------------------------------------------
@ -1433,8 +1517,8 @@ take n (Fold fstep finitial fextract) = Fold step initial extract
-- /Pre-release/
{-# INLINE duplicate #-}
duplicate :: Monad m => Fold m a b -> Fold m a (Fold m a b)
duplicate (Fold step1 initial1 extract1) =
Fold step initial (\s -> pure $ Fold step1 (pure $ Partial s) extract1)
duplicate (Fold step1 initial1 extract1 final1) =
Fold step initial extract final
where
@ -1442,6 +1526,11 @@ duplicate (Fold step1 initial1 extract1) =
step s a = second fromPure <$> step1 s a
-- Scanning may be problematic due to multiple finalizations.
extract = error "duplicate: scanning may be problematic"
final s = pure $ Fold step1 (pure $ Partial s) extract1 final1
-- If there were a finalize/flushing action in the stream type that would be
-- equivalent to running initialize in Fold. But we do not have a flushing
-- action in streams.
@ -1453,9 +1542,9 @@ duplicate (Fold step1 initial1 extract1) =
-- /Pre-release/
{-# INLINE reduce #-}
reduce :: Monad m => Fold m a b -> m (Fold m a b)
reduce (Fold step initial extract) = do
reduce (Fold step initial extract final) = do
i <- initial
return $ Fold step (return i) extract
return $ Fold step (return i) extract final
-- This is the dual of Stream @cons@.
@ -1465,7 +1554,8 @@ reduce (Fold step initial extract) = do
-- /Pre-release/
{-# INLINE snoclM #-}
snoclM :: Monad m => Fold m a b -> m a -> Fold m a b
snoclM (Fold fstep finitial fextract) action = Fold fstep initial fextract
snoclM (Fold fstep finitial fextract ffinal) action =
Fold fstep initial fextract ffinal
where
@ -1492,7 +1582,8 @@ snoclM (Fold fstep finitial fextract) action = Fold fstep initial fextract
{-# INLINE snocl #-}
snocl :: Monad m => Fold m a b -> a -> Fold m a b
-- snocl f = snoclM f . return
snocl (Fold fstep finitial fextract) a = Fold fstep initial fextract
snocl (Fold fstep finitial fextract ffinal) a =
Fold fstep initial fextract ffinal
where
@ -1512,12 +1603,12 @@ snocl (Fold fstep finitial fextract) a = Fold fstep initial fextract
-- /Pre-release/
{-# INLINE snocM #-}
snocM :: Monad m => Fold m a b -> m a -> m (Fold m a b)
snocM (Fold step initial extract) action = do
snocM (Fold step initial extract final) action = do
res <- initial
r <- case res of
Partial fs -> action >>= step fs
Done _ -> return res
return $ Fold step (return r) extract
return $ Fold step (return r) extract final
-- Definitions:
--
@ -1536,12 +1627,12 @@ snocM (Fold step initial extract) action = do
-- /Pre-release/
{-# INLINE snoc #-}
snoc :: Monad m => Fold m a b -> a -> m (Fold m a b)
snoc (Fold step initial extract) a = do
snoc (Fold step initial extract final) a = do
res <- initial
r <- case res of
Partial fs -> step fs a
Done _ -> return res
return $ Fold step (return r) extract
return $ Fold step (return r) extract final
-- | Append a singleton value to the fold.
--
@ -1569,7 +1660,7 @@ addOne = flip snoc
-- /Pre-release/
{-# INLINE extractM #-}
extractM :: Monad m => Fold m a b -> m b
extractM (Fold _ initial extract) = do
extractM (Fold _ initial extract _) = do
res <- initial
case res of
Partial fs -> extract fs
@ -1578,14 +1669,15 @@ extractM (Fold _ initial extract) = do
-- | Close a fold so that it does not accept any more input.
{-# INLINE close #-}
close :: Monad m => Fold m a b -> Fold m a b
close (Fold _ initial1 extract1) = Fold undefined initial undefined
close (Fold _ initial1 _ final1) =
Fold undefined initial undefined undefined
where
initial = do
res <- initial1
case res of
Partial s -> Done <$> extract1 s
Partial s -> Done <$> final1 s
Done b -> return $ Done b
-- Corresponds to the null check for streams.
@ -1595,7 +1687,7 @@ close (Fold _ initial1 extract1) = Fold undefined initial undefined
-- /Pre-release/
{-# INLINE isClosed #-}
isClosed :: Monad m => Fold m a b -> m Bool
isClosed (Fold _ initial _) = do
isClosed (Fold _ initial _ _) = do
res <- initial
return $ case res of
Partial _ -> False
@ -1629,8 +1721,10 @@ data ManyState s1 s2
--
{-# INLINE many #-}
many :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
many (Fold sstep sinitial sextract) (Fold cstep cinitial cextract) =
Fold step initial extract
many
(Fold sstep sinitial sextract sfinal)
(Fold cstep cinitial cextract cfinal) =
Fold step initial extract final
where
@ -1679,6 +1773,13 @@ many (Fold sstep sinitial sextract) (Fold cstep cinitial cextract) =
Partial s -> cextract s
Done b -> return b
final (ManyFirst ss cs) = sfinal ss *> cfinal cs
final (ManyLoop ss cs) = do
cres <- sfinal ss >>= cstep cs
case cres of
Partial s -> cfinal s
Done b -> return b
-- | Like many, but the "first" fold emits an output at the end even if no
-- input is received.
--
@ -1688,8 +1789,10 @@ many (Fold sstep sinitial sextract) (Fold cstep cinitial cextract) =
--
{-# INLINE manyPost #-}
manyPost :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
manyPost (Fold sstep sinitial sextract) (Fold cstep cinitial cextract) =
Fold step initial extract
manyPost
(Fold sstep sinitial sextract sfinal)
(Fold cstep cinitial cextract cfinal) =
Fold step initial extract final
where
@ -1725,6 +1828,12 @@ manyPost (Fold sstep sinitial sextract) (Fold cstep cinitial cextract) =
Partial s -> cextract s
Done b -> return b
final (Tuple' ss cs) = do
cres <- sfinal ss >>= cstep cs
case cres of
Partial s -> cfinal s
Done b -> return b
-- | @groupsOf n split collect@ repeatedly applies the @split@ fold to chunks
-- of @n@ items in the input stream and supplies the result to the @collect@
-- fold.
@ -1753,7 +1862,10 @@ groupsOf n split = many (take n split)
--
{-# INLINE refoldMany #-}
refoldMany :: Monad m => Fold m a b -> Refold m x b c -> Refold m x a c
refoldMany (Fold sstep sinitial sextract) (Refold cstep cinject cextract) =
refoldMany
(Fold sstep sinitial sextract _sfinal)
-- XXX We will need a "final" in refold as well
(Refold cstep cinject cextract) =
Refold step inject extract
where
@ -1804,7 +1916,9 @@ data ConsumeManyState x cs ss = ConsumeMany x cs (Either ss ss)
-- /Internal/
{-# INLINE refoldMany1 #-}
refoldMany1 :: Monad m => Refold m x a b -> Fold m b c -> Refold m x a c
refoldMany1 (Refold sstep sinject sextract) (Fold cstep cinitial cextract) =
refoldMany1
(Refold sstep sinject sextract)
(Fold cstep cinitial cextract _cfinal) =
Refold step inject extract
where
@ -1855,7 +1969,7 @@ refoldMany1 (Refold sstep sinject sextract) (Fold cstep cinitial cextract) =
{-# INLINE refold #-}
refold :: Monad m => Refold m b a c -> Fold m a b -> Fold m a c
refold (Refold step inject extract) f =
Fold step (extractM f >>= inject) extract
Fold step (extractM f >>= inject) extract extract
------------------------------------------------------------------------------
-- morphInner
@ -1865,8 +1979,8 @@ refold (Refold step inject extract) f =
--
-- /Pre-release/
morphInner :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
morphInner f (Fold step initial extract) =
Fold (\x a -> f $ step x a) (f initial) (f . extract)
morphInner f (Fold step initial extract final) =
Fold (\x a -> f $ step x a) (f initial) (f . extract) (f . final)
-- | Adapt a pure fold to any monad.
--

6
core/src/Streamly/Internal/Data/Fold/Window.hs
View File

@ -148,7 +148,7 @@ windowRollingMap f = Fold.foldl' f1 initial
--
{-# INLINE windowSumInt #-}
windowSumInt :: forall m a. (Monad m, Integral a) => Fold m (a, Maybe a) a
windowSumInt = Fold step initial extract
windowSumInt = Fold step initial extract extract
where
@ -182,7 +182,7 @@ windowSumInt = Fold step initial extract
--
{-# INLINE windowSum #-}
windowSum :: forall m a. (Monad m, Num a) => Fold m (a, Maybe a) a
windowSum = Fold step initial extract
windowSum = Fold step initial extract extract
where
@ -267,7 +267,7 @@ windowPowerSumFrac p = windowLmap (** p) windowSum
--
{-# INLINE windowRange #-}
windowRange :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (Maybe (a, a))
windowRange n = Fold step initial extract
windowRange n = Fold step initial extract extract
where

4
core/src/Streamly/Internal/Data/IsMap.hs
View File

@ -28,6 +28,8 @@ class IsMap f where
mapDelete :: Key f -> f a -> f a
mapUnion :: f a -> f a -> f a
mapNull :: f a -> Bool
mapTraverseWithKey ::
Applicative t => (Key f -> a -> t b) -> f a -> t (f b)
instance Ord k => IsMap (Map k) where
type Key (Map k) = k
@ -39,6 +41,7 @@ instance Ord k => IsMap (Map k) where
mapDelete = Map.delete
mapUnion = Map.union
mapNull = Map.null
mapTraverseWithKey = Map.traverseWithKey
instance IsMap IntMap.IntMap where
type Key IntMap.IntMap = Int
@ -50,3 +53,4 @@ instance IsMap IntMap.IntMap where
mapDelete = IntMap.delete
mapUnion = IntMap.union
mapNull = IntMap.null
mapTraverseWithKey = IntMap.traverseWithKey

2
core/src/Streamly/Internal/Data/MutArray/Generic.hs
View File

@ -598,7 +598,7 @@ arrayChunkSize = 1024
-- /Pre-release/
{-# INLINE_NORMAL writeNUnsafe #-}
writeNUnsafe :: MonadIO m => Int -> Fold m a (MutArray a)
writeNUnsafe n = Fold step initial return
writeNUnsafe n = Fold step initial return return
where

241
core/src/Streamly/Internal/Data/Parser.hs
View File

@ -21,8 +21,8 @@ module Streamly.Internal.Data.Parser
, Step (..)
, Initial (..)
-- * Downgrade to Fold
, toFold
-- -- * Downgrade to Fold
-- , toFold
-- First order parsers
-- * Accumulators
@ -270,6 +270,10 @@ import Prelude hiding
-- Downgrade a parser to a Fold
-------------------------------------------------------------------------------
-- XXX Parsers cannot be converted to folds, because they do not have a
-- scanning function. Can we move the applicative folds to parsers instead?
-- need to measure the performance.
{-
-- | Make a 'Fold' from a 'Parser'. The fold just throws an exception if the
-- parser fails or tries to backtrack.
--
@ -281,7 +285,7 @@ import Prelude hiding
--
{-# INLINE toFold #-}
toFold :: Monad m => Parser a m b -> Fold m a b
toFold (Parser pstep pinitial pextract) = Fold step initial extract
toFold (Parser pstep pinitial pextract) = Fold step initial extract final
where
@ -317,6 +321,7 @@ toFold (Parser pstep pinitial pextract) = Fold step initial extract
Continue n _ -> cerror n
Done n _ -> derror n
Error err -> eerror err
-}
-------------------------------------------------------------------------------
-- Upgrade folds to parses
@ -327,7 +332,7 @@ toFold (Parser pstep pinitial pextract) = Fold step initial extract
--
{-# INLINE fromFold #-}
fromFold :: Monad m => Fold m a b -> Parser a m b
fromFold (Fold fstep finitial fextract) = Parser step initial extract
fromFold (Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -345,7 +350,7 @@ fromFold (Fold fstep finitial fextract) = Parser step initial extract
FL.Partial s1 -> Partial 0 s1
FL.Done b -> Done 0 b
extract = fmap (Done 0) . fextract
extract = fmap (Done 0) . ffinal
-- | Convert a Maybe returning fold to an error returning parser. The first
-- argument is the error message that the parser would return when the fold
@ -355,7 +360,7 @@ fromFold (Fold fstep finitial fextract) = Parser step initial extract
--
{-# INLINE fromFoldMaybe #-}
fromFoldMaybe :: Monad m => String -> Fold m a (Maybe b) -> Parser a m b
fromFoldMaybe errMsg (Fold fstep finitial fextract) =
fromFoldMaybe errMsg (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -381,7 +386,7 @@ fromFoldMaybe errMsg (Fold fstep finitial fextract) =
Nothing -> Error errMsg
extract s = do
res <- fextract s
res <- ffinal s
case res of
Just x -> return $ Done 0 x
Nothing -> return $ Error errMsg
@ -630,7 +635,7 @@ data Tuple'Fused a b = Tuple'Fused !a !b deriving Show
--
{-# INLINE takeBetween #-}
takeBetween :: Monad m => Int -> Int -> Fold m a b -> Parser a m b
takeBetween low high (Fold fstep finitial fextract) =
takeBetween low high (Fold fstep finitial _ ffinal) =
Parser step initial (extract streamErr)
@ -685,7 +690,7 @@ takeBetween low high (Fold fstep finitial fextract) =
then return $ Continue 0 s1
else if i1 < high
then return $ Partial 0 s1
else fmap (Done 0) (fextract s)
else fmap (Done 0) (ffinal s)
FL.Done b ->
return
$ if i1 >= low
@ -695,12 +700,12 @@ takeBetween low high (Fold fstep finitial fextract) =
step (Tuple'Fused i s) a = fstep s a >>= snext i
extract f (Tuple'Fused i s)
| i >= low && i <= high = fmap (Done 0) (fextract s)
| i >= low && i <= high = fmap (Done 0) (ffinal s)
| otherwise = return $ Error (f i)
-- XXX Need to make Initial return type Step to deduplicate this
iextract f (Tuple'Fused i s)
| i >= low && i <= high = fmap IDone (fextract s)
| i >= low && i <= high = fmap IDone (ffinal s)
| otherwise = return $ IError (f i)
-- | Stops after taking exactly @n@ input elements.
@ -717,7 +722,7 @@ takeBetween low high (Fold fstep finitial fextract) =
--
{-# INLINE takeEQ #-}
takeEQ :: Monad m => Int -> Fold m a b -> Parser a m b
takeEQ n (Fold fstep finitial fextract) = Parser step initial extract
takeEQ n (Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -727,7 +732,7 @@ takeEQ n (Fold fstep finitial fextract) = Parser step initial extract
FL.Partial s ->
if n > 0
then return $ IPartial $ Tuple'Fused 1 s
else fmap IDone (fextract s)
else fmap IDone (ffinal s)
FL.Done b -> return $
if n > 0
then IError
@ -751,7 +756,7 @@ takeEQ n (Fold fstep finitial fextract) = Parser step initial extract
-- assert (n == i1)
Done 0
<$> case res of
FL.Partial s -> fextract s
FL.Partial s -> ffinal s
FL.Done b -> return b
extract (Tuple'Fused i _) =
@ -783,7 +788,7 @@ data TakeGEState s =
--
{-# INLINE takeGE #-}
takeGE :: Monad m => Int -> Fold m a b -> Parser a m b
takeGE n (Fold fstep finitial fextract) = Parser step initial extract
takeGE n (Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -831,7 +836,7 @@ takeGE n (Fold fstep finitial fextract) = Parser step initial extract
$ Error
$ "takeGE: Expecting at least " ++ show n
++ " elements, input terminated on " ++ show (i - 1)
extract (TakeGEGE r) = fmap (Done 0) $ fextract r
extract (TakeGEGE r) = fmap (Done 0) $ ffinal r
-------------------------------------------------------------------------------
-- Conditional splitting
@ -892,7 +897,7 @@ takeWhileP predicate (Parser pstep pinitial pextract) =
{-# INLINE takeWhile #-}
takeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Parser a m b
-- takeWhile cond f = takeWhileP cond (fromFold f)
takeWhile predicate (Fold fstep finitial fextract) =
takeWhile predicate (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -911,9 +916,9 @@ takeWhile predicate (Fold fstep finitial fextract) =
$ case fres of
FL.Partial s1 -> Partial 0 s1
FL.Done b -> Done 0 b
else Done 1 <$> fextract s
else Done 1 <$> ffinal s
extract s = fmap (Done 0) (fextract s)
extract s = fmap (Done 0) (ffinal s)
{-
-- XXX This may not be composable because of the b argument. We can instead
@ -933,7 +938,7 @@ takeWhile1 acc cond f = undefined
{-# INLINE takeWhile1 #-}
takeWhile1 :: Monad m => (a -> Bool) -> Fold m a b -> Parser a m b
-- takeWhile1 cond f = takeWhileP cond (takeBetween 1 maxBound f)
takeWhile1 predicate (Fold fstep finitial fextract) =
takeWhile1 predicate (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -963,11 +968,11 @@ takeWhile1 predicate (Fold fstep finitial fextract) =
if predicate a
then process s a
else do
b <- fextract s
b <- ffinal s
return $ Done 1 b
extract (Left' _) = return $ Error "takeWhile1: end of input"
extract (Right' s) = fmap (Done 0) (fextract s)
extract (Right' s) = fmap (Done 0) (ffinal s)
-- | Drain the input as long as the predicate succeeds, running the effects and
-- discarding the results.
@ -996,7 +1001,7 @@ takeFramedByGeneric :: Monad m =>
-> Maybe (a -> Bool) -- is frame end?
-> Fold m a b
-> Parser a m b
takeFramedByGeneric esc begin end (Fold fstep finitial fextract) =
takeFramedByGeneric esc begin end (Fold fstep finitial _ ffinal) =
Parser step initial extract
@ -1028,20 +1033,20 @@ takeFramedByGeneric esc begin end (Fold fstep finitial fextract) =
if isEnd a
then
if n == 0
then Done 0 <$> fextract s
then Done 0 <$> ffinal s
else process s a (n - 1)
else
let n1 = if isBegin a then n + 1 else n
in process s a n1
Nothing -> -- takeEndBy case
if isEnd a
then Done 0 <$> fextract s
then Done 0 <$> ffinal s
else process s a n
Nothing -> -- takeStartBy case
case begin of
Just isBegin ->
if isBegin a
then Done 0 <$> fextract s
then Done 0 <$> ffinal s
else process s a n
Nothing ->
error $ "takeFramedByGeneric: "
@ -1066,7 +1071,7 @@ takeFramedByGeneric esc begin end (Fold fstep finitial fextract) =
case end of
Just isEnd ->
if isEnd a
then Done 0 <$> fextract s
then Done 0 <$> ffinal s
else processCheckEsc s a 0
Nothing ->
error "Both begin and end frame predicate missing"
@ -1081,7 +1086,7 @@ takeFramedByGeneric esc begin end (Fold fstep finitial fextract) =
case begin of
Just _ ->
case end of
Nothing -> fmap (Done 0) $ fextract s
Nothing -> fmap (Done 0) $ ffinal s
Just _ -> err "takeFramedByGeneric: missing frame end"
Nothing -> err "takeFramedByGeneric: missing closing frame"
extract (FrameEscEsc _ _) = err "takeFramedByGeneric: trailing escape"
@ -1120,7 +1125,7 @@ blockWithQuotes :: (Monad m, Eq a) =>
-> Fold m a b
-> Parser a m b
blockWithQuotes isEsc isQuote bopen bclose
(Fold fstep finitial fextract) =
(Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -1150,7 +1155,7 @@ blockWithQuotes isEsc isQuote bopen bclose
| a == bopen = process s a (BlockUnquoted (level + 1))
| a == bclose =
if level == 1
then fmap (Done 0) (fextract s)
then fmap (Done 0) (ffinal s)
else process s a (BlockUnquoted (level - 1))
| isQuote a = process s a (BlockQuoted level)
| otherwise = process s a (BlockUnquoted level)
@ -1164,7 +1169,7 @@ blockWithQuotes isEsc isQuote bopen bclose
err = return . Error
extract (BlockInit s) = fmap (Done 0) $ fextract s
extract (BlockInit s) = fmap (Done 0) $ ffinal s
extract (BlockUnquoted level _) =
err $ "blockWithQuotes: finished at block nest level " ++ show level
extract (BlockQuoted level _) =
@ -1299,7 +1304,7 @@ takeEitherSepBy _cond = undefined -- D.toParserK . D.takeEitherSepBy cond
--
{-# INLINE takeStartBy #-}
takeStartBy :: Monad m => (a -> Bool) -> Fold m a b -> Parser a m b
takeStartBy cond (Fold fstep finitial fextract) =
takeStartBy cond (Fold fstep finitial _ ffinal) =
Parser step initial extract
@ -1327,10 +1332,10 @@ takeStartBy cond (Fold fstep finitial fextract) =
step (Right' s) a =
if not (cond a)
then process s a
else Done 1 <$> fextract s
else Done 1 <$> ffinal s
extract (Left' s) = fmap (Done 0) $ fextract s
extract (Right' s) = fmap (Done 0) $ fextract s
extract (Left' s) = fmap (Done 0) $ ffinal s
extract (Right' s) = fmap (Done 0) $ ffinal s
-- | Like 'takeStartBy' but drops the separator.
--
@ -1366,7 +1371,7 @@ takeFramedByEsc_ :: Monad m =>
(a -> Bool) -> (a -> Bool) -> (a -> Bool) -> Fold m a b -> Parser a m b
-- takeFramedByEsc_ isEsc isEnd p =
-- takeFramedByGeneric (Just isEsc) Nothing (Just isEnd) (toFold p)
takeFramedByEsc_ isEsc isBegin isEnd (Fold fstep finitial fextract) =
takeFramedByEsc_ isEsc isBegin isEnd (Fold fstep finitial _ ffinal ) =
Parser step initial extract
@ -1402,7 +1407,7 @@ takeFramedByEsc_ isEsc isBegin isEnd (Fold fstep finitial fextract) =
in process s a n1
else
if n == 0
then Done 0 <$> fextract s
then Done 0 <$> ffinal s
else process s a (n - 1)
step (FrameEscEsc s n) a = process s a n
@ -1424,7 +1429,7 @@ takeFramedBy_ :: Monad m =>
(a -> Bool) -> (a -> Bool) -> Fold m a b -> Parser a m b
-- takeFramedBy_ isBegin isEnd =
-- takeFramedByGeneric (Just (const False)) (Just isBegin) (Just isEnd)
takeFramedBy_ isBegin isEnd (Fold fstep finitial fextract) =
takeFramedBy_ isBegin isEnd (Fold fstep finitial _ ffinal) =
Parser step initial extract
@ -1454,7 +1459,7 @@ takeFramedBy_ isBegin isEnd (Fold fstep finitial fextract) =
| not (isEnd a) =
let n1 = if isBegin a then n + 1 else n
in process s a n1
| n == 0 = Done 0 <$> fextract s
| n == 0 = Done 0 <$> ffinal s
| otherwise = process s a (n - 1)
err = return . Error
@ -1489,7 +1494,7 @@ data WordByState s b = WBLeft !s | WBWord !s | WBRight !b
--
{-# INLINE wordBy #-}
wordBy :: Monad m => (a -> Bool) -> Fold m a b -> Parser a m b
wordBy predicate (Fold fstep finitial fextract) = Parser step initial extract
wordBy predicate (Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -1516,7 +1521,7 @@ wordBy predicate (Fold fstep finitial fextract) = Parser step initial extract
if not (predicate a)
then worder s a
else do
b <- fextract s
b <- ffinal s
return $ Partial 0 $ WBRight b
step (WBRight b) a =
return
@ -1524,8 +1529,8 @@ wordBy predicate (Fold fstep finitial fextract) = Parser step initial extract
then Done 1 b
else Partial 0 $ WBRight b
extract (WBLeft s) = fmap (Done 0) $ fextract s
extract (WBWord s) = fmap (Done 0) $ fextract s
extract (WBLeft s) = fmap (Done 0) $ ffinal s
extract (WBWord s) = fmap (Done 0) $ ffinal s
extract (WBRight b) = return (Done 0 b)
data WordFramedState s b =
@ -1562,7 +1567,7 @@ wordFramedBy :: Monad m =>
-> Fold m a b
-> Parser a m b
wordFramedBy isEsc isBegin isEnd isSep
(Fold fstep finitial fextract) =
(Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -1593,7 +1598,7 @@ wordFramedBy isEsc isBegin isEnd isSep
step (WordFramedWord s n) a
| isEsc a = return $ Continue 0 $ WordFramedEsc s n
| n == 0 && isSep a = do
b <- fextract s
b <- ffinal s
return $ Partial 0 $ WordFramedSkipPost b
| otherwise = do
-- We need to use different order for checking begin and end for
@ -1626,10 +1631,10 @@ wordFramedBy isEsc isBegin isEnd isSep
err = return . Error
extract (WordFramedSkipPre s) = fmap (Done 0) $ fextract s
extract (WordFramedSkipPre s) = fmap (Done 0) $ ffinal s
extract (WordFramedWord s n) =
if n == 0
then fmap (Done 0) $ fextract s
then fmap (Done 0) $ ffinal s
else err "wordFramedBy: missing frame end"
extract (WordFramedEsc _ _) =
err "wordFramedBy: trailing escape"
@ -1716,7 +1721,7 @@ wordWithQuotes :: (Monad m, Eq a) =>
-> Fold m a b
-> Parser a m b
wordWithQuotes keepQuotes tr escChar toRight isSep
(Fold fstep finitial fextract) =
(Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -1767,7 +1772,7 @@ wordWithQuotes keepQuotes tr escChar toRight isSep
step (WordUnquotedWord s) a
| isEsc a = return $ Continue 0 $ WordUnquotedEsc s
| isSep a = do
b <- fextract s
b <- ffinal s
return $ Partial 0 $ WordQuotedSkipPost b
| otherwise = do
case toRight a of
@ -1815,11 +1820,11 @@ wordWithQuotes keepQuotes tr escChar toRight isSep
err = return . Error
extract (WordQuotedSkipPre s) = fmap (Done 0) $ fextract s
extract (WordUnquotedWord s) = fmap (Done 0) $ fextract s
extract (WordQuotedSkipPre s) = fmap (Done 0) $ ffinal s
extract (WordUnquotedWord s) = fmap (Done 0) $ ffinal s
extract (WordQuotedWord s n _ _) =
if n == 0
then fmap (Done 0) $ fextract s
then fmap (Done 0) $ ffinal s
else err "wordWithQuotes: missing frame end"
extract WordQuotedEsc {} =
err "wordWithQuotes: trailing escape"
@ -1897,7 +1902,7 @@ data GroupByState a s
--
{-# INLINE groupBy #-}
groupBy :: Monad m => (a -> a -> Bool) -> Fold m a b -> Parser a m b
groupBy eq (Fold fstep finitial fextract) = Parser step initial extract
groupBy eq (Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -1920,10 +1925,10 @@ groupBy eq (Fold fstep finitial fextract) = Parser step initial extract
step (GroupByGrouping a0 s) a =
if eq a0 a
then grouper s a0 a
else Done 1 <$> fextract s
else Done 1 <$> ffinal s
extract (GroupByInit s) = fmap (Done 0) $ fextract s
extract (GroupByGrouping _ s) = fmap (Done 0) $ fextract s
extract (GroupByInit s) = fmap (Done 0) $ ffinal s
extract (GroupByGrouping _ s) = fmap (Done 0) $ ffinal s
-- | Unlike 'groupBy' this combinator performs a rolling comparison of two
-- successive elements in the input stream. Assuming the input stream
@ -1957,7 +1962,7 @@ groupBy eq (Fold fstep finitial fextract) = Parser step initial extract
--
{-# INLINE groupByRolling #-}
groupByRolling :: Monad m => (a -> a -> Bool) -> Fold m a b -> Parser a m b
groupByRolling eq (Fold fstep finitial fextract) = Parser step initial extract
groupByRolling eq (Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -1980,10 +1985,10 @@ groupByRolling eq (Fold fstep finitial fextract) = Parser step initial extract
step (GroupByGrouping a0 s) a =
if eq a0 a
then grouper s a
else Done 1 <$> fextract s
else Done 1 <$> ffinal s
extract (GroupByInit s) = fmap (Done 0) $ fextract s
extract (GroupByGrouping _ s) = fmap (Done 0) $ fextract s
extract (GroupByInit s) = fmap (Done 0) $ ffinal s
extract (GroupByGrouping _ s) = fmap (Done 0) $ ffinal s
{-# ANN type GroupByStatePair Fuse #-}
data GroupByStatePair a s1 s2
@ -2007,8 +2012,8 @@ groupByRollingEither :: Monad m =>
(a -> a -> Bool) -> Fold m a b -> Fold m a c -> Parser a m (Either b c)
groupByRollingEither
eq
(Fold fstep1 finitial1 fextract1)
(Fold fstep2 finitial2 fextract2) = Parser step initial extract
(Fold fstep1 finitial1 _ ffinal1)
(Fold fstep2 finitial2 _ ffinal2) = Parser step initial extract
where
@ -2067,21 +2072,21 @@ groupByRollingEither
step (GroupByGroupingPairL a0 s1 s2) a =
if not (eq a0 a)
then grouperL2 s1 s2 a
else Done 1 . Left <$> fextract1 s1
else Done 1 . Left <$> ffinal1 s1
step (GroupByGroupingPairR a0 s1 s2) a =
if eq a0 a
then grouperR2 s1 s2 a
else Done 1 . Right <$> fextract2 s2
else Done 1 . Right <$> ffinal2 s2
extract (GroupByInitPair s1 _) = Done 0 . Left <$> fextract1 s1
extract (GroupByGroupingPairL _ s1 _) = Done 0 . Left <$> fextract1 s1
extract (GroupByGroupingPairR _ _ s2) = Done 0 . Right <$> fextract2 s2
extract (GroupByInitPair s1 _) = Done 0 . Left <$> ffinal1 s1
extract (GroupByGroupingPairL _ s1 _) = Done 0 . Left <$> ffinal1 s1
extract (GroupByGroupingPairR _ _ s2) = Done 0 . Right <$> ffinal2 s2
extract (GroupByGroupingPair a s1 _) = do
res <- fstep1 s1 a
case res of
FL.Done b -> return $ Done 0 (Left b)
FL.Partial s11 -> Done 0 . Left <$> fextract1 s11
FL.Partial s11 -> Done 0 . Left <$> ffinal1 s11
-- XXX use an Unfold instead of a list?
-- XXX custom combinators for matching list, array and stream?
@ -2234,7 +2239,7 @@ postscan = undefined
{-# INLINE zipWithM #-}
zipWithM :: Monad m =>
(a -> b -> m c) -> D.Stream m a -> Fold m c x -> Parser b m x
zipWithM zf (D.Stream sstep state) (Fold fstep finitial fextract) =
zipWithM zf (D.Stream sstep state) (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -2247,7 +2252,7 @@ zipWithM zf (D.Stream sstep state) (Fold fstep finitial fextract) =
case r of
D.Yield x s -> return $ IPartial (Just' x, s, fs)
D.Stop -> do
x <- fextract fs
x <- ffinal fs
return $ IDone x
-- Need Skip/Continue in initial to loop right here
D.Skip s -> return $ IPartial (Nothing', s, fs)
@ -2262,7 +2267,7 @@ zipWithM zf (D.Stream sstep state) (Fold fstep finitial fextract) =
case r of
D.Yield x1 s -> return $ Continue 0 (Just' x1, s, fs1)
D.Stop -> do
x <- fextract fs1
x <- ffinal fs1
return $ Done 0 x
D.Skip s -> return $ Continue 1 (Nothing', s, fs1)
FL.Done x -> return $ Done 0 x
@ -2277,7 +2282,7 @@ zipWithM zf (D.Stream sstep state) (Fold fstep finitial fextract) =
return $ Continue 0 (Nothing', s, fs1)
FL.Done x -> return $ Done 0 x
D.Stop -> do
x <- fextract fs
x <- ffinal fs
return $ Done 1 x
D.Skip s -> return $ Continue 1 (Nothing', s, fs)
@ -2561,7 +2566,7 @@ deintercalateAll :: Monad m =>
deintercalateAll
(Parser stepL initialL extractL)
(Parser stepR initialR _)
(Fold fstep finitial fextract) = Parser step initial extract
(Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -2632,9 +2637,9 @@ deintercalateAll
extractResult n fs r = do
res <- fstep fs r
case res of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
extract (DeintercalateAllInitL fs) = fmap (Done 0) $ fextract fs
extract (DeintercalateAllInitL fs) = fmap (Done 0) $ ffinal fs
extract (DeintercalateAllL fs sL) = do
r <- extractL sL
case r of
@ -2642,7 +2647,7 @@ deintercalateAll
Error err -> return $ Error err
Continue n s -> return $ Continue n (DeintercalateAllL fs s)
Partial _ _ -> error "Partial in extract"
extract (DeintercalateAllInitR fs) = fmap (Done 0) $ fextract fs
extract (DeintercalateAllInitR fs) = fmap (Done 0) $ ffinal fs
extract (DeintercalateAllR _ _) =
return $ Error "deintercalateAll: input ended at 'Right' value"
@ -2685,7 +2690,7 @@ deintercalate :: Monad m =>
deintercalate
(Parser stepL initialL extractL)
(Parser stepR initialR _)
(Fold fstep finitial fextract) = Parser step initial extract
(Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -2716,7 +2721,7 @@ deintercalate
Done n b ->
processL (fstep fs (Left b)) n DeintercalateInitR
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
{-# INLINE processR #-}
@ -2736,7 +2741,7 @@ deintercalate
Continue n s -> return $ Continue n (DeintercalateR (cnt1 - n) fs s)
Done n b -> processR (cnt1 - n) b fs n
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
step (DeintercalateInitL fs) a = do
@ -2771,17 +2776,17 @@ deintercalate
-- XXX We could have the fold accept pairs of (bR, bL)
FL.Done _ -> error "Fold terminated consuming partial input"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
{-# INLINE extractResult #-}
extractResult n fs r = do
res <- fstep fs r
case res of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
extract (DeintercalateInitL fs) = fmap (Done 0) $ fextract fs
extract (DeintercalateInitL fs) = fmap (Done 0) $ ffinal fs
extract (DeintercalateL cnt fs sL) = do
r <- extractL sL
case r of
@ -2789,10 +2794,10 @@ deintercalate
Continue n s -> return $ Continue n (DeintercalateL (cnt - n) fs s)
Partial _ _ -> error "Partial in extract"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt xs
extract (DeintercalateInitR fs) = fmap (Done 0) $ fextract fs
extract (DeintercalateR cnt fs _) = fmap (Done cnt) $ fextract fs
extract (DeintercalateInitR fs) = fmap (Done 0) $ ffinal fs
extract (DeintercalateR cnt fs _) = fmap (Done cnt) $ ffinal fs
extract (DeintercalateRL cnt bR fs sL) = do
r <- extractL sL
case r of
@ -2804,7 +2809,7 @@ deintercalate
Continue n s -> return $ Continue n (DeintercalateRL (cnt - n) bR fs s)
Partial _ _ -> error "Partial in extract"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt xs
{-# ANN type Deintercalate1State Fuse #-}
@ -2841,7 +2846,7 @@ deintercalate1 :: Monad m =>
deintercalate1
(Parser stepL initialL extractL)
(Parser stepR initialR _)
(Fold fstep finitial fextract) = Parser step initial extract
(Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -2895,7 +2900,7 @@ deintercalate1
Continue n s -> return $ Continue n (Deintercalate1R (cnt1 - n) fs s)
Done n b -> processR (cnt1 - n) b fs n
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
step (Deintercalate1InitL cnt fs sL) a = runStepInitL cnt fs sL a
@ -2924,14 +2929,14 @@ deintercalate1
-- XXX We could have the fold accept pairs of (bR, bL)
FL.Done _ -> error "Fold terminated consuming partial input"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
{-# INLINE extractResult #-}
extractResult n fs r = do
res <- fstep fs r
case res of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
extract (Deintercalate1InitL cnt fs sL) = do
@ -2941,8 +2946,8 @@ deintercalate1
Continue n s -> return $ Continue n (Deintercalate1InitL (cnt - n) fs s)
Partial _ _ -> error "Partial in extract"
Error err -> return $ Error err
extract (Deintercalate1InitR fs) = fmap (Done 0) $ fextract fs
extract (Deintercalate1R cnt fs _) = fmap (Done cnt) $ fextract fs
extract (Deintercalate1InitR fs) = fmap (Done 0) $ ffinal fs
extract (Deintercalate1R cnt fs _) = fmap (Done cnt) $ ffinal fs
extract (Deintercalate1RL cnt bR fs sL) = do
r <- extractL sL
case r of
@ -2954,7 +2959,7 @@ deintercalate1
Continue n s -> return $ Continue n (Deintercalate1RL (cnt - n) bR fs s)
Partial _ _ -> error "Partial in extract"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt xs
{-# ANN type SepByState Fuse #-}
@ -2997,7 +3002,7 @@ sepBy :: Monad m =>
sepBy
(Parser stepL initialL extractL)
(Parser stepR initialR _)
(Fold fstep finitial fextract) = Parser step initial extract
(Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -3028,7 +3033,7 @@ sepBy
Done n b ->
processL (fstep fs b) n SepByInitR
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
{-# INLINE processR #-}
@ -3048,7 +3053,7 @@ sepBy
Continue n s -> return $ Continue n (SepByR (cnt1 - n) fs s)
Done n _ -> processR (cnt1 - n) fs n
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
step (SepByInitL fs) a = do
@ -3070,10 +3075,10 @@ sepBy
extractResult n fs r = do
res <- fstep fs r
case res of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
extract (SepByInitL fs) = fmap (Done 0) $ fextract fs
extract (SepByInitL fs) = fmap (Done 0) $ ffinal fs
extract (SepByL cnt fs sL) = do
r <- extractL sL
case r of
@ -3081,10 +3086,10 @@ sepBy
Continue n s -> return $ Continue n (SepByL (cnt - n) fs s)
Partial _ _ -> error "Partial in extract"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt xs
extract (SepByInitR fs) = fmap (Done 0) $ fextract fs
extract (SepByR cnt fs _) = fmap (Done cnt) $ fextract fs
extract (SepByInitR fs) = fmap (Done 0) $ ffinal fs
extract (SepByR cnt fs _) = fmap (Done cnt) $ ffinal fs
-- | Non-backtracking version of sepBy. Several times faster.
{-# INLINE sepByAll #-}
@ -3139,7 +3144,7 @@ sepBy1 :: Monad m =>
sepBy1
(Parser stepL initialL extractL)
(Parser stepR initialR _)
(Fold fstep finitial fextract) = Parser step initial extract
(Fold fstep finitial _ ffinal) = Parser step initial extract
where
@ -3186,7 +3191,7 @@ sepBy1
Done n b ->
processL (fstep fs b) n SepBy1InitR
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
{-# INLINE processR #-}
@ -3206,7 +3211,7 @@ sepBy1
Continue n s -> return $ Continue n (SepBy1R (cnt1 - n) fs s)
Done n _ -> processR (cnt1 - n) fs n
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
step (SepBy1InitL cnt fs sL) a = runStepInitL cnt fs sL a
@ -3223,7 +3228,7 @@ sepBy1
extractResult n fs r = do
res <- fstep fs r
case res of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
extract (SepBy1InitL cnt fs sL) = do
@ -3240,10 +3245,10 @@ sepBy1
Continue n s -> return $ Continue n (SepBy1L (cnt - n) fs s)
Partial _ _ -> error "Partial in extract"
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt xs
extract (SepBy1InitR fs) = fmap (Done 0) $ fextract fs
extract (SepBy1R cnt fs _) = fmap (Done cnt) $ fextract fs
extract (SepBy1InitR fs) = fmap (Done 0) $ ffinal fs
extract (SepBy1R cnt fs _) = fmap (Done cnt) $ ffinal fs
-------------------------------------------------------------------------------
-- Interleaving a collection of parsers
@ -3273,7 +3278,7 @@ roundRobin _ps _f = undefined
{-# INLINE sequence #-}
sequence :: Monad m =>
D.Stream m (Parser a m b) -> Fold m b c -> Parser a m c
sequence (D.Stream sstep sstate) (Fold fstep finitial fextract) =
sequence (D.Stream sstep sstate) (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -3291,7 +3296,7 @@ sequence (D.Stream sstep sstate) (Fold fstep finitial fextract) =
case sres of
D.Yield p ss1 -> return $ Continue 1 (Just' p, ss1, fs)
D.Stop -> do
c <- fextract fs
c <- ffinal fs
return $ Done 1 c
D.Skip ss1 -> return $ Continue 1 (Nothing', ss1, fs)
@ -3327,7 +3332,7 @@ sequence (D.Stream sstep sstate) (Fold fstep finitial fextract) =
FL.Done c -> return $ Done 1 c
IError err -> return $ Error err
extract (Nothing', _, fs) = fmap (Done 0) $ fextract fs
extract (Nothing', _, fs) = fmap (Done 0) $ ffinal fs
extract (Just' (Parser pstep pinit pextr), ss, fs) = do
ps <- pinit
case ps of
@ -3337,7 +3342,7 @@ sequence (D.Stream sstep sstate) (Fold fstep finitial fextract) =
Done n b -> do
res <- fstep fs b
case res of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
Error err -> return $ Error err
Continue n s -> return $ Continue n (Just' (Parser pstep (return (IPartial s)) pextr), ss, fs)
@ -3345,7 +3350,7 @@ sequence (D.Stream sstep sstate) (Fold fstep finitial fextract) =
IDone b -> do
fres <- fstep fs b
case fres of
FL.Partial fs1 -> fmap (Done 0) $ fextract fs1
FL.Partial fs1 -> fmap (Done 0) $ ffinal fs1
FL.Done c -> return (Done 0 c)
IError err -> return $ Error err
@ -3480,7 +3485,7 @@ manyTill :: Monad m
=> Parser a m b -> Parser a m x -> Fold m b c -> Parser a m c
manyTill (Parser stepL initialL extractL)
(Parser stepR initialR _)
(Fold fstep finitial fextract) =
(Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -3502,7 +3507,7 @@ manyTill (Parser stepL initialL extractL)
resR <- initialR
case resR of
IPartial sr -> return $ p (ManyTillR 0 fs sr)
IDone _ -> d <$> fextract fs
IDone _ -> d <$> ffinal fs
IError _ -> scrutL fs p c d e
initial = do
@ -3519,7 +3524,7 @@ manyTill (Parser stepL initialL extractL)
assertM(cnt + 1 - n >= 0)
return $ Continue n (ManyTillR (cnt + 1 - n) fs s)
Done n _ -> do
b <- fextract fs
b <- ffinal fs
return $ Done n b
Error _ -> do
resL <- initialL
@ -3558,12 +3563,12 @@ manyTill (Parser stepL initialL extractL)
Done n b -> do
r <- fstep fs b
case r of
FL.Partial fs1 -> fmap (Done n) $ fextract fs1
FL.Partial fs1 -> fmap (Done n) $ ffinal fs1
FL.Done c -> return (Done n c)
Error err -> return $ Error err
Continue n s -> return $ Continue n (ManyTillL fs s)
Partial _ _ -> error "Partial in extract"
extract (ManyTillR _ fs _) = fmap (Done 0) $ fextract fs
extract (ManyTillR _ fs _) = fmap (Done 0) $ ffinal fs
-- | @manyThen f collect recover@ repeats the parser @collect@ on the input and
-- collects the output in the supplied fold. If the the parser @collect@ fails,

34
core/src/Streamly/Internal/Data/Parser/Type.hs
View File

@ -927,7 +927,7 @@ data Fused3 a b c = Fused3 !a !b !c
--
{-# INLINE splitMany #-}
splitMany :: Monad m => Parser a m b -> Fold m b c -> Parser a m c
splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
splitMany (Parser step1 initial1 extract1) (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -943,7 +943,7 @@ splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
IPartial ps -> return $ partial $ Fused3 ps 0 fs
IDone pb ->
runCollectorWith (handleCollect partial done) fs pb
IError _ -> done <$> fextract fs
IError _ -> done <$> ffinal fs
FL.Done fb -> return $ done fb
runCollectorWith cont fs pb = fstep fs pb >>= cont
@ -967,19 +967,19 @@ splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
assertM(cnt1 - n >= 0)
fstep fs b >>= handleCollect (Partial n) (Done n)
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt xs
extract (Fused3 _ 0 fs) = fmap (Done 0) (fextract fs)
extract (Fused3 _ 0 fs) = fmap (Done 0) (ffinal fs)
extract (Fused3 s cnt fs) = do
r <- extract1 s
case r of
Error _ -> fmap (Done cnt) (fextract fs)
Error _ -> fmap (Done cnt) (ffinal fs)
Done n b -> do
assertM(n <= cnt)
fs1 <- fstep fs b
case fs1 of
FL.Partial s1 -> fmap (Done n) (fextract s1)
FL.Partial s1 -> fmap (Done n) (ffinal s1)
FL.Done b1 -> return (Done n b1)
Partial _ _ -> error "splitMany: Partial in extract"
Continue n s1 -> do
@ -993,7 +993,7 @@ splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
--
{-# INLINE splitManyPost #-}
splitManyPost :: Monad m => Parser a m b -> Fold m b c -> Parser a m c
splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -1009,7 +1009,7 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
IPartial ps -> return $ partial $ Fused3 ps 0 fs
IDone pb ->
runCollectorWith (handleCollect partial done) fs pb
IError _ -> done <$> fextract fs
IError _ -> done <$> ffinal fs
FL.Done fb -> return $ done fb
runCollectorWith cont fs pb = fstep fs pb >>= cont
@ -1031,18 +1031,18 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
assertM(cnt1 - n >= 0)
fstep fs b >>= handleCollect (Partial n) (Done n)
Error _ -> do
xs <- fextract fs
xs <- ffinal fs
return $ Done cnt1 xs
extract (Fused3 s cnt fs) = do
r <- extract1 s
case r of
Error _ -> fmap (Done cnt) (fextract fs)
Error _ -> fmap (Done cnt) (ffinal fs)
Done n b -> do
assertM(n <= cnt)
fs1 <- fstep fs b
case fs1 of
FL.Partial s1 -> fmap (Done n) (fextract s1)
FL.Partial s1 -> fmap (Done n) (ffinal s1)
FL.Done b1 -> return (Done n b1)
Partial _ _ -> error "splitMany: Partial in extract"
Continue n s1 -> do
@ -1055,7 +1055,7 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
--
{-# INLINE splitSome #-}
splitSome :: Monad m => Parser a m b -> Fold m b c -> Parser a m c
splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
splitSome (Parser step1 initial1 extract1) (Fold fstep finitial _ ffinal) =
Parser step initial extract
where
@ -1071,7 +1071,7 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
IPartial ps -> return $ partial $ Fused3 ps 0 $ Right fs
IDone pb ->
runCollectorWith (handleCollect partial done) fs pb
IError _ -> done <$> fextract fs
IError _ -> done <$> ffinal fs
FL.Done fb -> return $ done fb
runCollectorWith cont fs pb = fstep fs pb >>= cont
@ -1121,7 +1121,7 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
Done n b -> do
assertM(cnt1 - n >= 0)
fstep fs b >>= handleCollect (Partial n) (Done n)
Error _ -> Done cnt1 <$> fextract fs
Error _ -> Done cnt1 <$> ffinal fs
extract (Fused3 s cnt (Left fs)) = do
r <- extract1 s
@ -1131,7 +1131,7 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
assertM(n <= cnt)
fs1 <- fstep fs b
case fs1 of
FL.Partial s1 -> fmap (Done n) (fextract s1)
FL.Partial s1 -> fmap (Done n) (ffinal s1)
FL.Done b1 -> return (Done n b1)
Partial _ _ -> error "splitSome: Partial in extract"
Continue n s1 -> do
@ -1140,12 +1140,12 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
extract (Fused3 s cnt (Right fs)) = do
r <- extract1 s
case r of
Error _ -> fmap (Done cnt) (fextract fs)
Error _ -> fmap (Done cnt) (ffinal fs)
Done n b -> do
assertM(n <= cnt)
fs1 <- fstep fs b
case fs1 of
FL.Partial s1 -> fmap (Done n) (fextract s1)
FL.Partial s1 -> fmap (Done n) (ffinal s1)
FL.Done b1 -> return (Done n b1)
Partial _ _ -> error "splitSome: Partial in extract"
Continue n s1 -> do

5
core/src/Streamly/Internal/Data/Ring.hs
View File

@ -558,7 +558,8 @@ data Tuple4' a b c d = Tuple4' !a !b !c !d deriving Show
{-# INLINE slidingWindowWith #-}
slidingWindowWith :: forall m a b. (MonadIO m, Storable a, Unbox a)
=> Int -> Fold m ((a, Maybe a), m (MutArray a)) b -> Fold m a b
slidingWindowWith n (Fold step1 initial1 extract1) = Fold step initial extract
slidingWindowWith n (Fold step1 initial1 extract1 final1) =
Fold step initial extract final
where
@ -601,6 +602,8 @@ slidingWindowWith n (Fold step1 initial1 extract1) = Fold step initial extract
extract (Tuple4' _ _ _ st) = extract1 st
final (Tuple4' _ _ _ st) = final1 st
-- | @slidingWindow collector@ is an incremental sliding window
-- fold that does not require all the intermediate elements in a computation.
-- This maintains @n@ elements in the window, when a new element comes it slides

2
core/src/Streamly/Internal/Data/Ring/Generic.hs
View File

@ -69,7 +69,7 @@ createRing count = liftIO $ do
-- the Ring.
{-# INLINE writeLastN #-}
writeLastN :: MonadIO m => Int -> Fold m a (Ring a)
writeLastN n = Fold step initial extract
writeLastN n = Fold step initial extract extract
where

8
core/src/Streamly/Internal/Data/Stream/Chunked.hs
View File

@ -327,7 +327,7 @@ splitOnSuffix byte = D.splitInnerBySuffix (A.breakOn byte) A.splice
{-# INLINE_NORMAL foldBreakD #-}
foldBreakD :: forall m a b. (MonadIO m, Unbox a) =>
Fold m a b -> D.Stream m (Array a) -> m (b, D.Stream m (Array a))
foldBreakD (FL.Fold fstep initial extract) stream@(D.Stream step state) = do
foldBreakD (FL.Fold fstep initial _ final) stream@(D.Stream step state) = do
res <- initial
case res of
FL.Partial fs -> go SPEC state fs
@ -344,7 +344,7 @@ foldBreakD (FL.Fold fstep initial extract) stream@(D.Stream step state) = do
in goArray SPEC s fp start fs
D.Skip s -> go SPEC s fs
D.Stop -> do
b <- extract fs
b <- final fs
return (b, D.nil)
goArray !_ s (Tuple' end _) !cur !fs
@ -363,7 +363,7 @@ foldBreakD (FL.Fold fstep initial extract) stream@(D.Stream step state) = do
{-# INLINE_NORMAL foldBreakK #-}
foldBreakK :: forall m a b. (MonadIO m, Unbox a) =>
Fold m a b -> K.StreamK m (Array a) -> m (b, K.StreamK m (Array a))
foldBreakK (FL.Fold fstep initial extract) stream = do
foldBreakK (FL.Fold fstep initial _ final) stream = do
res <- initial
case res of
FL.Partial fs -> go fs stream
@ -373,7 +373,7 @@ foldBreakK (FL.Fold fstep initial extract) stream = do
{-# INLINE go #-}
go !fs st = do
let stop = (, K.nil) <$> extract fs
let stop = (, K.nil) <$> final fs
single a = yieldk a K.nil
yieldk (Array contents start end) r =
let fp = Tuple' end contents

32
core/src/Streamly/Internal/Data/Stream/MutChunked.hs
View File

@ -131,8 +131,8 @@ packArraysChunksOf n (D.Stream step state) =
{-# INLINE_NORMAL lpackArraysChunksOf #-}
lpackArraysChunksOf :: (MonadIO m, Unbox a)
=> Int -> Fold m (MutArray a) () -> Fold m (MutArray a) ()
lpackArraysChunksOf n (Fold step1 initial1 extract1) =
Fold step initial extract
lpackArraysChunksOf n (Fold step1 initial1 _ final1) =
Fold step initial extract final
where
@ -145,13 +145,6 @@ lpackArraysChunksOf n (Fold step1 initial1 extract1) =
r <- initial1
return $ first (Tuple' Nothing) r
extract (Tuple' Nothing r1) = extract1 r1
extract (Tuple' (Just buf) r1) = do
r <- step1 r1 buf
case r of
FL.Partial rr -> extract1 rr
FL.Done _ -> return ()
step (Tuple' Nothing r1) arr =
let len = MArray.byteLength arr
in if len >= n
@ -160,7 +153,7 @@ lpackArraysChunksOf n (Fold step1 initial1 extract1) =
case r of
FL.Done _ -> return $ FL.Done ()
FL.Partial s -> do
extract1 s
_ <- final1 s
res <- initial1
return $ first (Tuple' Nothing) res
else return $ FL.Partial $ Tuple' (Just arr) r1
@ -179,11 +172,26 @@ lpackArraysChunksOf n (Fold step1 initial1 extract1) =
case r of
FL.Done _ -> return $ FL.Done ()
FL.Partial s -> do
extract1 s
_ <- final1 s
res <- initial1
return $ first (Tuple' Nothing) res
else return $ FL.Partial $ Tuple' (Just buf'') r1
-- XXX Several folds do extract >=> final, therefore, we need to make final
-- return "m b" rather than using extract post it if we want extract to be
-- partial.
--
-- extract forces the pending buffer to be sent to the fold which is not
-- what we want.
extract _ = error "lpackArraysChunksOf: not designed for scanning"
final (Tuple' Nothing r1) = final1 r1
final (Tuple' (Just buf) r1) = do
r <- step1 r1 buf
case r of
FL.Partial rr -> final1 rr
FL.Done _ -> return ()
-- XXX Same as compactLE, to be removed once that is implemented.
--
-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
@ -256,7 +264,7 @@ compactLEParserD n = ParserD.Parser step initial extract
compactGEFold ::
forall m a. (MonadIO m, Unbox a)
=> Int -> FL.Fold m (MutArray a) (MutArray a)
compactGEFold n = Fold step initial extract
compactGEFold n = Fold step initial extract extract
where

125
core/src/Streamly/Internal/Data/Stream/Nesting.hs
View File

@ -1242,6 +1242,7 @@ foldSequence _f _m = undefined
data FIterState s f m a b
= FIterInit s f
| forall fs. FIterStream s (fs -> a -> m (FL.Step fs b)) fs (fs -> m b)
(fs -> m b)
| FIterYield b (FIterState s f m a b)
| FIterStop
@ -1269,26 +1270,26 @@ foldIterateM func seed0 (Stream step state) =
where
{-# INLINE iterStep #-}
iterStep from st fstep extract = do
iterStep from st fstep extract final = do
res <- from
return
$ Skip
$ case res of
FL.Partial fs -> FIterStream st fstep fs extract
FL.Partial fs -> FIterStream st fstep fs extract final
FL.Done fb -> FIterYield fb $ FIterInit st (return fb)
{-# INLINE_LATE stepOuter #-}
stepOuter _ (FIterInit st seed) = do
(FL.Fold fstep initial extract) <- seed >>= func
iterStep initial st fstep extract
stepOuter gst (FIterStream st fstep fs extract) = do
(FL.Fold fstep initial extract final) <- seed >>= func
iterStep initial st fstep extract final
stepOuter gst (FIterStream st fstep fs extract final) = do
r <- step (adaptState gst) st
case r of
Yield x s -> do
iterStep (fstep fs x) s fstep extract
Skip s -> return $ Skip $ FIterStream s fstep fs extract
iterStep (fstep fs x) s fstep extract final
Skip s -> return $ Skip $ FIterStream s fstep fs extract final
Stop -> do
b <- extract fs
b <- final fs
return $ Skip $ FIterYield b FIterStop
stepOuter _ (FIterYield a next) = return $ Yield a next
stepOuter _ FIterStop = return Stop
@ -1338,7 +1339,7 @@ refoldIterateM (Refold fstep finject fextract) initial (Stream step state) =
-- "n" elements at the end are dropped by the fold.
{-# INLINE sliceBy #-}
sliceBy :: Monad m => Fold m a Int -> Int -> Refold m (Int, Int) a (Int, Int)
sliceBy (Fold step1 initial1 extract1) n = Refold step inject extract
sliceBy (Fold step1 initial1 extract1 _final) n = Refold step inject extract
where
@ -1916,7 +1917,7 @@ groupsWhile :: Monad m
{-
groupsWhile eq fld = parseMany (PRD.groupBy eq fld)
-}
groupsWhile cmp (Fold fstep initial done) (Stream step state) =
groupsWhile cmp (Fold fstep initial _ final) (Stream step state) =
Stream stepOuter (GroupingInit state)
where
@ -1939,7 +1940,7 @@ groupsWhile cmp (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC x s fs1
FL.Done b -> return $ Yield b (GroupingInit s)
Skip s -> return $ Skip $ GroupingDo s fs
Stop -> return Stop
Stop -> final fs >> return Stop
where
@ -1954,10 +1955,12 @@ groupsWhile cmp (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC prev s fs1
FL.Done b -> return $ Yield b (GroupingInit s)
else do
r <- done acc
r <- final acc
return $ Yield r (GroupingInitWith s x)
Skip s -> go SPEC prev s acc
Stop -> done acc >>= \r -> return $ Yield r GroupingDone
Stop -> do
r <- final acc
return $ Yield r GroupingDone
stepOuter _ (GroupingInitWith st x) = do
res <- initial
return
@ -1984,10 +1987,12 @@ groupsWhile cmp (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC s fs1
FL.Done b -> return $ Yield b (GroupingInit s)
else do
r <- done acc
r <- final acc
return $ Yield r (GroupingInitWith s x)
Skip s -> go SPEC s acc
Stop -> done acc >>= \r -> return $ Yield r GroupingDone
Stop -> do
r <- final acc
return $ Yield r GroupingDone
stepOuter _ (GroupingYield _ _) = error "groupsWhile: Unreachable"
stepOuter _ GroupingDone = return Stop
@ -2009,7 +2014,7 @@ groupsRollingBy :: Monad m
{-
groupsRollingBy eq fld = parseMany (PRD.groupByRolling eq fld)
-}
groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
groupsRollingBy cmp (Fold fstep initial _ final) (Stream step state) =
Stream stepOuter (GroupingInit state)
where
@ -2032,7 +2037,7 @@ groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC x s fs1
FL.Done fb -> return $ Yield fb (GroupingInit s)
Skip s -> return $ Skip $ GroupingDo s fs
Stop -> return Stop
Stop -> final fs >> return Stop
where
@ -2047,10 +2052,12 @@ groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC x s fs1
FL.Done b -> return $ Yield b (GroupingInit s)
else do
r <- done acc
r <- final acc
return $ Yield r (GroupingInitWith s x)
Skip s -> go SPEC prev s acc
Stop -> done acc >>= \r -> return $ Yield r GroupingDone
Stop -> do
r <- final acc
return $ Yield r GroupingDone
stepOuter _ (GroupingInitWith st x) = do
res <- initial
return
@ -2081,7 +2088,7 @@ groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
FL.Done b -> return $ Yield b (GroupingInit st)
else do
{-
r <- done acc
r <- final acc
return $ Yield r (GroupingInitWith s x)
-}
-- The code above does not let groupBy fuse. We use the
@ -2090,14 +2097,16 @@ groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
-- GroupingInitWith state here to help GHC with stream
-- fusion.
result <- initial
r <- done acc
r <- final acc
return
$ Yield r
$ case result of
FL.Partial fsi -> GroupingDoWith s fsi x
FL.Done b -> GroupingYield b (GroupingInit s)
Skip s -> go SPEC prev s acc
Stop -> done acc >>= \r -> return $ Yield r GroupingDone
Stop -> do
r <- final acc
return $ Yield r GroupingDone
stepOuter _ (GroupingYield r next) = return $ Yield r next
stepOuter _ GroupingDone = return Stop
@ -2119,7 +2128,7 @@ data WordsByState st fs b
{-# INLINE_NORMAL wordsBy #-}
wordsBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
wordsBy predicate (Fold fstep initial done) (Stream step state) =
wordsBy predicate (Fold fstep initial _ final) (Stream step state) =
Stream stepOuter (WordsByInit state)
where
@ -2149,7 +2158,7 @@ wordsBy predicate (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC s fs1
FL.Done b -> return $ Yield b (WordsByInit s)
Skip s -> return $ Skip $ WordsByDo s fs
Stop -> return Stop
Stop -> final fs >> return Stop
where
@ -2160,7 +2169,7 @@ wordsBy predicate (Fold fstep initial done) (Stream step state) =
if predicate x
then do
{-
r <- done acc
r <- final acc
return $ Yield r (WordsByInit s)
-}
-- The above code does not fuse well. Need to check why
@ -2169,7 +2178,7 @@ wordsBy predicate (Fold fstep initial done) (Stream step state) =
-- state always, we directly go to WordsByDo state in
-- the common case of Partial.
resi <- initial
r <- done acc
r <- final acc
return
$ Yield r
$ case resi of
@ -2181,7 +2190,9 @@ wordsBy predicate (Fold fstep initial done) (Stream step state) =
FL.Partial fs1 -> go SPEC s fs1
FL.Done b -> return $ Yield b (WordsByInit s)
Skip s -> go SPEC s acc
Stop -> done acc >>= \r -> return $ Yield r WordsByDone
Stop -> do
r <- final acc
return $ Yield r WordsByDone
stepOuter _ WordsByDone = return Stop
@ -2239,7 +2250,7 @@ splitOnSeq
-> Fold m a b
-> Stream m a
-> Stream m b
splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
splitOnSeq patArr (Fold fstep initial _ final) (Stream step state) =
Stream stepOuter SplitOnSeqInit
where
@ -2322,12 +2333,12 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
r <- fstep acc x
b1 <-
case r of
FL.Partial acc1 -> done acc1
FL.Partial acc1 -> final acc1
FL.Done b -> return b
let jump c = SplitOnSeqEmpty c s
in yieldProceed jump b1
Skip s -> skip (SplitOnSeqEmpty acc s)
Stop -> return Stop
Stop -> final acc >> return Stop
-----------------
-- Done
@ -2345,7 +2356,7 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
Yield x s -> do
let jump c = SplitOnSeqSingle c s pat
if pat == x
then done fs >>= yieldProceed jump
then final fs >>= yieldProceed jump
else do
r <- fstep fs x
case r of
@ -2353,7 +2364,7 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
FL.Done b -> yieldProceed jump b
Skip s -> return $ Skip $ SplitOnSeqSingle fs s pat
Stop -> do
r <- done fs
r <- final fs
return $ Skip $ SplitOnSeqYield r SplitOnSeqDone
---------------------------
@ -2361,7 +2372,7 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
---------------------------
stepOuter _ (SplitOnSeqWordDone 0 fs _) = do
r <- done fs
r <- final fs
skip $ SplitOnSeqYield r SplitOnSeqDone
stepOuter _ (SplitOnSeqWordDone n fs wrd) = do
let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
@ -2388,7 +2399,7 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
if wrd1 .&. wordMask == wordPat
then do
let jump c = SplitOnSeqWordInit c s
done fs >>= yieldProceed jump
final fs >>= yieldProceed jump
else skip $ SplitOnSeqWordLoop wrd1 s fs
else go SPEC (idx + 1) wrd1 s
Skip s -> go SPEC idx wrd s
@ -2396,7 +2407,7 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
if idx /= 0
then skip $ SplitOnSeqWordDone idx fs wrd
else do
r <- done fs
r <- final fs
skip $ SplitOnSeqYield r SplitOnSeqDone
stepOuter gst (SplitOnSeqWordLoop wrd0 st0 fs0) =
@ -2417,7 +2428,7 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
case r of
FL.Partial fs1 -> do
if wrd1 .&. wordMask == wordPat
then done fs1 >>= yieldProceed jump
then final fs1 >>= yieldProceed jump
else go SPEC wrd1 s fs1
FL.Done b -> yieldProceed jump b
Skip s -> go SPEC wrd s fs
@ -2502,14 +2513,14 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
stepOuter _ (SplitOnSeqKRCheck fs st rb rh) = do
if RB.unsafeEqArray rb rh patArr
then do
r <- done fs
r <- final fs
let rst = RB.startOf rb
jump c = SplitOnSeqKRInit 0 c st rb rst
yieldProceed jump r
else skip $ SplitOnSeqKRLoop fs st rb rh patHash
stepOuter _ (SplitOnSeqKRDone 0 fs _ _) = do
r <- done fs
r <- final fs
skip $ SplitOnSeqYield r SplitOnSeqDone
stepOuter _ (SplitOnSeqKRDone n fs rb rh) = do
old <- liftIO $ peek rh
@ -2553,7 +2564,7 @@ splitOnSuffixSeq
-> Fold m a b
-> Stream m a
-> Stream m b
splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
splitOnSuffixSeq withSep patArr (Fold fstep initial _ final) (Stream step state) =
Stream stepOuter SplitOnSuffixSeqInit
where
@ -2593,7 +2604,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
r <- if withSep then fstep fs x else return $ FL.Partial fs
b1 <-
case r of
FL.Partial fs1 -> done fs1
FL.Partial fs1 -> final fs1
FL.Done b -> return b
yieldProceed jump b1
else do
@ -2656,11 +2667,11 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
r <- fstep acc x
b1 <-
case r of
FL.Partial fs -> done fs
FL.Partial fs -> final fs
FL.Done b -> return b
yieldProceed jump b1
Skip s -> skip (SplitOnSuffixSeqEmpty acc s)
Stop -> return Stop
Stop -> final acc >> return Stop
-----------------
-- Done
@ -2677,7 +2688,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
case res of
Yield x s -> processYieldSingle pat x s fs
Skip s -> skip $ SplitOnSuffixSeqSingleInit fs s pat
Stop -> return Stop
Stop -> final fs >> return Stop
stepOuter gst (SplitOnSuffixSeqSingle fs st pat) = do
res <- step (adaptState gst) st
@ -2685,7 +2696,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
Yield x s -> processYieldSingle pat x s fs
Skip s -> skip $ SplitOnSuffixSeqSingle fs s pat
Stop -> do
r <- done fs
r <- final fs
skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
---------------------------
@ -2693,7 +2704,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
---------------------------
stepOuter _ (SplitOnSuffixSeqWordDone 0 fs _) = do
r <- done fs
r <- final fs
skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
stepOuter _ (SplitOnSuffixSeqWordDone n fs wrd) = do
let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
@ -2716,7 +2727,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
let jump c = SplitOnSuffixSeqWordInit c s
yieldProceed jump b
Skip s -> skip (SplitOnSuffixSeqWordInit fs0 s)
Stop -> return Stop
Stop -> final fs0 >> return Stop
where
@ -2734,7 +2745,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
then go SPEC (idx + 1) wrd1 s fs1
else if wrd1 .&. wordMask /= wordPat
then skip $ SplitOnSuffixSeqWordLoop wrd1 s fs1
else do done fs >>= yieldProceed jump
else do final fs >>= yieldProceed jump
FL.Done b -> yieldProceed jump b
Skip s -> go SPEC idx wrd s fs
Stop -> skip $ SplitOnSuffixSeqWordDone idx fs wrd
@ -2760,16 +2771,16 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
case r of
FL.Partial fs1 ->
if wrd1 .&. wordMask == wordPat
then done fs1 >>= yieldProceed jump
then final fs1 >>= yieldProceed jump
else go SPEC wrd1 s fs1
FL.Done b -> yieldProceed jump b
Skip s -> go SPEC wrd s fs
Stop ->
if wrd .&. wordMask == wordPat
then return Stop
then final fs >> return Stop
else if withSep
then do
r <- done fs
r <- final fs
skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
else skip $ SplitOnSuffixSeqWordDone patLen fs wrd
@ -2791,7 +2802,7 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
yieldProceed jump b
Skip s -> skip $ SplitOnSuffixSeqKRInit idx0 fs s rb rh0
Stop -> return Stop
Stop -> final fs >> return Stop
stepOuter gst (SplitOnSuffixSeqKRInit1 fs0 st0 rb rh0) = do
go SPEC 1 rh0 st0 fs0
@ -2823,10 +2834,10 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
Stop -> do
-- do not issue a blank segment when we end at pattern
if (idx == maxIndex) && RB.unsafeEqArray rb rh patArr
then return Stop
then final fs >> return Stop
else if withSep
then do
r <- done fs
r <- final fs
skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
else skip $ SplitOnSuffixSeqKRDone idx fs rb (RB.startOf rb)
@ -2855,24 +2866,24 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
Skip s -> go SPEC fs s rh cksum
Stop ->
if RB.unsafeEqArray rb rh patArr
then return Stop
then final fs >> return Stop
else if withSep
then do
r <- done fs
r <- final fs
skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
else skip $ SplitOnSuffixSeqKRDone patLen fs rb rh
stepOuter _ (SplitOnSuffixSeqKRCheck fs st rb rh) = do
if RB.unsafeEqArray rb rh patArr
then do
r <- done fs
r <- final fs
let rst = RB.startOf rb
jump c = SplitOnSuffixSeqKRInit 0 c st rb rst
yieldProceed jump r
else skip $ SplitOnSuffixSeqKRLoop fs st rb rh patHash
stepOuter _ (SplitOnSuffixSeqKRDone 0 fs _ _) = do
r <- done fs
r <- final fs
skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
stepOuter _ (SplitOnSuffixSeqKRDone n fs rb rh) = do
old <- liftIO $ peek rh

16
core/src/Streamly/Internal/Data/Stream/Transform.hs
View File

@ -300,7 +300,7 @@ data TapState fs st a
--
{-# INLINE tap #-}
tap :: Monad m => Fold m a b -> Stream m a -> Stream m a
tap (Fold fstep initial extract) (Stream step state) = Stream step' TapInit
tap (Fold fstep initial _ final) (Stream step state) = Stream step' TapInit
where
@ -323,7 +323,7 @@ tap (Fold fstep initial extract) (Stream step state) = Stream step' TapInit
FL.Done _ -> TapDone s
Skip s -> return $ Skip (Tapping acc s)
Stop -> do
void $ extract acc
void $ final acc
return Stop
step' gst (TapDone st) = do
r <- step gst st
@ -342,7 +342,7 @@ data TapOffState fs s a
{-# INLINE_NORMAL tapOffsetEvery #-}
tapOffsetEvery :: Monad m
=> Int -> Int -> Fold m a b -> Stream m a -> Stream m a
tapOffsetEvery offset n (Fold fstep initial extract) (Stream step state) =
tapOffsetEvery offset n (Fold fstep initial _ final) (Stream step state) =
Stream step' TapOffInit
where
@ -372,7 +372,7 @@ tapOffsetEvery offset n (Fold fstep initial extract) (Stream step state) =
return $ Yield x next
Skip s -> return $ Skip (TapOffTapping acc s count)
Stop -> do
void $ extract acc
void $ final acc
return Stop
step' gst (TapOffDone st) = do
r <- step gst st
@ -437,7 +437,7 @@ data ScanState s f = ScanInit s | ScanDo s !f | ScanDone
--
{-# INLINE_NORMAL postscan #-}
postscan :: Monad m => FL.Fold m a b -> Stream m a -> Stream m b
postscan (FL.Fold fstep initial extract) (Stream sstep state) =
postscan (FL.Fold fstep initial extract final) (Stream sstep state) =
Stream step (ScanInit state)
where
@ -460,13 +460,13 @@ postscan (FL.Fold fstep initial extract) (Stream sstep state) =
return $ Yield b $ ScanDo s fs1
FL.Done b -> return $ Yield b ScanDone
Skip s -> return $ Skip $ ScanDo s fs
Stop -> return Stop
Stop -> final fs >> return Stop
step _ ScanDone = return Stop
{-# INLINE scanWith #-}
scanWith :: Monad m
=> Bool -> Fold m a b -> Stream m a -> Stream m b
scanWith restart (Fold fstep initial extract) (Stream sstep state) =
scanWith restart (Fold fstep initial extract final) (Stream sstep state) =
Stream step (ScanInit state)
where
@ -489,7 +489,7 @@ scanWith restart (Fold fstep initial extract) (Stream sstep state) =
case res of
Yield x s -> runStep s (fstep fs x)
Skip s -> return $ Skip $ ScanDo s fs
Stop -> return Stop
Stop -> final fs >> return Stop
step _ ScanDone = return Stop
-- XXX It may be useful to have a version of scan where we can keep the

24
core/src/Streamly/Internal/Data/Stream/Type.hs
View File

@ -398,7 +398,7 @@ toStreamK (Stream step state) = go state
{-# INLINE_NORMAL foldEither #-}
foldEither :: Monad m =>
Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))
foldEither (Fold fstep begin done) (UnStream step state) = do
foldEither (Fold fstep begin done final) (UnStream step state) = do
res <- begin
case res of
FL.Partial fs -> go SPEC fs state
@ -416,7 +416,9 @@ foldEither (Fold fstep begin done) (UnStream step state) = do
FL.Done b -> return $! Right (b, Stream step s)
FL.Partial fs1 -> go SPEC fs1 s
Skip s -> go SPEC fs s
Stop -> return $! Left (Fold fstep (return $ FL.Partial fs) done)
Stop ->
let f = Fold fstep (return $ FL.Partial fs) done final
in return $! Left f
-- | Like 'fold' but also returns the remaining stream. The resulting stream
-- would be 'Stream.nil' if the stream finished before the fold.
@ -427,12 +429,12 @@ foldBreak fld strm = do
r <- foldEither fld strm
case r of
Right res -> return res
Left (Fold _ initial extract) -> do
Left (Fold _ initial _ final) -> do
res <- initial
case res of
FL.Done _ -> error "foldBreak: unreachable state"
FL.Partial s -> do
b <- extract s
b <- final s
return (b, nil)
where
@ -475,8 +477,8 @@ fold fld strm = do
--
{-# INLINE_NORMAL foldAddLazy #-}
foldAddLazy :: Monad m => Fold m a b -> Stream m a -> Fold m a b
foldAddLazy (Fold fstep finitial fextract) (Stream sstep state) =
Fold fstep initial fextract
foldAddLazy (Fold fstep finitial fextract ffinal) (Stream sstep state) =
Fold fstep initial fextract ffinal
where
@ -1796,7 +1798,7 @@ data FoldManyPost s fs b a
--
{-# INLINE_NORMAL foldManyPost #-}
foldManyPost :: Monad m => Fold m a b -> Stream m a -> Stream m b
foldManyPost (Fold fstep initial extract) (Stream step state) =
foldManyPost (Fold fstep initial _ final) (Stream step state) =
Stream step' (FoldManyPostStart state)
where
@ -1824,7 +1826,7 @@ foldManyPost (Fold fstep initial extract) (Stream step state) =
Yield x s -> consume x s fs
Skip s -> return $ Skip (FoldManyPostLoop s fs)
Stop -> do
b <- extract fs
b <- final fs
return $ Skip (FoldManyPostYield b FoldManyPostDone)
step' _ (FoldManyPostYield b next) = return $ Yield b next
step' _ FoldManyPostDone = return Stop
@ -1868,7 +1870,7 @@ data FoldMany s fs b a
--
{-# INLINE_NORMAL foldMany #-}
foldMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
foldMany (Fold fstep initial extract) (Stream step state) =
foldMany (Fold fstep initial _ final) (Stream step state) =
Stream step' (FoldManyStart state)
where
@ -1895,14 +1897,14 @@ foldMany (Fold fstep initial extract) (Stream step state) =
case r of
Yield x s -> consume x s fs
Skip s -> return $ Skip (FoldManyFirst fs s)
Stop -> return Stop
Stop -> final fs >> return Stop
step' gst (FoldManyLoop st fs) = do
r <- step (adaptState gst) st
case r of
Yield x s -> consume x s fs
Skip s -> return $ Skip (FoldManyLoop s fs)
Stop -> do
b <- extract fs
b <- final fs
return $ Skip (FoldManyYield b FoldManyDone)
step' _ (FoldManyYield b next) = return $ Yield b next
step' _ FoldManyDone = return Stop

25
core/src/Streamly/Internal/Data/StreamK.hs
View File

@ -336,7 +336,7 @@ foldlMx' step begin done = go begin
--
{-# INLINABLE fold #-}
fold :: Monad m => FL.Fold m a b -> StreamK m a -> m b
fold (FL.Fold step begin done) m = do
fold (FL.Fold step begin _ final) m = do
res <- begin
case res of
FL.Partial fs -> go fs m
@ -344,10 +344,10 @@ fold (FL.Fold step begin done) m = do
where
go !acc m1 =
let stop = done acc
let stop = final acc
single a = step acc a
>>= \case
FL.Partial s -> done s
FL.Partial s -> final s
FL.Done b1 -> return b1
yieldk a r = step acc a
>>= \case
@ -365,7 +365,7 @@ fold (FL.Fold step begin done) m = do
{-# INLINE foldEither #-}
foldEither :: Monad m =>
Fold m a b -> StreamK m a -> m (Either (Fold m a b) (b, StreamK m a))
foldEither (FL.Fold step begin done) m = do
foldEither (FL.Fold step begin done final) m = do
res <- begin
case res of
FL.Partial fs -> go fs m
@ -374,12 +374,15 @@ foldEither (FL.Fold step begin done) m = do
where
go !acc m1 =
let stop = return $ Left (Fold step (return $ FL.Partial acc) done)
let stop =
let f = Fold step (return $ FL.Partial acc) done final
in return $ Left f
single a =
step acc a
>>= \case
FL.Partial s ->
return $ Left (Fold step (return $ FL.Partial s) done)
let f = Fold step (return $ FL.Partial s) done final
in return $ Left f
FL.Done b1 -> return $ Right (b1, nil)
yieldk a r =
step acc a
@ -397,12 +400,12 @@ foldBreak fld strm = do
r <- foldEither fld strm
case r of
Right res -> return res
Left (Fold _ initial extract) -> do
Left (Fold _ initial _ final) -> do
res <- initial
case res of
FL.Done _ -> error "foldBreak: unreachable state"
FL.Partial s -> do
b <- extract s
b <- final s
return (b, nil)
-- XXX Array folds can be implemented using this.
@ -421,7 +424,7 @@ foldConcat :: Monad m =>
Producer m a b -> Fold m b c -> StreamK m a -> m (c, StreamK m a)
foldConcat
(Producer pstep pinject pextract)
(Fold fstep begin done)
(Fold fstep begin _ final)
stream = do
res <- begin
@ -433,14 +436,14 @@ foldConcat
go !acc m1 = do
let stop = do
r <- done acc
r <- final acc
return (r, nil)
single a = do
st <- pinject a
res <- go1 SPEC acc st
case res of
Left fs -> do
r <- done fs
r <- final fs
return (r, nil)
Right (b, s) -> do
x <- pextract s

18
core/src/Streamly/Internal/Data/Unfold.hs
View File

@ -205,7 +205,7 @@ swap = lmap Tuple.swap
--
{-# INLINE_NORMAL fold #-}
fold :: Monad m => Fold m b c -> Unfold m a b -> a -> m c
fold (Fold fstep initial extract) (Unfold ustep inject) a = do
fold (Fold fstep initial _ final) (Unfold ustep inject) a = do
res <- initial
case res of
FL.Partial x -> inject a >>= go SPEC x
@ -223,7 +223,7 @@ fold (Fold fstep initial extract) (Unfold ustep inject) a = do
FL.Partial fs1 -> go SPEC fs1 s
FL.Done c -> return c
Skip s -> go SPEC fs s
Stop -> extract fs
Stop -> final fs
-- {-# ANN type FoldMany Fuse #-}
data FoldMany s fs b a
@ -238,7 +238,7 @@ data FoldMany s fs b a
-- /Pre-release/
{-# INLINE_NORMAL foldMany #-}
foldMany :: Monad m => Fold m b c -> Unfold m a b -> Unfold m a c
foldMany (Fold fstep initial extract) (Unfold ustep inject1) =
foldMany (Fold fstep initial _ final) (Unfold ustep inject1) =
Unfold step inject
where
@ -269,14 +269,14 @@ foldMany (Fold fstep initial extract) (Unfold ustep inject1) =
case r of
Yield x s -> consume x s fs
Skip s -> return $ Skip (FoldManyFirst fs s)
Stop -> return Stop
Stop -> final fs >> return Stop
step (FoldManyLoop st fs) = do
r <- ustep st
case r of
Yield x s -> consume x s fs
Skip s -> return $ Skip (FoldManyLoop s fs)
Stop -> do
b <- extract fs
b <- final fs
return $ Skip (FoldManyYield b FoldManyDone)
step (FoldManyYield b next) = return $ Yield b next
step FoldManyDone = return Stop
@ -317,7 +317,7 @@ either (Unfold stepL injectL) (Unfold stepR injectR) = Unfold step inject
-- /Pre-release/
{-# INLINE_NORMAL postscan #-}
postscan :: Monad m => Fold m b c -> Unfold m a b -> Unfold m a c
postscan (Fold stepF initial extract) (Unfold stepU injectU) =
postscan (Fold stepF initial extract final) (Unfold stepU injectU) =
Unfold step inject
where
@ -340,7 +340,7 @@ postscan (Fold stepF initial extract) (Unfold stepU injectU) =
v <- extract fs1
return $ Yield v (Just (fs1, s))
Skip s -> return $ Skip (Just (fs, s))
Stop -> return Stop
Stop -> final fs >> return Stop
step Nothing = return Stop
@ -348,7 +348,7 @@ data ScanState s f = ScanInit s | ScanDo s !f | ScanDone
{-# INLINE_NORMAL scanWith #-}
scanWith :: Monad m => Bool -> Fold m b c -> Unfold m a b -> Unfold m a c
scanWith restart (Fold fstep initial extract) (Unfold stepU injectU) =
scanWith restart (Fold fstep initial extract final) (Unfold stepU injectU) =
Unfold step inject
where
@ -373,7 +373,7 @@ scanWith restart (Fold fstep initial extract) (Unfold stepU injectU) =
case res of
Yield x s -> runStep s (fstep fs x)
Skip s -> return $ Skip $ ScanDo s fs
Stop -> return Stop
Stop -> final fs >> return Stop
step ScanDone = return Stop
-- | Scan the output of an 'Unfold' to change it in a stateful manner.

11
core/src/Streamly/Internal/FileSystem/File.hs
View File

@ -438,7 +438,7 @@ write = toHandleWith A.defaultChunkSize
{-# INLINE writeChunks #-}
writeChunks :: (MonadIO m, MonadCatch m)
=> FilePath -> Fold m (Array a) ()
writeChunks path = Fold step initial extract
writeChunks path = Fold step initial extract final
where
initial = do
h <- liftIO (openFile path WriteMode)
@ -448,12 +448,15 @@ writeChunks path = Fold step initial extract
step (fld, h) x = do
r <- FL.snoc fld x `MC.onException` liftIO (hClose h)
return $ FL.Partial (r, h)
extract (Fold _ initial1 extract1, h) = do
extract _ = return ()
final (Fold _ initial1 _ final1, h) = do
liftIO $ hClose h
res <- initial1
case res of
FL.Partial fs -> extract1 fs
FL.Done fb -> return fb
FL.Partial fs -> final1 fs
FL.Done () -> return ()
-- | @writeWith chunkSize handle@ writes the input stream to @handle@.
-- Bytes in the input stream are collected into a buffer until we have a chunk

5
core/src/Streamly/Internal/Unicode/Stream.hs
View File

@ -111,6 +111,7 @@ import System.IO.Unsafe (unsafePerformIO)
import Streamly.Internal.Data.Array.Type (Array(..))
import Streamly.Internal.Data.MutArray.Type (MutableByteArray)
import Streamly.Internal.Data.Fold (Fold)
import Streamly.Internal.Data.Parser (Parser)
import Streamly.Internal.Data.Stream (Stream)
import Streamly.Internal.Data.Stream (Step (..))
import Streamly.Internal.Data.SVar.Type (adaptState)
@ -523,8 +524,8 @@ parseCharUtf8WithD cfm = ParserD.Parser (step' utf8d) initial extract
-- workflow requires backtracking 1 element. This can be revisited once "Fold"
-- supports backtracking.
{-# INLINE writeCharUtf8' #-}
writeCharUtf8' :: Monad m => Fold m Word8 Char
writeCharUtf8' = ParserD.toFold (parseCharUtf8WithD ErrorOnCodingFailure)
writeCharUtf8' :: Monad m => Parser Word8 m Char
writeCharUtf8' = parseCharUtf8WithD ErrorOnCodingFailure
-- XXX The initial idea was to have "parseCharUtf8" and offload the error
-- handling to another parser. So, say we had "parseCharUtf8'",

7
src/Streamly/Internal/Data/Fold/Async.hs
View File

@ -63,7 +63,8 @@ import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))
--
{-# INLINE takeInterval #-}
takeInterval :: MonadAsync m => Double -> Fold m a b -> Fold m a b
takeInterval n (Fold step initial done) = Fold step' initial' done'
takeInterval n (Fold step initial done final) =
Fold step' initial' done' final'
where
@ -90,7 +91,7 @@ takeInterval n (Fold step initial done) = Fold step' initial' done'
then do
res <- step s a
case res of
Partial sres -> Done <$> done sres
Partial sres -> Done <$> final sres
Done bres -> return $ Done bres
else do
res <- step s a
@ -100,6 +101,8 @@ takeInterval n (Fold step initial done) = Fold step' initial' done'
done' (Tuple3' s _ _) = done s
final' (Tuple3' s _ _) = final s
timerThread mv = do
liftIO $ threadDelay (round $ n * 1000000)
-- Use IORef + CAS? instead of MVar since its a Bool?

29
src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs
View File

@ -42,12 +42,13 @@ import Streamly.Internal.Data.Stream.Channel.Types
-- | Evaluate a fold asynchronously using a concurrent channel. The driver just
-- queues the input stream values to the fold channel buffer and returns. The
-- fold evaluates the queued values asynchronously.
-- fold evaluates the queued values asynchronously. On finalization, 'parEval'
-- waits for the asynchronous fold to complete before it returns.
--
{-# INLINABLE parEval #-}
parEval :: MonadAsync m => (Config -> Config) -> Fold m a b -> Fold m a b
parEval modifier f =
Fold step initial extract
Fold step initial extract final
where
@ -62,13 +63,33 @@ parEval modifier f =
--
-- A polled stream abstraction may be useful, it would consist of normal
-- events and tick events, latter are guaranteed to arrive.
--
-- XXX We can use the config to indicate if the fold is a scanning type or
-- one-shot, or use a separate parEvalScan for scanning. For a scanning
-- type fold the worker would always send the intermediate values back to
-- the driver. An intermediate value can be returned on an input, or the
-- driver can poll even without input, if we have the Skip input support.
-- When the buffer is full we can return "Skip" and then the next step
-- without input can wait for an output to arrive. Similarly, when "final"
-- is called it can return "Skip" to continue or "Done" to indicate
-- termination.
step chan a = do
status <- sendToWorker chan a
return $ case status of
Nothing -> Partial chan
Just b -> Done b
extract chan = do
-- XXX We can use a separate type for non-scanning folds that will
-- introduce a lot of complexity. Are there combinators that rely on the
-- "extract" function even in non-scanning use cases?
-- Instead of making such folds partial we can also make them return a
-- Maybe type.
extract _ = error "Concurrent folds do not support scanning"
-- XXX depending on the use case we may want to either wait for the result
-- or cancel the ongoing work. We can use the config to control that?
-- Currently it waits for the work to complete.
final chan = do
liftIO $ void
$ sendWithDoorBell
(outputQueue chan)
@ -84,7 +105,7 @@ parEval modifier f =
"parEval: waiting to drain"
$ takeMVar (outputDoorBellFromConsumer chan)
-- XXX remove recursion
extract chan
final chan
Just b -> do
when (svarInspectMode chan) $ liftIO $ do
t <- getTime Monotonic

10
src/Streamly/Internal/Data/Fold/SVar.hs
View File

@ -33,7 +33,7 @@ import Streamly.Internal.Data.SVar
--
{-# INLINE write #-}
write :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
write svar winfo = Fold step initial extract
write svar winfo = Fold step initial return final
where
@ -47,14 +47,14 @@ write svar winfo = Fold step initial extract
void $ send svar (ChildYield x)
return $ FL.Partial ()
extract () = liftIO $ sendStop svar winfo
final () = liftIO $ sendStop svar winfo
-- | Like write, but applies a yield limit.
--
{-# INLINE writeLimited #-}
writeLimited :: MonadIO m
=> SVar t m a -> Maybe WorkerInfo -> Fold m a ()
writeLimited svar winfo = Fold step initial extract
writeLimited svar winfo = Fold step initial (const (return ())) final
where
@ -74,5 +74,5 @@ writeLimited svar winfo = Fold step initial extract
return $ FL.Done ()
step False _ = return $ FL.Done ()
extract True = liftIO $ sendStop svar winfo
extract False = return ()
final True = liftIO $ sendStop svar winfo
final False = return ()

1
src/Streamly/Internal/Data/IsMap/HashMap.hs
View File

@ -34,3 +34,4 @@ instance (Hashable k, Eq k) => IsMap (HashMap.HashMap k) where
mapDelete = HashMap.delete
mapUnion = HashMap.union
mapNull = HashMap.null
mapTraverseWithKey = HashMap.traverseWithKey

4
src/Streamly/Internal/Data/Stream/IsStream/Reduce.hs
View File

@ -1321,10 +1321,10 @@ classifySessionsByGeneric
-> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
-> t m (Key f, b) -- ^ session key, fold result
classifySessionsByGeneric _ tick reset ejectPred tmout
(Fold step initial extract) input =
(Fold step initial extract final) input =
Expand.unfoldMany
(Unfold.lmap (toStreamK . sessionOutputStream) Unfold.fromStreamK)
$ scanlMAfter' sstep (return szero) (flush extract)
$ scanlMAfter' sstep (return szero) (flush final)
$ interjectSuffix tick (return Nothing)
$ map Just input

4
src/Streamly/Internal/Data/Stream/Time.hs
View File

@ -536,9 +536,9 @@ classifySessionsByGeneric
-- data)
-> Stream m (Key f, b) -- ^ session key, fold result
classifySessionsByGeneric _ tick reset ejectPred tmout
(Fold step initial extract) input =
(Fold step initial extract final) input =
Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
$ Stream.scanlMAfter' sstep (return szero) (flush extract)
$ Stream.scanlMAfter' sstep (return szero) (flush final)
$ interject (return Nothing) tick
$ fmap Just input

11
src/Streamly/Internal/Network/Inet/TCP.hs
View File

@ -358,7 +358,7 @@ writeChunks
=> (Word8, Word8, Word8, Word8)
-> PortNumber
-> Fold m (Array Word8) ()
writeChunks addr port = Fold step initial extract
writeChunks addr port = Fold step initial extract final
where
initial = do
skt <- liftIO (connect addr port)
@ -368,12 +368,15 @@ writeChunks addr port = Fold step initial extract
step (Tuple' fld skt) x = do
r <- FL.addOne x fld `MC.onException` liftIO (Net.close skt)
return $ FL.Partial (Tuple' r skt)
extract (Tuple' (Fold _ initial1 extract1) skt) = do
extract _ = return ()
final (Tuple' (Fold _ initial1 _ final1) skt) = do
liftIO $ Net.close skt
res <- initial1
case res of
FL.Partial fs -> extract1 fs
FL.Done fb -> return fb
FL.Partial fs -> final1 fs
FL.Done () -> return ()
-- | Like 'write' but provides control over the write buffer. Output will
-- be written to the IO device as soon as we collect the specified number of

4
test/Streamly/Test/Data/Parser.hs
View File

@ -719,7 +719,7 @@ many :: Property
many =
forAll (listOf (chooseInt (0, 1))) $ \ls ->
let fldstp conL currL = return $ FL.Partial (conL ++ currL)
concatFold = FL.Fold fldstp (return (FL.Partial [])) return
concatFold = FL.Fold fldstp (return (FL.Partial [])) return return
prsr =
flip P.many concatFold
$ P.fromFold $ FL.takeEndBy_ (== 1) FL.toList
@ -740,7 +740,7 @@ some =
let
ls = 0 : genLs
fldstp conL currL = return $ FL.Partial $ conL ++ currL
concatFold = FL.Fold fldstp (return (FL.Partial [])) return
concatFold = FL.Fold fldstp (return (FL.Partial [])) return return
prsr =
flip P.some concatFold
$ P.fromFold $ FL.takeEndBy_ (== 1) FL.toList

4
test/Streamly/Test/Data/ParserK.hs
View File

@ -532,7 +532,7 @@ many =
$ \ls ->
let fldstp conL currL = return $ FL.Partial (conL ++ currL)
concatFold =
FL.Fold fldstp (return (FL.Partial [])) return
FL.Fold fldstp (return (FL.Partial [])) return return
prsr =
flip P.many concatFold
$ P.fromFold $ FL.takeEndBy_ (== 1) FL.toList
@ -552,7 +552,7 @@ some =
forAll (listOf (chooseInt (0, 1)))
$ \ls ->
let fldstp conL currL = return $ FL.Partial $ conL ++ currL
concatFold = FL.Fold fldstp (return (FL.Partial [])) return
concatFold = FL.Fold fldstp (return (FL.Partial [])) return return
prsr =
flip P.some concatFold
$ P.fromFold $ FL.takeEndBy_ (== 1) FL.toList