53a6a87 · streamly · git.iko.soy

git.iko.soy/streamly

Add a splitInnerBy combinator to split chunked streams

commit53a6a873c0644a3cca74dafaa93086db49ee436a authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 13 Sep 2019, 18:08 (7 years ago) parent4a367b8

5 files changed+94−5

compare with master

unifiedsplit

src/Streamly/Internal.hs+1 src/Streamly/Memory/Array/Types.hs+7−4 src/Streamly/Memory/ArrayStream.hs+3−1 src/Streamly/Prelude/Internal.hs+28 src/Streamly/Streams/StreamD.hs+55

src/Streamly/Internal.hs+1

@@ -46,6 +46,7 @@ module Streamly.Internal

4646 , splitOnSuffixSeq

4747 , splitBySeq

4848 , splitWithSuffixSeq

49+ , splitInnerBy

4950

5051 , tapAsync

5152

src/Streamly/Memory/Array/Types.hs+7−4

@@ -28,6 +28,7 @@ module Streamly.Memory.Array.Types

2828 , unsafeSnoc

2929 , snoc

3030 , spliceWithDoubling

31+ , spliceTwo

3132

3233 , fromList

3334 , fromListN

@@ -45,6 +46,7 @@ module Streamly.Memory.Array.Types

4546 , groupIOVecsOf

4647#endif

4748 , splitOn

49+ , breakOn

4850

4951 -- * Elimination

5052 , unsafeIndexIO

@@ -942,6 +944,7 @@ unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)

942944 return $ D.Yield x (InnerLoop st startf

943945 (p `plusPtr` (sizeOf (undefined :: a))) end)

944946

947+-- Splice two immutable arrays creating a new array.

945948{-# INLINE spliceTwo #-}

946949spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)

947950spliceTwo arr1 arr2 = do

@@ -960,8 +963,8 @@ spliceTwo arr1 arr2 = do

960963 touchForeignPtr (aStart arr2)

961964 return arr { aEnd = dst `plusPtr` (len1 + len2) }

962965

963−-- Splice a new array into a pre-reserved array. The user must ensure that

964−-- there is enough space in the array.

966+-- Splice an array into a pre-reserved mutable array. The user must ensure

967+-- that there is enough space in the mutable array.

965968{-# INLINE spliceWith #-}

966969spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)

967970spliceWith dst@(Array _ end bound) src = liftIO $ do

@@ -975,8 +978,8 @@ spliceWith dst@(Array _ end bound) src = liftIO $ do

975978 memcpy (castPtr pdst) (castPtr psrc) srcLen

976979 return $ dst { aEnd = pdst `plusPtr` srcLen }

977980

978−-- Splice a new array into a preallocated array, doubling the space if there is

979−-- no space in the target array.

981+-- Splice a new array into a preallocated mutable array, doubling the space if

982+-- there is no space in the target array.

980983{-# INLINE spliceWithDoubling #-}

981984spliceWithDoubling :: (MonadIO m, Storable a)

982985 => Array a -> Array a -> m (Array a)

src/Streamly/Memory/ArrayStream.hs+3−1

@@ -109,7 +109,9 @@ splitOn

109109 => Word8

110110 -> t m (Array Word8)

111111 -> t m (Array Word8)

112−splitOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s

112+-- splitOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s

113+splitOn byte s =

114+ D.fromStreamD $ D.splitInnerBy (A.breakOn byte) A.spliceTwo $ D.toStreamD s

113115

114116-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a

115117-- maximum specified size in bytes.

src/Streamly/Prelude/Internal.hs+28

@@ -300,6 +300,9 @@ module Streamly.Prelude.Internal

300300 -- , splitOnAnySuffixSeq

301301 -- , splitOnAnyPrefixSeq

302302

303+ -- Nested splitting

304+ , splitInnerBy

305+

303306 -- ** Grouping

304307 , groups

305308 , groupsBy

@@ -3012,6 +3015,31 @@ splitSuffixOnAny subseq f m = undefined

30123015 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)

