1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE MagicHash #-}
4{-# LANGUAGE RecordWildCards #-}
5{-# LANGUAGE UnboxedTuples #-}
6{-# LANGUAGE ScopedTypeVariables #-}
8#include "Streams/inline.hs"
11-- Module : Streamly.Array
12-- Copyright : (c) 2019 Harendra Kumar
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
19-- Arrays are the computing duals of streams. Streams are good at sequential
20-- access, immutable transformations of in-transit data whereas arrays are good
21-- at random access, in-place transformations of buffered data. Unlike streams
22-- which are potentially infinite, arrays are necessarily /finite/. Arrays can
23-- be used as an efficient interface between streams and external storage
24-- systems like memory, files and network. Streams and arrays complete each
25-- other to provide a general purpose computing system. The design of streamly
26-- as a general purpose computing framework is centered around these two
27-- fundamental aspects of computing and storage.
29-- Even though the implementation allows for in-place mutation, the current
30-- implementation of arrays is immutable. Arrays are of fixed size. The size of
31-- an array is fixed at creation and cannot be changed later without copying.
32-- Arrays are chunks of memory that hold a sequence of 'Storable' values of a
33-- given type. They are designed to store serializable data, they cannot store
34-- non-serializable data like functions, this is reflected by the 'Storable'
35-- constraint. For efficient buffering of data, arrays use pinned memory and
36-- therefore can hold arbitrary number of elements with a single movable
37-- pointer visible to GC, therefore, adding no pressure to GC. Moreover,
38-- pinned memory allows communication with foreign consumers and producers
39-- (e.g. file or network IO) without copying the data.
41-- By design, there are no transformation operations provided in this module,
42-- only stream conversion and IO routines are provided. An array is purely a
43-- data store, therefore, it is not a 'Functor'. Arrays can be operated upon
44-- /efficiently/ by converting them into a stream, applying the desired
45-- transformations on the stream and then converting it back to an array.
46-- 'Foldable' instance is not provided because the implementation would be much
47-- less efficient compared to folding via streams. 'Semigroup' and 'Monoid'
48-- instances are deliberately not provided to avoid misuse; concatenating
49-- arrays using binary operations can be highly inefficient. Instead, use
50-- 'spliceArrays' to concatenate N arrays at once.
54-- * Arrays are finite and fixed in size
55-- * provide /O(1)/ access to elements
56-- * store only data and not functions
57-- * provide efficient IO interfacing
59-- 'ByteString' data type from the 'bytestring' package and the 'Text' data
60-- type from the 'text' package are just special cases of arrays. 'ByteString'
61-- is equivalent to @Array Word8@ and 'Text' is equivalent to a @utf16@ encoded
62-- @Array Word8@. All the 'bytestring' and 'text' operations can be performed
63-- on arrays with equivalent or better performance by converting them to and
66-- This module is designed to be imported qualified:
68-- > import qualified Streamly.Array as A
70-- Each array is one pointer visible to the GC. Too many small arrays (e.g.
71-- single byte) are only as good as holding those elements in a Haskell list.
72-- However, small arrays can be compacted into large ones to reduce the
73-- overhead. To hold 32GB memory in 32k sized buffers we need 1 million arrays
74-- if we use one array for each chunk. This is still significant to add
75-- pressure to GC. However, we can create arrays of arrays (trees) to scale to
76-- arbitrarily large amounts of memory but still using small chunks of
86 -- | When performance matters, the fastest way to generate an array is
87 -- 'fromStreamN'. For regular use, 'IsList' and 'IsString' instances can be
88 -- used to conveniently construct arrays from literal values.
89 -- 'OverloadedLists' extension or 'fromList' can be used to construct an
90 -- array from a list literal. Similarly, 'OverloadedStrings' extension or
91 -- 'fromList' can be used to construct an array from a string literal.
107 -- 'GHC.Exts.toList' from "GHC.Exts" can be used to convert an array to a
114 -- , flattenArraysRev
121 -- * IO: Single Array
124 -- , fromHandlePosUpto
128 -- * IO: Stream of Arrays
129 -- , fromHandleArraysUpto
130 -- , fromHandleArraysOf
135import Control.Monad.IO.Class (MonadIO(..))
136import Data.Word (Word8)
137import Foreign.ForeignPtr (withForeignPtr)
138import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
139import Foreign.Ptr (minusPtr, plusPtr, castPtr)
140import Foreign.Storable (Storable(..))
141import System.IO (Handle, hGetBufSome, hPutBuf)
142import Prelude hiding (length, null, last, map, (!!))
144import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
145import GHC.IO (unsafeDupablePerformIO)
147import Streamly.Streams.Serial (SerialT)
148import Streamly.Streams.StreamK.Type (IsStream, mkStream)
149import Streamly.Array.Types hiding (flattenArrays)
150import qualified Streamly.Array.Types as A
152import qualified Streamly.Streams.StreamD as D
153import qualified Streamly.Prelude as S
155-------------------------------------------------------------------------------
157-------------------------------------------------------------------------------
159-- | Create an 'Array' from the first N elements of a stream. The array is
160-- allocated to size N, if the stream terminates before N elements then the
161-- array may hold less than N elements.
164{-# INLINE fromStreamN #-}
165fromStreamN :: (Monad m, Storable a) => Int -> SerialT m a -> m (Array a)
166fromStreamN n m = fromStreamDN n $ D.toStreamD m
168-------------------------------------------------------------------------------
170-------------------------------------------------------------------------------
172-- | Convert an 'Array' into a stream.
175{-# INLINE_EARLY toStream #-}
176toStream :: (Monad m, IsStream t, Storable a) => Array a -> t m a
177toStream = D.fromStreamD . toStreamD
178-- XXX add fallback to StreamK rule
179-- {-# RULES "Streamly.Array.toStream fallback to StreamK" [1]
180-- forall a. S.toStreamK (toStream a) = K.fromArray a #-}
182-- | Convert an 'Array' into a stream in reverse order.
185{-# INLINE_EARLY toStreamRev #-}
186toStreamRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
187toStreamRev = D.fromStreamD . toStreamDRev
188-- XXX add fallback to StreamK rule
189-- {-# RULES "Streamly.Array.toStreamRev fallback to StreamK" [1]
190-- forall a. S.toStreamK (toStreamRev a) = K.revFromArray a #-}
193null :: Storable a => Array a -> Bool
194null arr = length arr == 0
197_last :: forall a. Storable a => Array a -> Maybe a
198_last arr = arr !! (length arr - 1)
200-------------------------------------------------------------------------------
202-------------------------------------------------------------------------------
204-- | /O(1)/ Lookup the element at the given index.
208(!!) :: Storable a => Array a -> Int -> Maybe a
210 if i < 0 || i > length arr - 1
212 else Just $ unsafeIndex arr i
214-------------------------------------------------------------------------------
216-------------------------------------------------------------------------------
218-- | Convert a stream of arrays into a stream of their elements.
221{-# INLINE flattenArrays #-}
222flattenArrays :: (IsStream t, Monad m, Storable a) => t m (Array a) -> t m a
223flattenArrays m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)
225-- XXX should we have a reverseArrays API to reverse the stream of arrays
228-- | Convert a stream of arrays into a stream of their elements reversing the
229-- contents of each array before flattening.
232{-# INLINE _flattenArraysRev #-}
233_flattenArraysRev :: (IsStream t, Monad m, Storable a) => t m (Array a) -> t m a
234_flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
237-- > arraysOf n = FL.groupsOf n (FL.toArrayN n)
239-- Groups the elements in an input stream into arrays of given size.
242{-# INLINE arraysOf #-}
243arraysOf :: (IsStream t, Monad m, Storable a)
244 => Int -> t m a -> t m (Array a)
246 D.fromStreamD $ fromStreamDArraysOf n (D.toStreamD str)
248-- | Given a stream of arrays, splice them all together to generate a single
249-- array. The stream must be /finite/.
252{-# INLINABLE spliceArrays #-}
253spliceArrays :: (Monad m, Storable a) => SerialT m (Array a) -> m (Array a)
255 buffered <- S.foldr S.cons S.nil s
256 len <- S.sum (S.map length buffered)
258 let !arr = unsafeDupablePerformIO $ unsafeNew len
259 end <- S.foldl' write (aEnd arr) buffered
260 return $ arr {aEnd = end}
264 write dst Array{..} =
265 let !dst' = unsafeInlineIO $ withForeignPtr aStart $ \src -> do
266 let len = aEnd `minusPtr` src
267 memcpy (castPtr dst) (castPtr src) len
268 return $ dst `plusPtr` len
271-- | Create an 'Array' from a stream. This is useful when we want to create a
272-- single array from a stream of unknown size. 'fromStreamN' is at least twice
273-- as efficient when the size is already known.
275-- Note that if the input stream is too large memory allocation for the array
276-- may fail. When the stream size is not known, `arraysOf` followed by
277-- processing of indvidual arrays in the resulting stream should be preferred.
279{-# INLINE fromStream #-}
280fromStream :: (Monad m, Storable a) => SerialT m a -> m (Array a)
281fromStream m = A.fromStreamD $ D.toStreamD m
284-- Move this to FileIO
285-- XXX we should use overWrite/write
286-- | Write a stream of arrays to a handle.
287{-# INLINE toHandleArrays #-}
288toHandleArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()
289toHandleArrays h m = S.mapM_ (liftIO . toHandle h) m
292-------------------------------------------------------------------------------
294-------------------------------------------------------------------------------
296-- | Read a 'ByteArray' from a file handle. If no data is available on the
297-- handle it blocks until some data becomes available. If data is available
298-- then it immediately returns that data without blocking. It reads a maximum
299-- of up to the size requested.
300{-# INLINABLE fromHandleUpto #-}
301fromHandleUpto :: Int -> Handle -> IO (Array Word8)
302fromHandleUpto size h = do
303 ptr <- mallocPlainForeignPtrBytes size
304 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
305 withForeignPtr ptr $ \p -> do
306 n <- hGetBufSome h p size
309 , aEnd = p `plusPtr` n
310 , aBound = p `plusPtr` size
312 -- XXX shrink only if the diff is significant
315-------------------------------------------------------------------------------
317-------------------------------------------------------------------------------
319-- | Write an Array to a file handle.
322{-# INLINABLE toHandle #-}
323toHandle :: Storable a => Handle -> Array a -> IO ()
324toHandle _ arr | null arr = return ()
325toHandle h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
328 let p = unsafeForeignPtrToPtr aStart
331-------------------------------------------------------------------------------
332-- Stream of Arrays IO
333-------------------------------------------------------------------------------
335-- | @fromHandleArraysUpto size h@ reads a stream of arrays from file handle @h@.
336-- The maximum size of a single array is limited to @size@.
337-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
339{-# INLINABLE fromHandleArraysUpto #-}
340fromHandleArraysUpto :: (IsStream t, MonadIO m)
341 => Int -> Handle -> t m (Array Word8)
342fromHandleArraysUpto size h = go
344 -- XXX use cons/nil instead
345 go = mkStream $ \_ yld sng _ -> do
346 vec <- liftIO $ fromHandleUpto size h
351-- XXX read 'Array a' instead of Word8
353-- | @fromHandleArrays h@ reads a stream of arrays from file handle @h@.
354-- The maximum size of a single array is limited to @defaultChunkSize@.
355-- 'fromHandleArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
359{-# INLINE fromHandleArrays #-}
360fromHandleArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
361fromHandleArrays = fromHandleArraysUpto defaultChunkSize
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list