@@ -1,12 +1,10 @@
1−{-# LANGUAGE UnboxedTuples #-}
2−
31#include "inline.hs"
42
53-- |
64-- Module : Streamly.Internal.Memory.Array.Types
7−-- Copyright : (c) 2019 Composewell Technologies
5+-- Copyright : (c) 2020 Composewell Technologies
86--
9−-- License : BSD3
7+-- License : BSD3-3-Clause
108-- Maintainer : streamly@composewell.com
119-- Stability : experimental
1210-- Portability : GHC
@@ -15,28 +13,25 @@ module Streamly.Internal.Memory.Array.Types
1513 (
1614 Array (..)
1715
16+ -- * Freezing and Thawing
17+ , unsafeFreeze
18+ , unsafeFreezeWithShrink
19+ , unsafeThaw
20+
1821 -- * Construction
19− , withNewArray
20− , newArray
21− , unsafeSnoc
22− , snoc
23− , spliceWithDoubling
2422 , spliceTwo
2523
26− , fromAddr#
27− , fromString#
2824 , fromList
2925 , fromListN
3026 , fromStreamDN
31− -- , fromStreamD
3227
3328 -- * Streams of arrays
3429 , fromStreamDArraysOf
35− , FlattenState (..) -- for inspection testing
30+ , MA.FlattenState (..) -- for inspection testing
3631 , flattenArrays
3732 , flattenArraysRev
3833 , packArraysChunksOf
39− , SpliceState (..)
34+ , MA.SpliceState (..)
4035 , lpackArraysChunksOf
4136#if !defined(mingw32_HOST_OS)
4237 , groupIOVecsOf
@@ -47,9 +42,9 @@ module Streamly.Internal.Memory.Array.Types
4742 -- * Elimination
4843 , unsafeIndexIO
4944 , unsafeIndex
50− , length
5145 , byteLength
52− , byteCapacity
46+ , length
47+
5348 , foldl'
5449 , foldr
5550 , splitAt
@@ -59,78 +54,60 @@ module Streamly.Internal.Memory.Array.Types
5954 , toStreamK
6055 , toStreamKRev
6156 , toList
57+
58+ , unlines
59+
60+ -- * Folds
6261 , toArrayMinChunk
6362 , writeN
6463 , writeNUnsafe
65− , ArrayUnsafe (..)
64+ , MA.ArrayUnsafe (..)
6665 , writeNAligned
6766 , writeNAlignedUnmanaged
6867 , write
6968 , writeAligned
7069
7170 -- * Utilities
72− , defaultChunkSize
73− , mkChunkSize
74− , mkChunkSizeKB
75− , unsafeInlineIO
76− , realloc
77− , shrinkToFit
78− , memcpy
79− , memcmp
80− , bytesToElemCount
81−
82− , unlines
71+ , MA.defaultChunkSize
72+ , MA.mkChunkSize
73+ , MA.mkChunkSizeKB
74+ , MA.unsafeInlineIO
75+
76+ , MA.memcpy
77+ , MA.memcmp
78+ , MA.bytesToElemCount
8379 )
8480where
8581
86−import Control.Exception (assert)
8782import Control.DeepSeq (NFData(..))
88−import Control.Monad (when, void)
8983import Control.Monad.IO.Class (MonadIO(..))
90−import Data.Functor.Identity (runIdentity)
91−#if __GLASGOW_HASKELL__ < 808
92−import Data.Semigroup (Semigroup(..))
93−#endif
9484import Data.Word (Word8)
95−import Foreign.C.String (CString)
96−import Foreign.C.Types (CSize(..), CInt(..))
97−import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
98−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
99−import Foreign.Ptr (plusPtr, minusPtr, castPtr, nullPtr)
10085import Foreign.Storable (Storable(..))
101−import Prelude hiding (length, foldr, read, unlines, splitAt)
102−import Text.Read (readPrec, readListPrec, readListPrecDefault)
103−
104−import GHC.Base (Addr#, nullAddr#, realWorld#, build)
86+import GHC.Base (nullAddr#)
10587import GHC.Exts (IsList, IsString(..))
106−import GHC.ForeignPtr (ForeignPtr(..), newForeignPtr_)
107−import GHC.IO (IO(IO), unsafePerformIO)
88+import GHC.ForeignPtr (ForeignPtr(..))
89+import GHC.IO (unsafePerformIO)
10890import GHC.Ptr (Ptr(..))
91+import Streamly.Internal.Data.Fold.Types (Fold(..), lmap)
92+import Text.Read (readPrec, readListPrec, readListPrecDefault)
10993
110−import Streamly.Internal.Data.Fold.Types (Fold(..))
111−import Streamly.Internal.Data.Strict (Tuple'(..))
112−import Streamly.Internal.Data.SVar (adaptState)
113−
114−#if !defined(mingw32_HOST_OS)
115−import Streamly.FileSystem.FDIO (IOVec(..))
116−#endif
94+import Prelude hiding (length, foldr, read, unlines, splitAt)
11795
118−import qualified Streamly.Memory.Malloc as Malloc
11996import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
12097import qualified Streamly.Internal.Data.Stream.StreamK as K
98+import qualified Streamly.Internal.Memory.Mutable.Array.Types as MA
12199import qualified GHC.Exts as Exts
122100
123−#ifdef DEVBUILD
124−import qualified Data.Foldable as F
101+#if __GLASGOW_HASKELL__ < 808
102+import Data.Semigroup (Semigroup(..))
125103#endif
126104
127−#if MIN_VERSION_base(4,10,0)
128−import Foreign.ForeignPtr (plusForeignPtr)
129−#else
130−import GHC.Base (Int(..), plusAddr#)
131−import GHC.ForeignPtr (ForeignPtr(..))
132−plusForeignPtr :: ForeignPtr a -> Int -> ForeignPtr b
133−plusForeignPtr (ForeignPtr addr c) (I# d) = ForeignPtr (plusAddr# addr d) c
105+#if !defined(mingw32_HOST_OS)
106+import Streamly.FileSystem.FDIO (IOVec(..))
107+#endif
108+
109+#ifdef DEVBUILD
110+import qualified Data.Foldable as F
134111#endif
135112
136113-------------------------------------------------------------------------------
@@ -186,253 +163,140 @@ data Array a =
186163#endif
187164 Array
188165 { aStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
189− , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused address
190− , aBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
166+ , aEnd :: {-# UNPACK #-} !(Ptr a) -- first unused addres
191167 }
192168
193169-------------------------------------------------------------------------------
194−-- Utility functions
170+-- Freezing and Thawing
195171-------------------------------------------------------------------------------
196172
197−foreign import ccall unsafe "string.h memcpy" c_memcpy
198− :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)
199−
200−foreign import ccall unsafe "string.h strlen" c_strlen
201− :: CString -> IO CSize
202−
203−foreign import ccall unsafe "string.h memchr" c_memchr
204− :: Ptr Word8 -> Word8 -> CSize -> IO (Ptr Word8)
173+-- | Returns an immutable array using the same underlying pointers of the
174+-- mutable array. If the underlying array is mutated, the immutable promise is
175+-- lost. Please make sure that the mutable array is never used after freezing it
176+-- using /unsafeFreeze/.
177+{-# INLINE unsafeFreeze #-}
178+unsafeFreeze :: MA.Array a -> Array a
179+unsafeFreeze (MA.Array as ae _) = Array as ae
180+
181+-- | Similar to 'unsafeFreeze' but uses 'MA.shrinkToFit' on the mutable array
182+-- first.
183+{-# INLINE unsafeFreezeWithShrink #-}
184+unsafeFreezeWithShrink :: Storable a => MA.Array a -> Array a
185+unsafeFreezeWithShrink arr = unsafePerformIO $ do
186+ MA.Array as ae _ <- MA.shrinkToFit arr
187+ return $ Array as ae
188+
189+-- | Returns a mutable array using the same underlying pointers of the immutable
190+-- array. If the resulting array is mutated, the older immutable array is
191+-- mutated as well. Please make sure that the immutable array is never used
192+-- after thawing it using /unsafeThaw/.
193+{-# INLINE unsafeThaw #-}
194+unsafeThaw :: Array a -> MA.Array a
195+unsafeThaw (Array as ae) = MA.Array as ae ae
205196
206−-- XXX we are converting Int to CSize
207−memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()
208−memcpy dst src len = void (c_memcpy dst src (fromIntegral len))
197+-------------------------------------------------------------------------------
198+-- Construction
199+-------------------------------------------------------------------------------
209200
210−foreign import ccall unsafe "string.h memcmp" c_memcmp
211− :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt
201+-- Splice two immutable arrays creating a new array.
202+{-# INLINE spliceTwo #-}
203+spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
204+spliceTwo arr1 arr2 =
205+ unsafeFreeze <$> MA.spliceTwo (unsafeThaw arr1) (unsafeThaw arr2)
212206
213−-- XXX we are converting Int to CSize
214−-- return True if the memory locations have identical contents
215−{-# INLINE memcmp #-}
216−memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool
217−memcmp p1 p2 len = do
218− r <- c_memcmp p1 p2 (fromIntegral len)
219− return $ r == 0
207+-- | Create an 'Array' from the first N elements of a list. The array is
208+-- allocated to size N, if the list terminates before N elements then the
209+-- array may hold less than N elements.
210+--
211+-- @since 0.7.0
212+{-# INLINABLE fromListN #-}
213+fromListN :: Storable a => Int -> [a] -> Array a
214+fromListN n xs = unsafeFreeze $ MA.fromListN n xs
220215
221−{-# INLINE unsafeInlineIO #-}
222−unsafeInlineIO :: IO a -> a
223−unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r
216+-- | Create an 'Array' from a list. The list must be of finite size.
217+--
218+-- @since 0.7.0
219+{-# INLINABLE fromList #-}
220+fromList :: Storable a => [a] -> Array a
221+fromList xs = unsafeFreeze $ MA.fromList xs
224222
225−{-# INLINE bytesToElemCount #-}
226−bytesToElemCount :: Storable a => a -> Int -> Int
227−bytesToElemCount x n =
228− let elemSize = sizeOf x
229− in n + elemSize - 1 `div` elemSize
223+{-# INLINE_NORMAL fromStreamDN #-}
224+fromStreamDN :: forall m a. (MonadIO m, Storable a)
225+ => Int -> D.Stream m a -> m (Array a)
226+fromStreamDN limit str = unsafeFreeze <$> MA.fromStreamDN limit str
230227
231228-------------------------------------------------------------------------------
232−-- Construction
229+-- Streams of arrays
233230-------------------------------------------------------------------------------
234231
235−-- | allocate a new array using the provided allocator function.
236−{-# INLINE newArrayAlignedAllocWith #-}
237−newArrayAlignedAllocWith :: forall a. Storable a
238− => (Int -> Int -> IO (ForeignPtr a)) -> Int -> Int -> IO (Array a)
239−newArrayAlignedAllocWith alloc alignSize count = do
240− let size = count * sizeOf (undefined :: a)
241− fptr <- alloc size alignSize
242− let p = unsafeForeignPtrToPtr fptr
243− return $ Array
244− { aStart = fptr
245− , aEnd = p
246− , aBound = p `plusPtr` size
247− }
248−
249−-- | Allocate a new array aligned to the specified alignmend and using
250−-- unmanaged pinned memory. The memory will not be automatically freed by GHC.
251−-- This could be useful in allocate once global data structures. Use carefully
252−-- as incorrect use can lead to memory leak.
253−{-# INLINE newArrayAlignedUnmanaged #-}
254−newArrayAlignedUnmanaged :: forall a. Storable a => Int -> Int -> IO (Array a)
255−newArrayAlignedUnmanaged =
256− newArrayAlignedAllocWith Malloc.mallocForeignPtrAlignedUnmanagedBytes
257−
258−{-# INLINE newArrayAligned #-}
259−newArrayAligned :: forall a. Storable a => Int -> Int -> IO (Array a)
260−newArrayAligned = newArrayAlignedAllocWith Malloc.mallocForeignPtrAlignedBytes
261−
262−-- XXX can unaligned allocation be more efficient when alignment is not needed?
263−--
264−-- | Allocate an array that can hold 'count' items. The memory of the array is
265−-- uninitialized.
266−--
267−-- Note that this is internal routine, the reference to this array cannot be
268−-- given out until the array has been written to and frozen.
269−{-# INLINE newArray #-}
270−newArray :: forall a. Storable a => Int -> IO (Array a)
271−newArray = newArrayAligned (alignment (undefined :: a))
272−
273−-- | Allocate an Array of the given size and run an IO action passing the array
274−-- start pointer.
275−{-# INLINE withNewArray #-}
276−withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)
277−withNewArray count f = do
278− arr <- newArray count
279− withForeignPtr (aStart arr) $ \p -> f p >> return arr
280−
281−-- XXX grow the array when we are beyond bound.
282−--
283−-- Internal routine for when the array is being created. Appends one item at
284−-- the end of the array. Useful when sequentially writing a stream to the
285−-- array.
286−{-# INLINE unsafeSnoc #-}
287−unsafeSnoc :: forall a. Storable a => Array a -> a -> IO (Array a)
288−unsafeSnoc arr@Array{..} x = do
289− when (aEnd == aBound) $
290− error "BUG: unsafeSnoc: writing beyond array bounds"
291− poke aEnd x
292− touchForeignPtr aStart
293− return $ arr {aEnd = aEnd `plusPtr` sizeOf (undefined :: a)}
294−
295−{-# INLINE snoc #-}
296−snoc :: forall a. Storable a => Array a -> a -> IO (Array a)
297−snoc arr@Array {..} x =
298− if aEnd == aBound
299− then do
300− let oldStart = unsafeForeignPtrToPtr aStart
301− size = aEnd `minusPtr` oldStart
302− newSize = size + sizeOf (undefined :: a)
303− newPtr <- Malloc.mallocForeignPtrAlignedBytes
304− newSize (alignment (undefined :: a))
305− withForeignPtr newPtr $ \pNew -> do
306− memcpy (castPtr pNew) (castPtr oldStart) size
307− poke (pNew `plusPtr` size) x
308− touchForeignPtr aStart
309− return $ Array
310− { aStart = newPtr
311− , aEnd = pNew `plusPtr` (size + sizeOf (undefined :: a))
312− , aBound = pNew `plusPtr` newSize
313− }
314− else do
315− poke aEnd x
316− touchForeignPtr aStart
317− return $ arr {aEnd = aEnd `plusPtr` sizeOf (undefined :: a)}
318−
319−-- | Reallocate the array to the specified size in bytes. If the size is less
320−-- than the original array the array gets truncated.
321−{-# NOINLINE reallocAligned #-}
322−reallocAligned :: Int -> Int -> Array a -> IO (Array a)
323−reallocAligned alignSize newSize Array{..} = do
324− assert (aEnd <= aBound) (return ())
325− let oldStart = unsafeForeignPtrToPtr aStart
326− let size = aEnd `minusPtr` oldStart
327− newPtr <- Malloc.mallocForeignPtrAlignedBytes newSize alignSize
328− withForeignPtr newPtr $ \pNew -> do
329− memcpy (castPtr pNew) (castPtr oldStart) size
330− touchForeignPtr aStart
331− return $ Array
332− { aStart = newPtr
333− , aEnd = pNew `plusPtr` size
334− , aBound = pNew `plusPtr` newSize
335− }
336−
337−-- XXX can unaligned allocation be more efficient when alignment is not needed?
338−{-# INLINABLE realloc #-}
339−realloc :: forall a. Storable a => Int -> Array a -> IO (Array a)
340−realloc = reallocAligned (alignment (undefined :: a))
341−
342−-- | Remove the free space from an Array.
343−shrinkToFit :: forall a. Storable a => Array a -> IO (Array a)
344−shrinkToFit arr@Array{..} = do
345− assert (aEnd <= aBound) (return ())
346− let start = unsafeForeignPtrToPtr aStart
347− let used = aEnd `minusPtr` start
348− waste = aBound `minusPtr` aEnd
349− -- if used == waste == 0 then do not realloc
350− -- if the wastage is more than 25% of the array then realloc
351− if used < 3 * waste
352− then realloc used arr
353− else return arr
354−
355−-- See also: https://gitlab.haskell.org/ghc/ghc/issues/5218 (Add
356−-- unpackCStringLen# to create Strings from string literals)
357−--
358−-- TBD: We can also add template haskell quasiquotes to specify arrays of other
359−-- literal types. TH will encode them into a string literal and we read that as
360−-- an array of the required type. With template Haskell we can provide a safe
361−-- version of fromString#.
362−--
363−-- | Create an @Array Word8@ of the given length from a machine address
364−-- 'Addr#'. This API is unsafe for the following reasons:
365−--
366−-- 1. The address must point to pinned memory or foreign memory.
367−-- 2. The address must be legally accessible upto the given length.
368−-- 3. To guarantee that the array is immutable, the contents of the address
369−-- must be guaranteed to not change.
370−--
371−-- A common use case for this API is to create an array from a static unboxed
372−-- string literal. GHC string literals are of type 'Addr#', and must contain
373−-- characters that can be encoded in a byte i.e. characters or literal bytes in
374−-- the range from 0-255.
375−--
376−-- >>> fromAddr# 5 "hello world!"#
377−-- > [104,101,108,108,111]
378−--
379−-- >>> fromAddr# 3 "\255\NUL\255"#
380−-- > [255,0,255]
381−--
382−-- /See also: 'fromString#'/
383−--
384−-- /Unsafe/
385−--
386−-- /Time complexity: O(1)/
387−--
388−{-# INLINE fromAddr# #-}
389−fromAddr# :: Int -> Addr# -> IO (Array Word8)
390−fromAddr# n addr# = do
391− ptr <- newForeignPtr_ (castPtr $ Ptr addr#)
392− let p = unsafeForeignPtrToPtr ptr
393− let end = p `plusPtr` n
394− return $ Array
395− { aStart = ptr
396− , aEnd = end
397− , aBound = end
398− }
399−
400−-- | Like 'fromAddr#' but determines the length of the array upto and excluding
401−-- the first @0@ byte. The address must be legally accessible up to the first
402−-- NUL byte.
403−--
404−-- A common use case for this API is to create an array from an unboxed string
405−-- literal. Unboxed string literals are guaranteed to be terminated by a NUL
406−-- byte.
407−--
408−-- >>> fromString# "hello world!"#
409−-- > [104,101,108,108,111,32,119,111,114,108,100,33]
410−--
411−-- >>> fromString# "\255\NUL\255"#
412−-- > [255]
413−--
414−-- /See also: 'fromAddr#', 'fromString'/
415−--
416−-- /Unsafe/
232+-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
233+-- arrays of size n.
234+{-# INLINE_NORMAL fromStreamDArraysOf #-}
235+fromStreamDArraysOf :: forall m a. (MonadIO m, Storable a)
236+ => Int -> D.Stream m a -> D.Stream m (Array a)
237+-- fromStreamDArraysOf n str = D.groupsOf n (writeN n) str
238+fromStreamDArraysOf n str = D.map unsafeFreeze $ MA.fromStreamDArraysOf n str
239+
240+-- XXX concatMap does not seem to have the best possible performance so we have
241+-- a custom way to concat arrays.
242+{-# INLINE_NORMAL flattenArrays #-}
243+flattenArrays :: forall m a. (MonadIO m, Storable a)
244+ => D.Stream m (Array a) -> D.Stream m a
245+flattenArrays = MA.flattenArrays . D.map unsafeThaw
246+
247+{-# INLINE_NORMAL flattenArraysRev #-}
248+flattenArraysRev :: forall m a. (MonadIO m, Storable a)
249+ => D.Stream m (Array a) -> D.Stream m a
250+flattenArraysRev = MA.flattenArraysRev . D.map unsafeThaw
251+
252+
253+{-# INLINE_NORMAL packArraysChunksOf #-}
254+packArraysChunksOf :: (MonadIO m, Storable a)
255+ => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
256+packArraysChunksOf n str =
257+ D.map unsafeFreeze $ MA.packArraysChunksOf n $ D.map unsafeThaw str
258+
259+-- XXX instead of writing two different versions of this operation, we should
260+-- write it as a pipe.
261+{-# INLINE_NORMAL lpackArraysChunksOf #-}
262+lpackArraysChunksOf :: (MonadIO m, Storable a)
263+ => Int -> Fold m (Array a) () -> Fold m (Array a) ()
264+lpackArraysChunksOf n fld =
265+ lmap unsafeThaw $ MA.lpackArraysChunksOf n (lmap unsafeFreeze fld)
266+
267+#if !defined(mingw32_HOST_OS)
268+
269+-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
270+-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
271+-- each array and a maximum of @maxEntries@ entries in each array.
417272--
418−-- /Time complexity: O(n) (computes the length of the string)/
273+-- @since 0.7.0
274+{-# INLINE_NORMAL groupIOVecsOf #-}
275+groupIOVecsOf :: MonadIO m
276+ => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
277+groupIOVecsOf n maxIOVLen str =
278+ D.map unsafeFreeze $ MA.groupIOVecsOf n maxIOVLen $ D.map unsafeThaw str
279+#endif
280+
281+-- | Split a stream of arrays on a given separator byte, dropping the separator
282+-- and coalescing all the arrays between two separators into a single array.
419283--
420−{-# INLINE fromString# #-}
421−fromString# :: Addr# -> IO (Array Word8)
422−fromString# addr# = do
423− ptr <- newForeignPtr_ (castPtr cstr)
424− len <- c_strlen cstr
425− let n = fromIntegral len
426− let p = unsafeForeignPtrToPtr ptr
427− let end = p `plusPtr` n
428− return $ Array
429− { aStart = ptr
430− , aEnd = end
431− , aBound = end
432− }
433− where
434− cstr :: CString
435− cstr = Ptr addr#
284+-- @since 0.7.0
285+{-# INLINE_NORMAL splitOn #-}
286+splitOn
287+ :: MonadIO m
288+ => Word8
289+ -> D.Stream m (Array Word8)
290+ -> D.Stream m (Array Word8)
291+splitOn byte str = D.map unsafeFreeze $ MA.splitOn byte $ D.map unsafeThaw str
292+
293+-- Drops the separator byte
294+{-# INLINE breakOn #-}
295+breakOn :: MonadIO m
296+ => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
297+breakOn sep arr = do
298+ (a, b) <- MA.breakOn sep (unsafeThaw arr)
299+ return (unsafeFreeze a, unsafeFreeze <$> b)
436300
437301-------------------------------------------------------------------------------
438302-- Elimination
@@ -443,28 +307,19 @@ fromString# addr# = do
443307-- Unsafe because it does not check the bounds of the array.
444308{-# INLINE_NORMAL unsafeIndexIO #-}
445309unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a
446−unsafeIndexIO Array {..} i =
447− withForeignPtr aStart $ \p -> do
448− let elemSize = sizeOf (undefined :: a)
449− elemOff = p `plusPtr` (elemSize * i)
450− assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)
451− (return ())
452− peek elemOff
310+unsafeIndexIO arr = MA.unsafeIndexIO (unsafeThaw arr)
453311
454312-- | Return element at the specified index without checking the bounds.
455313{-# INLINE_NORMAL unsafeIndex #-}
456314unsafeIndex :: forall a. Storable a => Array a -> Int -> a
457−unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r
315+unsafeIndex arr = MA.unsafeIndex (unsafeThaw arr)
458316
459317-- | /O(1)/ Get the byte length of the array.
460318--
461319-- @since 0.7.0
462320{-# INLINE byteLength #-}
463321byteLength :: Array a -> Int
464−byteLength Array{..} =
465− let p = unsafeForeignPtrToPtr aStart
466− len = aEnd `minusPtr` p
467− in assert (len >= 0) len
322+byteLength = MA.byteLength . unsafeThaw
468323
469324-- | /O(1)/ Get the length of the array i.e. the number of elements in the
470325-- array.
@@ -472,116 +327,56 @@ byteLength Array{..} =
472327-- @since 0.7.0
473328{-# INLINE length #-}
474329length :: forall a. Storable a => Array a -> Int
475−length arr = byteLength arr `div` sizeOf (undefined :: a)
476−
477−{-# INLINE byteCapacity #-}
478−byteCapacity :: Array a -> Int
479−byteCapacity Array{..} =
480− let p = unsafeForeignPtrToPtr aStart
481− len = aBound `minusPtr` p
482− in assert (len >= 0) len
330+length arr = MA.length (unsafeThaw arr)
483331
484332{-# INLINE_NORMAL toStreamD #-}
485333toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
486−toStreamD Array{..} =
487− let p = unsafeForeignPtrToPtr aStart
488− in D.Stream step p
489−
490− where
491−
492− {-# INLINE_LATE step #-}
493− step _ p | p == aEnd = return D.Stop
494− step _ p = do
495− -- unsafeInlineIO allows us to run this in Identity monad for pure
496− -- toList/foldr case which makes them much faster due to not
497− -- accumulating the list and fusing better with the pure consumers.
498− --
499− -- This should be safe as the array contents are guaranteed to be
500− -- evaluated/written to before we peek at them.
501− let !x = unsafeInlineIO $ do
502− r <- peek p
503− touchForeignPtr aStart
504− return r
505− return $ D.Yield x (p `plusPtr` sizeOf (undefined :: a))
334+toStreamD arr = MA.toStreamD (unsafeThaw arr)
506335
507336{-# INLINE toStreamK #-}
508337toStreamK :: forall t m a. (K.IsStream t, Storable a) => Array a -> t m a
509−toStreamK Array{..} =
510− let p = unsafeForeignPtrToPtr aStart
511− in go p
512−
513− where
514−
515− go p | p == aEnd = K.nil
516− | otherwise =
517− -- See Note in toStreamD.
518− let !x = unsafeInlineIO $ do
519− r <- peek p
520− touchForeignPtr aStart
521− return r
522− in x `K.cons` go (p `plusPtr` sizeOf (undefined :: a))
338+toStreamK arr = MA.toStreamK (unsafeThaw arr)
523339
524340{-# INLINE_NORMAL toStreamDRev #-}
525341toStreamDRev :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a
526−toStreamDRev Array{..} =
527− let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
528− in D.Stream step p
529−
530− where
531−
532− {-# INLINE_LATE step #-}
533− step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop
534− step _ p = do
535− -- See comments in toStreamD for why we use unsafeInlineIO
536− let !x = unsafeInlineIO $ do
537− r <- peek p
538− touchForeignPtr aStart
539− return r
540− return $ D.Yield x (p `plusPtr` negate (sizeOf (undefined :: a)))
342+toStreamDRev arr = MA.toStreamDRev (unsafeThaw arr)
541343
542344{-# INLINE toStreamKRev #-}
543345toStreamKRev :: forall t m a. (K.IsStream t, Storable a) => Array a -> t m a
544−toStreamKRev Array {..} =
545− let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
546− in go p
547−
548− where
549−
550− go p | p < unsafeForeignPtrToPtr aStart = K.nil
551− | otherwise =
552− let !x = unsafeInlineIO $ do
553− r <- peek p
554− touchForeignPtr aStart
555− return r
556− in x `K.cons` go (p `plusPtr` negate (sizeOf (undefined :: a)))
346+toStreamKRev arr = MA.toStreamKRev (unsafeThaw arr)
557347
558348{-# INLINE_NORMAL foldl' #-}
559349foldl' :: forall a b. Storable a => (b -> a -> b) -> b -> Array a -> b
560−foldl' f z arr = runIdentity $ D.foldl' f z $ toStreamD arr
350+foldl' f z arr = MA.foldl' f z (unsafeThaw arr)
561351
562352{-# INLINE_NORMAL foldr #-}
563353foldr :: Storable a => (a -> b -> b) -> b -> Array a -> b
564−foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr
354+foldr f z arr = MA.foldr f z (unsafeThaw arr)
565355
566−-------------------------------------------------------------------------------
567−-- Instances
568−-------------------------------------------------------------------------------
356+-- | Create two slices of an array without copying the original array. The
357+-- specified index @i@ is the first index of the second slice.
358+--
359+-- @since 0.7.0
360+splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
361+splitAt i arr = (unsafeFreeze a, unsafeFreeze b)
362+ where
363+ (a, b) = MA.splitAt i (unsafeThaw arr)
569364
570−{-# INLINE_NORMAL writeNAllocWith #-}
571−writeNAllocWith :: forall m a. (MonadIO m, Storable a)
572− => (Int -> IO (Array a)) -> Int -> Fold m a (Array a)
573−writeNAllocWith alloc n = Fold step initial extract
365+-- | Convert an 'Array' into a list.
366+--
367+-- @since 0.7.0
368+{-# INLINE toList #-}
369+toList :: Storable a => Array a -> [a]
370+toList = MA.toList . unsafeThaw
574371
575− where
372+{-# INLINE_NORMAL unlines #-}
373+unlines :: forall m a. (MonadIO m, Storable a)
374+ => a -> D.Stream m (Array a) -> D.Stream m a
375+unlines sep str = MA.unlines sep (D.map unsafeThaw str)
576376
577− initial = liftIO $ alloc (max n 0)
578− step arr@(Array _ end bound) _ | end == bound = return arr
579− step (Array start end bound) x = do
580− liftIO $ poke end x
581− return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
582− -- XXX note that shirkToFit does not maintain alignment, in case we are
583− -- using aligned allocation.
584− extract = return -- liftIO . shrinkToFit
377+-------------------------------------------------------------------------------
378+-- Folds
379+-------------------------------------------------------------------------------
585380
586381-- | @writeN n@ folds a maximum of @n@ elements from the input stream to an
587382-- 'Array'.
@@ -589,7 +384,7 @@ writeNAllocWith alloc n = Fold step initial extract
589384-- @since 0.7.0
590385{-# INLINE_NORMAL writeN #-}
591386writeN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)
592−writeN = writeNAllocWith newArray
387+writeN = fmap unsafeFreeze . MA.writeN
593388
594389-- | @writeNAligned alignment n@ folds a maximum of @n@ elements from the input
595390-- stream to an 'Array' aligned to the given size.
@@ -599,7 +394,7 @@ writeN = writeNAllocWith newArray
599394{-# INLINE_NORMAL writeNAligned #-}
600395writeNAligned :: forall m a. (MonadIO m, Storable a)
601396 => Int -> Int -> Fold m a (Array a)
602−writeNAligned alignSize = writeNAllocWith (newArrayAligned alignSize)
397+writeNAligned alignSize = fmap unsafeFreeze . MA.writeNAligned alignSize
603398
604399-- | @writeNAlignedUnmanaged n@ folds a maximum of @n@ elements from the input
605400-- stream to an 'Array' aligned to the given size and using unmanaged memory.
@@ -612,11 +407,7 @@ writeNAligned alignSize = writeNAllocWith (newArrayAligned alignSize)
612407writeNAlignedUnmanaged :: forall m a. (MonadIO m, Storable a)
613408 => Int -> Int -> Fold m a (Array a)
614409writeNAlignedUnmanaged alignSize =
615− writeNAllocWith (newArrayAlignedUnmanaged alignSize)
616−
617−data ArrayUnsafe a = ArrayUnsafe
618− {-# UNPACK #-} !(ForeignPtr a) -- first address
619− {-# UNPACK #-} !(Ptr a) -- first unused address
410+ fmap unsafeFreeze . MA.writeNAlignedUnmanaged alignSize
620411
621412-- | Like 'writeN' but does not check the array bounds when writing. The fold
622413-- driver must not call the step function more than 'n' times otherwise it will
@@ -628,17 +419,7 @@ data ArrayUnsafe a = ArrayUnsafe
628419{-# INLINE_NORMAL writeNUnsafe #-}
629420writeNUnsafe :: forall m a. (MonadIO m, Storable a)
630421 => Int -> Fold m a (Array a)
631−writeNUnsafe n = Fold step initial extract
632−
633− where
634−
635− initial = do
636− (Array start end _) <- liftIO $ newArray (max n 0)
637− return $ ArrayUnsafe start end
638− step (ArrayUnsafe start end) x = do
639− liftIO $ poke end x
640− return $ ArrayUnsafe start (end `plusPtr` sizeOf (undefined :: a))
641− extract (ArrayUnsafe start end) = return $ Array start end end -- liftIO . shrinkToFit
422+writeNUnsafe n = unsafeFreeze <$> MA.writeNUnsafe n
642423
643424-- XXX The realloc based implementation needs to make one extra copy if we use
644425-- shrinkToFit. On the other hand, the stream of arrays implementation may
@@ -656,25 +437,8 @@ writeNUnsafe n = Fold step initial extract
656437toArrayMinChunk :: forall m a. (MonadIO m, Storable a)
657438 => Int -> Int -> Fold m a (Array a)
658439-- toArrayMinChunk n = FL.mapM spliceArrays $ toArraysOf n
659−toArrayMinChunk alignSize elemCount = Fold step initial extract
660−
661− where
662−
663− insertElem (Array start end bound) x = do
664− liftIO $ poke end x
665− return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound
666−
667− initial = do
668− when (elemCount < 0) $ error "toArrayMinChunk: elemCount is negative"
669− liftIO $ newArrayAligned alignSize elemCount
670− step arr@(Array start end bound) x | end == bound = do
671− let p = unsafeForeignPtrToPtr start
672− oldSize = end `minusPtr` p
673− newSize = max (oldSize * 2) 1
674− arr1 <- liftIO $ reallocAligned alignSize newSize arr
675− insertElem arr1 x
676− step arr x = insertElem arr x
677− extract = liftIO . shrinkToFit
440+toArrayMinChunk alignSize elemCount =
441+ unsafeFreeze <$> MA.toArrayMinChunk alignSize elemCount
678442
679443-- | Fold the whole input to a single array.
680444--
@@ -683,9 +447,7 @@ toArrayMinChunk alignSize elemCount = Fold step initial extract
683447-- @since 0.7.0
684448{-# INLINE write #-}
685449write :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)
686−write = toArrayMinChunk (alignment (undefined :: a))
687− (bytesToElemCount (undefined :: a)
688− (mkChunkSize 1024))
450+write = fmap unsafeFreeze MA.write
689451
690452-- | Like 'write' but the array memory is aligned according to the specified
691453-- alignment size. This could be useful when we have specific alignment, for
@@ -697,202 +459,16 @@ write = toArrayMinChunk (alignment (undefined :: a))
697459{-# INLINE writeAligned #-}
698460writeAligned :: forall m a. (MonadIO m, Storable a)
699461 => Int -> Fold m a (Array a)
700−writeAligned alignSize =
701− toArrayMinChunk alignSize
702− (bytesToElemCount (undefined :: a)
703− (mkChunkSize 1024))
704−
705−{-# INLINE_NORMAL fromStreamDN #-}
706−fromStreamDN :: forall m a. (MonadIO m, Storable a)
707− => Int -> D.Stream m a -> m (Array a)
708−fromStreamDN limit str = do
709− arr <- liftIO $ newArray limit
710− end <- D.foldlM' fwrite (return $ aEnd arr) $ D.take limit str
711− return $ arr {aEnd = end}
712−
713− where
714−
715− fwrite ptr x = do
716− liftIO $ poke ptr x
717− return $ ptr `plusPtr` sizeOf (undefined :: a)
718−
719−data GroupState s start end bound
720− = GroupStart s
721− | GroupBuffer s start end bound
722− | GroupYield start end bound (GroupState s start end bound)
723− | GroupFinish
724−
725−-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of
726−-- arrays of size n.
727−{-# INLINE_NORMAL fromStreamDArraysOf #-}
728−fromStreamDArraysOf :: forall m a. (MonadIO m, Storable a)
729− => Int -> D.Stream m a -> D.Stream m (Array a)
730−-- fromStreamDArraysOf n str = D.groupsOf n (writeN n) str
731−fromStreamDArraysOf n (D.Stream step state) =
732− D.Stream step' (GroupStart state)
733−
734− where
735−
736− {-# INLINE_LATE step' #-}
737− step' _ (GroupStart st) = do
738− when (n <= 0) $
739− -- XXX we can pass the module string from the higher level API
740− error $ "Streamly.Internal.Memory.Array.Types.fromStreamDArraysOf: the size of "
741− ++ "arrays [" ++ show n ++ "] must be a natural number"
742− Array start end bound <- liftIO $ newArray n
743− return $ D.Skip (GroupBuffer st start end bound)
744−
745− step' gst (GroupBuffer st start end bound) = do
746− r <- step (adaptState gst) st
747− case r of
748− D.Yield x s -> do
749− liftIO $ poke end x
750− let end' = end `plusPtr` sizeOf (undefined :: a)
751− return $
752− if end' >= bound
753− then D.Skip (GroupYield start end' bound (GroupStart s))
754− else D.Skip (GroupBuffer s start end' bound)
755− D.Skip s -> return $ D.Skip (GroupBuffer s start end bound)
756− D.Stop -> return $ D.Skip (GroupYield start end bound GroupFinish)
757−
758− step' _ (GroupYield start end bound next) =
759− return $ D.Yield (Array start end bound) next
760−
761− step' _ GroupFinish = return D.Stop
462+writeAligned = fmap unsafeFreeze . MA.writeAligned
762463
763−-- XXX concatMap does not seem to have the best possible performance so we have
764−-- a custom way to concat arrays.
765−data FlattenState s a =
766− OuterLoop s
767− | InnerLoop s !(ForeignPtr a) !(Ptr a) !(Ptr a)
768−
769−{-# INLINE_NORMAL flattenArrays #-}
770−flattenArrays :: forall m a. (MonadIO m, Storable a)
771− => D.Stream m (Array a) -> D.Stream m a
772−flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)
773−
774− where
775−
776− {-# INLINE_LATE step' #-}
777− step' gst (OuterLoop st) = do
778− r <- step (adaptState gst) st
779− return $ case r of
780− D.Yield Array{..} s ->
781− let p = unsafeForeignPtrToPtr aStart
782− in D.Skip (InnerLoop s aStart p aEnd)
783− D.Skip s -> D.Skip (OuterLoop s)
784− D.Stop -> D.Stop
785−
786− step' _ (InnerLoop st _ p end) | p == end =
787− return $ D.Skip $ OuterLoop st
788−
789− step' _ (InnerLoop st startf p end) = do
790− x <- liftIO $ do
791− r <- peek p
792− touchForeignPtr startf
793− return r
794− return $ D.Yield x (InnerLoop st startf
795− (p `plusPtr` sizeOf (undefined :: a)) end)
796−
797−{-# INLINE_NORMAL flattenArraysRev #-}
798−flattenArraysRev :: forall m a. (MonadIO m, Storable a)
799− => D.Stream m (Array a) -> D.Stream m a
800−flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)
801−
802− where
803−
804− {-# INLINE_LATE step' #-}
805− step' gst (OuterLoop st) = do
806− r <- step (adaptState gst) st
807− return $ case r of
808− D.Yield Array{..} s ->
809− let p = aEnd `plusPtr` negate (sizeOf (undefined :: a))
810− -- XXX we do not need aEnd
811− in D.Skip (InnerLoop s aStart p aEnd)
812− D.Skip s -> D.Skip (OuterLoop s)
813− D.Stop -> D.Stop
814−
815− step' _ (InnerLoop st start p _) | p < unsafeForeignPtrToPtr start =
816− return $ D.Skip $ OuterLoop st
817−
818− step' _ (InnerLoop st startf p end) = do
819− x <- liftIO $ do
820− r <- peek p
821− touchForeignPtr startf
822− return r
823− return $ D.Yield x (InnerLoop st startf
824− (p `plusPtr` negate (sizeOf (undefined :: a))) end)
825−
826−-- CAUTION: a very large number (millions) of arrays can degrade performance
827−-- due to GC overhead because we need to buffer the arrays before we flatten
828−-- all the arrays.
829−--
830−-- We could take the approach of doubling the memory allocation on each
831−-- overflow. This would result in more or less the same amount of copying as in
832−-- the chunking approach. However, if we have to shrink in the end then it may
833−-- result in an extra copy of the entire data.
834−--
835−{-# INLINE fromStreamD #-}
836−fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
837−fromStreamD m = do
838− let s = fromStreamDArraysOf defaultChunkSize m
839− buffered <- D.foldr K.cons K.nil s
840− len <- K.foldl' (+) 0 (K.map length buffered)
841− fromStreamDN len $ flattenArrays $ D.fromStreamK buffered
842−{-
843−fromStreamD m = runFold write m
844− where
845− runFold (Fold step begin done) = D.foldlMx' step begin done
846−-}
847−
848−-- Use foldr/build fusion to fuse with list consumers
849−-- This can be useful when using the IsList instance
850−{-# INLINE_LATE toListFB #-}
851−toListFB :: forall a b. Storable a => (a -> b -> b) -> b -> Array a -> b
852−toListFB c n Array{..} = go (unsafeForeignPtrToPtr aStart)
853− where
854−
855− go p | p == aEnd = n
856− go p =
857− -- unsafeInlineIO allows us to run this in Identity monad for pure
858− -- toList/foldr case which makes them much faster due to not
859− -- accumulating the list and fusing better with the pure consumers.
860− --
861− -- This should be safe as the array contents are guaranteed to be
862− -- evaluated/written to before we peek at them.
863− let !x = unsafeInlineIO $ do
864− r <- peek p
865− touchForeignPtr aStart
866− return r
867− in c x (go (p `plusPtr` sizeOf (undefined :: a)))
868−
869−-- | Convert an 'Array' into a list.
870−--
871−-- @since 0.7.0
872−{-# INLINE toList #-}
873−toList :: Storable a => Array a -> [a]
874−toList s = build (\c n -> toListFB c n s)
464+-------------------------------------------------------------------------------
465+-- Instances
466+-------------------------------------------------------------------------------
875467
876468instance (Show a, Storable a) => Show (Array a) where
877469 {-# INLINE showsPrec #-}
878470 showsPrec _ = shows . toList
879471
880−-- | Create an 'Array' from the first N elements of a list. The array is
881−-- allocated to size N, if the list terminates before N elements then the
882−-- array may hold less than N elements.
883−--
884−-- @since 0.7.0
885−{-# INLINABLE fromListN #-}
886−fromListN :: Storable a => Int -> [a] -> Array a
887−fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs
888−
889−-- | Create an 'Array' from a list. The list must be of finite size.
890−--
891−-- @since 0.7.0
892−{-# INLINABLE fromList #-}
893−fromList :: Storable a => [a] -> Array a
894−fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs
895−
896472instance (Storable a, Read a, Show a) => Read (Array a) where
897473 {-# INLINE readPrec #-}
898474 readPrec = fromList <$> readPrec
@@ -912,29 +488,11 @@ instance Storable a => IsList (Array a) where
912488 {-# INLINE toList #-}
913489 toList = toList
914490
915−{-# INLINE arrcmp #-}
916−arrcmp :: Array a -> Array a -> Bool
917−arrcmp arr1 arr2 =
918− let !res = unsafeInlineIO $ do
919− let ptr1 = unsafeForeignPtrToPtr $ aStart arr1
920− let ptr2 = unsafeForeignPtrToPtr $ aStart arr2
921− let len1 = aEnd arr1 `minusPtr` ptr1
922− let len2 = aEnd arr2 `minusPtr` ptr2
923−
924− if len1 == len2
925− then do
926− r <- memcmp (castPtr ptr1) (castPtr ptr2) len1
927− touchForeignPtr $ aStart arr1
928− touchForeignPtr $ aStart arr2
929− return r
930− else return False
931− in res
932−
933491-- XXX we are assuming that Storable equality means element equality. This may
934492-- or may not be correct? arrcmp is 40% faster compared to stream equality.
935493instance (Storable a, Eq a) => Eq (Array a) where
936494 {-# INLINE (==) #-}
937− (==) = arrcmp
495+ a == b = unsafeThaw a == unsafeThaw b
938496 -- arr1 == arr2 = runIdentity $ D.eqBy (==) (toStreamD arr1) (toStreamD arr2)
939497
940498instance (Storable a, NFData a) => NFData (Array a) where
@@ -943,8 +501,7 @@ instance (Storable a, NFData a) => NFData (Array a) where
943501
944502instance (Storable a, Ord a) => Ord (Array a) where
945503 {-# INLINE compare #-}
946− compare arr1 arr2 = unsafePerformIO $
947− D.cmpBy compare (toStreamD arr1) (toStreamD arr2)
504+ compare a b = compare (unsafeThaw a) (unsafeThaw b)
948505
949506 -- Default definitions defined in base do not have an INLINE on them, so we
950507 -- replicate them here with an INLINE.
@@ -1005,27 +562,6 @@ instance Foldable Array where
1005562-- Semigroup and Monoid
1006563-------------------------------------------------------------------------------
1007564
1008−-- Splice two immutable arrays creating a new array.
1009−{-# INLINE spliceTwo #-}
1010−spliceTwo :: (MonadIO m, Storable a) => Array a -> Array a -> m (Array a)
1011−spliceTwo arr1 arr2 = do
1012− let src1 = unsafeForeignPtrToPtr (aStart arr1)
1013− src2 = unsafeForeignPtrToPtr (aStart arr2)
1014− len1 = aEnd arr1 `minusPtr` src1
1015− len2 = aEnd arr2 `minusPtr` src2
1016−
1017− arr <- liftIO $ newArray (len1 + len2)
1018− let dst = unsafeForeignPtrToPtr (aStart arr)
1019−
1020− -- XXX Should we use copyMutableByteArray# instead? Is there an overhead to
1021− -- ccall?
1022− liftIO $ do
1023− memcpy (castPtr dst) (castPtr src1) len1
1024− touchForeignPtr (aStart arr1)
1025− memcpy (castPtr (dst `plusPtr` len1)) (castPtr src2) len2
1026− touchForeignPtr (aStart arr2)
1027− return arr { aEnd = dst `plusPtr` (len1 + len2) }
1028−
1029565instance Storable a => Semigroup (Array a) where
1030566 arr1 <> arr2 = unsafePerformIO $ spliceTwo arr1 arr2
1031567
@@ -1037,372 +573,8 @@ nil ::
1037573 Storable a =>
1038574#endif
1039575 Array a
1040−nil = Array nullForeignPtr (Ptr nullAddr#) (Ptr nullAddr#)
576+nil = Array nullForeignPtr (Ptr nullAddr#)
1041577
1042578instance Storable a => Monoid (Array a) where
1043579 mempty = nil
1044580 mappend = (<>)
1045−
1046−-------------------------------------------------------------------------------
1047−-- IO
1048−-------------------------------------------------------------------------------
1049−
1050−-- | GHC memory management allocation header overhead
1051−allocOverhead :: Int
1052−allocOverhead = 2 * sizeOf (undefined :: Int)
1053−
1054−mkChunkSize :: Int -> Int
1055−mkChunkSize n = let size = n - allocOverhead in max size 0
1056−
1057−mkChunkSizeKB :: Int -> Int
1058−mkChunkSizeKB n = mkChunkSize (n * k)
1059− where k = 1024
1060−
1061−-- | Default maximum buffer size in bytes, for reading from and writing to IO
1062−-- devices, the value is 32KB minus GHC allocation overhead, which is a few
1063−-- bytes, so that the actual allocation is 32KB.
1064−defaultChunkSize :: Int
1065−defaultChunkSize = mkChunkSizeKB 32
1066−
1067−{-# INLINE_NORMAL unlines #-}
1068−unlines :: forall m a. (MonadIO m, Storable a)
1069− => a -> D.Stream m (Array a) -> D.Stream m a
1070−unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
1071− where
1072− {-# INLINE_LATE step' #-}
1073− step' gst (OuterLoop st) = do
1074− r <- step (adaptState gst) st
1075− return $ case r of
1076− D.Yield Array{..} s ->
1077− let p = unsafeForeignPtrToPtr aStart
1078− in D.Skip (InnerLoop s aStart p aEnd)
1079− D.Skip s -> D.Skip (OuterLoop s)
1080− D.Stop -> D.Stop
1081−
1082− step' _ (InnerLoop st _ p end) | p == end =
1083− return $ D.Yield sep $ OuterLoop st
1084−
1085− step' _ (InnerLoop st startf p end) = do
1086− x <- liftIO $ do
1087− r <- peek p
1088− touchForeignPtr startf
1089− return r
1090− return $ D.Yield x (InnerLoop st startf
1091− (p `plusPtr` sizeOf (undefined :: a)) end)
1092−
1093−-- Splice an array into a pre-reserved mutable array. The user must ensure
1094−-- that there is enough space in the mutable array.
1095−{-# INLINE spliceWith #-}
1096−spliceWith :: (MonadIO m) => Array a -> Array a -> m (Array a)
1097−spliceWith dst@(Array _ end bound) src = liftIO $ do
1098− let srcLen = byteLength src
1099− if end `plusPtr` srcLen > bound
1100− then error "Bug: spliceIntoUnsafe: Not enough space in the target array"
1101− else
1102− withForeignPtr (aStart dst) $ \_ ->
1103− withForeignPtr (aStart src) $ \psrc -> do
1104− let pdst = aEnd dst
1105− memcpy (castPtr pdst) (castPtr psrc) srcLen
1106− return $ dst { aEnd = pdst `plusPtr` srcLen }
1107−
1108−-- Splice a new array into a preallocated mutable array, doubling the space if
1109−-- there is no space in the target array.
1110−{-# INLINE spliceWithDoubling #-}
1111−spliceWithDoubling :: (MonadIO m, Storable a)
1112− => Array a -> Array a -> m (Array a)
1113−spliceWithDoubling dst@(Array start end bound) src = do
1114− assert (end <= bound) (return ())
1115− let srcLen = aEnd src `minusPtr` unsafeForeignPtrToPtr (aStart src)
1116−
1117− dst1 <-
1118− if end `plusPtr` srcLen >= bound
1119− then do
1120− let oldStart = unsafeForeignPtrToPtr start
1121− oldSize = end `minusPtr` oldStart
1122− newSize = max (oldSize * 2) (oldSize + srcLen)
1123− liftIO $ realloc newSize dst
1124− else return dst
1125− spliceWith dst1 src
1126−
1127−data SpliceState s arr
1128− = SpliceInitial s
1129− | SpliceBuffering s arr
1130− | SpliceYielding arr (SpliceState s arr)
1131− | SpliceFinish
1132−
1133−-- XXX can use general grouping combinators to achieve this?
1134−-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
1135−-- maximum specified size. Note that if a single array is bigger than the
1136−-- specified size we do not split it to fit. When we coalesce multiple arrays
1137−-- if the size would exceed the specified size we do not coalesce therefore the
1138−-- actual array size may be less than the specified chunk size.
1139−--
1140−-- @since 0.7.0
1141−{-# INLINE_NORMAL packArraysChunksOf #-}
1142−packArraysChunksOf :: (MonadIO m, Storable a)
1143− => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
1144−packArraysChunksOf n (D.Stream step state) =
1145− D.Stream step' (SpliceInitial state)
1146−
1147− where
1148−
1149− {-# INLINE_LATE step' #-}
1150− step' gst (SpliceInitial st) = do
1151− when (n <= 0) $
1152− -- XXX we can pass the module string from the higher level API
1153− error $ "Streamly.Internal.Memory.Array.Types.packArraysChunksOf: the size of "
1154− ++ "arrays [" ++ show n ++ "] must be a natural number"
1155− r <- step gst st
1156− case r of
1157− D.Yield arr s -> return $
1158− let len = byteLength arr
1159− in if len >= n
1160− then D.Skip (SpliceYielding arr (SpliceInitial s))
1161− else D.Skip (SpliceBuffering s arr)
1162− D.Skip s -> return $ D.Skip (SpliceInitial s)
1163− D.Stop -> return D.Stop
1164−
1165− step' gst (SpliceBuffering st buf) = do
1166− r <- step gst st
1167− case r of
1168− D.Yield arr s -> do
1169− let len = byteLength buf + byteLength arr
1170− if len > n
1171− then return $
1172− D.Skip (SpliceYielding buf (SpliceBuffering s arr))
1173− else do
1174− buf' <- if byteCapacity buf < n
1175− then liftIO $ realloc n buf
1176− else return buf
1177− buf'' <- spliceWith buf' arr
1178− return $ D.Skip (SpliceBuffering s buf'')
1179− D.Skip s -> return $ D.Skip (SpliceBuffering s buf)
1180− D.Stop -> return $ D.Skip (SpliceYielding buf SpliceFinish)
1181−
1182− step' _ SpliceFinish = return D.Stop
1183−
1184− step' _ (SpliceYielding arr next) = return $ D.Yield arr next
1185−
1186−-- XXX instead of writing two different versions of this operation, we should
1187−-- write it as a pipe.
1188−{-# INLINE_NORMAL lpackArraysChunksOf #-}
1189−lpackArraysChunksOf :: (MonadIO m, Storable a)
1190− => Int -> Fold m (Array a) () -> Fold m (Array a) ()
1191−lpackArraysChunksOf n (Fold step1 initial1 extract1) =
1192− Fold step initial extract
1193−
1194− where
1195−
1196− initial = do
1197− when (n <= 0) $
1198− -- XXX we can pass the module string from the higher level API
1199− error $ "Streamly.Internal.Memory.Array.Types.packArraysChunksOf: the size of "
1200− ++ "arrays [" ++ show n ++ "] must be a natural number"
1201− r1 <- initial1
1202− return (Tuple' Nothing r1)
1203−
1204− extract (Tuple' Nothing r1) = extract1 r1
1205− extract (Tuple' (Just buf) r1) = do
1206− r <- step1 r1 buf
1207− extract1 r
1208−
1209− step (Tuple' Nothing r1) arr =
1210− let len = byteLength arr
1211− in if len >= n
1212− then do
1213− r <- step1 r1 arr
1214− extract1 r
1215− r1' <- initial1
1216− return (Tuple' Nothing r1')
1217− else return (Tuple' (Just arr) r1)
1218−
1219− step (Tuple' (Just buf) r1) arr = do
1220− let len = byteLength buf + byteLength arr
1221− buf' <- if byteCapacity buf < len
1222− then liftIO $ realloc (max n len) buf
1223− else return buf
1224− buf'' <- spliceWith buf' arr
1225−
1226− if len >= n
1227− then do
1228− r <- step1 r1 buf''
1229− extract1 r
1230− r1' <- initial1
1231− return (Tuple' Nothing r1')
1232− else return (Tuple' (Just buf'') r1)
1233−
1234−#if !defined(mingw32_HOST_OS)
1235−data GatherState s arr
1236− = GatherInitial s
1237− | GatherBuffering s arr Int
1238− | GatherYielding arr (GatherState s arr)
1239− | GatherFinish
1240−
1241−-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
1242−-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
1243−-- each array and a maximum of @maxEntries@ entries in each array.
1244−--
1245−-- @since 0.7.0
1246−{-# INLINE_NORMAL groupIOVecsOf #-}
1247−groupIOVecsOf :: MonadIO m
1248− => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
1249−groupIOVecsOf n maxIOVLen (D.Stream step state) =
1250− D.Stream step' (GatherInitial state)
1251−
1252− where
1253−
1254− {-# INLINE_LATE step' #-}
1255− step' gst (GatherInitial st) = do
1256− when (n <= 0) $
1257− -- XXX we can pass the module string from the higher level API
1258− error $ "Streamly.Internal.Memory.Array.Types.groupIOVecsOf: the size of "
1259− ++ "groups [" ++ show n ++ "] must be a natural number"
1260− when (maxIOVLen <= 0) $
1261− -- XXX we can pass the module string from the higher level API
1262− error $ "Streamly.Internal.Memory.Array.Types.groupIOVecsOf: the number of "
1263− ++ "IOVec entries [" ++ show n ++ "] must be a natural number"
1264− r <- step (adaptState gst) st
1265− case r of
1266− D.Yield arr s -> do
1267− let p = unsafeForeignPtrToPtr (aStart arr)
1268− len = byteLength arr
1269− iov <- liftIO $ newArray maxIOVLen
1270− iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
1271− (fromIntegral len))
1272− if len >= n
1273− then return $ D.Skip (GatherYielding iov' (GatherInitial s))
1274− else return $ D.Skip (GatherBuffering s iov' len)
1275− D.Skip s -> return $ D.Skip (GatherInitial s)
1276− D.Stop -> return D.Stop
1277−
1278− step' gst (GatherBuffering st iov len) = do
1279− r <- step (adaptState gst) st
1280− case r of
1281− D.Yield arr s -> do
1282− let p = unsafeForeignPtrToPtr (aStart arr)
1283− alen = byteLength arr
1284− len' = len + alen
1285− if len' > n || length iov >= maxIOVLen
1286− then do
1287− iov' <- liftIO $ newArray maxIOVLen
1288− iov'' <- liftIO $ unsafeSnoc iov' (IOVec (castPtr p)
1289− (fromIntegral alen))
1290− return $ D.Skip (GatherYielding iov
1291− (GatherBuffering s iov'' alen))
1292− else do
1293− iov' <- liftIO $ unsafeSnoc iov (IOVec (castPtr p)
1294− (fromIntegral alen))
1295− return $ D.Skip (GatherBuffering s iov' len')
1296− D.Skip s -> return $ D.Skip (GatherBuffering s iov len)
1297− D.Stop -> return $ D.Skip (GatherYielding iov GatherFinish)
1298−
1299− step' _ GatherFinish = return D.Stop
1300−
1301− step' _ (GatherYielding iov next) = return $ D.Yield iov next
1302−#endif
1303−
1304−-- | Create two slices of an array without copying the original array. The
1305−-- specified index @i@ is the first index of the second slice.
1306−--
1307−-- @since 0.7.0
1308−splitAt :: forall a. Storable a => Int -> Array a -> (Array a, Array a)
1309−splitAt i arr@Array{..} =
1310− let maxIndex = length arr - 1
1311− in if i < 0
1312− then error "sliceAt: negative array index"
1313− else if i > maxIndex
1314− then error $ "sliceAt: specified array index " ++ show i
1315− ++ " is beyond the maximum index " ++ show maxIndex
1316− else let off = i * sizeOf (undefined :: a)
1317− p = unsafeForeignPtrToPtr aStart `plusPtr` off
1318− in ( Array
1319− { aStart = aStart
1320− , aEnd = p
1321− , aBound = p
1322− }
1323− , Array
1324− { aStart = aStart `plusForeignPtr` off
1325− , aEnd = aEnd
1326− , aBound = aBound
1327− }
1328− )
1329−
1330−-- Drops the separator byte
1331−{-# INLINE breakOn #-}
1332−breakOn :: MonadIO m
1333− => Word8 -> Array Word8 -> m (Array Word8, Maybe (Array Word8))
1334−breakOn sep arr@Array{..} = liftIO $ do
1335− let p = unsafeForeignPtrToPtr aStart
1336− loc <- c_memchr p sep (fromIntegral $ aEnd `minusPtr` p)
1337− return $
1338− if loc == nullPtr
1339− then (arr, Nothing)
1340− else
1341− ( Array
1342− { aStart = aStart
1343− , aEnd = loc
1344− , aBound = loc
1345− }
1346− , Just $ Array
1347− { aStart = aStart `plusForeignPtr` (loc `minusPtr` p + 1)
1348− , aEnd = aEnd
1349− , aBound = aBound
1350− }
1351− )
1352−
1353−data SplitState s arr
1354− = Initial s
1355− | Buffering s arr
1356− | Splitting s arr
1357− | Yielding arr (SplitState s arr)
1358− | Finishing
1359−
1360−-- | Split a stream of arrays on a given separator byte, dropping the separator
1361−-- and coalescing all the arrays between two separators into a single array.
1362−--
1363−-- @since 0.7.0
1364−{-# INLINE_NORMAL splitOn #-}
1365−splitOn
1366− :: MonadIO m
1367− => Word8
1368− -> D.Stream m (Array Word8)
1369− -> D.Stream m (Array Word8)
1370−splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
1371−
1372− where
1373−
1374− {-# INLINE_LATE step' #-}
1375− step' gst (Initial st) = do
1376− r <- step gst st
1377− case r of
1378− D.Yield arr s -> do
1379− (arr1, marr2) <- breakOn byte arr
1380− return $ case marr2 of
1381− Nothing -> D.Skip (Buffering s arr1)
1382− Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
1383− D.Skip s -> return $ D.Skip (Initial s)
1384− D.Stop -> return D.Stop
1385−
1386− step' gst (Buffering st buf) = do
1387− r <- step gst st
1388− case r of
1389− D.Yield arr s -> do
1390− (arr1, marr2) <- breakOn byte arr
1391− buf' <- spliceTwo buf arr1
1392− return $ case marr2 of
1393− Nothing -> D.Skip (Buffering s buf')
1394− Just x -> D.Skip (Yielding buf' (Splitting s x))
1395− D.Skip s -> return $ D.Skip (Buffering s buf)
1396− D.Stop -> return $
1397− if byteLength buf == 0
1398− then D.Stop
1399− else D.Skip (Yielding buf Finishing)
1400−
1401− step' _ (Splitting st buf) = do
1402− (arr1, marr2) <- breakOn byte buf
1403− return $ case marr2 of
1404− Nothing -> D.Skip $ Buffering st arr1
1405− Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
1406−
1407− step' _ (Yielding arr next) = return $ D.Yield arr next
1408− step' _ Finishing = return D.Stop