git.iko.soy/streamly
on ffcfcac help sign in
streamly/src/Streamly/Internal/Memory/ArrayStream.hs
225 lines 7.4 kB history raw
1{-# OPTIONS_HADDOCK hide #-}
2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE CPP #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE UnboxedTuples #-}
7{-# LANGUAGE ScopedTypeVariables #-}
9#include "inline.hs"
11-- |
12-- Module : Streamly.Internal.Memory.ArrayStream
13-- Copyright : (c) 2019 Composewell Technologies
14--
15-- License : BSD3
16-- Maintainer : streamly@composewell.com
17-- Stability : experimental
18-- Portability : GHC
19--
20-- Combinators to efficiently manipulate streams of arrays.
21--
22module Streamly.Internal.Memory.ArrayStream
23 (
24 -- * Creation
25 arraysOf
27 -- * Flattening to elements
28 , concat
29 , concatRev
30 , interpose
31 , interposeSuffix
32 , intercalateSuffix
34 -- * Transformation
35 , splitOn
36 , splitOnSuffix
37 , compact -- compact
39 -- * Elimination
40 , toArray
41 )
42where
44import Control.Monad.IO.Class (MonadIO(..))
45-- import Data.Functor.Identity (Identity)
46import Data.Word (Word8)
47import Foreign.ForeignPtr (withForeignPtr)
48import Foreign.Ptr (minusPtr, plusPtr, castPtr)
49import Foreign.Storable (Storable(..))
50import Prelude hiding (length, null, last, map, (!!), read, concat)
52import Streamly.Internal.Memory.Array.Types (Array(..), length)
53import Streamly.Streams.Serial (SerialT)
54import Streamly.Streams.StreamK.Type (IsStream)
56import qualified Streamly.Internal.Memory.Array as A
57import qualified Streamly.Internal.Memory.Array.Types as A
58import qualified Streamly.Internal.Prelude as S
59import qualified Streamly.Streams.StreamD as D
60import qualified Streamly.Streams.Prelude as P
62-- XXX efficiently compare two streams of arrays. Two streams can have chunks
63-- of different sizes, we can handle that in the stream comparison abstraction.
64-- This could be useful e.g. to fast compare whether two files differ.
66-- | Convert a stream of arrays into a stream of their elements.
67--
68-- Same as the following but more efficient:
69--
70-- > concat = S.concatMap A.read
71--
72-- @since 0.7.0
73{-# INLINE concat #-}
74concat :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
75-- concat m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)
76-- concat m = D.fromStreamD $ D.concatMap A.toStreamD (D.toStreamD m)
77concat m = D.fromStreamD $ D.concatMapU A.read (D.toStreamD m)
79-- XXX should we have a reverseArrays API to reverse the stream of arrays
80-- instead?
81--
82-- | Convert a stream of arrays into a stream of their elements reversing the
83-- contents of each array before flattening.
84--
85-- @since 0.7.0
86{-# INLINE concatRev #-}
87concatRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
88concatRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
90-- | Flatten a stream of arrays after inserting the given element between
91-- arrays.
92--
93-- /Internal/
94{-# INLINE interpose #-}
95interpose :: (MonadIO m, IsStream t, Storable a) => a -> t m (Array a) -> t m a
96interpose x = S.interpose x A.read
98{-# INLINE intercalateSuffix #-}
99intercalateSuffix :: (MonadIO m, IsStream t, Storable a)
100 => Array a -> t m (Array a) -> t m a
101intercalateSuffix arr = S.intercalateSuffix arr A.read
103-- | Flatten a stream of arrays appending the given element after each
104-- array.
105--
106-- @since 0.7.0
107{-# INLINE interposeSuffix #-}
108interposeSuffix :: (MonadIO m, IsStream t, Storable a)
109 => a -> t m (Array a) -> t m a
110-- interposeSuffix x = D.fromStreamD . A.unlines x . D.toStreamD
111interposeSuffix x = S.interposeSuffix x A.read
113-- | Split a stream of arrays on a given separator byte, dropping the separator
114-- and coalescing all the arrays between two separators into a single array.
115--
116-- @since 0.7.0
117{-# INLINE splitOn #-}
118splitOn
119 :: (IsStream t, MonadIO m)
120 => Word8
121 -> t m (Array Word8)
122 -> t m (Array Word8)
123splitOn byte s =
124 D.fromStreamD $ D.splitInnerBy (A.breakOn byte) A.spliceTwo $ D.toStreamD s
126{-# INLINE splitOnSuffix #-}
127splitOnSuffix
128 :: (IsStream t, MonadIO m)
129 => Word8
130 -> t m (Array Word8)
131 -> t m (Array Word8)
132-- splitOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s
133splitOnSuffix byte s =
134 D.fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.spliceTwo $ D.toStreamD s
136-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
137-- maximum specified size in bytes.
138--
139-- @since 0.7.0
140{-# INLINE compact #-}
141compact :: (MonadIO m, Storable a)
142 => Int -> SerialT m (Array a) -> SerialT m (Array a)
143compact n xs = D.fromStreamD $ A.packArraysChunksOf n (D.toStreamD xs)
145-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
146-- @n@ elements each.
147--
148-- Same as the following but more efficient:
149--
150-- > arraysOf n = S.chunksOf n (A.writeN n)
151--
152-- @since 0.7.0
153{-# INLINE arraysOf #-}
154arraysOf :: (IsStream t, MonadIO m, Storable a)
155 => Int -> t m a -> t m (Array a)
156arraysOf n str =
157 D.fromStreamD $ A.fromStreamDArraysOf n (D.toStreamD str)
159-- XXX Both of these implementations of splicing seem to perform equally well.
160-- We need to perform benchmarks over a range of sizes though.
162-- CAUTION! length must more than equal to lengths of all the arrays in the
163-- stream.
164{-# INLINE spliceArraysLenUnsafe #-}
165spliceArraysLenUnsafe :: (MonadIO m, Storable a)
166 => Int -> SerialT m (Array a) -> m (Array a)
167spliceArraysLenUnsafe len buffered = do
168 arr <- liftIO $ A.newArray len
169 end <- S.foldlM' writeArr (aEnd arr) buffered
170 return $ arr {aEnd = end}
172 where
174 writeArr dst Array{..} =
175 liftIO $ withForeignPtr aStart $ \src -> do
176 let count = aEnd `minusPtr` src
177 A.memcpy (castPtr dst) (castPtr src) count
178 return $ dst `plusPtr` count
180{-# INLINE _spliceArraysBuffered #-}
181_spliceArraysBuffered :: (MonadIO m, Storable a)
182 => SerialT m (Array a) -> m (Array a)
183_spliceArraysBuffered s = do
184 buffered <- P.foldr S.cons S.nil s
185 len <- S.sum (S.map length buffered)
186 spliceArraysLenUnsafe len s
188{-# INLINE spliceArraysRealloced #-}
189spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)
190 => SerialT m (Array a) -> m (Array a)
191spliceArraysRealloced s = do
192 idst <- liftIO $ A.newArray (A.bytesToElemCount (undefined :: a)
193 (A.mkChunkSizeKB 4))
195 arr <- S.foldlM' A.spliceWithDoubling idst s
196 liftIO $ A.shrinkToFit arr
198-- | Given a stream of arrays, splice them all together to generate a single
199-- array. The stream must be /finite/.
200--
201-- @since 0.7.0
202{-# INLINE toArray #-}
203toArray :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)
204toArray = spliceArraysRealloced
205-- spliceArrays = _spliceArraysBuffered
207-- exponentially increasing sizes of the chunks upto the max limit.
208-- XXX this will be easier to implement with parsers/terminating folds
209-- With this we should be able to reduce the number of chunks/allocations.
210-- The reallocation/copy based toArray can also be implemented using this.
211--
212{-
213{-# INLINE toArraysInRange #-}
214toArraysInRange :: (IsStream t, MonadIO m, Storable a)
215 => Int -> Int -> Fold m (Array a) b -> Fold m a b
216toArraysInRange low high (Fold step initial extract) =
217-}
219{-
220-- | Fold the input to a pure buffered stream (List) of arrays.
221{-# INLINE _toArraysOf #-}
222_toArraysOf :: (MonadIO m, Storable a)
223 => Int -> Fold m a (SerialT Identity (Array a))
224_toArraysOf n = FL.lchunksOf n (A.writeNF n) FL.toStream
225-}