git.iko.soy/streamly
on master help sign in
Deprecate Array/Stream module, move parsing funcs to Array
901c6ebHarendra Kumar
core/src/DocTestDataArray.hs+2
compare with master
unifiedsplit
@@ -11,7 +11,9 @@
11
11
12>>> import qualified Streamly.Data.Array as Array
12>>> import qualified Streamly.Data.Array as Array
13>>> import qualified Streamly.Data.Fold as Fold
13>>> import qualified Streamly.Data.Fold as Fold
14+>>> import qualified Streamly.Data.ParserK as ParserK
14>>> import qualified Streamly.Data.Stream as Stream
15>>> import qualified Streamly.Data.Stream as Stream
16+>>> import qualified Streamly.Data.StreamK as StreamK
15
17
16For APIs that have not been released yet.
18For APIs that have not been released yet.
17
19
core/src/Streamly/Internal/Data/Array.hs+363−3
@@ -77,6 +77,11 @@ module Streamly.Internal.Data.Array
77 , compactOnByte
77 , compactOnByte
78 , compactOnByteSuffix
78 , compactOnByteSuffix
79
79
80+ , foldBreakChunks
81+ , foldChunks
82+ , foldBreakChunksK
83+ , parseBreakChunksK
84+
80 -- * Serialization
85 -- * Serialization
81 , encodeAs
86 , encodeAs
82 , serialize
87 , serialize
@@ -95,32 +100,42 @@ where
95#include "ArrayMacros.h"
100#include "ArrayMacros.h"
96
101
97import Control.Monad.IO.Class (MonadIO(..))
102import Control.Monad.IO.Class (MonadIO(..))
103+-- import Data.Bifunctor (first)
104+-- import Data.Either (fromRight)
98import Data.Functor.Identity (Identity)
105import Data.Functor.Identity (Identity)
99import Data.Proxy (Proxy(..))
106import Data.Proxy (Proxy(..))
100import Data.Word (Word8)
107import Data.Word (Word8)
101import Foreign.C.String (CString)
108import Foreign.C.String (CString)
102import Foreign.Ptr (castPtr)
109import Foreign.Ptr (castPtr)
103import Foreign.Storable (Storable)
110import Foreign.Storable (Storable)
111+import GHC.Types (SPEC(..))
104import Streamly.Internal.Data.Unbox (Unbox(..))
112import Streamly.Internal.Data.Unbox (Unbox(..))
105import Prelude hiding (length, null, last, map, (!!), read, concat)
113import Prelude hiding (length, null, last, map, (!!), read, concat)
106
114
107import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..), MutByteArray)
115import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..), MutByteArray)
108import Streamly.Internal.Data.Serialize.Type (Serialize)
116import Streamly.Internal.Data.Serialize.Type (Serialize)
109import Streamly.Internal.Data.Fold.Type (Fold(..))
117import Streamly.Internal.Data.Fold.Type (Fold(..))
110−import Streamly.Internal.Data.Stream (Stream)
118+import Streamly.Internal.Data.Parser (Parser(..), Initial(..), ParseError(..))
119+import Streamly.Internal.Data.Stream (Stream(..))
120+import Streamly.Internal.Data.StreamK (StreamK)
111−import Streamly.Internal.Data.SVar.Type (adaptState)
121+import Streamly.Internal.Data.SVar.Type (adaptState, defState)
112−import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))
122+import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3Fused'(..))
113import Streamly.Internal.Data.Unfold.Type (Unfold(..))
123import Streamly.Internal.Data.Unfold.Type (Unfold(..))
114import Streamly.Internal.System.IO (unsafeInlineIO)
124import Streamly.Internal.System.IO (unsafeInlineIO)
115
125
126+import qualified Streamly.Internal.Data.Fold.Type as Fold
116import qualified Streamly.Internal.Data.Serialize.Type as Serialize
127import qualified Streamly.Internal.Data.Serialize.Type as Serialize
117import qualified Streamly.Internal.Data.MutByteArray.Type as MBA
128import qualified Streamly.Internal.Data.MutByteArray.Type as MBA
118import qualified Streamly.Internal.Data.MutArray as MA
129import qualified Streamly.Internal.Data.MutArray as MA
119import qualified Streamly.Internal.Data.Fold as FL
130import qualified Streamly.Internal.Data.Fold as FL
120import qualified Streamly.Internal.Data.Ring as RB
131import qualified Streamly.Internal.Data.Ring as RB
132+import qualified Streamly.Internal.Data.Parser as Parser
133+-- import qualified Streamly.Internal.Data.ParserK as ParserK
121import qualified Streamly.Internal.Data.Stream as D
134import qualified Streamly.Internal.Data.Stream as D
122import qualified Streamly.Internal.Data.Stream as Stream
135import qualified Streamly.Internal.Data.Stream as Stream
136+import qualified Streamly.Internal.Data.StreamK as StreamK
123import qualified Streamly.Internal.Data.Unfold as Unfold
137import qualified Streamly.Internal.Data.Unfold as Unfold
138+import qualified Prelude
124
139
125import Streamly.Internal.Data.Array.Type
140import Streamly.Internal.Data.Array.Type
126
141
@@ -631,3 +646,348 @@ compactOnByteSuffix
631 -> Stream m (Array Word8)
646 -> Stream m (Array Word8)
632compactOnByteSuffix byte =
647compactOnByteSuffix byte =
633 fmap unsafeFreeze . MA.compactOnByteSuffix byte . fmap unsafeThaw
648 fmap unsafeFreeze . MA.compactOnByteSuffix byte . fmap unsafeThaw
649+
650+-------------------------------------------------------------------------------
651+-- Folding Streams of Arrays
652+-------------------------------------------------------------------------------
653+
654+-- XXX This should not be used for breaking a stream as the D.cons used in
655+-- reconstructing the stream could be very bad for performance. This can only
656+-- be useful in folding without breaking.
657+{-# INLINE_NORMAL foldBreakChunks #-}
658+foldBreakChunks :: forall m a b. (MonadIO m, Unbox a) =>
659+ Fold m a b -> Stream m (Array a) -> m (b, Stream m (Array a))
660+foldBreakChunks (Fold fstep initial _ final) stream@(Stream step state) = do
661+ res <- initial
662+ case res of
663+ Fold.Partial fs -> go SPEC state fs
664+ Fold.Done fb -> return $! (fb, stream)
665+
666+ where
667+
668+ {-# INLINE go #-}
669+ go !_ st !fs = do
670+ r <- step defState st
671+ case r of
672+ Stream.Yield (Array contents start end) s ->
673+ let fp = Tuple' end contents
674+ in goArray SPEC s fp start fs
675+ Stream.Skip s -> go SPEC s fs
676+ Stream.Stop -> do
677+ b <- final fs
678+ return (b, D.nil)
679+
680+ goArray !_ s (Tuple' end _) !cur !fs
681+ | cur == end = do
682+ go SPEC s fs
683+ goArray !_ st fp@(Tuple' end contents) !cur !fs = do
684+ x <- liftIO $ peekAt cur contents
685+ res <- fstep fs x
686+ let next = INDEX_NEXT(cur,a)
687+ case res of
688+ Fold.Done b -> do
689+ let arr = Array contents next end
690+ return $! (b, D.cons arr (D.Stream step st))
691+ Fold.Partial fs1 -> goArray SPEC st fp next fs1
692+
693+-- This may be more robust wrt fusion compared to unfoldMany?
694+
695+-- | Fold a stream of arrays using a 'Fold'. This is equivalent to the
696+-- following:
697+--
698+-- >>> foldChunks f = Stream.fold f . Stream.unfoldMany Array.reader
699+--
700+foldChunks :: (MonadIO m, Unbox a) => Fold m a b -> Stream m (Array a) -> m b
701+foldChunks f s = fmap fst (foldBreakChunks f s)
702+-- foldStream f = Stream.fold f . Stream.unfoldMany reader
703+
704+-- | Fold a stream of arrays using a 'Fold' and return the remaining stream.
705+--
706+-- The following alternative to this function allows composing the fold using
707+-- the parser Monad:
708+--
709+-- @
710+-- foldBreakStreamK f s =
711+-- fmap (first (fromRight undefined))
712+-- $ StreamK.parseBreakChunks (ParserK.adaptC (Parser.fromFold f)) s
713+-- @
714+--
715+-- We can compare perf and remove this one or define it in terms of that.
716+--
717+foldBreakChunksK :: forall m a b. (MonadIO m, Unbox a) =>
718+ Fold m a b -> StreamK m (Array a) -> m (b, StreamK m (Array a))
719+{-
720+foldBreakChunksK f s =
721+ fmap (first (fromRight undefined))
722+ $ StreamK.parseBreakChunks (ParserK.adaptC (Parser.fromFold f)) s
723+-}
724+foldBreakChunksK (Fold fstep initial _ final) stream = do
725+ res <- initial
726+ case res of
727+ Fold.Partial fs -> go fs stream
728+ Fold.Done fb -> return (fb, stream)
729+
730+ where
731+
732+ {-# INLINE go #-}
733+ go !fs st = do
734+ let stop = (, StreamK.nil) <$> final fs
735+ single a = yieldk a StreamK.nil
736+ yieldk (Array contents start end) r =
737+ let fp = Tuple' end contents
738+ in goArray fs r fp start
739+ in StreamK.foldStream defState yieldk single stop st
740+
741+ goArray !fs st (Tuple' end _) !cur
742+ | cur == end = do
743+ go fs st
744+ goArray !fs st fp@(Tuple' end contents) !cur = do
745+ x <- liftIO $ peekAt cur contents
746+ res <- fstep fs x
747+ let next = INDEX_NEXT(cur,a)
748+ case res of
749+ Fold.Done b -> do
750+ let arr = Array contents next end
751+ return $! (b, StreamK.cons arr st)
752+ Fold.Partial fs1 -> goArray fs1 st fp next
753+
754+{-
755+-- This can be generalized to any type provided it can be unfolded to a stream
756+-- and it can be combined using a semigroup operation.
757+--
758+{-# INLINE_NORMAL parseBreakD #-}
759+parseBreakD ::
760+ forall m a b. (MonadIO m, MonadThrow m, Unbox a)
761+ => PRD.Parser a m b
762+ -> D.Stream m (Array.Array a)
763+ -> m (b, D.Stream m (Array.Array a))
764+parseBreakD
765+ (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do
766+
767+ res <- initial
768+ case res of
769+ PRD.IPartial s -> go SPEC state (List []) s
770+ PRD.IDone b -> return (b, stream)
771+ PRD.IError err -> throwM $ ParseError err
772+
773+ where
774+
775+ -- "backBuf" contains last few items in the stream that we may have to
776+ -- backtrack to.
777+ --
778+ -- XXX currently we are using a dumb list based approach for backtracking
779+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
780+ -- That will allow us more efficient random back and forth movement.
781+ go !_ st backBuf !pst = do
782+ r <- step defState st
783+ case r of
784+ D.Yield (Array contents start end) s ->
785+ gobuf SPEC s backBuf
786+ (Tuple' end contents) start pst
787+ D.Skip s -> go SPEC s backBuf pst
788+ D.Stop -> do
789+ b <- extract pst
790+ return (b, D.nil)
791+
792+ -- Use strictness on "cur" to keep it unboxed
793+ gobuf !_ s backBuf (Tuple' end _) !cur !pst
794+ | cur == end = do
795+ go SPEC s backBuf pst
796+ gobuf !_ s backBuf fp@(Tuple' end contents) !cur !pst = do
797+ x <- liftIO $ peekByteIndex contents cur
798+ pRes <- pstep pst x
799+ let next = INDEX_NEXT(cur,a)
800+ case pRes of
801+ PR.Partial 0 pst1 ->
802+ gobuf SPEC s (List []) fp next pst1
803+ PR.Partial n pst1 -> do
804+ assert (n <= Prelude.length (x:getList backBuf)) (return ())
805+ let src0 = Prelude.take n (x:getList backBuf)
806+ arr0 = A.fromListN n (Prelude.reverse src0)
807+ arr1 = Array contents next end
808+ src = arr0 <> arr1
809+ let !(Array cont1 start end1) = src
810+ fp1 = Tuple' end1 cont1
811+ gobuf SPEC s (List []) fp1 start pst1
812+ PR.Continue 0 pst1 ->
813+ gobuf SPEC s (List (x:getList backBuf)) fp next pst1
814+ PR.Continue n pst1 -> do
815+ assert (n <= Prelude.length (x:getList backBuf)) (return ())
816+ let (src0, buf1) = splitAt n (x:getList backBuf)
817+ arr0 = A.fromListN n (Prelude.reverse src0)
818+ arr1 = Array contents next end
819+ src = arr0 <> arr1
820+ let !(Array cont1 start end1) = src
821+ fp1 = Tuple' end1 cont1
822+ gobuf SPEC s (List buf1) fp1 start pst1
823+ PR.Done 0 b -> do
824+ let arr = Array contents next end
825+ return (b, D.cons arr (D.Stream step s))
826+ PR.Done n b -> do
827+ assert (n <= Prelude.length (x:getList backBuf)) (return ())
828+ let src0 = Prelude.take n (x:getList backBuf)
829+ -- XXX create the array in reverse instead
830+ arr0 = A.fromListN n (Prelude.reverse src0)
831+ arr1 = Array contents next end
832+ -- XXX Use StreamK to avoid adding arbitrary layers of
833+ -- constructors every time.
834+ str = D.cons arr0 (D.cons arr1 (D.Stream step s))
835+ return (b, str)
836+ PR.Error err -> throwM $ ParseError err
837+-}
838+
839+-- | Parse an array stream using the supplied 'Parser'. Returns the parse
840+-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
841+--
842+-- The following alternative to this function allows composing the parser using
843+-- the parser Monad:
844+--
845+-- >>> parseBreakStreamK p = StreamK.parseBreakChunks (ParserK.adaptC p)
846+--
847+-- We can compare perf and remove this one or define it in terms of that.
848+--
849+-- /Internal/
850+--
851+{-# INLINE_NORMAL parseBreakChunksK #-}
852+parseBreakChunksK ::
853+ forall m a b. (MonadIO m, Unbox a)
854+ => Parser a m b
855+ -> StreamK m (Array a)
856+ -> m (Either ParseError b, StreamK m (Array a))
857+-- parseBreakStreamK p = StreamK.parseBreakChunks (ParserK.adaptC p)
858+parseBreakChunksK (Parser pstep initial extract) stream = do
859+ res <- initial
860+ case res of
861+ IPartial s -> go s stream []
862+ IDone b -> return (Right b, stream)
863+ IError err -> return (Left (ParseError err), stream)
864+
865+ where
866+
867+ -- "backBuf" contains last few items in the stream that we may have to
868+ -- backtrack to.
869+ --
870+ -- XXX currently we are using a dumb list based approach for backtracking
871+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
872+ -- That will allow us more efficient random back and forth movement.
873+ go !pst st backBuf = do
874+ let stop = goStop pst backBuf -- (, K.nil) <$> extract pst
875+ single a = yieldk a StreamK.nil
876+ yieldk arr r = goArray pst backBuf r arr
877+ in StreamK.foldStream defState yieldk single stop st
878+
879+ -- Use strictness on "cur" to keep it unboxed
880+ goArray !pst backBuf st (Array _ cur end) | cur == end = go pst st backBuf
881+ goArray !pst backBuf st (Array contents cur end) = do
882+ x <- liftIO $ peekAt cur contents
883+ pRes <- pstep pst x
884+ let next = INDEX_NEXT(cur,a)
885+ case pRes of
886+ Parser.Partial 0 s ->
887+ goArray s [] st (Array contents next end)
888+ Parser.Partial n s -> do
889+ assert (n <= Prelude.length (x:backBuf)) (return ())
890+ let src0 = Prelude.take n (x:backBuf)
891+ arr0 = fromListN n (Prelude.reverse src0)
892+ arr1 = Array contents next end
893+ src = arr0 <> arr1
894+ goArray s [] st src
895+ Parser.Continue 0 s ->
896+ goArray s (x:backBuf) st (Array contents next end)
897+ Parser.Continue n s -> do
898+ assert (n <= Prelude.length (x:backBuf)) (return ())
899+ let (src0, buf1) = Prelude.splitAt n (x:backBuf)
900+ arr0 = fromListN n (Prelude.reverse src0)
901+ arr1 = Array contents next end
902+ src = arr0 <> arr1
903+ goArray s buf1 st src
904+ Parser.Done 0 b -> do
905+ let arr = Array contents next end
906+ return (Right b, StreamK.cons arr st)
907+ Parser.Done n b -> do
908+ assert (n <= Prelude.length (x:backBuf)) (return ())
909+ let src0 = Prelude.take n (x:backBuf)
910+ -- XXX Use fromListRevN once implemented
911+ -- arr0 = A.fromListRevN n src0
912+ arr0 = fromListN n (Prelude.reverse src0)
913+ arr1 = Array contents next end
914+ str = StreamK.cons arr0 (StreamK.cons arr1 st)
915+ return (Right b, str)
916+ Parser.Error err -> do
917+ let n = Prelude.length backBuf
918+ arr0 = fromListN n (Prelude.reverse backBuf)
919+ arr1 = Array contents cur end
920+ str = StreamK.cons arr0 (StreamK.cons arr1 stream)
921+ return (Left (ParseError err), str)
922+
923+ -- This is a simplified goArray
924+ goExtract !pst backBuf (Array _ cur end)
925+ | cur == end = goStop pst backBuf
926+ goExtract !pst backBuf (Array contents cur end) = do
927+ x <- liftIO $ peekAt cur contents
928+ pRes <- pstep pst x
929+ let next = INDEX_NEXT(cur,a)
930+ case pRes of
931+ Parser.Partial 0 s ->
932+ goExtract s [] (Array contents next end)
933+ Parser.Partial n s -> do
934+ assert (n <= Prelude.length (x:backBuf)) (return ())
935+ let src0 = Prelude.take n (x:backBuf)
936+ arr0 = fromListN n (Prelude.reverse src0)
937+ arr1 = Array contents next end
938+ src = arr0 <> arr1
939+ goExtract s [] src
940+ Parser.Continue 0 s ->
941+ goExtract s backBuf (Array contents next end)
942+ Parser.Continue n s -> do
943+ assert (n <= Prelude.length (x:backBuf)) (return ())
944+ let (src0, buf1) = Prelude.splitAt n (x:backBuf)
945+ arr0 = fromListN n (Prelude.reverse src0)
946+ arr1 = Array contents next end
947+ src = arr0 <> arr1
948+ goExtract s buf1 src
949+ Parser.Done 0 b -> do
950+ let arr = Array contents next end
951+ return (Right b, StreamK.fromPure arr)
952+ Parser.Done n b -> do
953+ assert (n <= Prelude.length backBuf) (return ())
954+ let src0 = Prelude.take n backBuf
955+ -- XXX Use fromListRevN once implemented
956+ -- arr0 = A.fromListRevN n src0
957+ arr0 = fromListN n (Prelude.reverse src0)
958+ arr1 = Array contents next end
959+ str = StreamK.cons arr0 (StreamK.fromPure arr1)
960+ return (Right b, str)
961+ Parser.Error err -> do
962+ let n = Prelude.length backBuf
963+ arr0 = fromListN n (Prelude.reverse backBuf)
964+ arr1 = Array contents cur end
965+ str = StreamK.cons arr0 (StreamK.cons arr1 stream)
966+ return (Left (ParseError err), str)
967+
968+ -- This is a simplified goExtract
969+ {-# INLINE goStop #-}
970+ goStop !pst backBuf = do
971+ pRes <- extract pst
972+ case pRes of
973+ Parser.Partial _ _ -> error "Bug: parseBreak: Partial in extract"
974+ Parser.Continue 0 s ->
975+ goStop s backBuf
976+ Parser.Continue n s -> do
977+ assert (n <= Prelude.length backBuf) (return ())
978+ let (src0, buf1) = Prelude.splitAt n backBuf
979+ arr = fromListN n (Prelude.reverse src0)
980+ goExtract s buf1 arr
981+ Parser.Done 0 b ->
982+ return (Right b, StreamK.nil)
983+ Parser.Done n b -> do
984+ assert (n <= Prelude.length backBuf) (return ())
985+ let src0 = Prelude.take n backBuf
986+ -- XXX Use fromListRevN once implemented
987+ -- arr0 = A.fromListRevN n src0
988+ arr0 = fromListN n (Prelude.reverse src0)
989+ return (Right b, StreamK.fromPure arr0)
990+ Parser.Error err -> do
991+ let n = Prelude.length backBuf
992+ arr0 = fromListN n (Prelude.reverse backBuf)
993+ return (Left (ParseError err), StreamK.fromPure arr0)
core/src/Streamly/Internal/Data/Array/Stream.hs+4−299
@@ -12,6 +12,7 @@
12-- keep this as a separate module and release it.
12-- keep this as a separate module and release it.
13--
13--
14module Streamly.Internal.Data.Array.Stream
14module Streamly.Internal.Data.Array.Stream
15+{-# DEPRECATED "Please use \"Streamly.Internal.Data.Array\" instead." #-}
15 (
16 (
16 -- * Creation
17 -- * Creation
17 Array.chunksOf
18 Array.chunksOf
@@ -95,11 +96,9 @@ import Streamly.Internal.Data.Parser (ParseError(..))
95import Streamly.Internal.Data.Stream (Stream)
96import Streamly.Internal.Data.Stream (Stream)
96import Streamly.Internal.Data.StreamK (StreamK, fromStream, toStream)
97import Streamly.Internal.Data.StreamK (StreamK, fromStream, toStream)
97import Streamly.Internal.Data.SVar.Type (adaptState, defState)
98import Streamly.Internal.Data.SVar.Type (adaptState, defState)
98−import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
99
99
100import qualified Streamly.Internal.Data.Array as A
100import qualified Streamly.Internal.Data.Array as A
101import qualified Streamly.Internal.Data.Array as Array
101import qualified Streamly.Internal.Data.Array as Array
102−import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))
103import qualified Streamly.Internal.Data.Parser as PR
102import qualified Streamly.Internal.Data.Parser as PR
104import qualified Streamly.Internal.Data.Parser as PRD
103import qualified Streamly.Internal.Data.Parser as PRD
105 (Parser(..), Initial(..))
104 (Parser(..), Initial(..))
@@ -181,71 +180,7 @@ splitOnSuffix = Array.compactOnByteSuffix
181{-# INLINE_NORMAL foldBreakD #-}
180{-# INLINE_NORMAL foldBreakD #-}
182foldBreakD :: forall m a b. (MonadIO m, Unbox a) =>
181foldBreakD :: forall m a b. (MonadIO m, Unbox a) =>
183 Fold m a b -> D.Stream m (Array a) -> m (b, D.Stream m (Array a))
182 Fold m a b -> D.Stream m (Array a) -> m (b, D.Stream m (Array a))
184−foldBreakD (FL.Fold fstep initial _ final) stream@(D.Stream step state) = do
183+foldBreakD = Array.foldBreakChunks
185− res <- initial
186− case res of
187− FL.Partial fs -> go SPEC state fs
188− FL.Done fb -> return $! (fb, stream)
189−
190− where
191−
192− {-# INLINE go #-}
193− go !_ st !fs = do
194− r <- step defState st
195− case r of
196− D.Yield (Array contents start end) s ->
197− let fp = Tuple' end contents
198− in goArray SPEC s fp start fs
199− D.Skip s -> go SPEC s fs
200− D.Stop -> do
201− b <- final fs
202− return (b, D.nil)
203−
204− goArray !_ s (Tuple' end _) !cur !fs
205− | cur == end = do
206− go SPEC s fs
207− goArray !_ st fp@(Tuple' end contents) !cur !fs = do
208− x <- liftIO $ peekByteIndex cur contents
209− res <- fstep fs x
210− let next = INDEX_NEXT(cur,a)
211− case res of
212− FL.Done b -> do
213− let arr = Array contents next end
214− return $! (b, D.cons arr (D.Stream step st))
215− FL.Partial fs1 -> goArray SPEC st fp next fs1
216−
217−{-# INLINE_NORMAL foldBreakK #-}
218−foldBreakK :: forall m a b. (MonadIO m, Unbox a) =>
219− Fold m a b -> K.StreamK m (Array a) -> m (b, K.StreamK m (Array a))
220−foldBreakK (FL.Fold fstep initial _ final) stream = do
221− res <- initial
222− case res of
223− FL.Partial fs -> go fs stream
224− FL.Done fb -> return (fb, stream)
225−
226− where
227−
228− {-# INLINE go #-}
229− go !fs st = do
230− let stop = (, K.nil) <$> final fs
231− single a = yieldk a K.nil
232− yieldk (Array contents start end) r =
233− let fp = Tuple' end contents
234− in goArray fs r fp start
235− in K.foldStream defState yieldk single stop st
236−
237− goArray !fs st (Tuple' end _) !cur
238− | cur == end = do
239− go fs st
240− goArray !fs st fp@(Tuple' end contents) !cur = do
241− x <- liftIO $ peekByteIndex cur contents
242− res <- fstep fs x
243− let next = INDEX_NEXT(cur,a)
244− case res of
245− FL.Done b -> do
246− let arr = Array contents next end
247− return $! (b, K.cons arr st)
248− FL.Partial fs1 -> goArray fs1 st fp next
249
184
250-- | Fold an array stream using the supplied 'Fold'. Returns the fold result
185-- | Fold an array stream using the supplied 'Fold'. Returns the fold result
251-- and the unconsumed stream.
186-- and the unconsumed stream.
@@ -271,7 +206,7 @@ foldBreak ::
271 => Fold m a b
206 => Fold m a b
272 -> StreamK m (A.Array a)
207 -> StreamK m (A.Array a)
273 -> m (b, StreamK m (A.Array a))
208 -> m (b, StreamK m (A.Array a))
274−foldBreak = foldBreakK
209+foldBreak = Array.foldBreakChunksK
275--
210--
276-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)
211-- foldBreak f s = fmap fromStreamD <$> foldBreakD f (toStreamD s)
277--
212--
@@ -335,236 +270,6 @@ splitAtArrayListRev n ls
335{-# ANN type List NoSpecConstr #-}
270{-# ANN type List NoSpecConstr #-}
336newtype List a = List {getList :: [a]}
271newtype List a = List {getList :: [a]}
337
272
338−{-
339−-- This can be generalized to any type provided it can be unfolded to a stream
340−-- and it can be combined using a semigroup operation.
341−--
342−-- XXX This should be written using CPS (as parseK) if we want it to scale wrt
343−-- to the number of times it can be called on the same stream.
344−{-# INLINE_NORMAL parseBreakD #-}
345−parseBreakD ::
346− forall m a b. (MonadIO m, MonadThrow m, Unbox a)
347− => PRD.Parser a m b
348− -> D.Stream m (Array.Array a)
349− -> m (b, D.Stream m (Array.Array a))
350−parseBreakD
351− (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do
352−
353− res <- initial
354− case res of
355− PRD.IPartial s -> go SPEC state (List []) s
356− PRD.IDone b -> return (b, stream)
357− PRD.IError err -> throwM $ ParseError err
358−
359− where
360−
361− -- "backBuf" contains last few items in the stream that we may have to
362− -- backtrack to.
363− --
364− -- XXX currently we are using a dumb list based approach for backtracking
365− -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
366− -- That will allow us more efficient random back and forth movement.
367− go !_ st backBuf !pst = do
368− r <- step defState st
369− case r of
370− D.Yield (Array contents start end) s ->
371− gobuf SPEC s backBuf
372− (Tuple' end contents) start pst
373− D.Skip s -> go SPEC s backBuf pst
374− D.Stop -> do
375− b <- extract pst
376− return (b, D.nil)
377−
378− -- Use strictness on "cur" to keep it unboxed
379− gobuf !_ s backBuf (Tuple' end _) !cur !pst
380− | cur == end = do
381− go SPEC s backBuf pst
382− gobuf !_ s backBuf fp@(Tuple' end contents) !cur !pst = do
383− x <- liftIO $ peekByteIndex contents cur
384− pRes <- pstep pst x
385− let next = INDEX_NEXT(cur,a)
386− case pRes of
387− PR.Partial 0 pst1 ->
388− gobuf SPEC s (List []) fp next pst1
389− PR.Partial n pst1 -> do
390− assert (n <= Prelude.length (x:getList backBuf)) (return ())
391− let src0 = Prelude.take n (x:getList backBuf)
392− arr0 = A.fromListN n (Prelude.reverse src0)
393− arr1 = Array contents next end
394− src = arr0 <> arr1
395− let !(Array cont1 start end1) = src
396− fp1 = Tuple' end1 cont1
397− gobuf SPEC s (List []) fp1 start pst1
398− PR.Continue 0 pst1 ->
399− gobuf SPEC s (List (x:getList backBuf)) fp next pst1
400− PR.Continue n pst1 -> do
401− assert (n <= Prelude.length (x:getList backBuf)) (return ())
402− let (src0, buf1) = splitAt n (x:getList backBuf)
403− arr0 = A.fromListN n (Prelude.reverse src0)
404− arr1 = Array contents next end
405− src = arr0 <> arr1
406− let !(Array cont1 start end1) = src
407− fp1 = Tuple' end1 cont1
408− gobuf SPEC s (List buf1) fp1 start pst1
409− PR.Done 0 b -> do
410− let arr = Array contents next end
411− return (b, D.cons arr (D.Stream step s))
412− PR.Done n b -> do
413− assert (n <= Prelude.length (x:getList backBuf)) (return ())
414− let src0 = Prelude.take n (x:getList backBuf)
415− -- XXX create the array in reverse instead
416− arr0 = A.fromListN n (Prelude.reverse src0)
417− arr1 = Array contents next end
418− -- XXX Use StreamK to avoid adding arbitrary layers of
419− -- constructors every time.
420− str = D.cons arr0 (D.cons arr1 (D.Stream step s))
421− return (b, str)
422− PR.Error err -> throwM $ ParseError err
423−-}
424−
425−{-# INLINE_NORMAL parseBreakK #-}
426−parseBreakK ::
427− forall m a b. (MonadIO m, Unbox a)
428− => PRD.Parser a m b
429− -> K.StreamK m (Array.Array a)
430− -> m (Either ParseError b, K.StreamK m (Array.Array a))
431−parseBreakK (PRD.Parser pstep initial extract) stream = do
432− res <- initial
433− case res of
434− PRD.IPartial s -> go s stream []
435− PRD.IDone b -> return (Right b, stream)
436− PRD.IError err -> return (Left (ParseError err), stream)
437−
438− where
439−
440− -- "backBuf" contains last few items in the stream that we may have to
441− -- backtrack to.
442− --
443− -- XXX currently we are using a dumb list based approach for backtracking
444− -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
445− -- That will allow us more efficient random back and forth movement.
446− go !pst st backBuf = do
447− let stop = goStop pst backBuf -- (, K.nil) <$> extract pst
448− single a = yieldk a K.nil
449− yieldk arr r = goArray pst backBuf r arr
450− in K.foldStream defState yieldk single stop st
451−
452− -- Use strictness on "cur" to keep it unboxed
453− goArray !pst backBuf st (Array _ cur end) | cur == end = go pst st backBuf
454− goArray !pst backBuf st (Array contents cur end) = do
455− x <- liftIO $ peekByteIndex cur contents
456− pRes <- pstep pst x
457− let next = INDEX_NEXT(cur,a)
458− case pRes of
459− PR.Partial 0 s ->
460− goArray s [] st (Array contents next end)
461− PR.Partial n s -> do
462− assert (n <= Prelude.length (x:backBuf)) (return ())
463− let src0 = Prelude.take n (x:backBuf)
464− arr0 = A.fromListN n (Prelude.reverse src0)
465− arr1 = Array contents next end
466− src = arr0 <> arr1
467− goArray s [] st src
468− PR.Continue 0 s ->
469− goArray s (x:backBuf) st (Array contents next end)
470− PR.Continue n s -> do
471− assert (n <= Prelude.length (x:backBuf)) (return ())
472− let (src0, buf1) = splitAt n (x:backBuf)
473− arr0 = A.fromListN n (Prelude.reverse src0)
474− arr1 = Array contents next end
475− src = arr0 <> arr1
476− goArray s buf1 st src
477− PR.Done 0 b -> do
478− let arr = Array contents next end
479− return (Right b, K.cons arr st)
480− PR.Done n b -> do
481− assert (n <= Prelude.length (x:backBuf)) (return ())
482− let src0 = Prelude.take n (x:backBuf)
483− -- XXX Use fromListRevN once implemented
484− -- arr0 = A.fromListRevN n src0
485− arr0 = A.fromListN n (Prelude.reverse src0)
486− arr1 = Array contents next end
487− str = K.cons arr0 (K.cons arr1 st)
488− return (Right b, str)
489− PR.Error err -> do
490− let n = Prelude.length backBuf
491− arr0 = A.fromListN n (Prelude.reverse backBuf)
492− arr1 = Array contents cur end
493− str = K.cons arr0 (K.cons arr1 stream)
494− return (Left (ParseError err), str)
495−
496− -- This is a simplified goArray
497− goExtract !pst backBuf (Array _ cur end)
498− | cur == end = goStop pst backBuf
499− goExtract !pst backBuf (Array contents cur end) = do
500− x <- liftIO $ peekByteIndex cur contents
501− pRes <- pstep pst x
502− let next = INDEX_NEXT(cur,a)
503− case pRes of
504− PR.Partial 0 s ->
505− goExtract s [] (Array contents next end)
506− PR.Partial n s -> do
507− assert (n <= Prelude.length (x:backBuf)) (return ())
508− let src0 = Prelude.take n (x:backBuf)
509− arr0 = A.fromListN n (Prelude.reverse src0)
510− arr1 = Array contents next end
511− src = arr0 <> arr1
512− goExtract s [] src
513− PR.Continue 0 s ->
514− goExtract s backBuf (Array contents next end)
515− PR.Continue n s -> do
516− assert (n <= Prelude.length (x:backBuf)) (return ())
517− let (src0, buf1) = splitAt n (x:backBuf)
518− arr0 = A.fromListN n (Prelude.reverse src0)
519− arr1 = Array contents next end
520− src = arr0 <> arr1
521− goExtract s buf1 src
522− PR.Done 0 b -> do
523− let arr = Array contents next end
524− return (Right b, K.fromPure arr)
525− PR.Done n b -> do
526− assert (n <= Prelude.length backBuf) (return ())
527− let src0 = Prelude.take n backBuf
528− -- XXX Use fromListRevN once implemented
529− -- arr0 = A.fromListRevN n src0
530− arr0 = A.fromListN n (Prelude.reverse src0)
531− arr1 = Array contents next end
532− str = K.cons arr0 (K.fromPure arr1)
533− return (Right b, str)
534− PR.Error err -> do
535− let n = Prelude.length backBuf
536− arr0 = A.fromListN n (Prelude.reverse backBuf)
537− arr1 = Array contents cur end
538− str = K.cons arr0 (K.cons arr1 stream)
539− return (Left (ParseError err), str)
540−
541− -- This is a simplified goExtract
542− {-# INLINE goStop #-}
543− goStop !pst backBuf = do
544− pRes <- extract pst
545− case pRes of
546− PR.Partial _ _ -> error "Bug: parseBreak: Partial in extract"
547− PR.Continue 0 s ->
548− goStop s backBuf
549− PR.Continue n s -> do
550− assert (n <= Prelude.length backBuf) (return ())
551− let (src0, buf1) = splitAt n backBuf
552− arr = A.fromListN n (Prelude.reverse src0)
553− goExtract s buf1 arr
554− PR.Done 0 b ->
555− return (Right b, K.nil)
556− PR.Done n b -> do
557− assert (n <= Prelude.length backBuf) (return ())
558− let src0 = Prelude.take n backBuf
559− -- XXX Use fromListRevN once implemented
560− -- arr0 = A.fromListRevN n src0
561− arr0 = A.fromListN n (Prelude.reverse src0)
562− return (Right b, K.fromPure arr0)
563− PR.Error err -> do
564− let n = Prelude.length backBuf
565− arr0 = A.fromListN n (Prelude.reverse backBuf)
566− return (Left (ParseError err), K.fromPure arr0)
567−
568-- | Parse an array stream using the supplied 'Parser'. Returns the parse
273-- | Parse an array stream using the supplied 'Parser'. Returns the parse
569-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
274-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
570--
275--
@@ -588,7 +293,7 @@ parseBreak ::
588parseBreak p s =
293parseBreak p s =
589 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
294 fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
590-}
295-}
591−parseBreak = parseBreakK
296+parseBreak = Array.parseBreakChunksK
592-- parseBreak p = K.parseBreakChunks (ParserK.adaptC p)
297-- parseBreak p = K.parseBreakChunks (ParserK.adaptC p)
593
298
594-------------------------------------------------------------------------------
299-------------------------------------------------------------------------------
core/src/Streamly/Internal/Data/MutArray.hs+4
@@ -87,6 +87,10 @@ getSlicesFromLen :: forall m a. (Monad m, Unbox a)
87 -> Unfold m (MutArray a) (MutArray a)
87 -> Unfold m (MutArray a) (MutArray a)
88getSlicesFromLen = slicerFromLen
88getSlicesFromLen = slicerFromLen
89
89
90+-------------------------------------------------------------------------------
91+-- Compacting Streams of Arrays
92+-------------------------------------------------------------------------------
93+
90-- | Scan @compactLE n@ coalesces adjacent arrays in the input stream
94-- | Scan @compactLE n@ coalesces adjacent arrays in the input stream
91-- only if the combined size would be less than or equal to n.
95-- only if the combined size would be less than or equal to n.
92{-# INLINE compactLE #-}
96{-# INLINE compactLE #-}
core/src/Streamly/Internal/FileSystem/Handle.hs+2−6
@@ -133,14 +133,12 @@ import Streamly.Internal.Data.Unfold.Type (Unfold(..))
133import Streamly.Internal.Data.Array.Type
133import Streamly.Internal.Data.Array.Type
134 (Array(..), pinnedWriteNUnsafe, unsafeFreezeWithShrink, byteLength)
134 (Array(..), pinnedWriteNUnsafe, unsafeFreezeWithShrink, byteLength)
135import Streamly.Internal.Data.Stream.Type (Stream)
135import Streamly.Internal.Data.Stream.Type (Stream)
136−import Streamly.Internal.Data.Array.Stream (lpackArraysChunksOf)
137-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
136-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
138import Streamly.Internal.System.IO (defaultChunkSize)
137import Streamly.Internal.System.IO (defaultChunkSize)
139
138
140import qualified Streamly.Data.Fold as FL
139import qualified Streamly.Data.Fold as FL
141import qualified Streamly.Data.Array as A
140import qualified Streamly.Data.Array as A
142import qualified Streamly.Internal.Data.Array.Type as A
141import qualified Streamly.Internal.Data.Array.Type as A
143−import qualified Streamly.Internal.Data.Array.Stream as AS
144import qualified Streamly.Internal.Data.MutArray.Type as MArray
142import qualified Streamly.Internal.Data.MutArray.Type as MArray
145import qualified Streamly.Internal.Data.Refold.Type as Refold
143import qualified Streamly.Internal.Data.Refold.Type as Refold
146import qualified Streamly.Internal.Data.Fold.Type as FL(refoldMany)
144import qualified Streamly.Internal.Data.Fold.Type as FL(refoldMany)
@@ -421,7 +419,7 @@ putChunks h = S.fold (FL.drainMapM (putChunk h))
421{-# INLINE putChunksWith #-}
419{-# INLINE putChunksWith #-}
422putChunksWith :: (MonadIO m, Unbox a)
420putChunksWith :: (MonadIO m, Unbox a)
423 => Int -> Handle -> Stream m (Array a) -> m ()
421 => Int -> Handle -> Stream m (Array a) -> m ()
424−putChunksWith n h xs = putChunks h $ AS.compact n xs
422+putChunksWith n h xs = putChunks h $ A.compactLE n xs
425
423
426-- > putBytesWith n h m = Handle.putChunks h $ A.pinnedChunksOf n m
424-- > putBytesWith n h m = Handle.putChunks h $ A.pinnedChunksOf n m
427
425
@@ -461,8 +459,6 @@ writeChunks h = FL.drainMapM (putChunk h)
461chunkWriter :: MonadIO m => Refold m Handle (Array a) ()
459chunkWriter :: MonadIO m => Refold m Handle (Array a) ()
462chunkWriter = Refold.drainBy putChunk
460chunkWriter = Refold.drainBy putChunk
463
461
464−-- XXX lpackArraysChunksOf should be written idiomatically
465−
466-- | @writeChunksWith bufsize handle@ writes a stream of arrays
462-- | @writeChunksWith bufsize handle@ writes a stream of arrays
467-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
463-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
468-- We never split an array, if a single array is bigger than the specified size
464-- We never split an array, if a single array is bigger than the specified size
@@ -472,7 +468,7 @@ chunkWriter = Refold.drainBy putChunk
472{-# INLINE writeChunksWith #-}
468{-# INLINE writeChunksWith #-}
473writeChunksWith :: (MonadIO m, Unbox a)
469writeChunksWith :: (MonadIO m, Unbox a)
474 => Int -> Handle -> Fold m (Array a) ()
470 => Int -> Handle -> Fold m (Array a) ()
475−writeChunksWith n h = lpackArraysChunksOf n (writeChunks h)
471+writeChunksWith n h = A.lCompactGE n (writeChunks h)
476
472
477-- | Same as 'writeChunksWith'
473-- | Same as 'writeChunksWith'
478--
474--
src/Streamly/Internal/Network/Socket.hs+2−3
@@ -87,7 +87,6 @@ import Prelude hiding (read)
87import qualified Network.Socket as Net
87import qualified Network.Socket as Net
88
88
89import Streamly.Internal.Data.Array (Array(..))
89import Streamly.Internal.Data.Array (Array(..))
90−import Streamly.Internal.Data.Array.Stream (lpackArraysChunksOf)
91import Streamly.Data.Fold (Fold)
90import Streamly.Data.Fold (Fold)
92import Streamly.Data.Stream (Stream)
91import Streamly.Data.Stream (Stream)
93import Streamly.Internal.Data.Unfold (Unfold(..))
92import Streamly.Internal.Data.Unfold (Unfold(..))
@@ -100,7 +99,7 @@ import qualified Streamly.Data.Stream as S
100import qualified Streamly.Data.Unfold as UF
99import qualified Streamly.Data.Unfold as UF
101import qualified Streamly.Internal.Data.Array as A
100import qualified Streamly.Internal.Data.Array as A
102 ( unsafeFreeze, asPtrUnsafe, byteLength, pinnedChunksOf,
101 ( unsafeFreeze, asPtrUnsafe, byteLength, pinnedChunksOf,
103− pinnedWriteN, pinnedWriteNUnsafe )
102+ pinnedWriteN, pinnedWriteNUnsafe, lCompactGE )
104import qualified Streamly.Internal.Data.MutArray as MArray
103import qualified Streamly.Internal.Data.MutArray as MArray
105 (MutArray(..), asPtrUnsafe, pinnedNewBytes)
104 (MutArray(..), asPtrUnsafe, pinnedNewBytes)
106import qualified Streamly.Internal.Data.Stream as S (fromStreamK, Stream(..), Step(..))
105import qualified Streamly.Internal.Data.Stream as S (fromStreamK, Stream(..), Step(..))
@@ -473,7 +472,7 @@ writeChunks h = FL.drainMapM (liftIO . putChunk h)
473{-# INLINE writeChunksWith #-}
472{-# INLINE writeChunksWith #-}
474writeChunksWith :: (MonadIO m, Unbox a)
473writeChunksWith :: (MonadIO m, Unbox a)
475 => Int -> Socket -> Fold m (Array a) ()
474 => Int -> Socket -> Fold m (Array a) ()
476−writeChunksWith n h = lpackArraysChunksOf n (writeChunks h)
475+writeChunksWith n h = A.lCompactGE n (writeChunks h)
477
476
478-- | Same as 'writeChunksWith'
477-- | Same as 'writeChunksWith'
479--
478--