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
61import Control.Exception (assert)
62import Control.DeepSeq (NFData(..))
63import Control.Monad (when)
64import Control.Monad.IO.Class (MonadIO(..))
65import Data.Functor.Identity (runIdentity)
66import Data.Word (Word8)
67import Foreign.C.String (CString)
68import Foreign.C.Types (CSize(..), CInt(..))
69import Foreign.ForeignPtr
70 (ForeignPtr, withForeignPtr, touchForeignPtr)
71import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
72import Foreign.Ptr (plusPtr, minusPtr, castPtr)
73import Foreign.Storable (Storable(..))
74import Prelude hiding (length, foldr, read)
75import Text.Read (readPrec, readListPrec, readListPrecDefault)
77import GHC.Base (Addr#, realWorld#)
78import GHC.Exts (IsList, IsString(..))
79import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes, newForeignPtr_)
80import GHC.IO (IO(IO), unsafePerformIO)
81import GHC.Ptr (Ptr(..))
83import Streamly.Fold.Types (Fold(..))
84import Streamly.SVar (adaptState)
86import qualified Streamly.Streams.StreamD.Type as D
87import qualified Streamly.Streams.StreamK as K
88import qualified GHC.Exts as Exts
91import qualified Data.Foldable as F
94-------------------------------------------------------------------------------
96-------------------------------------------------------------------------------
98-- There are two goals that we want to fulfill with arrays. One, holding large
99-- amounts of data in non-GC memory and the ability to pass this data to and
100-- from the operating system without an extra copy overhead. Two, allow random
101-- access to elements based on their indices. The first one falls in the
102-- category of storage buffers while the second one falls in the category of
103-- maps/multisets/hashmaps.
105-- For the first requirement we use an array of Storables and store it in a
106-- ForeignPtr. We can have both immutable and mutable variants of this array
107-- using wrappers over the same underlying type.
109-- For the second requirement, we need a separate type for arrays of
110-- polymorphic values, for example vectors of handler functions, lookup tables.
111-- We can call this type a "vector" in contrast to arrays. It should not
112-- require Storable instance for the type. In that case we need to use an
113-- Array# instead of a ForeignPtr. This type of array would not reduce the GC
114-- overhead much because each element of the array still needs to be scanned by
115-- the GC. However, it can store polymorphic elements and allow random access
116-- to those. But in most cases random access means storage, and it means we
117-- need to avoid GC scanning except in cases of trivially small storage. One
118-- way to achieve that would be to put the array in a Compact region. However,
119-- if and when we mutate this, we will have to use a manual GC copying out to
120-- another CR and freeing the old one.
122-------------------------------------------------------------------------------
124-------------------------------------------------------------------------------
126-- We require that an array stores only Storable. Arrays are used for buffering
127-- data while streams are used for processing. If you want something to be
128-- buffered it better be Storable so that we can store it in non-GC memory.
130-- We can use a "Storable" constraint in the Array type and the Constraint can
131-- be automatically provided to a function that pattern matches on the Array
132-- type. However, it has huge performance cost, so we do not use it.
133-- XXX Can this cost be alleviated in GHC-8.10 specialization fix?
135-- XXX Another way to not require the Storable constraint in array operations
136-- is to store the elemSize in the array at construction and use that instead
137-- of using sizeOf. Need to charaterize perf cost of this.
139-- XXX rename the fields to "start, next, end".
146 { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
147 , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
148 , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
151-------------------------------------------------------------------------------
153-------------------------------------------------------------------------------
155foreign import ccall unsafe "string.h memcpy" c_memcpy
156 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)
158foreign import ccall unsafe "string.h strlen" c_strlen
159 :: CString -> IO CSize
161-- XXX we are converting Int to CSize
162memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
163memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
165foreign import ccall unsafe "string.h memcmp" c_memcmp
166 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt
168-- XXX we are converting Int to CSize
169-- return True if the memory locations have identical contents
171memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool
173 r <- c_memcmp p1 p2 (fromIntegral len)
176{-# INLINE unsafeInlineIO #-}
177unsafeInlineIO :: IO a -> a
178unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r
180-------------------------------------------------------------------------------
182-------------------------------------------------------------------------------
184-- | Allocate an array that can hold 'count' items. The memory of the array is
187-- Note that this is internal routine, the reference to this array cannot be
188-- given out until the array has been written to and frozen.
189{-# INLINE newArray #-}
190newArray :: forall a. Storable a => Int -> IO (Array a)
192 let size = count * sizeOf (undefined :: a)
193 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
194 let p = unsafeForeignPtrToPtr fptr
198 , aBound = p `plusPtr` size
201-- | Allocate an Array of the given size and run an IO action passing the array
203{-# INLINE withNewArray #-}
204withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)
205withNewArray count f = do
206 arr <- newArray count
207 withForeignPtr (aStart arr) $ \p -> f p >> return arr
209-- XXX grow the array when we are beyond bound.
211-- Internal routine for when the array is being created. Appends one item at
212-- the end of the array. Useful when sequentially writing a stream to the
213-- array. DOES NOT CHECK THE ARRAY BOUNDS.
214{-# INLINE unsafeAppend #-}
215unsafeAppend :: forall a. Storable a => Array a -> a -> IO (Array a)
216unsafeAppend arr@Array{..} x = do
217 when (aEnd == aBound) $
218 error "BUG: unsafeAppend: writing beyond array bounds"
220 touchForeignPtr aStart
221 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
223-- | Remove the free space from an Array.
224shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
225shrinkToFit arr@Array{..} = do
226 assert (aEnd <= aBound) (return ())
229 let oldStart = unsafeForeignPtrToPtr aStart
230 let size = aEnd `minusPtr` oldStart
231 newPtr <- mallocPlainForeignPtrAlignedBytes
232 size (alignment (undefined :: a))
233 withForeignPtr newPtr $ \pNew -> do
234 memcpy (castPtr pNew) (castPtr oldStart) size
235 touchForeignPtr aStart
236 let end = pNew `plusPtr` size
244-- XXX when converting an array of Word8 from a literal string we can simply
245-- refer to the literal string. Is it possible to write rules such that
246-- fromList Word8 can be rewritten so that GHC does not first convert the
247-- literal to [Char] and then we convert it back to an Array Word8?
249-- Note that the address must be a read-only address (meant to be used for
250-- read-only string literals) because we are sharing it, any modification to
251-- the original address would change our array. That's why this function is
253{-# INLINE _fromCStringAddrUnsafe #-}
254_fromCStringAddrUnsafe :: Addr# -> IO (Array Word8)
255_fromCStringAddrUnsafe addr# = do
256 ptr <- newForeignPtr_ (castPtr cstr)
258 let n = fromIntegral len
259 let p = unsafeForeignPtrToPtr ptr
260 let end = p `plusPtr` n
270-------------------------------------------------------------------------------
272-------------------------------------------------------------------------------
274-- | Return element at the specified index without checking the bounds.
276-- Unsafe because it does not check the bounds of the array.
277{-# INLINE_NORMAL unsafeIndexIO #-}
278unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
279unsafeIndexIO Array {..} i =
280 withForeignPtr aStart $ \p -> do
281 let elemSize = sizeOf (undefined :: a)
282 elemOff = p `plusPtr` (elemSize * i)
283 assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
287-- | Return element at the specified index without checking the bounds.
288{-# INLINE_NORMAL unsafeIndex #-}
289unsafeIndex :: forall a. Storable a => Array a -> Int -> a
290unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
292-- | /O(1)/ Get the length of the array.
296length :: forall a. Storable a => Array a -> Int
298 let p = unsafeForeignPtrToPtr aStart
299 aLen = aEnd `minusPtr` p
300 in assert (aLen >= 0) (aLen `div` sizeOf (undefined :: a))
302{-# INLINE_NORMAL toStreamD #-}
303toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
305 let p = unsafeForeignPtrToPtr aStart
310 {-# INLINE_LATE step #-}
311 step _ p | p == aEnd = return D.Stop
313 -- unsafeInlineIO allows us to run this in Identity monad for pure
314 -- toList/foldr case which makes them much faster due to not
315 -- accumulating the list and fusing better with the pure consumers.
317 -- This should be safe as the array contents are guaranteed to be
318 -- evaluated/written to before we peek at them.
319 let !x = unsafeInlineIO $ do
321 touchForeignPtr aStart
323 return $ D.Yield x (p `plusPtr` (sizeOf (undefined :: a)))
325{-# INLINE_NORMAL toStreamDRev #-}
326toStreamDRev :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
327toStreamDRev Array{..} =
328 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
333 {-# INLINE_LATE step #-}
334 step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop
336 -- See comments in toStreamD for why we use unsafeInlineIO
337 let !x = unsafeInlineIO $ do
339 touchForeignPtr aStart
341 return $ D.Yield x (p `plusPtr` negate (sizeOf (undefined :: a)))
343{-# INLINE_NORMAL foldl' #-}
344foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
345foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
347{-# INLINE_NORMAL foldr #-}
348foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
349foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
351-------------------------------------------------------------------------------
353-------------------------------------------------------------------------------
355-- | @toArrayN n@ folds a maximum of @n@ elements from the input stream to an
359{-# INLINE_NORMAL toArrayN #-}
360toArrayN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
361toArrayN n = Fold step initial extract
365 initial = liftIO $ newArray n
366 step arr@(Array _ end bound) _ | end == bound = return arr
367 step (Array start end bound) x = do
369 return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
373-- This can be implemented by combining the List of arrays from toArrays into a
375-- | Fold the whole input to a single array.
376{-# INLINE toArray #-}
377toArray :: forall m a. (Monad m, Storable a) => Fold m a (Array a)
378toArray = Fold step begin done
381{-# INLINE_NORMAL fromStreamDN #-}
382fromStreamDN :: forall m a. (MonadIO m, Storable a)
383 => Int -> D.Stream m a -> m (Array a)
384fromStreamDN limit str = do
385 arr <- liftIO $ newArray limit
386 end <- D.foldlM' write (aEnd arr) $ D.take limit str
387 return $ arr {aEnd = end}
393 return $ ptr `plusPtr` sizeOf (undefined :: a)
395-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
397{-# INLINE fromStreamDArraysOf #-}
398fromStreamDArraysOf :: (MonadIO m, Storable a)
399 => Int -> D.Stream m a -> D.Stream m (Array a)
400fromStreamDArraysOf n str = D.groupsOf n (toArrayN n) str
402-- XXX concatMap does not seem to have the best possible performance so we have
403-- a custom way to concat arrays.
404data FlattenState s a =
406 | InnerLoop s (ForeignPtr a) (Ptr a) (Ptr a)
408{-# INLINE_NORMAL flattenArrays #-}
409flattenArrays :: forall m a. (MonadIO m, Storable a)
410 => D.Stream m (Array a) -> D.Stream m a
411flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)
415 {-# INLINE_LATE step' #-}
416 step' gst (OuterLoop st) = do
417 r <- step (adaptState gst) st
419 D.Yield Array{..} s ->
420 let p = unsafeForeignPtrToPtr aStart
421 in D.Skip (InnerLoop s aStart p aEnd)
422 D.Skip s -> D.Skip (OuterLoop s)
425 step' _ (InnerLoop st _ p end) | p == end =
426 return $ D.Skip $ OuterLoop st
428 step' _ (InnerLoop st startf p end) = do
431 touchForeignPtr startf
433 return $ D.Yield x (InnerLoop st startf
434 (p `plusPtr` (sizeOf (undefined :: a))) end)
436{-# INLINE_NORMAL flattenArraysRev #-}
437flattenArraysRev :: forall m a. (MonadIO m, Storable a)
438 => D.Stream m (Array a) -> D.Stream m a
439flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
443 {-# INLINE_LATE step' #-}
444 step' gst (OuterLoop st) = do
445 r <- step (adaptState gst) st
447 D.Yield Array{..} s ->
448 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
449 -- XXX we do not need aEnd
450 in D.Skip (InnerLoop s aStart p aEnd)
451 D.Skip s -> D.Skip (OuterLoop s)
454 step' _ (InnerLoop st start p _) | p < unsafeForeignPtrToPtr start =
455 return $ D.Skip $ OuterLoop st
457 step' _ (InnerLoop st startf p end) = do
460 touchForeignPtr startf
462 return $ D.Yield x (InnerLoop st startf
463 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
465-- CAUTION: a very large number (millions) of arrays can degrade performance
466-- due to GC overhead because we need to buffer the arrays before we flatten
469-- We could take the approach of doubling the memory allocation on each
470-- overflow. This would result in more or less the same amount of copying as in
471-- the chunking approach. However, if we have to shrink in the end then it may
472-- result in an extra copy of the entire data.
474{-# INLINE fromStreamD #-}
475fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
477 let s = fromStreamDArraysOf defaultChunkSize m
478 buffered <- D.foldr K.cons K.nil s
479 len <- K.foldl' (+) 0 (K.map length buffered)
480 fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
482-- | Convert an 'Array' into a list.
485{-# INLINABLE toList #-}
486toList :: Storable a => Array a -> [a]
489instance (Show a, Storable a) => Show (Array a) where
490 {-# INLINE showsPrec #-}
491 showsPrec _ = shows . toList
493-- | Create an 'Array' from the first N elements of a list. The array is
494-- allocated to size N, if the list terminates before N elements then the
495-- array may hold less than N elements.
498{-# INLINABLE fromListN #-}
499fromListN :: Storable a => Int -> [a] -> Array a
500fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
502-- | Create an 'Array' from a list. The list must be of finite size.
505{-# INLINABLE fromList #-}
506fromList :: Storable a => [a] -> Array a
507fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
509instance (Storable a, Read a, Show a) => Read (Array a) where
510 {-# INLINE readPrec #-}
514 readListPrec = readListPrecDefault
516instance (a ~ Char) => IsString (Array a) where
517 {-# INLINE fromString #-}
518 fromString = fromList
520-- GHC versions 8.0 and below cannot derive IsList
521instance Storable a => IsList (Array a) where
522 type (Item (Array a)) = a
523 {-# INLINE fromList #-}
525 {-# INLINE fromListN #-}
526 fromListN = fromListN
527 {-# INLINE toList #-}
531arrcmp :: Array a -> Array a -> Bool
533 let !res = unsafeInlineIO $ do
534 let ptr1 = unsafeForeignPtrToPtr $ aStart arr1
535 let ptr2 = unsafeForeignPtrToPtr $ aStart arr2
536 let len1 = aEnd arr1 `minusPtr` ptr1
537 let len2 = aEnd arr2 `minusPtr` ptr2
541 r <- memcmp (castPtr ptr1) (castPtr ptr2) len1
542 touchForeignPtr $ aStart arr1
543 touchForeignPtr $ aStart arr2
548-- XXX we are assuming that Storable equality means element equality. This may
549-- or may not be correct? arrcmp is 40% faster compared to stream equality.
550instance (Storable a, Eq a) => Eq (Array a) where
553 -- arr1 == arr2 = runIdentity $ D.eqBy (==) (toStreamD arr1) (toStreamD arr2)
555instance (Storable a, NFData a) => NFData (Array a) where
557 rnf = foldl' (\_ x -> rnf x) ()
559instance (Storable a, Ord a) => Ord (Array a) where
560 {-# INLINE compare #-}
561 compare arr1 arr2 = unsafePerformIO $
562 D.cmpBy compare (toStreamD arr1) (toStreamD arr2)
564 -- Default definitions defined in base do not have an INLINE on them, so we
565 -- replicate them here with an INLINE.
567 x < y = case compare x y of { LT -> True; _ -> False }
570 x <= y = case compare x y of { GT -> False; _ -> True }
573 x > y = case compare x y of { GT -> True; _ -> False }
576 x >= y = case compare x y of { LT -> False; _ -> True }
578 -- These two default methods use '<=' rather than 'compare'
579 -- because the latter is often more expensive
581 max x y = if x <= y then y else x
584 min x y = if x <= y then x else y
587-- Definitions using the Storable constraint from the Array type. These are to
588-- make the Foldable instance possible though it is much slower (7x slower).
590{-# INLINE_NORMAL toStreamD_ #-}
591toStreamD_ :: forall m a. MonadIO m => Int -> Array a -> D.Stream m a
592toStreamD_ size Array{..} =
593 let p = unsafeForeignPtrToPtr aStart
598 {-# INLINE_LATE step #-}
599 step _ p | p == aEnd = return D.Stop
603 touchForeignPtr aStart
605 return $ D.Yield x (p `plusPtr` size)
607{-# INLINE_NORMAL _foldr #-}
608_foldr :: forall a b. (a -> b -> b) -> b -> Array a -> b
609_foldr f z arr@Array {..} =
610 let !n = sizeOf (undefined :: a)
611 in unsafePerformIO $ D.foldr f z $ toStreamD_ n arr
613-- | Note that the 'Foldable' instance is 7x slower than the direct
615instance Foldable Array where
619-- | Convert an 'Array' into a stream.
622{-# INLINE_EARLY read #-}
623read :: (Monad m, K.IsStream t, Storable a) => Array a -> t m a
624read = D.fromStreamD . toStreamD
625-- XXX add fallback to StreamK rule
626-- {-# RULES "Streamly.Array.read fallback to StreamK" [1]
627-- forall a. S.readK (read a) = K.fromArray a #-}
629-------------------------------------------------------------------------------
630-- Semigroup and Monoid
631-------------------------------------------------------------------------------
633-------------------------------------------------------------------------------
635-------------------------------------------------------------------------------
637-- | GHC memory management allocation header overhead
639allocOverhead = 2 * sizeOf (undefined :: Int)
641-- | Default maximum buffer size in bytes, for reading from and writing to IO
642-- devices, the value is 32KB minus GHC allocation overhead, which is a few
643-- bytes, so that the actual allocation is 32KB.
644defaultChunkSize :: Int
645defaultChunkSize = 32 * k - allocOverhead
648mkChunkSizeKB :: Int -> Int
649mkChunkSizeKB n = n * k - allocOverhead
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