@@ -1,71 +1,69 @@
1{-# LANGUAGE UnboxedTuples #-}
1{-# LANGUAGE UnboxedTuples #-}
6-- Module : Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types
4-- Module : Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types
7-- Copyright : (c) 2020 Composewell Technologies
5-- Copyright : (c) 2020 Composewell Technologies
9-- License : BSD3-3-Clause
6-- License : BSD3-3-Clause
10-- Maintainer : streamly@composewell.com
7-- Maintainer : streamly@composewell.com
11-- Stability : experimental
8-- Stability : experimental
12-- Portability : GHC
9-- Portability : GHC
11+-- Unboxed pinned mutable array type for 'Storable' types with an option to use
12+-- foreign (non-GHC) memory allocators. Fulfils the following goals:
14+-- * Random access (array)
15+-- * Efficient storage (unboxed)
16+-- * Performance (unboxed access)
17+-- * Performance - in-place operations (mutable)
18+-- * Performance - GC (pinned, mutable)
19+-- * interfacing with OS (pinned)
20+-- * Fragmentation control (foreign allocators)
22+-- Stream and Fold APIs allow easy, efficient and convenient operations on
14module Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types
25module Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types
16 Array (..)
29 Array (..)
19 -- * Construction
31 -- * Construction
20 , withNewArray
33 , withNewArray
21 , newArray
34 , newArray
22 , newArrayAligned
35 , newArrayAligned
23 , newArrayAlignedUnmanaged
36 , newArrayAlignedUnmanaged
24 , newArrayAlignedAllocWith
37 , newArrayAlignedAllocWith
31 , fromList
40 , fromList
32 , fromListN
41 , fromListN
33 , fromStreamDN
42 , fromStreamDN
34− -- , fromStreamD
43+ , fromStreamD
36− -- * Streams of arrays
45+ -- * Resizing
37− , fromStreamDArraysOf
46+ , realloc
38− , FlattenState (..) -- for inspection testing
47+ , shrinkToFit
43− , lpackArraysChunksOf
44−#if !defined(mingw32_HOST_OS)
50− -- * Elimination
49+ -- * Size
54 , byteLength
51 , byteLength
55 , byteCapacity
52 , byteCapacity
60− , toStreamD
54+ -- * Random access
61− , toStreamDRev
55+ , unsafeIndexIO
62− , toStreamK
56+ , unsafeIndex
64− , toList
58+ -- * Mutation
66− , unlines
62+ -- * Folding
68− -- * Folds
66+ -- * Composable Folds
69 , toArrayMinChunk
67 , toArrayMinChunk
70 , writeNAllocWith
68 , writeNAllocWith
@@ -76,24 +74,50 @@ module Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types
77 , writeAligned
75 , writeAligned
99+ -- * Stream of arrays
79 -- * Utilities
104 -- * Utilities
80 , defaultChunkSize
105 , defaultChunkSize
81 , mkChunkSize
106 , mkChunkSize
82 , mkChunkSizeKB
107 , mkChunkSizeKB
83 , unsafeInlineIO
109 , unsafeInlineIO
92import Control.Exception (assert)
117import Control.Exception (assert)
93import Control.DeepSeq (NFData(..))
118import Control.DeepSeq (NFData(..))
94import Control.Monad (when, void)
119import Control.Monad (when, void)
95import Control.Monad.IO.Class (MonadIO(..))
120import Control.Monad.IO.Class (MonadIO(..))
96−import Data.Bifunctor (first)
97import Data.Functor.Identity (runIdentity)
121import Data.Functor.Identity (runIdentity)
98#if __GLASGOW_HASKELL__ < 808
122#if __GLASGOW_HASKELL__ < 808
99import Data.Semigroup (Semigroup(..))
123import Data.Semigroup (Semigroup(..))
@@ -109,11 +133,8 @@ import GHC.Exts (IsList, IsString(..))
109import GHC.ForeignPtr (ForeignPtr(..))
133import GHC.ForeignPtr (ForeignPtr(..))
110import GHC.IO (IO(IO), unsafePerformIO)
134import GHC.IO (IO(IO), unsafePerformIO)
111import GHC.Ptr (Ptr(..))
135import GHC.Ptr (Ptr(..))
112−#if !defined(mingw32_HOST_OS)
113−import Streamly.FileSystem.FDIO (IOVec(..))
115import Streamly.Internal.Data.Fold.Types (Fold(..))
136import Streamly.Internal.Data.Fold.Types (Fold(..))
116−import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
137+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
117import Streamly.Internal.Data.SVar (adaptState)
138import Streamly.Internal.Data.SVar (adaptState)
118import Text.Read (readPrec, readListPrec, readListPrecDefault)
139import Text.Read (readPrec, readListPrec, readListPrecDefault)
@@ -137,61 +158,27 @@ plusForeignPtr :: ForeignPtr a -> Int -> ForeignPtr b
137plusForeignPtr (ForeignPtr addr c) (I# d) = ForeignPtr (plusAddr# addr d) c
158plusForeignPtr (ForeignPtr addr c) (I# d) = ForeignPtr (plusAddr# addr d) c
140−-------------------------------------------------------------------------------
142−-------------------------------------------------------------------------------
144−-- There are two goals that we want to fulfill with arrays. One, holding large
145−-- amounts of data in non-GC memory and the ability to pass this data to and
146−-- from the operating system without an extra copy overhead. Two, allow random
147−-- access to elements based on their indices. The first one falls in the
148−-- category of storage buffers while the second one falls in the category of
149−-- maps/multisets/hashmaps.
151−-- For the first requirement we use an array of Storables and store it in a
152−-- ForeignPtr. We can have both immutable and mutable variants of this array
153−-- using wrappers over the same underlying type.
155−-- For the second requirement, we need a separate type for arrays of
156−-- polymorphic values, for example vectors of handler functions, lookup tables.
157−-- We can call this type a "vector" in contrast to arrays. It should not
158−-- require Storable instance for the type. In that case we need to use an
159−-- Array# instead of a ForeignPtr. This type of array would not reduce the GC
160−-- overhead much because each element of the array still needs to be scanned by
161−-- the GC. However, it can store polymorphic elements and allow random access
162−-- to those. But in most cases random access means storage, and it means we
163−-- need to avoid GC scanning except in cases of trivially small storage. One
164−-- way to achieve that would be to put the array in a Compact region. However,
165−-- if and when we mutate this, we will have to use a manual GC copying out to
166−-- another CR and freeing the old one.
168-------------------------------------------------------------------------------
161-------------------------------------------------------------------------------
169-- Array Data Type
162-- Array Data Type
170-------------------------------------------------------------------------------
163-------------------------------------------------------------------------------
172−-- We require that an array stores only Storable. Arrays are used for buffering
165+-- $arrayNotes
173−-- data while streams are used for processing. If you want something to be
174−-- buffered it better be Storable so that we can store it in non-GC memory.
176−-- We can use a "Storable" constraint in the Array type and the Constraint can
167+-- We can use a 'Storable' constraint in the Array type and the constraint can
177-- be automatically provided to a function that pattern matches on the Array
168-- be automatically provided to a function that pattern matches on the Array
178-- type. However, it has huge performance cost, so we do not use it.
169-- type. However, it has huge performance cost, so we do not use it.
179−-- XXX Can this cost be alleviated in GHC-8.10 specialization fix?
170+-- Investigate a GHC improvement possiblity.
181−-- XXX Another way to not require the Storable constraint in array operations
172+-- XXX Rename the fields to better names.
182−-- is to store the elemSize in the array at construction and use that instead
183−-- of using sizeOf. Need to charaterize perf cost of this.
185−-- XXX rename the fields to "start, next, end".
187data Array a =
174data Array a =
188#ifdef DEVBUILD
175#ifdef DEVBUILD
189 Storable a =>
176 Storable a =>
192− { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
179+ { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- ^ first address
193− , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
180+ , aEnd :: {-# UNPACK #-} !(Ptr a) -- ^ first unused address
194− , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
181+ , aBound :: {-# UNPACK #-} !(Ptr a) -- ^ first address beyond allocated memory
197{-# INLINE mutableArray #-}
184{-# INLINE mutableArray #-}
@@ -318,6 +305,10 @@ withNewArray count f = do
318 arr <- newArray count
305 arr <- newArray count
319 withForeignPtr (aStart arr) $ \p -> f p >> return arr
306 withForeignPtr (aStart arr) $ \p -> f p >> return arr
308+-------------------------------------------------------------------------------
310+-------------------------------------------------------------------------------
321-- XXX grow the array when we are beyond bound.
312-- XXX grow the array when we are beyond bound.
323-- Internal routine for when the array is being created. Appends one item at
314-- Internal routine for when the array is being created. Appends one item at
@@ -358,6 +349,10 @@ snoc arr@Array {..} x =
358 touchForeignPtr aStart
349 touchForeignPtr aStart
359 return $ arr {aEnd = aEnd `plusPtr` sizeOf (undefined :: a)}
350 return $ arr {aEnd = aEnd `plusPtr` sizeOf (undefined :: a)}
352+-------------------------------------------------------------------------------
354+-------------------------------------------------------------------------------
361-- | Reallocate the array to the specified size in bytes. If the size is less
356-- | Reallocate the array to the specified size in bytes. If the size is less
362-- than the original array the array gets truncated.
357-- than the original array the array gets truncated.
363{-# NOINLINE reallocAligned #-}
358{-# NOINLINE reallocAligned #-}
@@ -381,117 +376,59 @@ reallocAligned alignSize newSize Array{..} = do
381realloc :: forall a. Storable a => Int -> Array a -> IO (Array a)
376realloc :: forall a. Storable a => Int -> Array a -> IO (Array a)
382realloc = reallocAligned (alignment (undefined :: a))
377realloc = reallocAligned (alignment (undefined :: a))
384−-- | Create an 'Array' from the first N elements of a list. The array is
379+-------------------------------------------------------------------------------
385−-- allocated to size N, if the list terminates before N elements then the
380+-- Elimination
386−-- array may hold less than N elements.
381+-------------------------------------------------------------------------------
389−{-# INLINABLE fromListN #-}
390−fromListN :: Storable a => Int -> [a] -> Array a
391−fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
393−-- | Create an 'Array' from a list. The list must be of finite size.
383+-- | Return element at the specified index without checking the bounds.
395−-- @since 0.7.0
385+-- Unsafe because it does not check the bounds of the array.
396−{-# INLINABLE fromList #-}
397−fromList :: Storable a => [a] -> Array a
398−fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
400−{-# INLINE_NORMAL fromStreamDN #-}
386+{-# INLINE_NORMAL unsafeIndexIO #-}
401−fromStreamDN :: forall m a. (MonadIO m, Storable a)
387+unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
402− => Int -> D.Stream m a -> m (Array a)
388+unsafeIndexIO Array {..} i =
403−fromStreamDN limit str = do
389+ withForeignPtr aStart $ \p -> do
404− arr <- liftIO $ newArray limit
390+ let elemSize = sizeOf (undefined :: a)
405− end <- D.foldlM' fwrite (return $ aEnd arr) $ D.take limit str
391+ elemOff = p `plusPtr` (elemSize * i)
406− return $ arr {aEnd = end}
392+ assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
408− where
394+ peek elemOff
410− fwrite ptr x = do
396+-- | Return element at the specified index without checking the bounds.
411− liftIO $ poke ptr x
397+{-# INLINE_NORMAL unsafeIndex #-}
412− return $ ptr `plusPtr` sizeOf (undefined :: a)
398+unsafeIndex :: forall a. Storable a => Array a -> Int -> a
399+unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
414−-- CAUTION: a very large number (millions) of arrays can degrade performance
401+-- | /O(1)/ Get the byte length of the array.
415−-- due to GC overhead because we need to buffer the arrays before we flatten
418−-- We could take the approach of doubling the memory allocation on each
419−-- overflow. This would result in more or less the same amount of copying as in
420−-- the chunking approach. However, if we have to shrink in the end then it may
421−-- result in an extra copy of the entire data.
423−{-# INLINE fromStreamD #-}
403+-- @since 0.7.0
424−fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
426− let s = fromStreamDArraysOf defaultChunkSize m
427− buffered <- D.foldr K.cons K.nil s
428− len <- K.foldl' (+) 0 (K.map length buffered)
429− fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
431−fromStreamD m = foldOnce write m
433− foldOnce (Fold step begin done) = D.foldlMx' step begin done
437−-- Splice two immutable arrays creating a new array.
438−{-# INLINE spliceTwo #-}
439−spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
440−spliceTwo arr1 arr2 = do
441− let src1 = unsafeForeignPtrToPtr (aStart arr1)
442− src2 = unsafeForeignPtrToPtr (aStart arr2)
443− len1 = aEnd arr1 `minusPtr` src1
444− len2 = aEnd arr2 `minusPtr` src2
446− arr <- liftIO $ newArray (len1 + len2)
447− let dst = unsafeForeignPtrToPtr (aStart arr)
449− -- XXX Should we use copyMutableByteArray# instead? Is there an overhead to
452− memcpy (castPtr dst) (castPtr src1) len1
453− touchForeignPtr (aStart arr1)
454− memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2
455− touchForeignPtr (aStart arr2)
456− return arr { aEnd = dst `plusPtr` (len1 + len2) }
458−-- Splice an array into a pre-reserved mutable array. The user must ensure
459−-- that there is enough space in the mutable array.
460−{-# INLINE spliceWith #-}
461−spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)
462−spliceWith dst@(Array _ end bound) src =
464− let srcLen = byteLength src
404+{-# INLINE byteLength #-}
465− if end `plusPtr` srcLen > bound
405+byteLength :: Array a -> Int
466− then error
406+byteLength Array{..} =
467− "Bug: spliceIntoUnsafe: Not enough space in the target array"
407+ let p = unsafeForeignPtrToPtr aStart
468− else withForeignPtr (aStart dst) $ \_ ->
408+ len = aEnd `minusPtr` p
469− withForeignPtr (aStart src) $ \psrc -> do
409+ in assert (len >= 0) len
471− memcpy (castPtr pdst) (castPtr psrc) srcLen
472− return $ dst {aEnd = pdst `plusPtr` srcLen}
474−-- Splice a new array into a preallocated mutable array, doubling the space if
411+-- | /O(1)/ Get the length of the array i.e. the number of elements in the
475−-- there is no space in the target array.
412+-- array.
476−{-# INLINE spliceWithDoubling #-}
413+--
477−spliceWithDoubling :: (MonadIO m, Storable a)
414+-- @since 0.7.0
478− => Array a -> Array a -> m (Array a)
415+{-# INLINE length #-}
479−spliceWithDoubling dst@(Array start end bound) src = do
416+length :: forall a. Storable a => Array a -> Int
480− assert (end <= bound) (return ())
417+length arr = byteLength arr `div` sizeOf (undefined :: a)
481− let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
483− dst1 <-
419+-- | Get the total capacity of an array. An array may have space reserved
484− if end `plusPtr` srcLen >= bound
420+-- beyond the current used length of the array.
423+{-# INLINE byteCapacity #-}
424+byteCapacity :: Array a -> Int
425+byteCapacity Array{..} =
486− let oldStart = unsafeForeignPtrToPtr start
426+ let p = unsafeForeignPtrToPtr aStart
487− oldSize = end `minusPtr` oldStart
427+ len = aBound `minusPtr` p
488− newSize = max (oldSize * 2) (oldSize + srcLen)
428+ in assert (len >= 0) len
489− liftIO $ realloc newSize dst
493-------------------------------------------------------------------------------
430-------------------------------------------------------------------------------
494−-- Streams of arrays
431+-- Streams of arrays - Creation
495-------------------------------------------------------------------------------
432-------------------------------------------------------------------------------
497data GroupState s start end bound
434data GroupState s start end bound
@@ -500,13 +437,20 @@ data GroupState s start end bound
500 | GroupYield start end bound (GroupState s start end bound)
437 | GroupYield start end bound (GroupState s start end bound)
501 | GroupFinish
438 | GroupFinish
503−-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
440+-- | @arraysOf n stream@ groups the input stream into a stream of
504-- arrays of size n.
441-- arrays of size n.
443+-- @arraysOf n = StreamD.foldMany (Array.writeN n)@
505−{-# INLINE_NORMAL fromStreamDArraysOf #-}
446+{-# INLINE_NORMAL arraysOf #-}
506−fromStreamDArraysOf :: forall m a. (MonadIO m, Storable a)
447+arraysOf :: forall m a. (MonadIO m, Storable a)
507 => Int -> D.Stream m a -> D.Stream m (Array a)
448 => Int -> D.Stream m a -> D.Stream m (Array a)
508−-- fromStreamDArraysOf n str = D.groupsOf n (writeN n) str
449+-- XXX the idiomatic implementation leads to large regression in the D.reverse'
450+-- benchmark. It seems it has difficulty producing optimized code when
451+-- converting to StreamK. Investigate GHC optimizations.
452+-- arraysOf n = D.foldMany (writeN n)
509−fromStreamDArraysOf n (D.Stream step state) =
453+arraysOf n (D.Stream step state) =
510 D.Stream step' (GroupStart state)
454 D.Stream step' (GroupStart state)
@@ -538,12 +482,57 @@ fromStreamDArraysOf n (D.Stream step state) =
539 step' _ GroupFinish = return D.Stop
483 step' _ GroupFinish = return D.Stop
541−-- XXX concatMap does not seem to have the best possible performance so we have
485+-- XXX buffer to a list instead?
542−-- a custom way to concat arrays.
486+-- | Buffer the stream into arrays in memory.
487+{-# INLINE bufferChunks #-}
488+bufferChunks :: (MonadIO m, Storable a) =>
489+ D.Stream m a -> m (K.Stream m (Array a))
490+bufferChunks m = D.foldr K.cons K.nil $ arraysOf defaultChunkSize m
492+-------------------------------------------------------------------------------
493+-- Streams of arrays - flatten
494+-------------------------------------------------------------------------------
496+data ReadUState a = ReadUState
497+ {-# UNPACK #-} !(ForeignPtr a) -- foreign ptr with end of array pointer
498+ {-# UNPACK #-} !(Ptr a) -- current pointer
500+-- | Unfold an array into a stream.
503+{-# INLINE_NORMAL read #-}
504+read :: forall m a. (Monad m, Storable a) => Unfold m (Array a) a
505+read = Unfold step inject
508+ inject (Array (ForeignPtr start contents) (Ptr end) _) =
509+ return $ ReadUState (ForeignPtr end contents) (Ptr start)
511+ {-# INLINE_LATE step #-}
512+ step (ReadUState fp@(ForeignPtr end _) p) | p == Ptr end =
513+ let x = unsafeInlineIO $ touchForeignPtr fp
514+ in x `seq` return D.Stop
515+ step (ReadUState fp p) = do
516+ -- unsafeInlineIO allows us to run this in Identity monad for pure
517+ -- toList/foldr case which makes them much faster due to not
518+ -- accumulating the list and fusing better with the pure consumers.
520+ -- This should be safe as the array contents are guaranteed to be
521+ -- evaluated/written to before we peek at them.
522+ let !x = unsafeInlineIO $ peek p
524+ (ReadUState fp (p `plusPtr` sizeOf (undefined :: a)))
543data FlattenState s a =
526data FlattenState s a =
544 OuterLoop s
527 OuterLoop s
545 | InnerLoop s !(ForeignPtr a) !(Ptr a) !(Ptr a)
528 | InnerLoop s !(ForeignPtr a) !(Ptr a) !(Ptr a)
530+-- | Use the "read" unfold instead.
532+-- @flattenArrays = concatUnfold read@
534+-- We can try this if there are any fusion issues in the unfold.
547{-# INLINE_NORMAL flattenArrays #-}
536{-# INLINE_NORMAL flattenArrays #-}
548flattenArrays :: forall m a. (MonadIO m, Storable a)
537flattenArrays :: forall m a. (MonadIO m, Storable a)
549 => D.Stream m (Array a) -> D.Stream m a
538 => D.Stream m (Array a) -> D.Stream m a
@@ -601,350 +590,43 @@ flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
601 return $ D.Yield x (InnerLoop st startf
590 return $ D.Yield x (InnerLoop st startf
602 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
591 (p `plusPtr` negate (sizeOf (undefined :: a))) end)
604−data SpliceState s arr
593+-------------------------------------------------------------------------------
605− = SpliceInitial s
594+-- to Lists and streams
606− | SpliceBuffering s arr
595+-------------------------------------------------------------------------------
607− | SpliceYielding arr (SpliceState s arr)
610−-- XXX can use general grouping combinators to achieve this?
611−-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
612−-- maximum specified size. Note that if a single array is bigger than the
613−-- specified size we do not split it to fit. When we coalesce multiple arrays
614−-- if the size would exceed the specified size we do not coalesce therefore the
615−-- actual array size may be less than the specified chunk size.
618−{-# INLINE_NORMAL packArraysChunksOf #-}
619−packArraysChunksOf :: (MonadIO m, Storable a)
620− => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
621−packArraysChunksOf n (D.Stream step state) =
622− D.Stream step' (SpliceInitial state)
597+-- Use foldr/build fusion to fuse with list consumers
598+-- This can be useful when using the IsList instance
599+{-# INLINE_LATE toListFB #-}
600+toListFB :: forall a b. Storable a => (a -> b -> b) -> b -> Array a -> b
601+toListFB c n Array{..} = go (unsafeForeignPtrToPtr aStart)
626− {-# INLINE_LATE step' #-}
604+ go p | p == aEnd = n
627− step' gst (SpliceInitial st) = do
605+ go p =
628− when (n <= 0) $
606+ -- unsafeInlineIO allows us to run this in Identity monad for pure
629− -- XXX we can pass the module string from the higher level API
607+ -- toList/foldr case which makes them much faster due to not
630− error $ "Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types.packArraysChunksOf: the size of "
608+ -- accumulating the list and fusing better with the pure consumers.
631− ++ "arrays [" ++ show n ++ "] must be a natural number"
609+ --
632− r <- step gst st
610+ -- This should be safe as the array contents are guaranteed to be
633− case r of
611+ -- evaluated/written to before we peek at them.
634− D.Yield arr s -> return $
612+ let !x = unsafeInlineIO $ do
635− let len = byteLength arr
613+ r <- peek p
636− in if len >= n
614+ touchForeignPtr aStart
637− then D.Skip (SpliceYielding arr (SpliceInitial s))
615+ return r
638− else D.Skip (SpliceBuffering s arr)
616+ in c x (go (p `plusPtr` sizeOf (undefined :: a)))
639− D.Skip s -> return $ D.Skip (SpliceInitial s)
640− D.Stop -> return D.Stop
642− step' gst (SpliceBuffering st buf) = do
646− let len = byteLength buf + byteLength arr
649− D.Skip (SpliceYielding buf (SpliceBuffering s arr))
651− buf' <- if byteCapacity buf < n
652− then liftIO $ realloc n buf
654− buf'' <- spliceWith buf' arr
655− return $ D.Skip (SpliceBuffering s buf'')
656− D.Skip s -> return $ D.Skip (SpliceBuffering s buf)
657− D.Stop -> return $ D.Skip (SpliceYielding buf SpliceFinish)
659− step' _ SpliceFinish = return D.Stop
661− step' _ (SpliceYielding arr next) = return $ D.Yield arr next
663−-- XXX instead of writing two different versions of this operation, we should
664−-- write it as a pipe.
665−{-# INLINE_NORMAL lpackArraysChunksOf #-}
666−lpackArraysChunksOf :: (MonadIO m, Storable a)
667− => Int -> Fold m (Array a) () -> Fold m (Array a) ()
668−lpackArraysChunksOf n (Fold step1 initial1 extract1) =
669− Fold step initial extract
671− where
618+-- | Convert an 'Array' into a list.
621+{-# INLINE toList #-}
622+toList :: Storable a => Array a -> [a]
623+toList s = build (\c n -> toListFB c n s)
673− initial = do
625+-- | Use the 'read' unfold instead.
675− -- XXX we can pass the module string from the higher level API
676− error $ "Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types.packArraysChunksOf: the size of "
677− ++ "arrays [" ++ show n ++ "] must be a natural number"
680− return $ first (Tuple' Nothing) r
682− extract (Tuple' Nothing r1) = extract1 r1
683− extract (Tuple' (Just buf) r1) = do
686− FL.Partial rr -> extract1 rr
687− FL.Done _ -> return ()
689− step (Tuple' Nothing r1) arr =
690− let len = byteLength arr
695− FL.Done _ -> return $ FL.Done ()
699− return $ first (Tuple' Nothing) res
700− else return $ FL.Partial $ Tuple' (Just arr) r1
702− step (Tuple' (Just buf) r1) arr = do
703− let len = byteLength buf + byteLength arr
704− buf' <- if byteCapacity buf < len
705− then liftIO $ realloc (max n len) buf
707− buf'' <- spliceWith buf' arr
709− -- XXX this is common in both the equations of step
714− FL.Done _ -> return $ FL.Done ()
718− return $ first (Tuple' Nothing) res
719− else return $ FL.Partial $ Tuple' (Just buf'') r1
721−#if !defined(mingw32_HOST_OS)
722−data GatherState s arr
724− | GatherBuffering s arr Int
725− | GatherYielding arr (GatherState s arr)
728−-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
729−-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
730−-- each array and a maximum of @maxEntries@ entries in each array.
733−{-# INLINE_NORMAL groupIOVecsOf #-}
734−groupIOVecsOf :: MonadIO m
735− => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
736−groupIOVecsOf n maxIOVLen (D.Stream step state) =
737− D.Stream step' (GatherInitial state)
741− {-# INLINE_LATE step' #-}
742− step' gst (GatherInitial st) = do
744− -- XXX we can pass the module string from the higher level API
745− error $ "Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types.groupIOVecsOf: the size of "
746− ++ "groups [" ++ show n ++ "] must be a natural number"
747− when (maxIOVLen <= 0) $
748− -- XXX we can pass the module string from the higher level API
749− error $ "Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types.groupIOVecsOf: the number of "
750− ++ "IOVec entries [" ++ show n ++ "] must be a natural number"
751− r <- step (adaptState gst) st
754− let p = unsafeForeignPtrToPtr (aStart arr)
755− len = byteLength arr
756− iov <- liftIO $ newArray maxIOVLen
757− iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
760− then return $ D.Skip (GatherYielding iov' (GatherInitial s))
761− else return $ D.Skip (GatherBuffering s iov' len)
762− D.Skip s -> return $ D.Skip (GatherInitial s)
763− D.Stop -> return D.Stop
765− step' gst (GatherBuffering st iov len) = do
766− r <- step (adaptState gst) st
769− let p = unsafeForeignPtrToPtr (aStart arr)
770− alen = byteLength arr
772− if len' > n || length iov >= maxIOVLen
774− iov' <- liftIO $ newArray maxIOVLen
775− iov'' <- liftIO $ unsafeSnoc iov' (IOVec (castPtr p)
776− (fromIntegral alen))
777− return $ D.Skip (GatherYielding iov
778− (GatherBuffering s iov'' alen))
780− iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
781− (fromIntegral alen))
782− return $ D.Skip (GatherBuffering s iov' len')
783− D.Skip s -> return $ D.Skip (GatherBuffering s iov len)
784− D.Stop -> return $ D.Skip (GatherYielding iov GatherFinish)
786− step' _ GatherFinish = return D.Stop
788− step' _ (GatherYielding iov next) = return $ D.Yield iov next
792−data SplitState s arr
796− | Yielding arr (SplitState s arr)
799−-- | Split a stream of arrays on a given separator byte, dropping the separator
800−-- and coalescing all the arrays between two separators into a single array.
803−{-# INLINE_NORMAL splitOn #-}
807− -> D.Stream m (Array Word8)
808− -> D.Stream m (Array Word8)
809−splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
813− {-# INLINE_LATE step' #-}
814− step' gst (Initial st) = do
818− (arr1, marr2) <- breakOn byte arr
819− return $ case marr2 of
820− Nothing -> D.Skip (Buffering s arr1)
821− Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
822− D.Skip s -> return $ D.Skip (Initial s)
823− D.Stop -> return D.Stop
825− step' gst (Buffering st buf) = do
829− (arr1, marr2) <- breakOn byte arr
830− buf' <- spliceTwo buf arr1
831− return $ case marr2 of
832− Nothing -> D.Skip (Buffering s buf')
833− Just x -> D.Skip (Yielding buf' (Splitting s x))
834− D.Skip s -> return $ D.Skip (Buffering s buf)
836− if byteLength buf == 0
838− else D.Skip (Yielding buf Finishing)
840− step' _ (Splitting st buf) = do
841− (arr1, marr2) <- breakOn byte buf
842− return $ case marr2 of
843− Nothing -> D.Skip $ Buffering st arr1
844− Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
846− step' _ (Yielding arr next) = return $ D.Yield arr next
847− step' _ Finishing = return D.Stop
850−-- Drops the separator byte
851−{-# INLINE breakOn #-}
853− => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
854−breakOn sep arr@Array{..} = liftIO $ do
855− let p = unsafeForeignPtrToPtr aStart
856− loc <- c_memchr p sep (fromIntegral $ aEnd `minusPtr` p)
867− { aStart = aStart `plusForeignPtr` (loc `minusPtr` p + 1)
873−-------------------------------------------------------------------------------
875−-------------------------------------------------------------------------------
877−-- | Return element at the specified index without checking the bounds.
879−-- Unsafe because it does not check the bounds of the array.
627+-- @toStreamD = D.unfold read@
880−{-# INLINE_NORMAL unsafeIndexIO #-}
881−unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
882−unsafeIndexIO Array {..} i =
883− withForeignPtr aStart $ \p -> do
884− let elemSize = sizeOf (undefined :: a)
885− elemOff = p `plusPtr` (elemSize * i)
886− assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
890−-- | Return element at the specified index without checking the bounds.
891−{-# INLINE_NORMAL unsafeIndex #-}
892−unsafeIndex :: forall a. Storable a => Array a -> Int -> a
893−unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
895−-- | /O(1)/ Get the byte length of the array.
898−{-# INLINE byteLength #-}
899−byteLength :: Array a -> Int
900−byteLength Array{..} =
901− let p = unsafeForeignPtrToPtr aStart
902− len = aEnd `minusPtr` p
903− in assert (len >= 0) len
905−-- | /O(1)/ Get the length of the array i.e. the number of elements in the
908−-- @since 0.7.0
629+-- We can try this if the unfold has any performance issues.
909−{-# INLINE length #-}
910−length :: forall a. Storable a => Array a -> Int
911−length arr = byteLength arr `div` sizeOf (undefined :: a)
913−{-# INLINE byteCapacity #-}
914−byteCapacity :: Array a -> Int
915−byteCapacity Array{..} =
916− let p = unsafeForeignPtrToPtr aStart
917− len = aBound `minusPtr` p
918− in assert (len >= 0) len
920−-- Use foldr/build fusion to fuse with list consumers
921−-- This can be useful when using the IsList instance
922−{-# INLINE_LATE toListFB #-}
923−toListFB :: forall a b. Storable a => (a -> b -> b) -> b -> Array a -> b
924−toListFB c n Array{..} = go (unsafeForeignPtrToPtr aStart)
927− go p | p == aEnd = n
929− -- unsafeInlineIO allows us to run this in Identity monad for pure
930− -- toList/foldr case which makes them much faster due to not
931− -- accumulating the list and fusing better with the pure consumers.
933− -- This should be safe as the array contents are guaranteed to be
934− -- evaluated/written to before we peek at them.
935− let !x = unsafeInlineIO $ do
937− touchForeignPtr aStart
939− in c x (go (p `plusPtr` sizeOf (undefined :: a)))
941−-- | Convert an 'Array' into a list.
944−{-# INLINE toList #-}
945−toList :: Storable a => Array a -> [a]
946−toList s = build (\c n -> toListFB c n s)
948{-# INLINE_NORMAL toStreamD #-}
630{-# INLINE_NORMAL toStreamD #-}
949toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
631toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
950toStreamD Array{..} =
632toStreamD Array{..} =
@@ -1019,70 +701,22 @@ toStreamKRev Array {..} =
1019 return r
701 return r
1020 in x `K.cons` go (p `plusPtr` negate (sizeOf (undefined :: a)))
702 in x `K.cons` go (p `plusPtr` negate (sizeOf (undefined :: a)))
704+-------------------------------------------------------------------------------
706+-------------------------------------------------------------------------------
708+-- | Strict left fold of an array.
1022{-# INLINE_NORMAL foldl' #-}
709{-# INLINE_NORMAL foldl' #-}
1023foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
710foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
1024foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
711foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
713+-- | Right fold of an array.
1026{-# INLINE_NORMAL foldr #-}
714{-# INLINE_NORMAL foldr #-}
1027foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
715foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
1028foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
716foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
1031−-- | Create two slices of an array without copying the original array. The
1032−-- specified index @i@ is the first index of the second slice.
1035−splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
1036−splitAt i arr@Array{..} =
1037− let maxIndex = length arr - 1
1039− then error "sliceAt: negative array index"
1040− else if i > maxIndex
1041− then error $ "sliceAt: specified array index " ++ show i
1042− ++ " is beyond the maximum index " ++ show maxIndex
1043− else let off = i * sizeOf (undefined :: a)
1044− p = unsafeForeignPtrToPtr aStart `plusPtr` off
1051− { aStart = aStart `plusForeignPtr` off
1058−{-# INLINE_NORMAL unlines #-}
1059−unlines :: forall m a. (MonadIO m, Storable a)
1060− => a -> D.Stream m (Array a) -> D.Stream m a
1061−unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
1063− {-# INLINE_LATE step' #-}
1064− step' gst (OuterLoop st) = do
1065− r <- step (adaptState gst) st
1067− D.Yield Array{..} s ->
1068− let p = unsafeForeignPtrToPtr aStart
1069− in D.Skip (InnerLoop s aStart p aEnd)
1070− D.Skip s -> D.Skip (OuterLoop s)
1073− step' _ (InnerLoop st _ p end) | p == end =
1074− return $ D.Yield sep $ OuterLoop st
1076− step' _ (InnerLoop st startf p end) = do
1079− touchForeignPtr startf
1081− return $ D.Yield x (InnerLoop st startf
1082− (p `plusPtr` sizeOf (undefined :: a)) end)
1084-------------------------------------------------------------------------------
718-------------------------------------------------------------------------------
1085−-- Folds
719+-- Write Folds to fold streams into arrays
1086-------------------------------------------------------------------------------
720-------------------------------------------------------------------------------
1088{-# INLINE_NORMAL writeNAllocWith #-}
722{-# INLINE_NORMAL writeNAllocWith #-}
@@ -1104,6 +738,8 @@ writeNAllocWith alloc n = Fold step initial extract
1104-- | @writeN n@ folds a maximum of @n@ elements from the input stream to an
738-- | @writeN n@ folds a maximum of @n@ elements from the input stream to an
1105-- 'Array'.
739-- 'Array'.
741+-- @writeN n = Fold.takeLE n writeNUnsafe@
1107-- @since 0.7.0
743-- @since 0.7.0
1108{-# INLINE_NORMAL writeN #-}
744{-# INLINE_NORMAL writeN #-}
1109writeN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
745writeN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
@@ -1162,6 +798,26 @@ writeNUnsafe n = Fold step initial extract
1163 extract (ArrayUnsafe start end) = return $ Array start end end -- liftIO . shrinkToFit
799 extract (ArrayUnsafe start end) = return $ Array start end end -- liftIO . shrinkToFit
801+-- XXX Buffer to a list instead?
803+-- | Buffer a stream into a stream of arrays.
805+-- @writeChunks = Fold.many Fold.toStream (Array.writeN n)@
807+-- See 'bufferChunks'.
811+{-# INLINE_NORMAL writeChunks #-}
812+writeChunks :: -- (MonadIO m, Storable a) =>
813+ Int -> Fold m a (D.Stream m (Array a))
814+writeChunks = undefined -- Fold.many Fold.toStream (Array.writeN n)
816+-- XXX Compare toArrayMinChunk with fromStreamD which uses an array of streams
817+-- implementation. We can write this using writeChunks above if that is faster.
818+-- If toArrayMinChunk is faster then we should use that to implement
1165-- XXX The realloc based implementation needs to make one extra copy if we use
821-- XXX The realloc based implementation needs to make one extra copy if we use
1166-- shrinkToFit. On the other hand, the stream of arrays implementation may
822-- shrinkToFit. On the other hand, the stream of arrays implementation may
1167-- buffer the array chunk pointers in memory but it does not have to shrink as
823-- buffer the array chunk pointers in memory but it does not have to shrink as
@@ -1224,6 +880,184 @@ writeAligned alignSize =
1224 (bytesToElemCount (undefined :: a)
880 (bytesToElemCount (undefined :: a)
1225 (mkChunkSize 1024))
881 (mkChunkSize 1024))
883+-------------------------------------------------------------------------------
884+-- construct from streams, known size
885+-------------------------------------------------------------------------------
887+-- | Use the 'writeN' fold instead.
889+-- @fromStreamDN n = D.fold (writeN n)@
891+{-# INLINE_NORMAL fromStreamDN #-}
892+fromStreamDN :: forall m a. (MonadIO m, Storable a)
893+ => Int -> D.Stream m a -> m (Array a)
894+fromStreamDN limit str = do
895+ arr <- liftIO $ newArray limit
896+ end <- D.foldlM' fwrite (return $ aEnd arr) $ D.take limit str
897+ return $ arr {aEnd = end}
903+ return $ ptr `plusPtr` sizeOf (undefined :: a)
905+-- | Create an 'Array' from the first N elements of a list. The array is
906+-- allocated to size N, if the list terminates before N elements then the
907+-- array may hold less than N elements.
910+{-# INLINABLE fromListN #-}
911+fromListN :: Storable a => Int -> [a] -> Array a
912+fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
914+-------------------------------------------------------------------------------
915+-- convert stream to a single array
916+-------------------------------------------------------------------------------
918+-- CAUTION: a very large number (millions) of arrays can degrade performance
919+-- due to GC overhead because we need to buffer the arrays before we flatten
922+-- XXX Compare if this is faster or "fold write".
924+-- | We could take the approach of doubling the memory allocation on each
925+-- overflow. This would result in more or less the same amount of copying as in
926+-- the chunking approach. However, if we have to shrink in the end then it may
927+-- result in an extra copy of the entire data.
930+-- fromStreamD = StreamD.fold Array.write
933+{-# INLINE fromStreamD #-}
934+fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
936+ buffered <- bufferChunks m
937+ len <- K.foldl' (+) 0 (K.map length buffered)
938+ fromStreamDN len $ D.concatUnfold read $ D.fromStreamK buffered
940+-- | Create an 'Array' from a list. The list must be of finite size.
943+{-# INLINABLE fromList #-}
944+fromList :: Storable a => [a] -> Array a
945+fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
947+-------------------------------------------------------------------------------
949+-------------------------------------------------------------------------------
951+-- | Copy two arrays into a newly allocated array.
952+{-# INLINE spliceTwo #-}
953+spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
954+spliceTwo arr1 arr2 = do
955+ let src1 = unsafeForeignPtrToPtr (aStart arr1)
956+ src2 = unsafeForeignPtrToPtr (aStart arr2)
957+ len1 = aEnd arr1 `minusPtr` src1
958+ len2 = aEnd arr2 `minusPtr` src2
960+ arr <- liftIO $ newArray (len1 + len2)
961+ let dst = unsafeForeignPtrToPtr (aStart arr)
963+ -- XXX Should we use copyMutableByteArray# instead? Is there an overhead to
966+ memcpy (castPtr dst) (castPtr src1) len1
967+ touchForeignPtr (aStart arr1)
968+ memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2
969+ touchForeignPtr (aStart arr2)
970+ return arr { aEnd = dst `plusPtr` (len1 + len2) }
972+-- | Splice an array into a pre-reserved mutable array. The user must ensure
973+-- that there is enough space in the mutable array, otherwise the splicing
975+{-# INLINE spliceWith #-}
976+spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)
977+spliceWith dst@(Array _ end bound) src =
979+ let srcLen = byteLength src
980+ if end `plusPtr` srcLen > bound
982+ "Bug: spliceWith: Not enough space in the target array"
983+ else withForeignPtr (aStart dst) $ \_ ->
984+ withForeignPtr (aStart src) $ \psrc -> do
986+ memcpy (castPtr pdst) (castPtr psrc) srcLen
987+ return $ dst {aEnd = pdst `plusPtr` srcLen}
989+-- | Splice a new array into a preallocated mutable array, doubling the space
990+-- if there is no space in the target array.
991+{-# INLINE spliceWithDoubling #-}
992+spliceWithDoubling :: (MonadIO m, Storable a)
993+ => Array a -> Array a -> m (Array a)
994+spliceWithDoubling dst@(Array start end bound) src = do
995+ assert (end <= bound) (return ())
996+ let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
999+ if end `plusPtr` srcLen >= bound
1001+ let oldStart = unsafeForeignPtrToPtr start
1002+ oldSize = end `minusPtr` oldStart
1003+ newSize = max (oldSize * 2) (oldSize + srcLen)
1004+ liftIO $ realloc newSize dst
1006+ spliceWith dst1 src
1008+-------------------------------------------------------------------------------
1010+-------------------------------------------------------------------------------
1012+-- | Drops the separator byte
1013+{-# INLINE breakOn #-}
1014+breakOn :: MonadIO m
1015+ => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
1016+breakOn sep arr@Array{..} = liftIO $ do
1017+ let p = unsafeForeignPtrToPtr aStart
1018+ loc <- c_memchr p sep (fromIntegral $ aEnd `minusPtr` p)
1021+ then (arr, Nothing)
1029+ { aStart = aStart `plusForeignPtr` (loc `minusPtr` p + 1)
1035+-- | Create two slices of an array without copying the original array. The
1036+-- specified index @i@ is the first index of the second slice.
1039+splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
1040+splitAt i arr@Array{..} =
1041+ let maxIndex = length arr - 1
1043+ then error "sliceAt: negative array index"
1044+ else if i > maxIndex
1045+ then error $ "sliceAt: specified array index " ++ show i
1046+ ++ " is beyond the maximum index " ++ show maxIndex
1047+ else let off = i * sizeOf (undefined :: a)
1048+ p = unsafeForeignPtrToPtr aStart `plusPtr` off
1055+ { aStart = aStart `plusForeignPtr` off
1227-------------------------------------------------------------------------------
1061-------------------------------------------------------------------------------
1228-- Instances
1062-- Instances
1229-------------------------------------------------------------------------------
1063-------------------------------------------------------------------------------
@@ -1346,6 +1180,9 @@ instance Foldable Array where
1346-- Semigroup and Monoid
1180-- Semigroup and Monoid
1347-------------------------------------------------------------------------------
1181-------------------------------------------------------------------------------
1183+-- Note: we cannot splice the second array into the free space of the first
1184+-- array because that would require a monadic API due to mutation.
1185+-- | Copies the two arrays into a newly allocated array.
1349instance Storable a => Semigroup (Array a) where
1186instance Storable a => Semigroup (Array a) where
1350 {-# INLINE (<>) #-}
1187 {-# INLINE (<>) #-}
1351 arr1 <> arr2 = unsafePerformIO $ spliceTwo arr1 arr2
1188 arr1 <> arr2 = unsafePerformIO $ spliceTwo arr1 arr2