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
37 -- * Streams of arrays
77import Control.Exception (assert)
78import Control.DeepSeq (NFData(..))
79import Control.Monad (when)
80import Control.Monad.IO.Class (MonadIO(..))
81import Data.Functor.Identity (runIdentity)
82import Data.Word (Word8)
83import Foreign.C.String (CString)
84import Foreign.C.Types (CSize(..), CInt(..))
85import Foreign.ForeignPtr
86 (ForeignPtr, withForeignPtr, touchForeignPtr, plusForeignPtr)
87import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
88import Foreign.Ptr (plusPtr, minusPtr, castPtr, nullPtr)
89import Foreign.Storable (Storable(..))
90import Prelude hiding (length, foldr, read, unlines, splitAt)
91import Text.Read (readPrec, readListPrec, readListPrecDefault)
93import GHC.Base (Addr#, realWorld#)
94import GHC.Exts (IsList, IsString(..))
95import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes, newForeignPtr_)
96import GHC.IO (IO(IO), unsafePerformIO)
97import GHC.Ptr (Ptr(..))
99import Streamly.Fold.Types (Fold(..))
100import Streamly.SVar (adaptState)
101import Streamly.FileSystem.FDIO (IOVec(..))
103import qualified Streamly.Streams.StreamD.Type as D
104import qualified Streamly.Streams.StreamK as K
105import qualified GHC.Exts as Exts
108import qualified Data.Foldable as F
111-------------------------------------------------------------------------------
113-------------------------------------------------------------------------------
115-- There are two goals that we want to fulfill with arrays. One, holding large
116-- amounts of data in non-GC memory and the ability to pass this data to and
117-- from the operating system without an extra copy overhead. Two, allow random
118-- access to elements based on their indices. The first one falls in the
119-- category of storage buffers while the second one falls in the category of
120-- maps/multisets/hashmaps.
122-- For the first requirement we use an array of Storables and store it in a
123-- ForeignPtr. We can have both immutable and mutable variants of this array
124-- using wrappers over the same underlying type.
126-- For the second requirement, we need a separate type for arrays of
127-- polymorphic values, for example vectors of handler functions, lookup tables.
128-- We can call this type a "vector" in contrast to arrays. It should not
129-- require Storable instance for the type. In that case we need to use an
130-- Array# instead of a ForeignPtr. This type of array would not reduce the GC
131-- overhead much because each element of the array still needs to be scanned by
132-- the GC. However, it can store polymorphic elements and allow random access
133-- to those. But in most cases random access means storage, and it means we
134-- need to avoid GC scanning except in cases of trivially small storage. One
135-- way to achieve that would be to put the array in a Compact region. However,
136-- if and when we mutate this, we will have to use a manual GC copying out to
137-- another CR and freeing the old one.
139-------------------------------------------------------------------------------
141-------------------------------------------------------------------------------
143-- We require that an array stores only Storable. Arrays are used for buffering
144-- data while streams are used for processing. If you want something to be
145-- buffered it better be Storable so that we can store it in non-GC memory.
147-- We can use a "Storable" constraint in the Array type and the Constraint can
148-- be automatically provided to a function that pattern matches on the Array
149-- type. However, it has huge performance cost, so we do not use it.
150-- XXX Can this cost be alleviated in GHC-8.10 specialization fix?
152-- XXX Another way to not require the Storable constraint in array operations
153-- is to store the elemSize in the array at construction and use that instead
154-- of using sizeOf. Need to charaterize perf cost of this.
156-- XXX rename the fields to "start, next, end".
163 { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
164 , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
165 , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
168-------------------------------------------------------------------------------
170-------------------------------------------------------------------------------
172foreign import ccall unsafe "string.h memcpy" c_memcpy
173 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)
175foreign import ccall unsafe "string.h strlen" c_strlen
176 :: CString -> IO CSize
178foreign import ccall unsafe "string.h memchr" c_memchr
179 :: Ptr Word8 -> Word8 -> CSize -> IO (Ptr Word8)
181-- XXX we are converting Int to CSize
182memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
183memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
185foreign import ccall unsafe "string.h memcmp" c_memcmp
186 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt
188-- XXX we are converting Int to CSize
189-- return True if the memory locations have identical contents
191memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool
193 r <- c_memcmp p1 p2 (fromIntegral len)
196{-# INLINE unsafeInlineIO #-}
197unsafeInlineIO :: IO a -> a
198unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r
200-------------------------------------------------------------------------------
202-------------------------------------------------------------------------------
204-- | Allocate an array that can hold 'count' items. The memory of the array is
207-- Note that this is internal routine, the reference to this array cannot be
208-- given out until the array has been written to and frozen.
209{-# INLINE newArray #-}
210newArray :: forall a. Storable a => Int -> IO (Array a)
212 let size = count * sizeOf (undefined :: a)
213 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
214 let p = unsafeForeignPtrToPtr fptr
218 , aBound = p `plusPtr` size
221-- | Allocate an Array of the given size and run an IO action passing the array
223{-# INLINE withNewArray #-}
224withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)
225withNewArray count f = do
226 arr <- newArray count
227 withForeignPtr (aStart arr) $ \p -> f p >> return arr
229-- XXX grow the array when we are beyond bound.
231-- Internal routine for when the array is being created. Appends one item at
232-- the end of the array. Useful when sequentially writing a stream to the
233-- array. DOES NOT CHECK THE ARRAY BOUNDS.
234{-# INLINE unsafeSnoc #-}
235unsafeSnoc :: forall a. Storable a => Array a -> a -> IO (Array a)
236unsafeSnoc arr@Array{..} x = do
237 when (aEnd == aBound) $
238 error "BUG: unsafeSnoc: writing beyond array bounds"
240 touchForeignPtr aStart
241 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
244snoc :: forall a. Storable a => Array a -> a -> IO (Array a)
245snoc arr@Array {..} x = do
248 let oldStart = unsafeForeignPtrToPtr aStart
249 size = aEnd `minusPtr` oldStart
250 newSize = (size + (sizeOf (undefined :: a)))
251 newPtr <- mallocPlainForeignPtrAlignedBytes newSize (alignment (undefined :: a))
252 withForeignPtr newPtr $ \pNew -> do
253 memcpy (castPtr pNew) (castPtr oldStart) size
254 poke (pNew `plusPtr` size) x
255 touchForeignPtr aStart
258 , aEnd = pNew `plusPtr` (size + sizeOf (undefined :: a))
259 , aBound = pNew `plusPtr` newSize
263 touchForeignPtr aStart
264 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
266-- | Reallocate the array to the specified size in bytes. If the size is less
267-- than the original array the array gets truncated.
268{-# NOINLINE realloc #-}
269realloc :: forall a. Storable a => Int -> Array a -> IO (Array a)
270realloc newSize Array{..} = do
271 assert (aEnd <= aBound) (return ())
272 let oldStart = unsafeForeignPtrToPtr aStart
273 let size = aEnd `minusPtr` oldStart
274 newPtr <- mallocPlainForeignPtrAlignedBytes
275 newSize (alignment (undefined :: a))
276 withForeignPtr newPtr $ \pNew -> do
277 memcpy (castPtr pNew) (castPtr oldStart) size
278 touchForeignPtr aStart
281 , aEnd = pNew `plusPtr` size
282 , aBound = pNew `plusPtr` newSize
285-- | Remove the free space from an Array.
286shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
287shrinkToFit arr@Array{..} = do
288 assert (aEnd <= aBound) (return ())
291 let oldStart = unsafeForeignPtrToPtr aStart
292 let size = aEnd `minusPtr` oldStart
296-- XXX when converting an array of Word8 from a literal string we can simply
297-- refer to the literal string. Is it possible to write rules such that
298-- fromList Word8 can be rewritten so that GHC does not first convert the
299-- literal to [Char] and then we convert it back to an Array Word8?
301-- Note that the address must be a read-only address (meant to be used for
302-- read-only string literals) because we are sharing it, any modification to
303-- the original address would change our array. That's why this function is
305{-# INLINE _fromCStringAddrUnsafe #-}
306_fromCStringAddrUnsafe :: Addr# -> IO (Array Word8)
307_fromCStringAddrUnsafe addr# = do
308 ptr <- newForeignPtr_ (castPtr cstr)
310 let n = fromIntegral len
311 let p = unsafeForeignPtrToPtr ptr
312 let end = p `plusPtr` n
322-------------------------------------------------------------------------------
324-------------------------------------------------------------------------------
326-- | Return element at the specified index without checking the bounds.
328-- Unsafe because it does not check the bounds of the array.
329{-# INLINE_NORMAL unsafeIndexIO #-}
330unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
331unsafeIndexIO Array {..} i =
332 withForeignPtr aStart $ \p -> do
333 let elemSize = sizeOf (undefined :: a)
334 elemOff = p `plusPtr` (elemSize * i)
335 assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
339-- | Return element at the specified index without checking the bounds.
340{-# INLINE_NORMAL unsafeIndex #-}
341unsafeIndex :: forall a. Storable a => Array a -> Int -> a
342unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
344-- | /O(1)/ Get the byte length of the array.
347{-# INLINE byteLength #-}
348byteLength :: Array a -> Int
349byteLength Array{..} =
350 let p = unsafeForeignPtrToPtr aStart
351 len = aEnd `minusPtr` p
352 in assert (len >= 0) len
354-- | /O(1)/ Get the length of the array i.e. the number of elements in the
359length :: forall a. Storable a => Array a -> Int
360length arr = byteLength arr `div` sizeOf (undefined :: a)
362{-# INLINE byteCapacity #-}
363byteCapacity :: Array a -> Int
364byteCapacity Array{..} =
365 let p = unsafeForeignPtrToPtr aStart
366 len = aBound `minusPtr` p
367 in assert (len >= 0) len
369{-# INLINE_NORMAL toStreamD #-}
370toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
372 let p = unsafeForeignPtrToPtr aStart
377 {-# INLINE_LATE step #-}
378 step _ p | p == aEnd = return D.Stop
380 -- unsafeInlineIO allows us to run this in Identity monad for pure
381 -- toList/foldr case which makes them much faster due to not
382 -- accumulating the list and fusing better with the pure consumers.
384 -- This should be safe as the array contents are guaranteed to be
385 -- evaluated/written to before we peek at them.
386 let !x = unsafeInlineIO $ do
388 touchForeignPtr aStart
390 return $ D.Yield x (p `plusPtr` (sizeOf (undefined :: a)))
392{-# INLINE toStreamK #-}
393toStreamK :: forall t m a. (K.IsStream t, Storable a) => Array a -> t m a
395 let p = unsafeForeignPtrToPtr aStart
400 go p | p == aEnd = K.nil
402 -- See Note in toStreamD.
403 let !x = unsafeInlineIO $ do
405 touchForeignPtr aStart
407 in x `K.cons` go (p `plusPtr` (sizeOf (undefined :: a)))
409{-# INLINE_NORMAL toStreamDRev #-}
410toStreamDRev :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
411toStreamDRev Array{..} =
412 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
417 {-# INLINE_LATE step #-}
418 step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop
420 -- See comments in toStreamD for why we use unsafeInlineIO
421 let !x = unsafeInlineIO $ do
423 touchForeignPtr aStart
425 return $ D.Yield x (p `plusPtr` negate (sizeOf (undefined :: a)))
427{-# INLINE toStreamKRev #-}
428toStreamKRev :: forall t m a. (K.IsStream t, Storable a) => Array a -> t m a
429toStreamKRev Array {..} =
430 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
435 go p | p < unsafeForeignPtrToPtr aStart = K.nil
437 let !x = unsafeInlineIO $ do
439 touchForeignPtr aStart
441 in x `K.cons` go (p `plusPtr` negate (sizeOf (undefined :: a)))
443{-# INLINE_NORMAL foldl' #-}
444foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
445foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
447{-# INLINE_NORMAL foldr #-}
448foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
449foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
451-------------------------------------------------------------------------------
453-------------------------------------------------------------------------------
455-- | @toArrayN n@ folds a maximum of @n@ elements from the input stream to an
459{-# INLINE_NORMAL toArrayN #-}
460toArrayN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
461toArrayN n = Fold step initial extract
466 if n < 0 then error "toArrayN: negative count specified" else return ()
468 step arr@(Array _ end bound) _ | end == bound = return arr
469 step (Array start end bound) x = do
471 return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
472 extract = liftIO . shrinkToFit
474{-# INLINE_NORMAL fromStreamDN #-}
475fromStreamDN :: forall m a. (MonadIO m, Storable a)
476 => Int -> D.Stream m a -> m (Array a)
477fromStreamDN limit str = do
478 arr <- liftIO $ newArray limit
479 end <- D.foldlM' write (aEnd arr) $ D.take limit str
480 return $ arr {aEnd = end}
486 return $ ptr `plusPtr` sizeOf (undefined :: a)
488data GroupState s start end bound
490 | GroupBuffer s start end bound
491 | GroupYield start end bound (GroupState s start end bound)
494-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
496{-# INLINE_NORMAL fromStreamDArraysOf #-}
497fromStreamDArraysOf :: (MonadIO m, Storable a)
498 => Int -> D.Stream m a -> D.Stream m (Array a)
499-- fromStreamDArraysOf n str = D.groupsOf n (toArrayN n) str
500fromStreamDArraysOf n (D.Stream step state) =
501 D.Stream step' (GroupStart state)
505 {-# INLINE_LATE step' #-}
506 step' _ (GroupStart st) = do
508 -- XXX we can pass the module string from the higher level API
509 error $ "Streamly.Mem.Array.Types.fromStreamDArraysOf: the size of "
510 ++ "arrays [" ++ show n ++ "] must be a natural number"
511 Array start end bound <- liftIO $ newArray n
512 return $ D.Skip (GroupBuffer st start end bound)
514 step' gst (GroupBuffer st start end bound) = do
515 r <- step (adaptState gst) st
519 let end' = end `plusPtr` 1
522 then D.Skip (GroupYield start end' bound (GroupStart s))
523 else D.Skip (GroupBuffer s start end' bound)
524 D.Skip s -> return $ D.Skip (GroupBuffer s start end bound)
525 D.Stop -> return $ D.Skip (GroupYield start end bound GroupFinish)
527 step' _ (GroupYield start end bound next) =
528 return $ D.Yield (Array start end bound) next
530 step' _ GroupFinish = return D.Stop
532-- XXX concatMap does not seem to have the best possible performance so we have
533-- a custom way to concat arrays.
534data FlattenState s a =
536 | InnerLoop s (ForeignPtr a) (Ptr a) (Ptr a)
538{-# INLINE_NORMAL flattenArrays #-}
539flattenArrays :: forall m a. (MonadIO m, Storable a)
540 => D.Stream m (Array a) -> D.Stream m a
541flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)
545 {-# INLINE_LATE step' #-}
546 step' gst (OuterLoop st) = do
547 r <- step (adaptState gst) st
549 D.Yield Array{..} s ->
550 let p = unsafeForeignPtrToPtr aStart
551 in D.Skip (InnerLoop s aStart p aEnd)
552 D.Skip s -> D.Skip (OuterLoop s)
555 step' _ (InnerLoop st _ p end) | p == end =
556 return $ D.Skip $ OuterLoop st
558 step' _ (InnerLoop st startf p end) = do
561 touchForeignPtr startf
563 return $ D.Yield x (InnerLoop st startf
564 (p `plusPtr` (sizeOf (undefined :: a))) end)
566{-# INLINE_NORMAL flattenArraysRev #-}
567flattenArraysRev :: forall m a. (MonadIO m, Storable a)
568 => D.Stream m (Array a) -> D.Stream m a
569flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
573 {-# INLINE_LATE step' #-}
574 step' gst (OuterLoop st) = do
575 r <- step (adaptState gst) st
577 D.Yield Array{..} s ->
578 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
579 -- XXX we do not need aEnd
580 in D.Skip (InnerLoop s aStart p aEnd)
581 D.Skip s -> D.Skip (OuterLoop s)
584 step' _ (InnerLoop st start p _) | p < unsafeForeignPtrToPtr start =
585 return $ D.Skip $ OuterLoop st
587 step' _ (InnerLoop st startf p end) = do
590 touchForeignPtr startf
592 return $ D.Yield x (InnerLoop st startf
593 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
595-- CAUTION: a very large number (millions) of arrays can degrade performance
596-- due to GC overhead because we need to buffer the arrays before we flatten
599-- We could take the approach of doubling the memory allocation on each
600-- overflow. This would result in more or less the same amount of copying as in
601-- the chunking approach. However, if we have to shrink in the end then it may
602-- result in an extra copy of the entire data.
604{-# INLINE fromStreamD #-}
605fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
607 let s = fromStreamDArraysOf defaultChunkSize m
608 buffered <- D.foldr K.cons K.nil s
609 len <- K.foldl' (+) 0 (K.map length buffered)
610 fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
612-- | Convert an 'Array' into a list.
615{-# INLINABLE toList #-}
616toList :: Storable a => Array a -> [a]
619instance (Show a, Storable a) => Show (Array a) where
620 {-# INLINE showsPrec #-}
621 showsPrec _ = shows . toList
623-- | Create an 'Array' from the first N elements of a list. The array is
624-- allocated to size N, if the list terminates before N elements then the
625-- array may hold less than N elements.
628{-# INLINABLE fromListN #-}
629fromListN :: Storable a => Int -> [a] -> Array a
630fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
632-- | Create an 'Array' from a list. The list must be of finite size.
635{-# INLINABLE fromList #-}
636fromList :: Storable a => [a] -> Array a
637fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
639instance (Storable a, Read a, Show a) => Read (Array a) where
640 {-# INLINE readPrec #-}
644 readListPrec = readListPrecDefault
646instance (a ~ Char) => IsString (Array a) where
647 {-# INLINE fromString #-}
648 fromString = fromList
650-- GHC versions 8.0 and below cannot derive IsList
651instance Storable a => IsList (Array a) where
652 type (Item (Array a)) = a
653 {-# INLINE fromList #-}
655 {-# INLINE fromListN #-}
656 fromListN = fromListN
657 {-# INLINE toList #-}
661arrcmp :: Array a -> Array a -> Bool
663 let !res = unsafeInlineIO $ do
664 let ptr1 = unsafeForeignPtrToPtr $ aStart arr1
665 let ptr2 = unsafeForeignPtrToPtr $ aStart arr2
666 let len1 = aEnd arr1 `minusPtr` ptr1
667 let len2 = aEnd arr2 `minusPtr` ptr2
671 r <- memcmp (castPtr ptr1) (castPtr ptr2) len1
672 touchForeignPtr $ aStart arr1
673 touchForeignPtr $ aStart arr2
678-- XXX we are assuming that Storable equality means element equality. This may
679-- or may not be correct? arrcmp is 40% faster compared to stream equality.
680instance (Storable a, Eq a) => Eq (Array a) where
683 -- arr1 == arr2 = runIdentity $ D.eqBy (==) (toStreamD arr1) (toStreamD arr2)
685instance (Storable a, NFData a) => NFData (Array a) where
687 rnf = foldl' (\_ x -> rnf x) ()
689instance (Storable a, Ord a) => Ord (Array a) where
690 {-# INLINE compare #-}
691 compare arr1 arr2 = unsafePerformIO $
692 D.cmpBy compare (toStreamD arr1) (toStreamD arr2)
694 -- Default definitions defined in base do not have an INLINE on them, so we
695 -- replicate them here with an INLINE.
697 x < y = case compare x y of { LT -> True; _ -> False }
700 x <= y = case compare x y of { GT -> False; _ -> True }
703 x > y = case compare x y of { GT -> True; _ -> False }
706 x >= y = case compare x y of { LT -> False; _ -> True }
708 -- These two default methods use '<=' rather than 'compare'
709 -- because the latter is often more expensive
711 max x y = if x <= y then y else x
714 min x y = if x <= y then x else y
717-- Definitions using the Storable constraint from the Array type. These are to
718-- make the Foldable instance possible though it is much slower (7x slower).
720{-# INLINE_NORMAL toStreamD_ #-}
721toStreamD_ :: forall m a. MonadIO m => Int -> Array a -> D.Stream m a
722toStreamD_ size Array{..} =
723 let p = unsafeForeignPtrToPtr aStart
728 {-# INLINE_LATE step #-}
729 step _ p | p == aEnd = return D.Stop
733 touchForeignPtr aStart
735 return $ D.Yield x (p `plusPtr` size)
737{-# INLINE_NORMAL _foldr #-}
738_foldr :: forall a b. (a -> b -> b) -> b -> Array a -> b
739_foldr f z arr@Array {..} =
740 let !n = sizeOf (undefined :: a)
741 in unsafePerformIO $ D.foldr f z $ toStreamD_ n arr
743-- | Note that the 'Foldable' instance is 7x slower than the direct
745instance Foldable Array where
749-- | Convert an 'Array' into a stream.
752{-# INLINE_EARLY read #-}
753read :: (Monad m, K.IsStream t, Storable a) => Array a -> t m a
754read = D.fromStreamD . toStreamD
755-- XXX add fallback to StreamK rule
756-- {-# RULES "Streamly.Array.read fallback to StreamK" [1]
757-- forall a. S.readK (read a) = K.fromArray a #-}
759-------------------------------------------------------------------------------
760-- Semigroup and Monoid
761-------------------------------------------------------------------------------
763-------------------------------------------------------------------------------
765-------------------------------------------------------------------------------
767-- | GHC memory management allocation header overhead
769allocOverhead = 2 * sizeOf (undefined :: Int)
771mkChunkSize :: Int -> Int
772mkChunkSize n = let size = n - allocOverhead in max size 0
774mkChunkSizeKB :: Int -> Int
775mkChunkSizeKB n = mkChunkSize (n * k)
778-- | Default maximum buffer size in bytes, for reading from and writing to IO
779-- devices, the value is 32KB minus GHC allocation overhead, which is a few
780-- bytes, so that the actual allocation is 32KB.
781defaultChunkSize :: Int
782defaultChunkSize = mkChunkSizeKB 32
784{-# INLINE_NORMAL unlines #-}
785unlines :: forall m a. (MonadIO m, Storable a)
786 => a -> D.Stream m (Array a) -> D.Stream m a
787unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
789 {-# INLINE_LATE step' #-}
790 step' gst (OuterLoop st) = do
791 r <- step (adaptState gst) st
793 D.Yield Array{..} s ->
794 let p = unsafeForeignPtrToPtr aStart
795 in D.Skip (InnerLoop s aStart p aEnd)
796 D.Skip s -> D.Skip (OuterLoop s)
799 step' _ (InnerLoop st _ p end) | p == end =
800 return $ D.Yield sep $ OuterLoop st
802 step' _ (InnerLoop st startf p end) = do
805 touchForeignPtr startf
807 return $ D.Yield x (InnerLoop st startf
808 (p `plusPtr` (sizeOf (undefined :: a))) end)
810{-# INLINE spliceTwo #-}
811spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
812spliceTwo arr1 arr2 = do
813 let src1 = unsafeForeignPtrToPtr (aStart arr1)
814 src2 = unsafeForeignPtrToPtr (aStart arr2)
815 len1 = aEnd arr1 `minusPtr` src1
816 len2 = aEnd arr2 `minusPtr` src2
818 arr <- liftIO $ newArray (len1 + len2)
819 let dst = unsafeForeignPtrToPtr (aStart arr)
822 memcpy (castPtr dst) (castPtr src1) len1
823 touchForeignPtr (aStart arr1)
824 memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2
825 touchForeignPtr (aStart arr2)
826 return arr { aEnd = dst `plusPtr` (len1 + len2) }
828-- Splice a new array into a pre-reserved array. The user must ensure that
829-- there is enough space in the array.
830{-# INLINE spliceWith #-}
831spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)
832spliceWith dst@(Array _ end bound) src = liftIO $ do
833 let srcLen = byteLength src
834 if end `plusPtr` srcLen > bound
835 then error "Bug: spliceIntoUnsafe: Not enough space in the target array"
837 withForeignPtr (aStart dst) $ \_ -> do
838 withForeignPtr (aStart src) $ \psrc -> do
840 memcpy (castPtr pdst) (castPtr psrc) srcLen
841 return $ dst { aEnd = pdst `plusPtr` srcLen }
843-- Splice a new array into a preallocated array, doubling the space if there is
844-- no space in the target array.
845{-# INLINE spliceWithDoubling #-}
846spliceWithDoubling :: (MonadIO m, Storable a)
847 => Array a -> Array a -> m (Array a)
848spliceWithDoubling dst@(Array start end bound) src = do
849 assert (end <= bound) (return ())
850 let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
853 if end `plusPtr` srcLen >= bound
855 let oldStart = unsafeForeignPtrToPtr start
856 oldSize = end `minusPtr` oldStart
857 newSize = max (oldSize * 2) (oldSize + srcLen)
858 liftIO $ realloc newSize dst
862data SpliceState s arr
864 | SpliceBuffering s arr
865 | SpliceYielding arr (SpliceState s arr)
868-- XXX can use general grouping combinators to achieve this?
869-- | Coalesce adajcent arrays in incoming stream to form bigger arrays of a
870-- maximum specified size.
873{-# INLINE_NORMAL packArraysChunksOf #-}
874packArraysChunksOf :: (MonadIO m, Storable a)
875 => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
876packArraysChunksOf n (D.Stream step state) =
877 D.Stream step' (SpliceInitial state)
881 {-# INLINE_LATE step' #-}
882 step' gst (SpliceInitial st) = do
884 -- XXX we can pass the module string from the higher level API
885 error $ "Streamly.Mem.Array.Types.packArraysChunksOf: the size of "
886 ++ "arrays [" ++ show n ++ "] must be a natural number"
889 D.Yield arr s -> return $
890 let len = byteLength arr
892 then D.Skip (SpliceYielding arr (SpliceInitial s))
893 else D.Skip (SpliceBuffering s arr)
894 D.Skip s -> return $ D.Skip (SpliceInitial s)
895 D.Stop -> return $ D.Stop
897 step' gst (SpliceBuffering st buf) = do
901 let len = byteLength buf + byteLength arr
904 D.Skip (SpliceYielding buf (SpliceBuffering s arr))
906 buf' <- if byteCapacity buf < n
907 then liftIO $ realloc n buf
909 buf'' <- spliceWith buf' arr
910 return $ D.Skip (SpliceBuffering s buf'')
911 D.Skip s -> return $ D.Skip (SpliceBuffering s buf)
912 D.Stop -> return $ D.Skip (SpliceYielding buf SpliceFinish)
914 step' _ SpliceFinish = return D.Stop
916 step' _ (SpliceYielding arr next) = return $ D.Yield arr next
918data GatherState s arr
920 | GatherBuffering s arr Int
921 | GatherYielding arr (GatherState s arr)
924-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
925-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
926-- each array and a maximum of @maxEntries@ entries in each array.
929{-# INLINE_NORMAL groupIOVecsOf #-}
930groupIOVecsOf :: MonadIO m
931 => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
932groupIOVecsOf n maxIOVLen (D.Stream step state) =
933 D.Stream step' (GatherInitial state)
937 {-# INLINE_LATE step' #-}
938 step' gst (GatherInitial st) = do
940 -- XXX we can pass the module string from the higher level API
941 error $ "Streamly.Mem.Array.Types.groupIOVecsOf: the size of "
942 ++ "groups [" ++ show n ++ "] must be a natural number"
943 when (maxIOVLen <= 0) $
944 -- XXX we can pass the module string from the higher level API
945 error $ "Streamly.Mem.Array.Types.groupIOVecsOf: the number of "
946 ++ "IOVec entries [" ++ show n ++ "] must be a natural number"
947 r <- step (adaptState gst) st
950 let p = unsafeForeignPtrToPtr (aStart arr)
952 iov <- liftIO $ newArray maxIOVLen
953 iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
956 then return $ D.Skip (GatherYielding iov' (GatherInitial s))
957 else return $ D.Skip (GatherBuffering s iov' len)
958 D.Skip s -> return $ D.Skip (GatherInitial s)
959 D.Stop -> return $ D.Stop
961 step' gst (GatherBuffering st iov len) = do
962 r <- step (adaptState gst) st
965 let p = unsafeForeignPtrToPtr (aStart arr)
966 alen = byteLength arr
968 if len' > n || length iov >= maxIOVLen
970 iov' <- liftIO $ newArray maxIOVLen
971 iov'' <- liftIO $ unsafeSnoc iov' (IOVec (castPtr p)
973 return $ D.Skip (GatherYielding iov
974 (GatherBuffering s iov'' alen))
976 iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
978 return $ D.Skip (GatherBuffering s iov' len')
979 D.Skip s -> return $ D.Skip (GatherBuffering s iov len)
980 D.Stop -> return $ D.Skip (GatherYielding iov GatherFinish)
982 step' _ GatherFinish = return D.Stop
984 step' _ (GatherYielding iov next) = return $ D.Yield iov next
986-- | Create two slices of an array without copying the original array. The
987-- specified index @i@ is the first index of the second slice.
990splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
991splitAt i arr@Array{..} =
992 let maxIndex = length arr - 1
994 then error "sliceAt: negative array index"
996 then error $ "sliceAt: specified array index " ++ show i
997 ++ " is beyond the maximum index " ++ show maxIndex
998 else let off = i * sizeOf (undefined :: a)
999 p = unsafeForeignPtrToPtr aStart `plusPtr` off
1006 { aStart = aStart `plusForeignPtr` off
1012-- Drops the separator byte
1013{-# INLINE breakOn #-}
1015 => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
1016breakOn sep arr@Array{..} = liftIO $ do
1017 let p = unsafeForeignPtrToPtr aStart
1018 loc <- c_memchr p sep (fromIntegral $ aEnd `minusPtr` p)
1029 { aStart = aStart `plusForeignPtr` (loc `minusPtr` p + 1)
1035data SplitState s arr
1039 | Yielding arr (SplitState s arr)
1042-- | Split a stream of arrays on a given separator byte, dropping the separator
1043-- and coalescing all the arrays between two separators into a single array.
1046{-# INLINE_NORMAL splitOn #-}
1050 -> D.Stream m (Array Word8)
1051 -> D.Stream m (Array Word8)
1052splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
1056 {-# INLINE_LATE step' #-}
1057 step' gst (Initial st) = do
1061 (arr1, marr2) <- breakOn byte arr
1062 return $ case marr2 of
1063 Nothing -> D.Skip (Buffering s arr1)
1064 Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
1065 D.Skip s -> return $ D.Skip (Initial s)
1066 D.Stop -> return $ D.Stop
1068 step' gst (Buffering st buf) = do
1072 (arr1, marr2) <- breakOn byte arr
1073 buf' <- spliceTwo buf arr1
1074 return $ case marr2 of
1075 Nothing -> D.Skip (Buffering s buf')
1076 Just x -> D.Skip (Yielding buf' (Splitting s x))
1077 D.Skip s -> return $ D.Skip (Buffering s buf)
1079 if byteLength buf == 0
1081 else D.Skip (Yielding buf Finishing)
1083 step' _ (Splitting st buf) = do
1084 (arr1, marr2) <- breakOn byte buf
1085 return $ case marr2 of
1086 Nothing -> D.Skip $ Buffering st arr1
1087 Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
1089 step' _ (Yielding arr next) = return $ D.Yield arr next
1090 step' _ Finishing = return $ D.Stop
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