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
71#include "ArrayMacros.h"
74import Control.Exception (assert)
75import Control.Monad.IO.Class (MonadIO(..))
76import Data.Word (Word8)
77import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
78import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
79import Foreign.Ptr (plusPtr, minusPtr, castPtr)
80import Foreign.Storable (Storable(..))
81import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
82import GHC.Ptr (Ptr(..))
83import Streamly.Internal.Data.Array.Foreign.Mut.Type (Array, memcmp)
84import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..))
85import Streamly.Internal.Data.Stream.Serial (SerialT(..))
86import Streamly.Internal.Data.Unfold.Type (Unfold(..))
87import Streamly.Internal.System.IO (unsafeInlineIO)
89import qualified Streamly.Internal.Data.Array.Foreign.Type as A
91import Prelude hiding (length, concat, read)
95-- >>> import qualified Streamly.Internal.Data.Ring.Foreign as Ring
97-- | A ring buffer is a mutable array of fixed size. Initially the array is
98-- empty, with ringStart pointing at the start of allocated memory. We call the
99-- next location to be written in the ring as ringHead. Initially ringHead ==
100-- ringStart. When the first item is added, ringHead points to ringStart +
101-- sizeof item. When the buffer becomes full ringHead would wrap around to
102-- ringStart. When the buffer is full, ringHead always points at the oldest
103-- item in the ring and the newest item added always overwrites the oldest
106-- When using it we should keep in mind that a ringBuffer is a mutable data
107-- structure. We should not leak out references to it for immutable use.
110 { ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
111 , ringBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
114-------------------------------------------------------------------------------
116-------------------------------------------------------------------------------
118-- | Get the first address of the ring as a pointer.
119startOf :: Ring a -> Ptr a
120startOf = unsafeForeignPtrToPtr . ringStart
122-- | Create a new ringbuffer and return the ring buffer and the ringHead.
123-- Returns the ring and the ringHead, the ringHead is same as ringStart.
125new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
127 let size = count * SIZE_OF(a)
128 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
129 let p = unsafeForeignPtrToPtr fptr
132 , ringBound = p `plusPtr` size
135-- XXX Rename this to "new".
137-- | @newRing count@ allocates an empty array that can hold 'count' items. The
138-- memory of the array is uninitialized and the allocation is aligned as per
139-- the 'Storable' instance of the type.
142{-# INLINE newRing #-}
143newRing :: Int -> m (Ring a)
146-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
148{-# INLINE advance #-}
149advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
150advance Ring{..} ringHead =
151 let ptr = PTR_NEXT(ringHead,a)
152 in if ptr < ringBound
154 else unsafeForeignPtrToPtr ringStart
156-- | Move the ringHead by n items. The direction depends on the sign on whether
157-- n is positive or negative. Wrap around if we hit the beginning or end of the
160moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
161moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead
165 elemSize = SIZE_OF(a)
166 ringStartPtr = unsafeForeignPtrToPtr ringStart
167 lenInBytes = ringBound `minusPtr` ringStartPtr
168 offInBytes = ringHead `minusPtr` ringStartPtr
169 len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize
170 off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize
171 advanceFromHead = (off + by `mod` len) * elemSize
173-- XXX Move the writeLastN from array module here.
175-- | @writeN n@ is a rolling fold that keeps the last n elements of the stream
180writeN :: -- (Storable a, MonadIO m) =>
181 Int -> Fold m a (Ring a)
184-------------------------------------------------------------------------------
186-------------------------------------------------------------------------------
188-- | Cast a mutable array to a ring array.
189fromArray :: Array a -> Ring a
192-------------------------------------------------------------------------------
193-- Conversion to/from array
194-------------------------------------------------------------------------------
196-- | Modify a given index of a ring array using a modifier function.
199modifyIndex :: -- forall m a b. (MonadIO m, Storable a) =>
200 Ring a -> Int -> (a -> (a, b)) -> m b
201modifyIndex = undefined
203-- | /O(1)/ Write the given element at the given index in the ring array.
204-- Performs in-place mutation of the array.
206-- >>> putIndex arr ix val = Ring.modifyIndex arr ix (const (val, ()))
209{-# INLINE putIndex #-}
210putIndex :: -- (MonadIO m, Storable a) =>
211 Ring a -> Int -> a -> m ()
214-- | Insert an item at the head of the ring, when the ring is full this
215-- replaces the oldest item in the ring with the new item. This is unsafe
216-- beause ringHead supplied is not verified to be within the Ring. Also,
217-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
218{-# INLINE unsafeInsert #-}
219unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
220unsafeInsert rb ringHead newVal = do
222 -- touchForeignPtr (ringStart rb)
223 return $ advance rb ringHead
225-- | Insert an item at the head of the ring, when the ring is full this
226-- replaces the oldest item in the ring with the new item.
229slide :: -- forall m a. (MonadIO m, Storable a) =>
230 Ring a -> a -> m (Ring a)
233-------------------------------------------------------------------------------
235-------------------------------------------------------------------------------
237-- | Return the element at the specified index without checking the bounds.
239-- Unsafe because it does not check the bounds of the ring array.
240{-# INLINE_NORMAL getIndexUnsafe #-}
241getIndexUnsafe :: -- forall m a. (MonadIO m, Storable a) =>
243getIndexUnsafe = undefined
245-- | /O(1)/ Lookup the element at the given index. Index starts from 0.
247{-# INLINE getIndex #-}
248getIndex :: -- (MonadIO m, Storable a) =>
252-- | /O(1)/ Lookup the element at the given index from the end of the array.
253-- Index starts from 0.
255-- Slightly faster than computing the forward index and using getIndex.
257{-# INLINE getIndexRev #-}
258getIndexRev :: -- (MonadIO m, Storable a) =>
260getIndexRev = undefined
262-------------------------------------------------------------------------------
264-------------------------------------------------------------------------------
266-- | /O(1)/ Get the byte length of the array.
269{-# INLINE byteLength #-}
270byteLength :: Ring a -> Int
271byteLength = undefined
273-- | /O(1)/ Get the length of the array i.e. the number of elements in the
276-- Note that 'byteLength' is less expensive than this operation, as 'length'
277-- involves a costly division operation.
281length :: -- forall a. Storable a =>
285-- | Get the total capacity of an array. An array may have space reserved
286-- beyond the current used length of the array.
289{-# INLINE byteCapacity #-}
290byteCapacity :: Ring a -> Int
291byteCapacity = undefined
293-- | The remaining capacity in the array for appending more elements without
297{-# INLINE bytesFree #-}
298bytesFree :: Ring a -> Int
301-------------------------------------------------------------------------------
303-------------------------------------------------------------------------------
305-- | Unfold a ring array into a stream.
308{-# INLINE_NORMAL read #-}
309read :: -- forall m a. (MonadIO m, Storable a) =>
313-- | Unfold a ring array into a stream in reverse order.
316{-# INLINE_NORMAL readRev #-}
317readRev :: -- forall m a. (MonadIO m, Storable a) =>
321-------------------------------------------------------------------------------
323-------------------------------------------------------------------------------
325-- XXX Move this module to a lower level Ring/Type module and move ringsOf to a
326-- higher level ring module where we can import "scan".
328-- | @ringsOf n stream@ groups the input stream into a stream of
329-- ring arrays of size n. Each ring is a sliding window of size n.
332{-# INLINE_NORMAL ringsOf #-}
333ringsOf :: -- forall m a. (MonadIO m, Storable a) =>
334 Int -> SerialT m a -> SerialT m (Array a)
335ringsOf = undefined -- Stream.scan (writeN n)
337-------------------------------------------------------------------------------
339-------------------------------------------------------------------------------
341-- | Cast an array having elements of type @a@ into an array having elements of
342-- type @b@. The array size must be a multiple of the size of type @b@.
346castUnsafe :: Ring a -> Ring b
347castUnsafe = undefined
349-- | Cast an @Array a@ into an @Array Word8@.
353asBytes :: Ring a -> Ring Word8
356-- | Cast an array having elements of type @a@ into an array having elements of
357-- type @b@. The length of the array should be a multiple of the size of the
358-- target element otherwise 'Nothing' is returned.
362cast :: forall a b. Storable b => Ring a -> Maybe (Ring b)
364 let len = byteLength arr
365 r = len `mod` SIZE_OF(b)
368 else Just $ castUnsafe arr
370-------------------------------------------------------------------------------
372-------------------------------------------------------------------------------
374-- XXX remove all usage of unsafeInlineIO
376-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
377-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
379{-# INLINE unsafeEqArrayN #-}
380unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
381unsafeEqArrayN Ring{..} rh A.Array{..} n =
382 let !res = unsafeInlineIO $ do
383 let rs = unsafeForeignPtrToPtr ringStart
385 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())
386 let len = ringBound `minusPtr` rh
387 r1 <- memcmp (castPtr rh) (castPtr as) (min len n)
389 then memcmp (castPtr rs) (castPtr (as `plusPtr` len))
390 (min (rh `minusPtr` rs) (n - len))
392 -- XXX enable these, check perf impact
393 -- touchForeignPtr ringStart
394 -- touchForeignPtr aStart
398-- | Byte compare the entire length of ringBuffer with the given array,
399-- starting at the supplied ringHead pointer. Returns true if the Array and
400-- the ringBuffer have identical contents.
402-- This is unsafe because the ringHead Ptr is not checked to be in range. The
403-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
405{-# INLINE unsafeEqArray #-}
406unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
407unsafeEqArray Ring{..} rh A.Array{..} =
408 let !res = unsafeInlineIO $ do
409 let rs = unsafeForeignPtrToPtr ringStart
411 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)
413 let len = ringBound `minusPtr` rh
414 r1 <- memcmp (castPtr rh) (castPtr as) len
415 r2 <- memcmp (castPtr rs) (castPtr (as `plusPtr` len))
417 -- XXX enable these, check perf impact
418 -- touchForeignPtr ringStart
419 -- touchForeignPtr aStart
423-------------------------------------------------------------------------------
425-------------------------------------------------------------------------------
427-- XXX We can unfold it into a stream and fold the stream instead.
430-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
431-- step function. This is useful to fold the items in the ring when the ring is
432-- not full. The supplied pointer is usually the end of the ring.
434-- Unsafe because the supplied Ptr is not checked to be in range.
435{-# INLINE unsafeFoldRing #-}
436unsafeFoldRing :: forall a b. Storable a
437 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
438unsafeFoldRing ptr f z Ring{..} =
439 let !res = unsafeInlineIO $ withForeignPtr ringStart $ \p ->
444 | p == q = return acc
447 go (f acc x) (PTR_NEXT(p,a)) q
449-- XXX Can we remove MonadIO here?
450withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
451withForeignPtrM fp fn = do
452 r <- fn $ unsafeForeignPtrToPtr fp
453 liftIO $ touchForeignPtr fp
456-- | Like unsafeFoldRing but with a monadic step function.
457{-# INLINE unsafeFoldRingM #-}
458unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
459 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
460unsafeFoldRingM ptr f z Ring {..} =
461 withForeignPtrM ringStart $ \x -> go z x ptr
464 | start == end = return acc
466 let !x = unsafeInlineIO $ peek start
468 go acc1 (PTR_NEXT(start,a)) end
470-- | Fold the entire length of a ring buffer starting at the supplied ringHead
471-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
472-- this would fold the ring starting from the oldest item to the newest item in
475-- Note, this will crash on ring of 0 size.
477{-# INLINE unsafeFoldRingFullM #-}
478unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
479 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
480unsafeFoldRingFullM rh f z rb@Ring {..} =
481 withForeignPtrM ringStart $ \_ -> go z rh
484 let !x = unsafeInlineIO $ peek start
486 let ptr = advance rb start
491-- | Fold @Int@ items in the ring starting at @Ptr a@. Won't fold more
492-- than the length of the ring.
494-- Note, this will crash on ring of 0 size.
496{-# INLINE unsafeFoldRingNM #-}
497unsafeFoldRingNM :: forall m a b. (MonadIO m, Storable a)
498 => Int -> Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
499unsafeFoldRingNM count rh f z rb@Ring {..} =
500 withForeignPtrM ringStart $ \_ -> go count z rh
504 go 0 acc _ = return acc
505 go !n !acc !start = do
506 let !x = unsafeInlineIO $ peek start
508 let ptr = advance rb start
509 if ptr == rh || n == 0
511 else go (n - 1) acc' ptr
513data Tuple4' a b c d = Tuple4' !a !b !c !d deriving Show
515-- | @slidingWindow collector@ is an incremental sliding window
516-- fold that does not require all the intermediate elements in a computation.
517-- This maintains n elements in the window, when a new element comes it slides
518-- out the oldest element and the new element along with the old element are
519-- supplied to the collector fold.
521-- The Maybe is for the case when initially the window is filling and
522-- there is no old element.
524{-# INLINE slidingWindow #-}
525slidingWindow :: forall m a b. (MonadIO m, Storable a)
526 => Int -> Fold m (a, Maybe a) b -> Fold m a b
527slidingWindow n (Fold step1 initial1 extract1)= Fold step initial extract
533 (rb, rh) <- liftIO $ new n
536 Partial s -> Partial $ Tuple4' rb rh (0 :: Int) s
539 step (Tuple4' rb rh i st) a
541 rh1 <- liftIO $ unsafeInsert rb rh a
542 liftIO $ touchForeignPtr (ringStart rb)
543 r <- step1 st (a, Nothing)
546 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
549 old <- liftIO $ peek rh
550 rh1 <- liftIO $ unsafeInsert rb rh a
551 liftIO $ touchForeignPtr (ringStart rb)
552 r <- step1 st (a, Just old)
555 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
558 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