2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ExistentialQuantification #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE ScopedTypeVariables #-}
7{-# LANGUAGE TypeFamilies #-}
8{-# LANGUAGE UnboxedTuples #-}
9{-# LANGUAGE FlexibleContexts #-}
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
65import Control.Exception (assert)
66import Control.DeepSeq (NFData(..))
67import Control.Monad (when)
68import Control.Monad.IO.Class (MonadIO(..))
69import Data.Functor.Identity (runIdentity)
70import Data.Word (Word8)
71import Foreign.C.String (CString)
72import Foreign.C.Types (CSize(..), CInt(..))
73import Foreign.ForeignPtr
74 (ForeignPtr, withForeignPtr, touchForeignPtr)
75import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
76import Foreign.Ptr (plusPtr, minusPtr, castPtr)
77import Foreign.Storable (Storable(..))
78import Prelude hiding (length, foldr, read, unlines, unwords)
79import Text.Read (readPrec, readListPrec, readListPrecDefault)
81import GHC.Base (Addr#, realWorld#)
82import GHC.Exts (IsList, IsString(..))
83import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes, newForeignPtr_)
84import GHC.IO (IO(IO), unsafePerformIO)
85import GHC.Ptr (Ptr(..))
87import Streamly.Fold.Types (Fold(..))
88import Streamly.SVar (adaptState)
90import qualified Streamly.Streams.StreamD.Type as D
91import qualified Streamly.Streams.StreamK as K
92import qualified GHC.Exts as Exts
95import qualified Data.Foldable as F
98-------------------------------------------------------------------------------
100-------------------------------------------------------------------------------
102-- There are two goals that we want to fulfill with arrays. One, holding large
103-- amounts of data in non-GC memory and the ability to pass this data to and
104-- from the operating system without an extra copy overhead. Two, allow random
105-- access to elements based on their indices. The first one falls in the
106-- category of storage buffers while the second one falls in the category of
107-- maps/multisets/hashmaps.
109-- For the first requirement we use an array of Storables and store it in a
110-- ForeignPtr. We can have both immutable and mutable variants of this array
111-- using wrappers over the same underlying type.
113-- For the second requirement, we need a separate type for arrays of
114-- polymorphic values, for example vectors of handler functions, lookup tables.
115-- We can call this type a "vector" in contrast to arrays. It should not
116-- require Storable instance for the type. In that case we need to use an
117-- Array# instead of a ForeignPtr. This type of array would not reduce the GC
118-- overhead much because each element of the array still needs to be scanned by
119-- the GC. However, it can store polymorphic elements and allow random access
120-- to those. But in most cases random access means storage, and it means we
121-- need to avoid GC scanning except in cases of trivially small storage. One
122-- way to achieve that would be to put the array in a Compact region. However,
123-- if and when we mutate this, we will have to use a manual GC copying out to
124-- another CR and freeing the old one.
126-------------------------------------------------------------------------------
128-------------------------------------------------------------------------------
130-- We require that an array stores only Storable. Arrays are used for buffering
131-- data while streams are used for processing. If you want something to be
132-- buffered it better be Storable so that we can store it in non-GC memory.
134-- We can use a "Storable" constraint in the Array type and the Constraint can
135-- be automatically provided to a function that pattern matches on the Array
136-- type. However, it has huge performance cost, so we do not use it.
137-- XXX Can this cost be alleviated in GHC-8.10 specialization fix?
139-- XXX Another way to not require the Storable constraint in array operations
140-- is to store the elemSize in the array at construction and use that instead
141-- of using sizeOf. Need to charaterize perf cost of this.
143-- XXX rename the fields to "start, next, end".
150 { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
151 , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
152 , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
155-------------------------------------------------------------------------------
157-------------------------------------------------------------------------------
159foreign import ccall unsafe "string.h memcpy" c_memcpy
160 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)
162foreign import ccall unsafe "string.h strlen" c_strlen
163 :: CString -> IO CSize
165-- XXX we are converting Int to CSize
166memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
167memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
169foreign import ccall unsafe "string.h memcmp" c_memcmp
170 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt
172-- XXX we are converting Int to CSize
173-- return True if the memory locations have identical contents
175memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool
177 r <- c_memcmp p1 p2 (fromIntegral len)
180{-# INLINE unsafeInlineIO #-}
181unsafeInlineIO :: IO a -> a
182unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r
184-------------------------------------------------------------------------------
186-------------------------------------------------------------------------------
188-- | Allocate an array that can hold 'count' items. The memory of the array is
191-- Note that this is internal routine, the reference to this array cannot be
192-- given out until the array has been written to and frozen.
193{-# INLINE newArray #-}
194newArray :: forall a. Storable a => Int -> IO (Array a)
196 let size = count * sizeOf (undefined :: a)
197 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
198 let p = unsafeForeignPtrToPtr fptr
202 , aBound = p `plusPtr` size
205-- | Allocate an Array of the given size and run an IO action passing the array
207{-# INLINE withNewArray #-}
208withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)
209withNewArray count f = do
210 arr <- newArray count
211 withForeignPtr (aStart arr) $ \p -> f p >> return arr
213-- XXX grow the array when we are beyond bound.
215-- Internal routine for when the array is being created. Appends one item at
216-- the end of the array. Useful when sequentially writing a stream to the
217-- array. DOES NOT CHECK THE ARRAY BOUNDS.
218{-# INLINE unsafeAppend #-}
219unsafeAppend :: forall a. Storable a => Array a -> a -> IO (Array a)
220unsafeAppend arr@Array{..} x = do
221 when (aEnd == aBound) $
222 error "BUG: unsafeAppend: writing beyond array bounds"
224 touchForeignPtr aStart
225 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
227-- | Remove the free space from an Array.
228shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
229shrinkToFit arr@Array{..} = do
230 assert (aEnd <= aBound) (return ())
233 let oldStart = unsafeForeignPtrToPtr aStart
234 let size = aEnd `minusPtr` oldStart
235 newPtr <- mallocPlainForeignPtrAlignedBytes
236 size (alignment (undefined :: a))
237 withForeignPtr newPtr $ \pNew -> do
238 memcpy (castPtr pNew) (castPtr oldStart) size
239 touchForeignPtr aStart
240 let end = pNew `plusPtr` size
248-- | Expand the free space in the array doubling the size with a minimum
249-- exapnsion of the specified amount of bytes.
250{-# NOINLINE reallocDouble #-}
251reallocDouble :: forall a. Storable a => Int -> Array a -> IO (Array a)
252reallocDouble minIncrease Array{..} = do
253 assert (aEnd <= aBound) (return ())
254 let oldStart = unsafeForeignPtrToPtr aStart
255 let size = aEnd `minusPtr` oldStart
256 newSize = max (size * 2) (size + minIncrease)
257 newPtr <- mallocPlainForeignPtrAlignedBytes
258 newSize (alignment (undefined :: a))
259 withForeignPtr newPtr $ \pNew -> do
260 memcpy (castPtr pNew) (castPtr oldStart) size
261 touchForeignPtr aStart
264 , aEnd = pNew `plusPtr` size
265 , aBound = pNew `plusPtr` newSize
268-- XXX when converting an array of Word8 from a literal string we can simply
269-- refer to the literal string. Is it possible to write rules such that
270-- fromList Word8 can be rewritten so that GHC does not first convert the
271-- literal to [Char] and then we convert it back to an Array Word8?
273-- Note that the address must be a read-only address (meant to be used for
274-- read-only string literals) because we are sharing it, any modification to
275-- the original address would change our array. That's why this function is
277{-# INLINE _fromCStringAddrUnsafe #-}
278_fromCStringAddrUnsafe :: Addr# -> IO (Array Word8)
279_fromCStringAddrUnsafe addr# = do
280 ptr <- newForeignPtr_ (castPtr cstr)
282 let n = fromIntegral len
283 let p = unsafeForeignPtrToPtr ptr
284 let end = p `plusPtr` n
294-------------------------------------------------------------------------------
296-------------------------------------------------------------------------------
298-- | Return element at the specified index without checking the bounds.
300-- Unsafe because it does not check the bounds of the array.
301{-# INLINE_NORMAL unsafeIndexIO #-}
302unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
303unsafeIndexIO Array {..} i =
304 withForeignPtr aStart $ \p -> do
305 let elemSize = sizeOf (undefined :: a)
306 elemOff = p `plusPtr` (elemSize * i)
307 assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
311-- | Return element at the specified index without checking the bounds.
312{-# INLINE_NORMAL unsafeIndex #-}
313unsafeIndex :: forall a. Storable a => Array a -> Int -> a
314unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
316-- | /O(1)/ Get the length of the array.
320length :: forall a. Storable a => Array a -> Int
322 let p = unsafeForeignPtrToPtr aStart
323 aLen = aEnd `minusPtr` p
324 in assert (aLen >= 0) (aLen `div` sizeOf (undefined :: a))
326{-# INLINE_NORMAL toStreamD #-}
327toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
329 let p = unsafeForeignPtrToPtr aStart
334 {-# INLINE_LATE step #-}
335 step _ p | p == aEnd = return D.Stop
337 -- unsafeInlineIO allows us to run this in Identity monad for pure
338 -- toList/foldr case which makes them much faster due to not
339 -- accumulating the list and fusing better with the pure consumers.
341 -- This should be safe as the array contents are guaranteed to be
342 -- evaluated/written to before we peek at them.
343 let !x = unsafeInlineIO $ do
345 touchForeignPtr aStart
347 return $ D.Yield x (p `plusPtr` (sizeOf (undefined :: a)))
350{-# INLINE toStreamK #-}
351toStreamK :: forall t m a. (K.IsStream t, Monad m, Storable a) => Array a -> t m a
353 let p = unsafeForeignPtrToPtr aStart
356 go p | p == aEnd = K.nil
358 -- See Note in toStreamD.
359 let !x = unsafeInlineIO $ do
361 touchForeignPtr aStart
363 in x `K.cons` go (p `plusPtr` (sizeOf (undefined :: a)))
365{-# INLINE_NORMAL toStreamDRev #-}
366toStreamDRev :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
367toStreamDRev Array{..} =
368 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
373 {-# INLINE_LATE step #-}
374 step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop
376 -- See comments in toStreamD for why we use unsafeInlineIO
377 let !x = unsafeInlineIO $ do
379 touchForeignPtr aStart
381 return $ D.Yield x (p `plusPtr` negate (sizeOf (undefined :: a)))
383{-# INLINE_NORMAL foldl' #-}
384foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
385foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
387{-# INLINE_NORMAL foldr #-}
388foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
389foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
391-------------------------------------------------------------------------------
393-------------------------------------------------------------------------------
395-- | @toArrayN n@ folds a maximum of @n@ elements from the input stream to an
399{-# INLINE_NORMAL toArrayN #-}
400toArrayN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
401toArrayN n = Fold step initial extract
406 if n < 0 then error "toArrayN: negative count specified" else return ()
408 step arr@(Array _ end bound) _ | end == bound = return arr
409 step (Array start end bound) x = do
411 return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
412 extract = liftIO . shrinkToFit
414{-# INLINE_NORMAL fromStreamDN #-}
415fromStreamDN :: forall m a. (MonadIO m, Storable a)
416 => Int -> D.Stream m a -> m (Array a)
417fromStreamDN limit str = do
418 arr <- liftIO $ newArray limit
419 end <- D.foldlM' write (aEnd arr) $ D.take limit str
420 return $ arr {aEnd = end}
426 return $ ptr `plusPtr` sizeOf (undefined :: a)
428-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
430{-# INLINE fromStreamDArraysOf #-}
431fromStreamDArraysOf :: (MonadIO m, Storable a)
432 => Int -> D.Stream m a -> D.Stream m (Array a)
433fromStreamDArraysOf n str = D.groupsOf n (toArrayN n) str
435-- XXX concatMap does not seem to have the best possible performance so we have
436-- a custom way to concat arrays.
437data FlattenState s a =
439 | InnerLoop s (ForeignPtr a) (Ptr a) (Ptr a)
441{-# INLINE_NORMAL flattenArrays #-}
442flattenArrays :: forall m a. (MonadIO m, Storable a)
443 => D.Stream m (Array a) -> D.Stream m a
444flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)
448 {-# INLINE_LATE step' #-}
449 step' gst (OuterLoop st) = do
450 r <- step (adaptState gst) st
452 D.Yield Array{..} s ->
453 let p = unsafeForeignPtrToPtr aStart
454 in D.Skip (InnerLoop s aStart p aEnd)
455 D.Skip s -> D.Skip (OuterLoop s)
458 step' _ (InnerLoop st _ p end) | p == end =
459 return $ D.Skip $ OuterLoop st
461 step' _ (InnerLoop st startf p end) = do
464 touchForeignPtr startf
466 return $ D.Yield x (InnerLoop st startf
467 (p `plusPtr` (sizeOf (undefined :: a))) end)
469{-# INLINE_NORMAL flattenArraysRev #-}
470flattenArraysRev :: forall m a. (MonadIO m, Storable a)
471 => D.Stream m (Array a) -> D.Stream m a
472flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
476 {-# INLINE_LATE step' #-}
477 step' gst (OuterLoop st) = do
478 r <- step (adaptState gst) st
480 D.Yield Array{..} s ->
481 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
482 -- XXX we do not need aEnd
483 in D.Skip (InnerLoop s aStart p aEnd)
484 D.Skip s -> D.Skip (OuterLoop s)
487 step' _ (InnerLoop st start p _) | p < unsafeForeignPtrToPtr start =
488 return $ D.Skip $ OuterLoop st
490 step' _ (InnerLoop st startf p end) = do
493 touchForeignPtr startf
495 return $ D.Yield x (InnerLoop st startf
496 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
498-- CAUTION: a very large number (millions) of arrays can degrade performance
499-- due to GC overhead because we need to buffer the arrays before we flatten
502-- We could take the approach of doubling the memory allocation on each
503-- overflow. This would result in more or less the same amount of copying as in
504-- the chunking approach. However, if we have to shrink in the end then it may
505-- result in an extra copy of the entire data.
507{-# INLINE fromStreamD #-}
508fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
510 let s = fromStreamDArraysOf defaultChunkSize m
511 buffered <- D.foldr K.cons K.nil s
512 len <- K.foldl' (+) 0 (K.map length buffered)
513 fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
515-- | Convert an 'Array' into a list.
518{-# INLINABLE toList #-}
519toList :: Storable a => Array a -> [a]
522instance (Show a, Storable a) => Show (Array a) where
523 {-# INLINE showsPrec #-}
524 showsPrec _ = shows . toList
526-- | Create an 'Array' from the first N elements of a list. The array is
527-- allocated to size N, if the list terminates before N elements then the
528-- array may hold less than N elements.
531{-# INLINABLE fromListN #-}
532fromListN :: Storable a => Int -> [a] -> Array a
533fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
535-- | Create an 'Array' from a list. The list must be of finite size.
538{-# INLINABLE fromList #-}
539fromList :: Storable a => [a] -> Array a
540fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
542instance (Storable a, Read a, Show a) => Read (Array a) where
543 {-# INLINE readPrec #-}
547 readListPrec = readListPrecDefault
549instance (a ~ Char) => IsString (Array a) where
550 {-# INLINE fromString #-}
551 fromString = fromList
553-- GHC versions 8.0 and below cannot derive IsList
554instance Storable a => IsList (Array a) where
555 type (Item (Array a)) = a
556 {-# INLINE fromList #-}
558 {-# INLINE fromListN #-}
559 fromListN = fromListN
560 {-# INLINE toList #-}
564arrcmp :: Array a -> Array a -> Bool
566 let !res = unsafeInlineIO $ do
567 let ptr1 = unsafeForeignPtrToPtr $ aStart arr1
568 let ptr2 = unsafeForeignPtrToPtr $ aStart arr2
569 let len1 = aEnd arr1 `minusPtr` ptr1
570 let len2 = aEnd arr2 `minusPtr` ptr2
574 r <- memcmp (castPtr ptr1) (castPtr ptr2) len1
575 touchForeignPtr $ aStart arr1
576 touchForeignPtr $ aStart arr2
581-- XXX we are assuming that Storable equality means element equality. This may
582-- or may not be correct? arrcmp is 40% faster compared to stream equality.
583instance (Storable a, Eq a) => Eq (Array a) where
586 -- arr1 == arr2 = runIdentity $ D.eqBy (==) (toStreamD arr1) (toStreamD arr2)
588instance (Storable a, NFData a) => NFData (Array a) where
590 rnf = foldl' (\_ x -> rnf x) ()
592instance (Storable a, Ord a) => Ord (Array a) where
593 {-# INLINE compare #-}
594 compare arr1 arr2 = unsafePerformIO $
595 D.cmpBy compare (toStreamD arr1) (toStreamD arr2)
597 -- Default definitions defined in base do not have an INLINE on them, so we
598 -- replicate them here with an INLINE.
600 x < y = case compare x y of { LT -> True; _ -> False }
603 x <= y = case compare x y of { GT -> False; _ -> True }
606 x > y = case compare x y of { GT -> True; _ -> False }
609 x >= y = case compare x y of { LT -> False; _ -> True }
611 -- These two default methods use '<=' rather than 'compare'
612 -- because the latter is often more expensive
614 max x y = if x <= y then y else x
617 min x y = if x <= y then x else y
620-- Definitions using the Storable constraint from the Array type. These are to
621-- make the Foldable instance possible though it is much slower (7x slower).
623{-# INLINE_NORMAL toStreamD_ #-}
624toStreamD_ :: forall m a. MonadIO m => Int -> Array a -> D.Stream m a
625toStreamD_ size Array{..} =
626 let p = unsafeForeignPtrToPtr aStart
631 {-# INLINE_LATE step #-}
632 step _ p | p == aEnd = return D.Stop
636 touchForeignPtr aStart
638 return $ D.Yield x (p `plusPtr` size)
640{-# INLINE_NORMAL _foldr #-}
641_foldr :: forall a b. (a -> b -> b) -> b -> Array a -> b
642_foldr f z arr@Array {..} =
643 let !n = sizeOf (undefined :: a)
644 in unsafePerformIO $ D.foldr f z $ toStreamD_ n arr
646-- | Note that the 'Foldable' instance is 7x slower than the direct
648instance Foldable Array where
652-- | Convert an 'Array' into a stream.
655{-# INLINE_EARLY read #-}
656read :: (Monad m, K.IsStream t, Storable a) => Array a -> t m a
657read = D.fromStreamD . toStreamD
658-- XXX add fallback to StreamK rule
659-- {-# RULES "Streamly.Array.read fallback to StreamK" [1]
660-- forall a. S.readK (read a) = K.fromArray a #-}
662-------------------------------------------------------------------------------
663-- Semigroup and Monoid
664-------------------------------------------------------------------------------
666-------------------------------------------------------------------------------
668-------------------------------------------------------------------------------
670-- | GHC memory management allocation header overhead
672allocOverhead = 2 * sizeOf (undefined :: Int)
674mkChunkSize :: Int -> Int
675mkChunkSize n = let size = n - allocOverhead in max size 0
677mkChunkSizeKB :: Int -> Int
678mkChunkSizeKB n = mkChunkSize (n * k)
681-- | Default maximum buffer size in bytes, for reading from and writing to IO
682-- devices, the value is 32KB minus GHC allocation overhead, which is a few
683-- bytes, so that the actual allocation is 32KB.
684defaultChunkSize :: Int
685defaultChunkSize = mkChunkSizeKB 32
687{-# INLINE_NORMAL unlines #-}
688unlines :: MonadIO m => D.Stream m (Array Char) -> D.Stream m Char
689unlines (D.Stream step state) = D.Stream step' (OuterLoop state)
691 {-# INLINE_LATE step' #-}
692 step' gst (OuterLoop st) = do
693 r <- step (adaptState gst) st
695 D.Yield Array{..} s ->
696 let p = unsafeForeignPtrToPtr aStart
697 in D.Skip (InnerLoop s aStart p aEnd)
698 D.Skip s -> D.Skip (OuterLoop s)
701 step' _ (InnerLoop st _ p end) | p == end =
702 return $ D.Yield '\n' $ OuterLoop st
704 step' _ (InnerLoop st startf p end) = do
707 touchForeignPtr startf
709 return $ D.Yield x (InnerLoop st startf
710 (p `plusPtr` (sizeOf (undefined :: Char))) end)
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