2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ExistentialQuantification #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE ScopedTypeVariables #-}
7{-# LANGUAGE TypeFamilies #-}
8{-# LANGUAGE UnboxedTuples #-}
13-- Module : Streamly.Mem.Array.Types
14-- Copyright : (c) 2019 Composewell Technologies
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
21module Streamly.Mem.Array.Types
63import Control.Exception (assert)
64import Control.DeepSeq (NFData(..))
65import Control.Monad (when)
66import Control.Monad.IO.Class (MonadIO(..))
67import Data.Functor.Identity (runIdentity)
68import Data.Word (Word8)
69import Foreign.C.String (CString)
70import Foreign.C.Types (CSize(..), CInt(..))
71import Foreign.ForeignPtr
72 (ForeignPtr, withForeignPtr, touchForeignPtr)
73import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
74import Foreign.Ptr (plusPtr, minusPtr, castPtr)
75import Foreign.Storable (Storable(..))
76import Prelude hiding (length, foldr, read)
77import Text.Read (readPrec, readListPrec, readListPrecDefault)
79import GHC.Base (Addr#, realWorld#)
80import GHC.Exts (IsList, IsString(..))
81import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes, newForeignPtr_)
82import GHC.IO (IO(IO), unsafePerformIO)
83import GHC.Ptr (Ptr(..))
85import Streamly.Fold.Types (Fold(..))
86import Streamly.SVar (adaptState)
88import qualified Streamly.Streams.StreamD.Type as D
89import qualified Streamly.Streams.StreamK as K
90import qualified GHC.Exts as Exts
93import qualified Data.Foldable as F
96-------------------------------------------------------------------------------
98-------------------------------------------------------------------------------
100-- There are two goals that we want to fulfill with arrays. One, holding large
101-- amounts of data in non-GC memory and the ability to pass this data to and
102-- from the operating system without an extra copy overhead. Two, allow random
103-- access to elements based on their indices. The first one falls in the
104-- category of storage buffers while the second one falls in the category of
105-- maps/multisets/hashmaps.
107-- For the first requirement we use an array of Storables and store it in a
108-- ForeignPtr. We can have both immutable and mutable variants of this array
109-- using wrappers over the same underlying type.
111-- For the second requirement, we need a separate type for arrays of
112-- polymorphic values, for example vectors of handler functions, lookup tables.
113-- We can call this type a "vector" in contrast to arrays. It should not
114-- require Storable instance for the type. In that case we need to use an
115-- Array# instead of a ForeignPtr. This type of array would not reduce the GC
116-- overhead much because each element of the array still needs to be scanned by
117-- the GC. However, it can store polymorphic elements and allow random access
118-- to those. But in most cases random access means storage, and it means we
119-- need to avoid GC scanning except in cases of trivially small storage. One
120-- way to achieve that would be to put the array in a Compact region. However,
121-- if and when we mutate this, we will have to use a manual GC copying out to
122-- another CR and freeing the old one.
124-------------------------------------------------------------------------------
126-------------------------------------------------------------------------------
128-- We require that an array stores only Storable. Arrays are used for buffering
129-- data while streams are used for processing. If you want something to be
130-- buffered it better be Storable so that we can store it in non-GC memory.
132-- We can use a "Storable" constraint in the Array type and the Constraint can
133-- be automatically provided to a function that pattern matches on the Array
134-- type. However, it has huge performance cost, so we do not use it.
135-- XXX Can this cost be alleviated in GHC-8.10 specialization fix?
137-- XXX Another way to not require the Storable constraint in array operations
138-- is to store the elemSize in the array at construction and use that instead
139-- of using sizeOf. Need to charaterize perf cost of this.
141-- XXX rename the fields to "start, next, end".
148 { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
149 , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
150 , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
153-------------------------------------------------------------------------------
155-------------------------------------------------------------------------------
157foreign import ccall unsafe "string.h memcpy" c_memcpy
158 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)
160foreign import ccall unsafe "string.h strlen" c_strlen
161 :: CString -> IO CSize
163-- XXX we are converting Int to CSize
164memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
165memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
167foreign import ccall unsafe "string.h memcmp" c_memcmp
168 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt
170-- XXX we are converting Int to CSize
171-- return True if the memory locations have identical contents
173memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool
175 r <- c_memcmp p1 p2 (fromIntegral len)
178{-# INLINE unsafeInlineIO #-}
179unsafeInlineIO :: IO a -> a
180unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r
182-------------------------------------------------------------------------------
184-------------------------------------------------------------------------------
186-- | Allocate an array that can hold 'count' items. The memory of the array is
189-- Note that this is internal routine, the reference to this array cannot be
190-- given out until the array has been written to and frozen.
191{-# INLINE newArray #-}
192newArray :: forall a. Storable a => Int -> IO (Array a)
194 let size = count * sizeOf (undefined :: a)
195 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
196 let p = unsafeForeignPtrToPtr fptr
200 , aBound = p `plusPtr` size
203-- | Allocate an Array of the given size and run an IO action passing the array
205{-# INLINE withNewArray #-}
206withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)
207withNewArray count f = do
208 arr <- newArray count
209 withForeignPtr (aStart arr) $ \p -> f p >> return arr
211-- XXX grow the array when we are beyond bound.
213-- Internal routine for when the array is being created. Appends one item at
214-- the end of the array. Useful when sequentially writing a stream to the
215-- array. DOES NOT CHECK THE ARRAY BOUNDS.
216{-# INLINE unsafeAppend #-}
217unsafeAppend :: forall a. Storable a => Array a -> a -> IO (Array a)
218unsafeAppend arr@Array{..} x = do
219 when (aEnd == aBound) $
220 error "BUG: unsafeAppend: writing beyond array bounds"
222 touchForeignPtr aStart
223 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
225-- | Remove the free space from an Array.
226shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
227shrinkToFit arr@Array{..} = do
228 assert (aEnd <= aBound) (return ())
231 let oldStart = unsafeForeignPtrToPtr aStart
232 let size = aEnd `minusPtr` oldStart
233 newPtr <- mallocPlainForeignPtrAlignedBytes
234 size (alignment (undefined :: a))
235 withForeignPtr newPtr $ \pNew -> do
236 memcpy (castPtr pNew) (castPtr oldStart) size
237 touchForeignPtr aStart
238 let end = pNew `plusPtr` size
246-- | Expand the free space in the array doubling the size with a minimum
247-- exapnsion of the specified amount of bytes.
248{-# NOINLINE reallocDouble #-}
249reallocDouble :: forall a. Storable a => Int -> Array a -> IO (Array a)
250reallocDouble minIncrease Array{..} = do
251 assert (aEnd <= aBound) (return ())
252 let oldStart = unsafeForeignPtrToPtr aStart
253 let size = aEnd `minusPtr` oldStart
254 newSize = max (size * 2) (size + minIncrease)
255 newPtr <- mallocPlainForeignPtrAlignedBytes
256 newSize (alignment (undefined :: a))
257 withForeignPtr newPtr $ \pNew -> do
258 memcpy (castPtr pNew) (castPtr oldStart) size
259 touchForeignPtr aStart
262 , aEnd = pNew `plusPtr` size
263 , aBound = pNew `plusPtr` newSize
266-- XXX when converting an array of Word8 from a literal string we can simply
267-- refer to the literal string. Is it possible to write rules such that
268-- fromList Word8 can be rewritten so that GHC does not first convert the
269-- literal to [Char] and then we convert it back to an Array Word8?
271-- Note that the address must be a read-only address (meant to be used for
272-- read-only string literals) because we are sharing it, any modification to
273-- the original address would change our array. That's why this function is
275{-# INLINE _fromCStringAddrUnsafe #-}
276_fromCStringAddrUnsafe :: Addr# -> IO (Array Word8)
277_fromCStringAddrUnsafe addr# = do
278 ptr <- newForeignPtr_ (castPtr cstr)
280 let n = fromIntegral len
281 let p = unsafeForeignPtrToPtr ptr
282 let end = p `plusPtr` n
292-------------------------------------------------------------------------------
294-------------------------------------------------------------------------------
296-- | Return element at the specified index without checking the bounds.
298-- Unsafe because it does not check the bounds of the array.
299{-# INLINE_NORMAL unsafeIndexIO #-}
300unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
301unsafeIndexIO Array {..} i =
302 withForeignPtr aStart $ \p -> do
303 let elemSize = sizeOf (undefined :: a)
304 elemOff = p `plusPtr` (elemSize * i)
305 assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
309-- | Return element at the specified index without checking the bounds.
310{-# INLINE_NORMAL unsafeIndex #-}
311unsafeIndex :: forall a. Storable a => Array a -> Int -> a
312unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
314-- | /O(1)/ Get the length of the array.
318length :: forall a. Storable a => Array a -> Int
320 let p = unsafeForeignPtrToPtr aStart
321 aLen = aEnd `minusPtr` p
322 in assert (aLen >= 0) (aLen `div` sizeOf (undefined :: a))
324{-# INLINE_NORMAL toStreamD #-}
325toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
327 let p = unsafeForeignPtrToPtr aStart
332 {-# INLINE_LATE step #-}
333 step _ p | p == aEnd = return D.Stop
335 -- unsafeInlineIO allows us to run this in Identity monad for pure
336 -- toList/foldr case which makes them much faster due to not
337 -- accumulating the list and fusing better with the pure consumers.
339 -- This should be safe as the array contents are guaranteed to be
340 -- evaluated/written to before we peek at them.
341 let !x = unsafeInlineIO $ do
343 touchForeignPtr aStart
345 return $ D.Yield x (p `plusPtr` (sizeOf (undefined :: a)))
347{-# INLINE_NORMAL toStreamDRev #-}
348toStreamDRev :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
349toStreamDRev Array{..} =
350 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
355 {-# INLINE_LATE step #-}
356 step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop
358 -- See comments in toStreamD for why we use unsafeInlineIO
359 let !x = unsafeInlineIO $ do
361 touchForeignPtr aStart
363 return $ D.Yield x (p `plusPtr` negate (sizeOf (undefined :: a)))
365{-# INLINE_NORMAL foldl' #-}
366foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
367foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
369{-# INLINE_NORMAL foldr #-}
370foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
371foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
373-------------------------------------------------------------------------------
375-------------------------------------------------------------------------------
377-- | @toArrayN n@ folds a maximum of @n@ elements from the input stream to an
381{-# INLINE_NORMAL toArrayN #-}
382toArrayN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
383toArrayN n = Fold step initial extract
388 if n < 0 then error "toArrayN: negative count specified" else return ()
390 step arr@(Array _ end bound) _ | end == bound = return arr
391 step (Array start end bound) x = do
393 return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
394 extract = liftIO . shrinkToFit
396{-# INLINE_NORMAL fromStreamDN #-}
397fromStreamDN :: forall m a. (MonadIO m, Storable a)
398 => Int -> D.Stream m a -> m (Array a)
399fromStreamDN limit str = do
400 arr <- liftIO $ newArray limit
401 end <- D.foldlM' write (aEnd arr) $ D.take limit str
402 return $ arr {aEnd = end}
408 return $ ptr `plusPtr` sizeOf (undefined :: a)
410-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
412{-# INLINE fromStreamDArraysOf #-}
413fromStreamDArraysOf :: (MonadIO m, Storable a)
414 => Int -> D.Stream m a -> D.Stream m (Array a)
415fromStreamDArraysOf n str = D.groupsOf n (toArrayN n) str
417-- XXX concatMap does not seem to have the best possible performance so we have
418-- a custom way to concat arrays.
419data FlattenState s a =
421 | InnerLoop s (ForeignPtr a) (Ptr a) (Ptr a)
423{-# INLINE_NORMAL flattenArrays #-}
424flattenArrays :: forall m a. (MonadIO m, Storable a)
425 => D.Stream m (Array a) -> D.Stream m a
426flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)
430 {-# INLINE_LATE step' #-}
431 step' gst (OuterLoop st) = do
432 r <- step (adaptState gst) st
434 D.Yield Array{..} s ->
435 let p = unsafeForeignPtrToPtr aStart
436 in D.Skip (InnerLoop s aStart p aEnd)
437 D.Skip s -> D.Skip (OuterLoop s)
440 step' _ (InnerLoop st _ p end) | p == end =
441 return $ D.Skip $ OuterLoop st
443 step' _ (InnerLoop st startf p end) = do
446 touchForeignPtr startf
448 return $ D.Yield x (InnerLoop st startf
449 (p `plusPtr` (sizeOf (undefined :: a))) end)
451{-# INLINE_NORMAL flattenArraysRev #-}
452flattenArraysRev :: forall m a. (MonadIO m, Storable a)
453 => D.Stream m (Array a) -> D.Stream m a
454flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
458 {-# INLINE_LATE step' #-}
459 step' gst (OuterLoop st) = do
460 r <- step (adaptState gst) st
462 D.Yield Array{..} s ->
463 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
464 -- XXX we do not need aEnd
465 in D.Skip (InnerLoop s aStart p aEnd)
466 D.Skip s -> D.Skip (OuterLoop s)
469 step' _ (InnerLoop st start p _) | p < unsafeForeignPtrToPtr start =
470 return $ D.Skip $ OuterLoop st
472 step' _ (InnerLoop st startf p end) = do
475 touchForeignPtr startf
477 return $ D.Yield x (InnerLoop st startf
478 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
480-- CAUTION: a very large number (millions) of arrays can degrade performance
481-- due to GC overhead because we need to buffer the arrays before we flatten
484-- We could take the approach of doubling the memory allocation on each
485-- overflow. This would result in more or less the same amount of copying as in
486-- the chunking approach. However, if we have to shrink in the end then it may
487-- result in an extra copy of the entire data.
489{-# INLINE fromStreamD #-}
490fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
492 let s = fromStreamDArraysOf defaultChunkSize m
493 buffered <- D.foldr K.cons K.nil s
494 len <- K.foldl' (+) 0 (K.map length buffered)
495 fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
497-- | Convert an 'Array' into a list.
500{-# INLINABLE toList #-}
501toList :: Storable a => Array a -> [a]
504instance (Show a, Storable a) => Show (Array a) where
505 {-# INLINE showsPrec #-}
506 showsPrec _ = shows . toList
508-- | Create an 'Array' from the first N elements of a list. The array is
509-- allocated to size N, if the list terminates before N elements then the
510-- array may hold less than N elements.
513{-# INLINABLE fromListN #-}
514fromListN :: Storable a => Int -> [a] -> Array a
515fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
517-- | Create an 'Array' from a list. The list must be of finite size.
520{-# INLINABLE fromList #-}
521fromList :: Storable a => [a] -> Array a
522fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
524instance (Storable a, Read a, Show a) => Read (Array a) where
525 {-# INLINE readPrec #-}
529 readListPrec = readListPrecDefault
531instance (a ~ Char) => IsString (Array a) where
532 {-# INLINE fromString #-}
533 fromString = fromList
535-- GHC versions 8.0 and below cannot derive IsList
536instance Storable a => IsList (Array a) where
537 type (Item (Array a)) = a
538 {-# INLINE fromList #-}
540 {-# INLINE fromListN #-}
541 fromListN = fromListN
542 {-# INLINE toList #-}
546arrcmp :: Array a -> Array a -> Bool
548 let !res = unsafeInlineIO $ do
549 let ptr1 = unsafeForeignPtrToPtr $ aStart arr1
550 let ptr2 = unsafeForeignPtrToPtr $ aStart arr2
551 let len1 = aEnd arr1 `minusPtr` ptr1
552 let len2 = aEnd arr2 `minusPtr` ptr2
556 r <- memcmp (castPtr ptr1) (castPtr ptr2) len1
557 touchForeignPtr $ aStart arr1
558 touchForeignPtr $ aStart arr2
563-- XXX we are assuming that Storable equality means element equality. This may
564-- or may not be correct? arrcmp is 40% faster compared to stream equality.
565instance (Storable a, Eq a) => Eq (Array a) where
568 -- arr1 == arr2 = runIdentity $ D.eqBy (==) (toStreamD arr1) (toStreamD arr2)
570instance (Storable a, NFData a) => NFData (Array a) where
572 rnf = foldl' (\_ x -> rnf x) ()
574instance (Storable a, Ord a) => Ord (Array a) where
575 {-# INLINE compare #-}
576 compare arr1 arr2 = unsafePerformIO $
577 D.cmpBy compare (toStreamD arr1) (toStreamD arr2)
579 -- Default definitions defined in base do not have an INLINE on them, so we
580 -- replicate them here with an INLINE.
582 x < y = case compare x y of { LT -> True; _ -> False }
585 x <= y = case compare x y of { GT -> False; _ -> True }
588 x > y = case compare x y of { GT -> True; _ -> False }
591 x >= y = case compare x y of { LT -> False; _ -> True }
593 -- These two default methods use '<=' rather than 'compare'
594 -- because the latter is often more expensive
596 max x y = if x <= y then y else x
599 min x y = if x <= y then x else y
602-- Definitions using the Storable constraint from the Array type. These are to
603-- make the Foldable instance possible though it is much slower (7x slower).
605{-# INLINE_NORMAL toStreamD_ #-}
606toStreamD_ :: forall m a. MonadIO m => Int -> Array a -> D.Stream m a
607toStreamD_ size Array{..} =
608 let p = unsafeForeignPtrToPtr aStart
613 {-# INLINE_LATE step #-}
614 step _ p | p == aEnd = return D.Stop
618 touchForeignPtr aStart
620 return $ D.Yield x (p `plusPtr` size)
622{-# INLINE_NORMAL _foldr #-}
623_foldr :: forall a b. (a -> b -> b) -> b -> Array a -> b
624_foldr f z arr@Array {..} =
625 let !n = sizeOf (undefined :: a)
626 in unsafePerformIO $ D.foldr f z $ toStreamD_ n arr
628-- | Note that the 'Foldable' instance is 7x slower than the direct
630instance Foldable Array where
634-- | Convert an 'Array' into a stream.
637{-# INLINE_EARLY read #-}
638read :: (Monad m, K.IsStream t, Storable a) => Array a -> t m a
639read = D.fromStreamD . toStreamD
640-- XXX add fallback to StreamK rule
641-- {-# RULES "Streamly.Array.read fallback to StreamK" [1]
642-- forall a. S.readK (read a) = K.fromArray a #-}
644-------------------------------------------------------------------------------
645-- Semigroup and Monoid
646-------------------------------------------------------------------------------
648-------------------------------------------------------------------------------
650-------------------------------------------------------------------------------
652-- | GHC memory management allocation header overhead
654allocOverhead = 2 * sizeOf (undefined :: Int)
656mkChunkSize :: Int -> Int
657mkChunkSize n = let size = n - allocOverhead in max size 0
659mkChunkSizeKB :: Int -> Int
660mkChunkSizeKB n = mkChunkSize (n * k)
663-- | Default maximum buffer size in bytes, for reading from and writing to IO
664-- devices, the value is 32KB minus GHC allocation overhead, which is a few
665-- bytes, so that the actual allocation is 32KB.
666defaultChunkSize :: Int
667defaultChunkSize = mkChunkSizeKB 32
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