3-- Module : Streamly.Internal.Data.Array
4-- Copyright : (c) 2019 Composewell Technologies
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11module Streamly.Internal.Data.Array
20 module Streamly.Internal.Data.Array.Type
33 , producer -- experimental
39 , last -- XXX getIndexLast?
41 , getIndicesFromThenTo
42 -- , getIndicesFrom -- read from a given position to the end of file
43 -- , getIndicesUpto -- read from beginning up to the given position
45 -- , getIndicesFromRev -- read from a given position to the beginning of file
46 -- , getIndicesUptoRev -- read from end to the given position in file
70 -- * Streaming Operations
87#include "ArrayMacros.h"
89import Control.Monad (when)
90import Control.Monad.IO.Class (MonadIO(..))
91import Data.Functor.Identity (Identity)
92import Data.Proxy (Proxy(..))
93import Data.Word (Word8)
94import Foreign.C.String (CString)
95import Foreign.Ptr (castPtr)
96import Foreign.Storable (Storable)
97import Streamly.Internal.Data.Unbox (Unbox(..))
98import Prelude hiding (length, null, last, map, (!!), read, concat)
100import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..))
101import Streamly.Internal.Data.Serialize.Type (Serialize)
102import Streamly.Internal.Data.Fold.Type (Fold(..))
103import Streamly.Internal.Data.Producer.Type (Producer(..))
104import Streamly.Internal.Data.Stream (Stream)
105import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))
106import Streamly.Internal.Data.Unfold.Type (Unfold(..))
107import Streamly.Internal.System.IO (unsafeInlineIO)
109import qualified Streamly.Internal.Data.Serialize.Type as Serialize
110import qualified Streamly.Internal.Data.MutByteArray.Type as MBA
111import qualified Streamly.Internal.Data.MutArray as MA
112import qualified Streamly.Internal.Data.Array.Type as A
113import qualified Streamly.Internal.Data.Fold as FL
114import qualified Streamly.Internal.Data.Producer.Type as Producer
115import qualified Streamly.Internal.Data.Producer as Producer
116import qualified Streamly.Internal.Data.Ring as RB
117import qualified Streamly.Internal.Data.Stream as D
118import qualified Streamly.Internal.Data.Stream as Stream
119import qualified Streamly.Internal.Data.Unfold as Unfold
121import Streamly.Internal.Data.Array.Type
123#include "DocTestDataArray.hs"
129-- * Arrays are finite and fixed in size
130-- * provide /O(1)/ access to elements
131-- * store only data and not functions
132-- * provide efficient IO interfacing
134-- 'Foldable' instance is not provided because the implementation would be much
135-- less efficient compared to folding via streams. 'Semigroup' and 'Monoid'
136-- instances should be used with care; concatenating arrays using binary
137-- operations can be highly inefficient. Instead, use
138-- 'Streamly.Internal.Data.Stream.Chunked.toArray' to concatenate N
141-- Each array is one pointer visible to the GC. Too many small arrays (e.g.
142-- single byte) are only as good as holding those elements in a Haskell list.
143-- However, small arrays can be compacted into large ones to reduce the
144-- overhead. To hold 32GB memory in 32k sized buffers we need 1 million arrays
145-- if we use one array for each chunk. This is still significant to add
148-------------------------------------------------------------------------------
150-------------------------------------------------------------------------------
152-- | Create an 'Array' from the first N elements of a stream. The array is
153-- allocated to size N, if the stream terminates before N elements then the
154-- array may hold less than N elements.
157{-# INLINE fromStreamN #-}
158fromStreamN :: (MonadIO m, Unbox a) => Int -> Stream m a -> m (Array a)
160 when (n < 0) $ error "writeN: negative write count specified"
163-- | Create an 'Array' from a stream. This is useful when we want to create a
164-- single array from a stream of unknown size. 'writeN' is at least twice
165-- as efficient when the size is already known.
167-- Note that if the input stream is too large memory allocation for the array
168-- may fail. When the stream size is not known, `chunksOf` followed by
169-- processing of indvidual arrays in the resulting stream should be preferred.
172{-# INLINE fromStream #-}
173fromStream :: (MonadIO m, Unbox a) => Stream m a -> m (Array a)
174fromStream = Stream.fold A.write
175-- write m = A.fromStreamD $ D.fromStreamK m
177-------------------------------------------------------------------------------
179-------------------------------------------------------------------------------
181{-# INLINE_NORMAL producer #-}
182producer :: forall m a. (Monad m, Unbox a) => Producer m (Array a) a
184 Producer.translate A.unsafeThaw A.unsafeFreeze
185 $ MA.producerWith (return . unsafeInlineIO)
187-- | Unfold an array into a stream.
189{-# INLINE_NORMAL reader #-}
190reader :: forall m a. (Monad m, Unbox a) => Unfold m (Array a) a
191reader = Producer.simplify producer
193-- | Unfold an array into a stream, does not check the end of the array, the
194-- user is responsible for terminating the stream within the array bounds. For
195-- high performance application where the end condition can be determined by
196-- a terminating fold.
198-- Written in the hope that it may be faster than "read", however, in the case
199-- for which this was written, "read" proves to be faster even though the core
200-- generated with unsafeRead looks simpler.
204{-# INLINE_NORMAL readerUnsafe #-}
205readerUnsafe :: forall m a. (Monad m, Unbox a) => Unfold m (Array a) a
206readerUnsafe = Unfold step inject
209 inject (Array contents start end) =
210 return (ArrayUnsafe contents end start)
212 {-# INLINE_LATE step #-}
213 step (ArrayUnsafe contents end p) = do
214 -- unsafeInlineIO allows us to run this in Identity monad for pure
215 -- toList/foldr case which makes them much faster due to not
216 -- accumulating the list and fusing better with the pure consumers.
218 -- This should be safe as the array contents are guaranteed to be
219 -- evaluated/written to before we peek at them.
220 let !x = unsafeInlineIO $ peekAt p contents
221 let !p1 = INDEX_NEXT(p,a)
222 return $ D.Yield x (ArrayUnsafe contents end p1)
226-- >>> null arr = Array.byteLength arr == 0
230null :: Array a -> Bool
231null arr = A.byteLength arr == 0
233-- | Like 'getIndex' but indexes the array in reverse from the end.
236{-# INLINE getIndexRev #-}
237getIndexRev :: forall a. Unbox a => Int -> Array a -> Maybe a
241 let elemPtr = RINDEX_OF(arrEnd arr, i, a)
242 if i >= 0 && elemPtr >= arrStart arr
243 then Just <$> peekAt elemPtr (arrContents arr)
248-- >>> last arr = Array.getIndexRev arr 0
252last :: Unbox a => Array a -> Maybe a
255-------------------------------------------------------------------------------
256-- Folds with Array as the container
257-------------------------------------------------------------------------------
259-- | @writeLastN n@ folds a maximum of @n@ elements from the end of the input
260-- stream to an 'Array'.
262{-# INLINE writeLastN #-}
264 (Storable a, Unbox a, MonadIO m) => Int -> Fold m a (Array a)
266 | n <= 0 = fmap (const mempty) FL.drain
267 | otherwise = A.unsafeFreeze <$> Fold step initial done done
271 step (Tuple3Fused' rb rh i) a = do
272 rh1 <- liftIO $ RB.unsafeInsert rb rh a
273 return $ FL.Partial $ Tuple3Fused' rb rh1 (i + 1)
276 let f (a, b) = FL.Partial $ Tuple3Fused' a b (0 :: Int)
277 in fmap f $ liftIO $ RB.new n
279 done (Tuple3Fused' rb rh i) = do
280 arr <- liftIO $ MA.new n
281 foldFunc i rh snoc' arr rb
283 -- XXX We should write a read unfold for ring.
284 snoc' b a = liftIO $ MA.snocUnsafe b a
287 | i < n = RB.unsafeFoldRingM
288 | otherwise = RB.unsafeFoldRingFullM
290-------------------------------------------------------------------------------
292-------------------------------------------------------------------------------
294-------------------------------------------------------------------------------
296-------------------------------------------------------------------------------
298-- | Given a sorted array, perform a binary search to find the given element.
299-- Returns the index of the element if found.
302{-# INLINE binarySearch #-}
303binarySearch :: a -> Array a -> Maybe Int
304binarySearch = undefined
306-- find/findIndex etc can potentially be implemented more efficiently on arrays
307-- compared to streams by using SIMD instructions.
308-- We can also return a bit array instead.
310-- | Perform a linear search to find all the indices where a given element is
311-- present in an array.
314findIndicesOf :: (a -> Bool) -> Unfold Identity (Array a) Int
315findIndicesOf = undefined
318findIndexOf :: (a -> Bool) -> Array a -> Maybe Int
319findIndexOf p = Unfold.fold Fold.one . Stream.unfold (findIndicesOf p)
321find :: (a -> Bool) -> Array a -> Bool
322find = Unfold.fold Fold.null . Stream.unfold (findIndicesOf p)
325-------------------------------------------------------------------------------
327-------------------------------------------------------------------------------
329-- XXX We can potentially use SIMD instructions on arrays to fold faster.
331-------------------------------------------------------------------------------
333-------------------------------------------------------------------------------
335-- | /O(1)/ Slice an array in constant time.
337-- Caution: The bounds of the slice are not checked.
342{-# INLINE getSliceUnsafe #-}
345 => Int -- ^ starting index
346 -> Int -- ^ length of the slice
349getSliceUnsafe index len (Array contents start e) =
350 let size = SIZE_OF(a)
351 start1 = start + (index * size)
352 end1 = start1 + (len * size)
353 in assert (end1 <= e) (Array contents start1 end1)
355-- | Split the array into a stream of slices using a predicate. The element
356-- matching the predicate is dropped.
359{-# INLINE splitOn #-}
360splitOn :: (Monad m, Unbox a) =>
361 (a -> Bool) -> Array a -> Stream m (Array a)
362splitOn predicate arr =
363 fmap (\(i, len) -> getSliceUnsafe i len arr)
364 $ D.sliceOnSuffix predicate (A.toStreamD arr)
366{-# INLINE genSlicesFromLen #-}
367genSlicesFromLen :: forall m a. (Monad m, Unbox a)
368 => Int -- ^ from index
369 -> Int -- ^ length of the slice
370 -> Unfold m (Array a) (Int, Int)
371genSlicesFromLen from len =
372 Unfold.lmap A.unsafeThaw (MA.genSlicesFromLen from len)
374-- | Generate a stream of slices of specified length from an array, starting
375-- from the supplied array index. The last slice may be shorter than the
379{-# INLINE getSlicesFromLen #-}
380getSlicesFromLen :: forall m a. (Monad m, Unbox a)
381 => Int -- ^ from index
382 -> Int -- ^ length of the slice
383 -> Unfold m (Array a) (Array a)
384getSlicesFromLen from len =
386 $ Unfold.lmap A.unsafeThaw (MA.getSlicesFromLen from len)
388-------------------------------------------------------------------------------
389-- Random reads and writes
390-------------------------------------------------------------------------------
392-- XXX Change this to a partial function instead of a Maybe type? And use
393-- MA.getIndex instead.
395-- | /O(1)/ Lookup the element at the given index. Index starts from 0.
397{-# INLINE getIndex #-}
398getIndex :: forall a. Unbox a => Int -> Array a -> Maybe a
402 let elemPtr = INDEX_OF(arrStart arr, i, a)
403 if i >= 0 && INDEX_VALID(elemPtr, arrEnd arr, a)
404 then Just <$> peekAt elemPtr (arrContents arr)
407-- | Given a stream of array indices, read the elements on those indices from
408-- the supplied Array. An exception is thrown if an index is out of bounds.
410-- This is the most general operation. We can implement other operations in
415-- let u = lmap (\arr -> (0, length arr - 1)) Unfold.enumerateFromTo
416-- in Unfold.lmap f (getIndices arr)
419-- let i = length arr - 1
420-- in Unfold.lmap f (getIndicesFromThenTo i (i - 1) 0)
424{-# INLINE getIndices #-}
425getIndices :: (Monad m, Unbox a) => Stream m Int -> Unfold m (Array a) a
427 let unf = MA.getIndicesD (return . unsafeInlineIO) m
428 in Unfold.lmap A.unsafeThaw unf
430-- | Unfolds @(from, then, to, array)@ generating a finite stream whose first
431-- element is the array value from the index @from@ and the successive elements
432-- are from the indices in increments of @then@ up to @to@. Index enumeration
433-- can occur downwards or upwards depending on whether @then@ comes before or
437-- getIndicesFromThenTo =
438-- let f (from, next, to, arr) =
439-- (Stream.enumerateFromThenTo from next to, arr)
440-- in Unfold.lmap f getIndices
444{-# INLINE getIndicesFromThenTo #-}
445getIndicesFromThenTo :: Unfold m (Int, Int, Int, Array a) a
446getIndicesFromThenTo = undefined
448-------------------------------------------------------------------------------
449-- Transform via stream operations
450-------------------------------------------------------------------------------
452-- for non-length changing operations we can use the original length for
453-- allocation. If we can predict the length then we can use the prediction for
454-- new allocation. Otherwise we can use a hint and adjust dynamically.
457-- | Transform an array into another array using a pipe transformation
460{-# INLINE runPipe #-}
461runPipe :: (MonadIO m, Unbox a, Unbox b)
462 => Pipe m a b -> Array a -> m (Array b)
463runPipe f arr = P.runPipe (toArrayMinChunk (length arr)) $ f (A.read arr)
466-- XXX For transformations that cannot change the number of elements e.g. "map"
467-- we can use a predetermined array length.
469-- | Transform an array into another array using a stream transformation
473{-# INLINE streamTransform #-}
474streamTransform :: forall m a b. (MonadIO m, Unbox a, Unbox b)
475 => (Stream m a -> Stream m b) -> Array a -> m (Array b)
476streamTransform f arr =
477 Stream.fold (A.writeWith (length arr)) $ f (A.read arr)
479-------------------------------------------------------------------------------
481-------------------------------------------------------------------------------
483-- | Cast an array having elements of type @a@ into an array having elements of
484-- type @b@. The array size must be a multiple of the size of type @b@
485-- otherwise accessing the last element of the array may result into a crash or
495castUnsafe (Array contents start end) =
496 Array contents start end
498-- | Cast an @Array a@ into an @Array Word8@.
501asBytes :: Array a -> Array Word8
504-- | Cast an array having elements of type @a@ into an array having elements of
505-- type @b@. The length of the array should be a multiple of the size of the
506-- target element otherwise 'Nothing' is returned.
509cast :: forall a b. (Unbox b) => Array a -> Maybe (Array b)
511 let len = A.byteLength arr
512 r = len `mod` SIZE_OF(b)
515 else Just $ castUnsafe arr
517-- | Convert an array of any type into a null terminated CString Ptr. If the
518-- array is unpinned it is first converted to a pinned array which requires a
523-- /O(n) Time: (creates a copy of the array)/
527asCStringUnsafe :: Array a -> (CString -> IO b) -> IO b
528asCStringUnsafe arr act = do
529 let arr1 = asBytes arr <> A.fromList [0]
530 -- asPtrUnsafe makes sure the array is pinned
531 asPtrUnsafe arr1 $ \ptr -> act (castPtr ptr)
533-------------------------------------------------------------------------------
535-------------------------------------------------------------------------------
537-- XXX We can directly use toStreamD and D.fold here.
539-- | Fold an array using a 'Fold'.
543fold :: forall m a b. (Monad m, Unbox a) => Fold m a b -> Array a -> m b
544fold f arr = Stream.fold f (A.read arr)
546-- | Fold an array using a stream fold operation.
549{-# INLINE streamFold #-}
550streamFold :: (Monad m, Unbox a) => (Stream m a -> m b) -> Array a -> m b
551streamFold f arr = f (A.read arr)
553--------------------------------------------------------------------------------
555--------------------------------------------------------------------------------
557{-# INLINE encodeAs #-}
558encodeAs :: forall a. Serialize a => PinnedState -> a -> Array Word8
561 let len = Serialize.addSizeTo 0 a
562 mbarr <- MBA.newBytesAs ps len
563 off <- Serialize.serializeAt 0 mbarr a
565 pure $ Array mbarr 0 off
567{-# INLINE serialize #-}
568serialize :: Serialize a => a -> Array Word8
569serialize = encodeAs Unpinned
571-- | Serialize a Haskell type to a pinned byte array. The array is allocated
572-- using pinned memory so that it can be used directly in OS APIs for writing
573-- to file or sending over the network.
574{-# INLINE pinnedSerialize #-}
575pinnedSerialize :: Serialize a => a -> Array Word8
576pinnedSerialize = encodeAs Pinned
578-- | Decode a Haskell type from a byte array containing its serialized
580{-# INLINE deserialize #-}
581deserialize :: Serialize a => Array Word8 -> a
582deserialize arr@(Array {..}) = unsafeInlineIO $ do
583 let lenArr = length arr
585 Serialize.deserializeAt arrStart arrContents (arrStart + lenArr)
586 assertM(off == arrStart + lenArr)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list