0b2f615 · streamly · git.iko.soy

git.iko.soy/streamly

Add splitInnerBySuffix to split lines in inner container

commit0b2f615930c8c3197d8f4046626cb03f0ee1b404 authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 15 Sep 2019, 23:07 (7 years ago) parent53a6a87

8 files changed+125−24

compare with master

unifiedsplit

src/Streamly/Benchmark/FileIO/Array.hs+2−2 src/Streamly/Internal.hs+1 src/Streamly/Memory/Array.hs+1 src/Streamly/Memory/Array/Types.hs+38−20 src/Streamly/Memory/ArrayStream.hs+11−1 src/Streamly/Prelude/Internal.hs+15 src/Streamly/Streams/StreamD.hs+56 test/String.hs+1−1

src/Streamly/Benchmark/FileIO/Array.hs+2−2

@@ -76,7 +76,7 @@ inspect $ 'countBytes `hasNoType` ''Step

7676-- | Count the number of lines in a file.

7777{-# INLINE countLines #-}

7878countLines :: Handle -> IO Int

79−countLines = S.length . A.splitOn 10 . FH.readArrays

79+countLines = S.length . A.splitOnSuffix 10 . FH.readArrays

8080

8181#ifdef INSPECTION

8282inspect $ hasNoTypeClasses 'countLines

@@ -125,7 +125,7 @@ linesUnlinesCopy :: Handle -> Handle -> IO ()

125125linesUnlinesCopy inh outh =

126126 S.runFold (FH.writeArraysInChunksOf (1024*1024) outh)

127127 $ Internal.insertAfterEach (return $ A.fromList [10])

128− $ A.splitOn 10

128+ $ A.splitOnSuffix 10

129129 $ FH.readArraysOf (1024*1024) inh

130130

131131#ifdef INSPECTION

src/Streamly/Internal.hs+1

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

4747 , splitBySeq

4848 , splitWithSuffixSeq

4949 , splitInnerBy

50+ , splitInnerBySuffix

5051

5152 , tapAsync

5253

src/Streamly/Memory/Array.hs+1

@@ -132,6 +132,7 @@ module Streamly.Memory.Array

132132

133133 -- ** Transformation

134134 , AS.splitOn

135+ , AS.splitOnSuffix

135136 -- , AS.compact -- compact

136137 )

137138where

src/Streamly/Memory/Array/Types.hs+38−20

@@ -90,6 +90,9 @@ import Control.DeepSeq (NFData(..))

9090import Control.Monad (when)

9191import Control.Monad.IO.Class (MonadIO(..))

9292import Data.Functor.Identity (runIdentity)

93+#if __GLASGOW_HASKELL__ < 808

94+import Data.Semigroup (Semigroup(..))

95+#endif

9396import Data.Word (Word8)

9497import Foreign.C.String (CString)

9598import Foreign.C.Types (CSize(..), CInt(..))

@@ -100,7 +103,7 @@ import Foreign.Storable (Storable(..))

100103import Prelude hiding (length, foldr, read, unlines, splitAt)

101104import Text.Read (readPrec, readListPrec, readListPrecDefault)

102105

