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
42#if !defined(mingw32_HOST_OS)
80import Control.Exception (assert)
81import Control.DeepSeq (NFData(..))
82import Control.Monad (when)
83import Control.Monad.IO.Class (MonadIO(..))
84import Data.Functor.Identity (runIdentity)
85import Data.Word (Word8)
86import Foreign.C.String (CString)
87import Foreign.C.Types (CSize(..), CInt(..))
88import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
89import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
90import Foreign.Ptr (plusPtr, minusPtr, castPtr, nullPtr)
91import Foreign.Storable (Storable(..))
92import Prelude hiding (length, foldr, read, unlines, splitAt)
93import Text.Read (readPrec, readListPrec, readListPrecDefault)
95import GHC.Base (Addr#, realWorld#)
96import GHC.Exts (IsList, IsString(..))
97import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes, newForeignPtr_)
98import GHC.IO (IO(IO), unsafePerformIO)
99import GHC.Ptr (Ptr(..))
101import Streamly.Fold.Types (Fold(..))
102import Streamly.SVar (adaptState)
104#if !defined(mingw32_HOST_OS)
105import Streamly.FileSystem.FDIO (IOVec(..))
108import qualified Streamly.Streams.StreamD.Type as D
109import qualified Streamly.Streams.StreamK as K
110import qualified GHC.Exts as Exts
113import qualified Data.Foldable as F
116#if MIN_VERSION_base(4,10,0)
117import Foreign.ForeignPtr (plusForeignPtr)
119import GHC.Base (Int(..), plusAddr#)
120import GHC.ForeignPtr (ForeignPtr(..))
121plusForeignPtr :: ForeignPtr a -> Int -> ForeignPtr b
122plusForeignPtr (ForeignPtr addr c) (I# d) = ForeignPtr (plusAddr# addr d) c
125-------------------------------------------------------------------------------
127-------------------------------------------------------------------------------
129-- There are two goals that we want to fulfill with arrays. One, holding large
130-- amounts of data in non-GC memory and the ability to pass this data to and
131-- from the operating system without an extra copy overhead. Two, allow random
132-- access to elements based on their indices. The first one falls in the
133-- category of storage buffers while the second one falls in the category of
134-- maps/multisets/hashmaps.
136-- For the first requirement we use an array of Storables and store it in a
137-- ForeignPtr. We can have both immutable and mutable variants of this array
138-- using wrappers over the same underlying type.
140-- For the second requirement, we need a separate type for arrays of
141-- polymorphic values, for example vectors of handler functions, lookup tables.
142-- We can call this type a "vector" in contrast to arrays. It should not
143-- require Storable instance for the type. In that case we need to use an
144-- Array# instead of a ForeignPtr. This type of array would not reduce the GC
145-- overhead much because each element of the array still needs to be scanned by
146-- the GC. However, it can store polymorphic elements and allow random access
147-- to those. But in most cases random access means storage, and it means we
148-- need to avoid GC scanning except in cases of trivially small storage. One
149-- way to achieve that would be to put the array in a Compact region. However,
150-- if and when we mutate this, we will have to use a manual GC copying out to
151-- another CR and freeing the old one.
153-------------------------------------------------------------------------------
155-------------------------------------------------------------------------------
157-- We require that an array stores only Storable. Arrays are used for buffering
158-- data while streams are used for processing. If you want something to be
159-- buffered it better be Storable so that we can store it in non-GC memory.
161-- We can use a "Storable" constraint in the Array type and the Constraint can
162-- be automatically provided to a function that pattern matches on the Array
163-- type. However, it has huge performance cost, so we do not use it.
164-- XXX Can this cost be alleviated in GHC-8.10 specialization fix?
166-- XXX Another way to not require the Storable constraint in array operations
167-- is to store the elemSize in the array at construction and use that instead
168-- of using sizeOf. Need to charaterize perf cost of this.
170-- XXX rename the fields to "start, next, end".
177 { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
178 , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
179 , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
182-------------------------------------------------------------------------------
184-------------------------------------------------------------------------------
186foreign import ccall unsafe "string.h memcpy" c_memcpy
187 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)
189foreign import ccall unsafe "string.h strlen" c_strlen
190 :: CString -> IO CSize
192foreign import ccall unsafe "string.h memchr" c_memchr
193 :: Ptr Word8 -> Word8 -> CSize -> IO (Ptr Word8)
195-- XXX we are converting Int to CSize
196memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
197memcpy dst src len = c_memcpy dst src (fromIntegral len) >> return ()
199foreign import ccall unsafe "string.h memcmp" c_memcmp
200 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt
202-- XXX we are converting Int to CSize
203-- return True if the memory locations have identical contents
205memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool
207 r <- c_memcmp p1 p2 (fromIntegral len)
210{-# INLINE unsafeInlineIO #-}
211unsafeInlineIO :: IO a -> a
212unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r
214{-# INLINE bytesToCount #-}
215bytesToCount :: Storable a => a -> Int -> Int
217 let elemSize = sizeOf x
218 in n + elemSize - 1 `div` elemSize
220-------------------------------------------------------------------------------
222-------------------------------------------------------------------------------
224-- | Allocate an array that can hold 'count' items. The memory of the array is
227-- Note that this is internal routine, the reference to this array cannot be
228-- given out until the array has been written to and frozen.
229{-# INLINE newArray #-}
230newArray :: forall a. Storable a => Int -> IO (Array a)
232 let size = count * sizeOf (undefined :: a)
233 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
234 let p = unsafeForeignPtrToPtr fptr
238 , aBound = p `plusPtr` size
241-- | Allocate an Array of the given size and run an IO action passing the array
243{-# INLINE withNewArray #-}
244withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)
245withNewArray count f = do
246 arr <- newArray count
247 withForeignPtr (aStart arr) $ \p -> f p >> return arr
249-- XXX grow the array when we are beyond bound.
251-- Internal routine for when the array is being created. Appends one item at
252-- the end of the array. Useful when sequentially writing a stream to the
253-- array. DOES NOT CHECK THE ARRAY BOUNDS.
254{-# INLINE unsafeSnoc #-}
255unsafeSnoc :: forall a. Storable a => Array a -> a -> IO (Array a)
256unsafeSnoc arr@Array{..} x = do
257 when (aEnd == aBound) $
258 error "BUG: unsafeSnoc: writing beyond array bounds"
260 touchForeignPtr aStart
261 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
264snoc :: forall a. Storable a => Array a -> a -> IO (Array a)
265snoc arr@Array {..} x = do
268 let oldStart = unsafeForeignPtrToPtr aStart
269 size = aEnd `minusPtr` oldStart
270 newSize = (size + (sizeOf (undefined :: a)))
271 newPtr <- mallocPlainForeignPtrAlignedBytes newSize (alignment (undefined :: a))
272 withForeignPtr newPtr $ \pNew -> do
273 memcpy (castPtr pNew) (castPtr oldStart) size
274 poke (pNew `plusPtr` size) x
275 touchForeignPtr aStart
278 , aEnd = pNew `plusPtr` (size + sizeOf (undefined :: a))
279 , aBound = pNew `plusPtr` newSize
283 touchForeignPtr aStart
284 return $ arr {aEnd = aEnd `plusPtr` (sizeOf (undefined :: a))}
286-- | Reallocate the array to the specified size in bytes. If the size is less
287-- than the original array the array gets truncated.
288{-# NOINLINE realloc #-}
289realloc :: forall a. Storable a => Int -> Array a -> IO (Array a)
290realloc newSize Array{..} = do
291 assert (aEnd <= aBound) (return ())
292 let oldStart = unsafeForeignPtrToPtr aStart
293 let size = aEnd `minusPtr` oldStart
294 newPtr <- mallocPlainForeignPtrAlignedBytes
295 newSize (alignment (undefined :: a))
296 withForeignPtr newPtr $ \pNew -> do
297 memcpy (castPtr pNew) (castPtr oldStart) size
298 touchForeignPtr aStart
301 , aEnd = pNew `plusPtr` size
302 , aBound = pNew `plusPtr` newSize
305-- | Remove the free space from an Array.
306shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
307shrinkToFit arr@Array{..} = do
308 assert (aEnd <= aBound) (return ())
311 let oldStart = unsafeForeignPtrToPtr aStart
312 let size = aEnd `minusPtr` oldStart
316-- XXX when converting an array of Word8 from a literal string we can simply
317-- refer to the literal string. Is it possible to write rules such that
318-- fromList Word8 can be rewritten so that GHC does not first convert the
319-- literal to [Char] and then we convert it back to an Array Word8?
321-- Note that the address must be a read-only address (meant to be used for
322-- read-only string literals) because we are sharing it, any modification to
323-- the original address would change our array. That's why this function is
325{-# INLINE _fromCStringAddrUnsafe #-}
326_fromCStringAddrUnsafe :: Addr# -> IO (Array Word8)
327_fromCStringAddrUnsafe addr# = do
328 ptr <- newForeignPtr_ (castPtr cstr)
330 let n = fromIntegral len
331 let p = unsafeForeignPtrToPtr ptr
332 let end = p `plusPtr` n
342-------------------------------------------------------------------------------
344-------------------------------------------------------------------------------
346-- | Return element at the specified index without checking the bounds.
348-- Unsafe because it does not check the bounds of the array.
349{-# INLINE_NORMAL unsafeIndexIO #-}
350unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
351unsafeIndexIO Array {..} i =
352 withForeignPtr aStart $ \p -> do
353 let elemSize = sizeOf (undefined :: a)
354 elemOff = p `plusPtr` (elemSize * i)
355 assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
359-- | Return element at the specified index without checking the bounds.
360{-# INLINE_NORMAL unsafeIndex #-}
361unsafeIndex :: forall a. Storable a => Array a -> Int -> a
362unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
364-- | /O(1)/ Get the byte length of the array.
367{-# INLINE byteLength #-}
368byteLength :: Array a -> Int
369byteLength Array{..} =
370 let p = unsafeForeignPtrToPtr aStart
371 len = aEnd `minusPtr` p
372 in assert (len >= 0) len
374-- | /O(1)/ Get the length of the array i.e. the number of elements in the
379length :: forall a. Storable a => Array a -> Int
380length arr = byteLength arr `div` sizeOf (undefined :: a)
382{-# INLINE byteCapacity #-}
383byteCapacity :: Array a -> Int
384byteCapacity Array{..} =
385 let p = unsafeForeignPtrToPtr aStart
386 len = aBound `minusPtr` p
387 in assert (len >= 0) len
389{-# INLINE_NORMAL toStreamD #-}
390toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
392 let p = unsafeForeignPtrToPtr aStart
397 {-# INLINE_LATE step #-}
398 step _ p | p == aEnd = return D.Stop
400 -- unsafeInlineIO allows us to run this in Identity monad for pure
401 -- toList/foldr case which makes them much faster due to not
402 -- accumulating the list and fusing better with the pure consumers.
404 -- This should be safe as the array contents are guaranteed to be
405 -- evaluated/written to before we peek at them.
406 let !x = unsafeInlineIO $ do
408 touchForeignPtr aStart
410 return $ D.Yield x (p `plusPtr` (sizeOf (undefined :: a)))
412{-# INLINE toStreamK #-}
413toStreamK :: forall t m a. (K.IsStream t, Storable a) => Array a -> t m a
415 let p = unsafeForeignPtrToPtr aStart
420 go p | p == aEnd = K.nil
422 -- See Note in toStreamD.
423 let !x = unsafeInlineIO $ do
425 touchForeignPtr aStart
427 in x `K.cons` go (p `plusPtr` (sizeOf (undefined :: a)))
429{-# INLINE_NORMAL toStreamDRev #-}
430toStreamDRev :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
431toStreamDRev Array{..} =
432 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
437 {-# INLINE_LATE step #-}
438 step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop
440 -- See comments in toStreamD for why we use unsafeInlineIO
441 let !x = unsafeInlineIO $ do
443 touchForeignPtr aStart
445 return $ D.Yield x (p `plusPtr` negate (sizeOf (undefined :: a)))
447{-# INLINE toStreamKRev #-}
448toStreamKRev :: forall t m a. (K.IsStream t, Storable a) => Array a -> t m a
449toStreamKRev Array {..} =
450 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
455 go p | p < unsafeForeignPtrToPtr aStart = K.nil
457 let !x = unsafeInlineIO $ do
459 touchForeignPtr aStart
461 in x `K.cons` go (p `plusPtr` negate (sizeOf (undefined :: a)))
463{-# INLINE_NORMAL foldl' #-}
464foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
465foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
467{-# INLINE_NORMAL foldr #-}
468foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
469foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
471-------------------------------------------------------------------------------
473-------------------------------------------------------------------------------
475-- | @toArrayN n@ folds a maximum of @n@ elements from the input stream to an
479{-# INLINE_NORMAL toArrayN #-}
480toArrayN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
481toArrayN n = Fold step initial extract
486 if n < 0 then error "toArrayN: negative count specified" else return ()
488 step arr@(Array _ end bound) _ | end == bound = return arr
489 step (Array start end bound) x = do
491 return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
492 extract = liftIO . shrinkToFit
494{-# INLINE_NORMAL fromStreamDN #-}
495fromStreamDN :: forall m a. (MonadIO m, Storable a)
496 => Int -> D.Stream m a -> m (Array a)
497fromStreamDN limit str = do
498 arr <- liftIO $ newArray limit
499 end <- D.foldlM' write (aEnd arr) $ D.take limit str
500 return $ arr {aEnd = end}
506 return $ ptr `plusPtr` sizeOf (undefined :: a)
508data GroupState s start end bound
510 | GroupBuffer s start end bound
511 | GroupYield start end bound (GroupState s start end bound)
514-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
516{-# INLINE_NORMAL fromStreamDArraysOf #-}
517fromStreamDArraysOf :: forall m a. (MonadIO m, Storable a)
518 => Int -> D.Stream m a -> D.Stream m (Array a)
519-- fromStreamDArraysOf n str = D.groupsOf n (toArrayN n) str
520fromStreamDArraysOf n (D.Stream step state) =
521 D.Stream step' (GroupStart state)
525 {-# INLINE_LATE step' #-}
526 step' _ (GroupStart st) = do
528 -- XXX we can pass the module string from the higher level API
529 error $ "Streamly.Mem.Array.Types.fromStreamDArraysOf: the size of "
530 ++ "arrays [" ++ show n ++ "] must be a natural number"
531 Array start end bound <- liftIO $ newArray n
532 return $ D.Skip (GroupBuffer st start end bound)
534 step' gst (GroupBuffer st start end bound) = do
535 r <- step (adaptState gst) st
539 let end' = end `plusPtr` sizeOf (undefined :: a)
542 then D.Skip (GroupYield start end' bound (GroupStart s))
543 else D.Skip (GroupBuffer s start end' bound)
544 D.Skip s -> return $ D.Skip (GroupBuffer s start end bound)
545 D.Stop -> return $ D.Skip (GroupYield start end bound GroupFinish)
547 step' _ (GroupYield start end bound next) =
548 return $ D.Yield (Array start end bound) next
550 step' _ GroupFinish = return D.Stop
552-- XXX concatMap does not seem to have the best possible performance so we have
553-- a custom way to concat arrays.
554data FlattenState s a =
556 | InnerLoop s (ForeignPtr a) (Ptr a) (Ptr a)
558{-# INLINE_NORMAL flattenArrays #-}
559flattenArrays :: forall m a. (MonadIO m, Storable a)
560 => D.Stream m (Array a) -> D.Stream m a
561flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)
565 {-# INLINE_LATE step' #-}
566 step' gst (OuterLoop st) = do
567 r <- step (adaptState gst) st
569 D.Yield Array{..} s ->
570 let p = unsafeForeignPtrToPtr aStart
571 in D.Skip (InnerLoop s aStart p aEnd)
572 D.Skip s -> D.Skip (OuterLoop s)
575 step' _ (InnerLoop st _ p end) | p == end =
576 return $ D.Skip $ OuterLoop st
578 step' _ (InnerLoop st startf p end) = do
581 touchForeignPtr startf
583 return $ D.Yield x (InnerLoop st startf
584 (p `plusPtr` (sizeOf (undefined :: a))) end)
586{-# INLINE_NORMAL flattenArraysRev #-}
587flattenArraysRev :: forall m a. (MonadIO m, Storable a)
588 => D.Stream m (Array a) -> D.Stream m a
589flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
593 {-# INLINE_LATE step' #-}
594 step' gst (OuterLoop st) = do
595 r <- step (adaptState gst) st
597 D.Yield Array{..} s ->
598 let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
599 -- XXX we do not need aEnd
600 in D.Skip (InnerLoop s aStart p aEnd)
601 D.Skip s -> D.Skip (OuterLoop s)
604 step' _ (InnerLoop st start p _) | p < unsafeForeignPtrToPtr start =
605 return $ D.Skip $ OuterLoop st
607 step' _ (InnerLoop st startf p end) = do
610 touchForeignPtr startf
612 return $ D.Yield x (InnerLoop st startf
613 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
615-- CAUTION: a very large number (millions) of arrays can degrade performance
616-- due to GC overhead because we need to buffer the arrays before we flatten
619-- We could take the approach of doubling the memory allocation on each
620-- overflow. This would result in more or less the same amount of copying as in
621-- the chunking approach. However, if we have to shrink in the end then it may
622-- result in an extra copy of the entire data.
624{-# INLINE fromStreamD #-}
625fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
627 let s = fromStreamDArraysOf defaultChunkSize m
628 buffered <- D.foldr K.cons K.nil s
629 len <- K.foldl' (+) 0 (K.map length buffered)
630 fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
632-- | Convert an 'Array' into a list.
635{-# INLINABLE toList #-}
636toList :: Storable a => Array a -> [a]
639instance (Show a, Storable a) => Show (Array a) where
640 {-# INLINE showsPrec #-}
641 showsPrec _ = shows . toList
643-- | Create an 'Array' from the first N elements of a list. The array is
644-- allocated to size N, if the list terminates before N elements then the
645-- array may hold less than N elements.
648{-# INLINABLE fromListN #-}
649fromListN :: Storable a => Int -> [a] -> Array a
650fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
652-- | Create an 'Array' from a list. The list must be of finite size.
655{-# INLINABLE fromList #-}
656fromList :: Storable a => [a] -> Array a
657fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
659instance (Storable a, Read a, Show a) => Read (Array a) where
660 {-# INLINE readPrec #-}
664 readListPrec = readListPrecDefault
666instance (a ~ Char) => IsString (Array a) where
667 {-# INLINE fromString #-}
668 fromString = fromList
670-- GHC versions 8.0 and below cannot derive IsList
671instance Storable a => IsList (Array a) where
672 type (Item (Array a)) = a
673 {-# INLINE fromList #-}
675 {-# INLINE fromListN #-}
676 fromListN = fromListN
677 {-# INLINE toList #-}
681arrcmp :: Array a -> Array a -> Bool
683 let !res = unsafeInlineIO $ do
684 let ptr1 = unsafeForeignPtrToPtr $ aStart arr1
685 let ptr2 = unsafeForeignPtrToPtr $ aStart arr2
686 let len1 = aEnd arr1 `minusPtr` ptr1
687 let len2 = aEnd arr2 `minusPtr` ptr2
691 r <- memcmp (castPtr ptr1) (castPtr ptr2) len1
692 touchForeignPtr $ aStart arr1
693 touchForeignPtr $ aStart arr2
698-- XXX we are assuming that Storable equality means element equality. This may
699-- or may not be correct? arrcmp is 40% faster compared to stream equality.
700instance (Storable a, Eq a) => Eq (Array a) where
703 -- arr1 == arr2 = runIdentity $ D.eqBy (==) (toStreamD arr1) (toStreamD arr2)
705instance (Storable a, NFData a) => NFData (Array a) where
707 rnf = foldl' (\_ x -> rnf x) ()
709instance (Storable a, Ord a) => Ord (Array a) where
710 {-# INLINE compare #-}
711 compare arr1 arr2 = unsafePerformIO $
712 D.cmpBy compare (toStreamD arr1) (toStreamD arr2)
714 -- Default definitions defined in base do not have an INLINE on them, so we
715 -- replicate them here with an INLINE.
717 x < y = case compare x y of { LT -> True; _ -> False }
720 x <= y = case compare x y of { GT -> False; _ -> True }
723 x > y = case compare x y of { GT -> True; _ -> False }
726 x >= y = case compare x y of { LT -> False; _ -> True }
728 -- These two default methods use '<=' rather than 'compare'
729 -- because the latter is often more expensive
731 max x y = if x <= y then y else x
734 min x y = if x <= y then x else y
737-- Definitions using the Storable constraint from the Array type. These are to
738-- make the Foldable instance possible though it is much slower (7x slower).
740{-# INLINE_NORMAL toStreamD_ #-}
741toStreamD_ :: forall m a. MonadIO m => Int -> Array a -> D.Stream m a
742toStreamD_ size Array{..} =
743 let p = unsafeForeignPtrToPtr aStart
748 {-# INLINE_LATE step #-}
749 step _ p | p == aEnd = return D.Stop
753 touchForeignPtr aStart
755 return $ D.Yield x (p `plusPtr` size)
757{-# INLINE_NORMAL _foldr #-}
758_foldr :: forall a b. (a -> b -> b) -> b -> Array a -> b
759_foldr f z arr@Array {..} =
760 let !n = sizeOf (undefined :: a)
761 in unsafePerformIO $ D.foldr f z $ toStreamD_ n arr
763-- | Note that the 'Foldable' instance is 7x slower than the direct
765instance Foldable Array where
769-- | Convert an 'Array' into a stream.
772{-# INLINE_EARLY read #-}
773read :: (Monad m, K.IsStream t, Storable a) => Array a -> t m a
774read = D.fromStreamD . toStreamD
775-- XXX add fallback to StreamK rule
776-- {-# RULES "Streamly.Array.read fallback to StreamK" [1]
777-- forall a. S.readK (read a) = K.fromArray a #-}
779-------------------------------------------------------------------------------
780-- Semigroup and Monoid
781-------------------------------------------------------------------------------
783-------------------------------------------------------------------------------
785-------------------------------------------------------------------------------
787-- | GHC memory management allocation header overhead
789allocOverhead = 2 * sizeOf (undefined :: Int)
791mkChunkSize :: Int -> Int
792mkChunkSize n = let size = n - allocOverhead in max size 0
794mkChunkSizeKB :: Int -> Int
795mkChunkSizeKB n = mkChunkSize (n * k)
798-- | Default maximum buffer size in bytes, for reading from and writing to IO
799-- devices, the value is 32KB minus GHC allocation overhead, which is a few
800-- bytes, so that the actual allocation is 32KB.
801defaultChunkSize :: Int
802defaultChunkSize = mkChunkSizeKB 32
804{-# INLINE_NORMAL unlines #-}
805unlines :: forall m a. (MonadIO m, Storable a)
806 => a -> D.Stream m (Array a) -> D.Stream m a
807unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
809 {-# INLINE_LATE step' #-}
810 step' gst (OuterLoop st) = do
811 r <- step (adaptState gst) st
813 D.Yield Array{..} s ->
814 let p = unsafeForeignPtrToPtr aStart
815 in D.Skip (InnerLoop s aStart p aEnd)
816 D.Skip s -> D.Skip (OuterLoop s)
819 step' _ (InnerLoop st _ p end) | p == end =
820 return $ D.Yield sep $ OuterLoop st
822 step' _ (InnerLoop st startf p end) = do
825 touchForeignPtr startf
827 return $ D.Yield x (InnerLoop st startf
828 (p `plusPtr` (sizeOf (undefined :: a))) end)
830{-# INLINE spliceTwo #-}
831spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
832spliceTwo arr1 arr2 = do
833 let src1 = unsafeForeignPtrToPtr (aStart arr1)
834 src2 = unsafeForeignPtrToPtr (aStart arr2)
835 len1 = aEnd arr1 `minusPtr` src1
836 len2 = aEnd arr2 `minusPtr` src2
838 arr <- liftIO $ newArray (len1 + len2)
839 let dst = unsafeForeignPtrToPtr (aStart arr)
842 memcpy (castPtr dst) (castPtr src1) len1
843 touchForeignPtr (aStart arr1)
844 memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2
845 touchForeignPtr (aStart arr2)
846 return arr { aEnd = dst `plusPtr` (len1 + len2) }
848-- Splice a new array into a pre-reserved array. The user must ensure that
849-- there is enough space in the array.
850{-# INLINE spliceWith #-}
851spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)
852spliceWith dst@(Array _ end bound) src = liftIO $ do
853 let srcLen = byteLength src
854 if end `plusPtr` srcLen > bound
855 then error "Bug: spliceIntoUnsafe: Not enough space in the target array"
857 withForeignPtr (aStart dst) $ \_ -> do
858 withForeignPtr (aStart src) $ \psrc -> do
860 memcpy (castPtr pdst) (castPtr psrc) srcLen
861 return $ dst { aEnd = pdst `plusPtr` srcLen }
863-- Splice a new array into a preallocated array, doubling the space if there is
864-- no space in the target array.
865{-# INLINE spliceWithDoubling #-}
866spliceWithDoubling :: (MonadIO m, Storable a)
867 => Array a -> Array a -> m (Array a)
868spliceWithDoubling dst@(Array start end bound) src = do
869 assert (end <= bound) (return ())
870 let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
873 if end `plusPtr` srcLen >= bound
875 let oldStart = unsafeForeignPtrToPtr start
876 oldSize = end `minusPtr` oldStart
877 newSize = max (oldSize * 2) (oldSize + srcLen)
878 liftIO $ realloc newSize dst
882data SpliceState s arr
884 | SpliceBuffering s arr
885 | SpliceYielding arr (SpliceState s arr)
888-- XXX can use general grouping combinators to achieve this?
889-- | Coalesce adajcent arrays in incoming stream to form bigger arrays of a
890-- maximum specified size.
893{-# INLINE_NORMAL packArraysChunksOf #-}
894packArraysChunksOf :: (MonadIO m, Storable a)
895 => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
896packArraysChunksOf n (D.Stream step state) =
897 D.Stream step' (SpliceInitial state)
901 {-# INLINE_LATE step' #-}
902 step' gst (SpliceInitial st) = do
904 -- XXX we can pass the module string from the higher level API
905 error $ "Streamly.Mem.Array.Types.packArraysChunksOf: the size of "
906 ++ "arrays [" ++ show n ++ "] must be a natural number"
909 D.Yield arr s -> return $
910 let len = byteLength arr
912 then D.Skip (SpliceYielding arr (SpliceInitial s))
913 else D.Skip (SpliceBuffering s arr)
914 D.Skip s -> return $ D.Skip (SpliceInitial s)
915 D.Stop -> return $ D.Stop
917 step' gst (SpliceBuffering st buf) = do
921 let len = byteLength buf + byteLength arr
924 D.Skip (SpliceYielding buf (SpliceBuffering s arr))
926 buf' <- if byteCapacity buf < n
927 then liftIO $ realloc n buf
929 buf'' <- spliceWith buf' arr
930 return $ D.Skip (SpliceBuffering s buf'')
931 D.Skip s -> return $ D.Skip (SpliceBuffering s buf)
932 D.Stop -> return $ D.Skip (SpliceYielding buf SpliceFinish)
934 step' _ SpliceFinish = return D.Stop
936 step' _ (SpliceYielding arr next) = return $ D.Yield arr next
938#if !defined(mingw32_HOST_OS)
939data GatherState s arr
941 | GatherBuffering s arr Int
942 | GatherYielding arr (GatherState s arr)
945-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
946-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
947-- each array and a maximum of @maxEntries@ entries in each array.
950{-# INLINE_NORMAL groupIOVecsOf #-}
951groupIOVecsOf :: MonadIO m
952 => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
953groupIOVecsOf n maxIOVLen (D.Stream step state) =
954 D.Stream step' (GatherInitial state)
958 {-# INLINE_LATE step' #-}
959 step' gst (GatherInitial st) = do
961 -- XXX we can pass the module string from the higher level API
962 error $ "Streamly.Mem.Array.Types.groupIOVecsOf: the size of "
963 ++ "groups [" ++ show n ++ "] must be a natural number"
964 when (maxIOVLen <= 0) $
965 -- XXX we can pass the module string from the higher level API
966 error $ "Streamly.Mem.Array.Types.groupIOVecsOf: the number of "
967 ++ "IOVec entries [" ++ show n ++ "] must be a natural number"
968 r <- step (adaptState gst) st
971 let p = unsafeForeignPtrToPtr (aStart arr)
973 iov <- liftIO $ newArray maxIOVLen
974 iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
977 then return $ D.Skip (GatherYielding iov' (GatherInitial s))
978 else return $ D.Skip (GatherBuffering s iov' len)
979 D.Skip s -> return $ D.Skip (GatherInitial s)
980 D.Stop -> return $ D.Stop
982 step' gst (GatherBuffering st iov len) = do
983 r <- step (adaptState gst) st
986 let p = unsafeForeignPtrToPtr (aStart arr)
987 alen = byteLength arr
989 if len' > n || length iov >= maxIOVLen
991 iov' <- liftIO $ newArray maxIOVLen
992 iov'' <- liftIO $ unsafeSnoc iov' (IOVec (castPtr p)
994 return $ D.Skip (GatherYielding iov
995 (GatherBuffering s iov'' alen))
997 iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
999 return $ D.Skip (GatherBuffering s iov' len')
1000 D.Skip s -> return $ D.Skip (GatherBuffering s iov len)
1001 D.Stop -> return $ D.Skip (GatherYielding iov GatherFinish)
1003 step' _ GatherFinish = return D.Stop
1005 step' _ (GatherYielding iov next) = return $ D.Yield iov next
1008-- | Create two slices of an array without copying the original array. The
1009-- specified index @i@ is the first index of the second slice.
1012splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
1013splitAt i arr@Array{..} =
1014 let maxIndex = length arr - 1
1016 then error "sliceAt: negative array index"
1017 else if i > maxIndex
1018 then error $ "sliceAt: specified array index " ++ show i
1019 ++ " is beyond the maximum index " ++ show maxIndex
1020 else let off = i * sizeOf (undefined :: a)
1021 p = unsafeForeignPtrToPtr aStart `plusPtr` off
1028 { aStart = aStart `plusForeignPtr` off
1034-- Drops the separator byte
1035{-# INLINE breakOn #-}
1037 => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
1038breakOn sep arr@Array{..} = liftIO $ do
1039 let p = unsafeForeignPtrToPtr aStart
1040 loc <- c_memchr p sep (fromIntegral $ aEnd `minusPtr` p)
1051 { aStart = aStart `plusForeignPtr` (loc `minusPtr` p + 1)
1057data SplitState s arr
1061 | Yielding arr (SplitState s arr)
1064-- | Split a stream of arrays on a given separator byte, dropping the separator
1065-- and coalescing all the arrays between two separators into a single array.
1068{-# INLINE_NORMAL splitOn #-}
1072 -> D.Stream m (Array Word8)
1073 -> D.Stream m (Array Word8)
1074splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
1078 {-# INLINE_LATE step' #-}
1079 step' gst (Initial st) = do
1083 (arr1, marr2) <- breakOn byte arr
1084 return $ case marr2 of
1085 Nothing -> D.Skip (Buffering s arr1)
1086 Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
1087 D.Skip s -> return $ D.Skip (Initial s)
1088 D.Stop -> return $ D.Stop
1090 step' gst (Buffering st buf) = do
1094 (arr1, marr2) <- breakOn byte arr
1095 buf' <- spliceTwo buf arr1
1096 return $ case marr2 of
1097 Nothing -> D.Skip (Buffering s buf')
1098 Just x -> D.Skip (Yielding buf' (Splitting s x))
1099 D.Skip s -> return $ D.Skip (Buffering s buf)
1101 if byteLength buf == 0
1103 else D.Skip (Yielding buf Finishing)
1105 step' _ (Splitting st buf) = do
1106 (arr1, marr2) <- breakOn byte buf
1107 return $ case marr2 of
1108 Nothing -> D.Skip $ Buffering st arr1
1109 Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
1111 step' _ (Yielding arr next) = return $ D.Yield arr next
1112 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