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