103−import GHC.Base (Addr#, realWorld#)

106+import GHC.Base (Addr#, nullAddr#, realWorld#)

104107import GHC.Exts (IsList, IsString(..))

105108import GHC.ForeignPtr (ForeignPtr(..), newForeignPtr_)

106109import GHC.IO (IO(IO), unsafePerformIO)

@@ -897,6 +900,40 @@ read = D.fromStreamD . toStreamD

897900-- Semigroup and Monoid

898901-------------------------------------------------------------------------------

899902

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

904+{-# INLINE spliceTwo #-}

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

906+spliceTwo arr1 arr2 = do

907+ let src1 = unsafeForeignPtrToPtr (aStart arr1)

908+ src2 = unsafeForeignPtrToPtr (aStart arr2)

909+ len1 = aEnd arr1 `minusPtr` src1

910+ len2 = aEnd arr2 `minusPtr` src2

911+

912+ arr <- liftIO $ newArray (len1 + len2)

913+ let dst = unsafeForeignPtrToPtr (aStart arr)

914+

915+ -- XXX Should we use copyMutableByteArray# instead? Is there an overhead to

916+ -- ccall?

917+ liftIO $ do

918+ memcpy (castPtr dst) (castPtr src1) len1

919+ touchForeignPtr (aStart arr1)

920+ memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2

921+ touchForeignPtr (aStart arr2)

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

923+

924+instance Storable a => Semigroup (Array a) where

925+ arr1 <> arr2 = unsafeInlineIO $ spliceTwo arr1 arr2

926+

927+nullForeignPtr :: ForeignPtr a

928+nullForeignPtr = ForeignPtr nullAddr# (error "nullForeignPtr")

929+

930+nil :: Array a

931+nil = Array nullForeignPtr (Ptr nullAddr#) (Ptr nullAddr#)

932+

933+instance Storable a => Monoid (Array a) where

934+ mempty = nil

935+ mappend = (<>)

936+

900937-------------------------------------------------------------------------------

901938-- IO

902939-------------------------------------------------------------------------------

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

944981 return $ D.Yield x (InnerLoop st startf

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

946983

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

948−{-# INLINE spliceTwo #-}

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

950−spliceTwo arr1 arr2 = do

951− let src1 = unsafeForeignPtrToPtr (aStart arr1)

952− src2 = unsafeForeignPtrToPtr (aStart arr2)

953− len1 = aEnd arr1 `minusPtr` src1

954− len2 = aEnd arr2 `minusPtr` src2

955−

956− arr <- liftIO $ newArray (len1 + len2)

957− let dst = unsafeForeignPtrToPtr (aStart arr)

958−

959− liftIO $ do

960− memcpy (castPtr dst) (castPtr src1) len1

961− touchForeignPtr (aStart arr1)

962− memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2

963− touchForeignPtr (aStart arr2)

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

965−

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

967985-- that there is enough space in the mutable array.

968986{-# INLINE spliceWith #-}

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

@@ -31,6 +31,7 @@ module Streamly.Memory.ArrayStream

3131

3232 -- * Transformation

3333 , splitOn

34+ , splitOnSuffix

3435 , compact -- compact

3536

3637 -- * Elimination

@@ -109,10 +110,19 @@ splitOn

109110 => Word8

110111 -> t m (Array Word8)

111112 -> t m (Array Word8)

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

113113splitOn byte s =

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

115115

116+{-# INLINE splitOnSuffix #-}

117+splitOnSuffix

118+ :: (IsStream t, MonadIO m)

119+ => Word8

120+ -> t m (Array Word8)

121+ -> t m (Array Word8)

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

123+splitOnSuffix byte s =

124+ D.fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.spliceTwo $ D.toStreamD s

125+

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

117127-- maximum specified size in bytes.

118128--

src/Streamly/Prelude/Internal.hs+15

@@ -302,6 +302,7 @@ module Streamly.Prelude.Internal

302302

303303 -- Nested splitting

304304 , splitInnerBy

305+ , splitInnerBySuffix

305306

306307 -- ** Grouping

307308 , groups

@@ -3040,6 +3041,20 @@ splitInnerBy

30403041splitInnerBy splitter joiner xs =

30413042 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs

30423043

3044+-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in

3045+-- a suffix style.

3046+--

3047+-- @since 0.7.0

3048+{-# INLINE splitInnerBySuffix #-}

3049+splitInnerBySuffix

3050+ :: (IsStream t, Monad m, Eq (f a), Monoid (f a))

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

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

3053+ -> t m (f a)

3054+ -> t m (f a)

3055+splitInnerBySuffix splitter joiner xs =

3056+ D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs

3057+

30433058------------------------------------------------------------------------------

30443059-- Reorder in sequence

30453060------------------------------------------------------------------------------

src/Streamly/Streams/StreamD.hs+56

@@ -164,6 +164,7 @@ module Streamly.Streams.StreamD

164164 , splitSuffixOn

165165

166166 , splitInnerBy

167+ , splitInnerBySuffix

167168

168169 -- ** Substreams

169170 , isPrefixOf

@@ -1637,6 +1638,12 @@ data SplitState s arr

16371638 | SplitYielding arr (SplitState s arr)

16381639 | SplitFinishing

16391640

1641+-- XXX An alternative approach would be to use a partial fold (Fold m a b) to

1642+-- split using a splitBy like combinator. The Fold would consume upto the

1643+-- separator and return any leftover which can then be fed to the next fold.

1644+--

1645+-- We can revisit this once we have partial folds/parsers.

1646+--

16401647-- | Performs infix separator style splitting.

16411648{-# INLINE_NORMAL splitInnerBy #-}

16421649splitInnerBy

@@ -1683,6 +1690,55 @@ splitInnerBy splitter joiner (Stream step1 state1) =

16831690 step _ (SplitYielding x next) = return $ Yield x next

16841691 step _ SplitFinishing = return $ Stop

16851692

1693+-- | Performs infix separator style splitting.

1694+{-# INLINE_NORMAL splitInnerBySuffix #-}

1695+splitInnerBySuffix

1696+ :: (Monad m, Eq (f a), Monoid (f a))

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

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

1699+ -> Stream m (f a)

1700+ -> Stream m (f a)

1701+splitInnerBySuffix splitter joiner (Stream step1 state1) =

1702+ (Stream step (SplitInitial state1))

1703+

1704+ where

1705+

1706+ {-# INLINE_LATE step #-}

1707+ step gst (SplitInitial st) = do

1708+ r <- step1 gst st

1709+ case r of

1710+ Yield x s -> do

1711+ (x1, mx2) <- splitter x

1712+ return $ case mx2 of

1713+ Nothing -> Skip (SplitBuffering s x1)

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

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

1716+ Stop -> return $ Stop

1717+

1718+ step gst (SplitBuffering st buf) = do

1719+ r <- step1 gst st

1720+ case r of

1721+ Yield x s -> do

1722+ (x1, mx2) <- splitter x

1723+ buf' <- joiner buf x1

1724+ return $ case mx2 of

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

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

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

1728+ Stop -> return $

1729+ if buf == mempty

1730+ then Stop

1731+ else Skip (SplitYielding buf SplitFinishing)

1732+

1733+ step _ (SplitSplitting st buf) = do

1734+ (x1, mx2) <- splitter buf

1735+ return $ case mx2 of

1736+ Nothing -> Skip $ SplitBuffering st x1

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

1738+

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

1740+ step _ SplitFinishing = return $ Stop

1741+

16861742------------------------------------------------------------------------------

16871743-- Substreams

16881744------------------------------------------------------------------------------

test/String.hs+1−1

@@ -57,7 +57,7 @@ testLinesArray =

5757 monadicIO $ do

5858 xs <- S.toList

5959 $ S.map A.toList

60− $ A.splitOn 10

60+ $ A.splitOnSuffix 10

6161 $ S.yield (A.fromList list)

6262 assert (xs == map (map (fromIntegral . ord))

6363 (lines (map (chr . fromIntegral) list)))