30133016-}

30143017

3018+------------------------------------------------------------------------------

3019+-- Nested Split

3020+------------------------------------------------------------------------------

3021+

3022+-- | Consider a chunked stream of container elements e.g. a stream of @Word8@

3023+-- chunked as a stream of arrays of @Word8@. @splitInnerBy splitter joiner

3024+-- stream@ splits the inner containers @f a@ using the @splitter@ function and

3025+-- joins back the resulting fragments from splitting across multiple containers

3026+-- using the @joiner@ function such that the transformed output stream is

3027+-- consolidated as one container per segment of the split.

3028+--

3029+-- CAUTION! This is not a true streaming function as the container size after

3030+-- the split and merge may not be bounded.

3031+--

3032+-- @since 0.7.0

3033+{-# INLINE splitInnerBy #-}

3034+splitInnerBy

3035+ :: (IsStream t, Monad m)

3036+ => (f a -> m (f a, Maybe (f a))) -- splitter

3037+ -> (f a -> f a -> m (f a)) -- joiner

3038+ -> t m (f a)

3039+ -> t m (f a)

3040+splitInnerBy splitter joiner xs =

3041+ D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs

3042+

30153043------------------------------------------------------------------------------

30163044-- Reorder in sequence

30173045------------------------------------------------------------------------------

src/Streamly/Streams/StreamD.hs+55

@@ -163,6 +163,8 @@ module Streamly.Streams.StreamD

163163 , splitOn

164164 , splitSuffixOn

165165

166+ , splitInnerBy

167+

166168 -- ** Substreams

167169 , isPrefixOf

168170 , isSubsequenceOf

@@ -1628,6 +1630,59 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

16281630

16291631 stepOuter _ GO_DONE = return Stop

16301632

1633+data SplitState s arr

1634+ = SplitInitial s

1635+ | SplitBuffering s arr

1636+ | SplitSplitting s arr

1637+ | SplitYielding arr (SplitState s arr)

1638+ | SplitFinishing

1639+

1640+-- | Performs infix separator style splitting.

1641+{-# INLINE_NORMAL splitInnerBy #-}

1642+splitInnerBy

1643+ :: Monad m

1644+ => (f a -> m (f a, Maybe (f a))) -- splitter

1645+ -> (f a -> f a -> m (f a)) -- joiner

1646+ -> Stream m (f a)

1647+ -> Stream m (f a)

1648+splitInnerBy splitter joiner (Stream step1 state1) =

1649+ (Stream step (SplitInitial state1))

1650+

1651+ where

1652+

1653+ {-# INLINE_LATE step #-}

1654+ step gst (SplitInitial st) = do

1655+ r <- step1 gst st

1656+ case r of

1657+ Yield x s -> do

1658+ (x1, mx2) <- splitter x

1659+ return $ case mx2 of

1660+ Nothing -> Skip (SplitBuffering s x1)

1661+ Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))

1662+ Skip s -> return $ Skip (SplitInitial s)

1663+ Stop -> return $ Stop

1664+

1665+ step gst (SplitBuffering st buf) = do

1666+ r <- step1 gst st

1667+ case r of

1668+ Yield x s -> do

1669+ (x1, mx2) <- splitter x

1670+ buf' <- joiner buf x1

1671+ return $ case mx2 of

1672+ Nothing -> Skip (SplitBuffering s buf')

1673+ Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))

1674+ Skip s -> return $ Skip (SplitBuffering s buf)

1675+ Stop -> return $ Skip (SplitYielding buf SplitFinishing)

1676+

1677+ step _ (SplitSplitting st buf) = do

1678+ (x1, mx2) <- splitter buf

1679+ return $ case mx2 of

1680+ Nothing -> Skip $ SplitBuffering st x1

1681+ Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)

1682+

1683+ step _ (SplitYielding x next) = return $ Yield x next

1684+ step _ SplitFinishing = return $ Stop

1685+

16311686------------------------------------------------------------------------------

16321687-- Substreams

16331688------------------------------------------------------------------------------