2-- Module : Streamly.Internal.Data.Ring.Foreign
3-- Copyright : (c) 2019 Composewell Technologies
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- A ring array is a circular mutable array.
11-- XXX Write benchmarks
12-- XXX Make the implementation similar to mutable array
13-- XXX Rename this module to Data.RingArray.Storable
15module Streamly.Internal.Data.Ring.Foreign
64 -- * Fast Byte Comparisons
73import Control.Exception (assert)
74import Control.Monad.IO.Class (MonadIO(..))
75import Data.Word (Word8)
76import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
77import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
78import Foreign.Ptr (plusPtr, minusPtr, castPtr)
79import Foreign.Storable (Storable(..))
80import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
81import GHC.Ptr (Ptr(..))
82import Streamly.Internal.Data.Array.Foreign.Mut.Type (Array, memcmp)
83import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..))
84import Streamly.Internal.Data.Stream.Serial (SerialT(..))
85import Streamly.Internal.Data.Unfold.Type (Unfold(..))
86import Streamly.Internal.System.IO (unsafeInlineIO)
88import qualified Streamly.Internal.Data.Array.Foreign.Type as A
90import Prelude hiding (length, concat, read)
94-- >>> import qualified Streamly.Internal.Data.Ring.Foreign as Ring
96-- | A ring buffer is a mutable array of fixed size. Initially the array is
97-- empty, with ringStart pointing at the start of allocated memory. We call the
98-- next location to be written in the ring as ringHead. Initially ringHead ==
99-- ringStart. When the first item is added, ringHead points to ringStart +
100-- sizeof item. When the buffer becomes full ringHead would wrap around to
101-- ringStart. When the buffer is full, ringHead always points at the oldest
102-- item in the ring and the newest item added always overwrites the oldest
105-- When using it we should keep in mind that a ringBuffer is a mutable data
106-- structure. We should not leak out references to it for immutable use.
109 { ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
110 , ringBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
113-------------------------------------------------------------------------------
115-------------------------------------------------------------------------------
117-- | Get the first address of the ring as a pointer.
118startOf :: Ring a -> Ptr a
119startOf = unsafeForeignPtrToPtr . ringStart
121-- | Create a new ringbuffer and return the ring buffer and the ringHead.
122-- Returns the ring and the ringHead, the ringHead is same as ringStart.
124new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
126 let size = count * sizeOf (undefined :: a)
127 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
128 let p = unsafeForeignPtrToPtr fptr
131 , ringBound = p `plusPtr` size
134-- XXX Rename this to "new".
136-- | @newRing count@ allocates an empty array that can hold 'count' items. The
137-- memory of the array is uninitialized and the allocation is aligned as per
138-- the 'Storable' instance of the type.
141{-# INLINE newRing #-}
142newRing :: Int -> m (Ring a)
145-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
147{-# INLINE advance #-}
148advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
149advance Ring{..} ringHead =
150 let ptr = ringHead `plusPtr` sizeOf (undefined :: a)
151 in if ptr < ringBound
153 else unsafeForeignPtrToPtr ringStart
155-- | Move the ringHead by n items. The direction depends on the sign on whether
156-- n is positive or negative. Wrap around if we hit the beginning or end of the
159moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
160moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead
164 elemSize = sizeOf (undefined :: a)
165 ringStartPtr = unsafeForeignPtrToPtr ringStart
166 lenInBytes = ringBound `minusPtr` ringStartPtr
167 offInBytes = ringHead `minusPtr` ringStartPtr
168 len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize
169 off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize
170 advanceFromHead = (off + by `mod` len) * elemSize
172-- XXX Move the writeLastN from array module here.
174-- | @writeN n@ is a rolling fold that keeps the last n elements of the stream
179writeN :: -- (Storable a, MonadIO m) =>
180 Int -> Fold m a (Ring a)
183-------------------------------------------------------------------------------
185-------------------------------------------------------------------------------
187-- | Cast a mutable array to a ring array.
188fromArray :: Array a -> Ring a
191-------------------------------------------------------------------------------
192-- Conversion to/from array
193-------------------------------------------------------------------------------
195-- | Modify a given index of a ring array using a modifier function.
198modifyIndex :: -- forall m a b. (MonadIO m, Storable a) =>
199 Ring a -> Int -> (a -> (a, b)) -> m b
200modifyIndex = undefined
202-- | /O(1)/ Write the given element at the given index in the ring array.
203-- Performs in-place mutation of the array.
205-- >>> putIndex arr ix val = Ring.modifyIndex arr ix (const (val, ()))
208{-# INLINE putIndex #-}
209putIndex :: -- (MonadIO m, Storable a) =>
210 Ring a -> Int -> a -> m ()
213-- | Insert an item at the head of the ring, when the ring is full this
214-- replaces the oldest item in the ring with the new item. This is unsafe
215-- beause ringHead supplied is not verified to be within the Ring. Also,
216-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
217{-# INLINE unsafeInsert #-}
218unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
219unsafeInsert rb ringHead newVal = do
221 -- touchForeignPtr (ringStart rb)
222 return $ advance rb ringHead
224-- | Insert an item at the head of the ring, when the ring is full this
225-- replaces the oldest item in the ring with the new item.
228slide :: -- forall m a. (MonadIO m, Storable a) =>
229 Ring a -> a -> m (Ring a)
232-------------------------------------------------------------------------------
234-------------------------------------------------------------------------------
236-- | Return the element at the specified index without checking the bounds.
238-- Unsafe because it does not check the bounds of the ring array.
239{-# INLINE_NORMAL getIndexUnsafe #-}
240getIndexUnsafe :: -- forall m a. (MonadIO m, Storable a) =>
242getIndexUnsafe = undefined
244-- | /O(1)/ Lookup the element at the given index. Index starts from 0.
246{-# INLINE getIndex #-}
247getIndex :: -- (MonadIO m, Storable a) =>
251-- | /O(1)/ Lookup the element at the given index from the end of the array.
252-- Index starts from 0.
254-- Slightly faster than computing the forward index and using getIndex.
256{-# INLINE getIndexRev #-}
257getIndexRev :: -- (MonadIO m, Storable a) =>
259getIndexRev = undefined
261-------------------------------------------------------------------------------
263-------------------------------------------------------------------------------
265-- | /O(1)/ Get the byte length of the array.
268{-# INLINE byteLength #-}
269byteLength :: Ring a -> Int
270byteLength = undefined
272-- | /O(1)/ Get the length of the array i.e. the number of elements in the
275-- Note that 'byteLength' is less expensive than this operation, as 'length'
276-- involves a costly division operation.
280length :: -- forall a. Storable a =>
284-- | Get the total capacity of an array. An array may have space reserved
285-- beyond the current used length of the array.
288{-# INLINE byteCapacity #-}
289byteCapacity :: Ring a -> Int
290byteCapacity = undefined
292-- | The remaining capacity in the array for appending more elements without
296{-# INLINE bytesFree #-}
297bytesFree :: Ring a -> Int
300-------------------------------------------------------------------------------
302-------------------------------------------------------------------------------
304-- | Unfold a ring array into a stream.
307{-# INLINE_NORMAL read #-}
308read :: -- forall m a. (MonadIO m, Storable a) =>
312-- | Unfold a ring array into a stream in reverse order.
315{-# INLINE_NORMAL readRev #-}
316readRev :: -- forall m a. (MonadIO m, Storable a) =>
320-------------------------------------------------------------------------------
322-------------------------------------------------------------------------------
324-- XXX Move this module to a lower level Ring/Type module and move ringsOf to a
325-- higher level ring module where we can import "scan".
327-- | @ringsOf n stream@ groups the input stream into a stream of
328-- ring arrays of size n. Each ring is a sliding window of size n.
331{-# INLINE_NORMAL ringsOf #-}
332ringsOf :: -- forall m a. (MonadIO m, Storable a) =>
333 Int -> SerialT m a -> SerialT m (Array a)
334ringsOf = undefined -- Stream.scan (writeN n)
336-------------------------------------------------------------------------------
338-------------------------------------------------------------------------------
340-- | Cast an array having elements of type @a@ into an array having elements of
341-- type @b@. The array size must be a multiple of the size of type @b@.
345castUnsafe :: Ring a -> Ring b
346castUnsafe = undefined
348-- | Cast an @Array a@ into an @Array Word8@.
352asBytes :: Ring a -> Ring Word8
355-- | Cast an array having elements of type @a@ into an array having elements of
356-- type @b@. The length of the array should be a multiple of the size of the
357-- target element otherwise 'Nothing' is returned.
361cast :: forall a b. Storable b => Ring a -> Maybe (Ring b)
363 let len = byteLength arr
364 r = len `mod` sizeOf (undefined :: b)
367 else Just $ castUnsafe arr
369-------------------------------------------------------------------------------
371-------------------------------------------------------------------------------
373-- XXX remove all usage of unsafeInlineIO
375-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
376-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
378{-# INLINE unsafeEqArrayN #-}
379unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
380unsafeEqArrayN Ring{..} rh A.Array{..} n =
381 let !res = unsafeInlineIO $ do
382 let rs = unsafeForeignPtrToPtr ringStart
384 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())
385 let len = ringBound `minusPtr` rh
386 r1 <- memcmp (castPtr rh) (castPtr as) (min len n)
388 then memcmp (castPtr rs) (castPtr (as `plusPtr` len))
389 (min (rh `minusPtr` rs) (n - len))
391 -- XXX enable these, check perf impact
392 -- touchForeignPtr ringStart
393 -- touchForeignPtr aStart
397-- | Byte compare the entire length of ringBuffer with the given array,
398-- starting at the supplied ringHead pointer. Returns true if the Array and
399-- the ringBuffer have identical contents.
401-- This is unsafe because the ringHead Ptr is not checked to be in range. The
402-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
404{-# INLINE unsafeEqArray #-}
405unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
406unsafeEqArray Ring{..} rh A.Array{..} =
407 let !res = unsafeInlineIO $ do
408 let rs = unsafeForeignPtrToPtr ringStart
410 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)
412 let len = ringBound `minusPtr` rh
413 r1 <- memcmp (castPtr rh) (castPtr as) len
414 r2 <- memcmp (castPtr rs) (castPtr (as `plusPtr` len))
416 -- XXX enable these, check perf impact
417 -- touchForeignPtr ringStart
418 -- touchForeignPtr aStart
422-------------------------------------------------------------------------------
424-------------------------------------------------------------------------------
426-- XXX We can unfold it into a stream and fold the stream instead.
429-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
430-- step function. This is useful to fold the items in the ring when the ring is
431-- not full. The supplied pointer is usually the end of the ring.
433-- Unsafe because the supplied Ptr is not checked to be in range.
434{-# INLINE unsafeFoldRing #-}
435unsafeFoldRing :: forall a b. Storable a
436 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
437unsafeFoldRing ptr f z Ring{..} =
438 let !res = unsafeInlineIO $ withForeignPtr ringStart $ \p ->
443 | p == q = return acc
446 go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q
448-- XXX Can we remove MonadIO here?
449withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
450withForeignPtrM fp fn = do
451 r <- fn $ unsafeForeignPtrToPtr fp
452 liftIO $ touchForeignPtr fp
455-- | Like unsafeFoldRing but with a monadic step function.
456{-# INLINE unsafeFoldRingM #-}
457unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
458 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
459unsafeFoldRingM ptr f z Ring {..} =
460 withForeignPtrM ringStart $ \x -> go z x ptr
463 | start == end = return acc
465 let !x = unsafeInlineIO $ peek start
467 go acc' (start `plusPtr` sizeOf (undefined :: a)) end
469-- | Fold the entire length of a ring buffer starting at the supplied ringHead
470-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
471-- this would fold the ring starting from the oldest item to the newest item in
474-- Note, this will crash on ring of 0 size.
476{-# INLINE unsafeFoldRingFullM #-}
477unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
478 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
479unsafeFoldRingFullM rh f z rb@Ring {..} =
480 withForeignPtrM ringStart $ \_ -> go z rh
483 let !x = unsafeInlineIO $ peek start
485 let ptr = advance rb start
490-- | Fold @Int@ items in the ring starting at @Ptr a@. Won't fold more
491-- than the length of the ring.
493-- Note, this will crash on ring of 0 size.
495{-# INLINE unsafeFoldRingNM #-}
496unsafeFoldRingNM :: forall m a b. (MonadIO m, Storable a)
497 => Int -> Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
498unsafeFoldRingNM count rh f z rb@Ring {..} =
499 withForeignPtrM ringStart $ \_ -> go count z rh
503 go 0 acc _ = return acc
504 go !n !acc !start = do
505 let !x = unsafeInlineIO $ peek start
507 let ptr = advance rb start
508 if ptr == rh || n == 0
510 else go (n - 1) acc' ptr
512data Tuple4' a b c d = Tuple4' !a !b !c !d deriving Show
514-- | @slidingWindow collector@ is an incremental sliding window
515-- fold that does not require all the intermediate elements in a computation.
516-- This maintains n elements in the window, when a new element comes it slides
517-- out the oldest element and the new element along with the old element are
518-- supplied to the collector fold.
520-- The Maybe is for the case when initially the window is filling and
521-- there is no old element.
523{-# INLINE slidingWindow #-}
524slidingWindow :: forall m a b. (MonadIO m, Storable a)
525 => Int -> Fold m (a, Maybe a) b -> Fold m a b
526slidingWindow n (Fold step1 initial1 extract1)= Fold step initial extract
532 (rb, rh) <- liftIO $ new n
535 Partial s -> Partial $ Tuple4' rb rh (0 :: Int) s
538 step (Tuple4' rb rh i st) a
540 rh1 <- liftIO $ unsafeInsert rb rh a
541 liftIO $ touchForeignPtr (ringStart rb)
542 r <- step1 st (a, Nothing)
545 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
548 old <- liftIO $ peek rh
549 rh1 <- liftIO $ unsafeInsert rb rh a
550 liftIO $ touchForeignPtr (ringStart rb)
551 r <- step1 st (a, Just old)
554 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
557 extract (Tuple4' _ _ _ st) = extract1 st
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