2-- Module : Streamly.Internal.Data.Ring
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
64 -- * Fast Byte Comparisons
72#include "ArrayMacros.h"
75import Control.Exception (assert)
76import Control.Monad.IO.Class (MonadIO(..))
77import Data.Word (Word8)
78import Foreign.Storable
79import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
80import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
81import Foreign.Ptr (plusPtr, minusPtr, castPtr)
82import Streamly.Internal.Data.Unbox as Unboxed (Unbox(peekByteIndex))
83import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
84import GHC.Ptr (Ptr(..))
85import Streamly.Internal.Data.MutArray.Type (MutArray)
86import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..), lmap)
87import Streamly.Internal.Data.Stream.Type (Stream)
88import Streamly.Internal.Data.Stream.Step (Step(..))
89import Streamly.Internal.Data.Unfold.Type (Unfold(..))
90import Streamly.Internal.System.IO (unsafeInlineIO)
92import qualified Streamly.Internal.Data.MutArray.Type as MA
93import qualified Streamly.Internal.Data.Array.Type as A
95import Prelude hiding (length, concat, read)
99-- >>> import qualified Streamly.Internal.Data.Ring as Ring
101-- | A ring buffer is a mutable array of fixed size. Initially the array is
102-- empty, with ringStart pointing at the start of allocated memory. We call the
103-- next location to be written in the ring as ringHead. Initially ringHead ==
104-- ringStart. When the first item is added, ringHead points to ringStart +
105-- sizeof item. When the buffer becomes full ringHead would wrap around to
106-- ringStart. When the buffer is full, ringHead always points at the oldest
107-- item in the ring and the newest item added always overwrites the oldest
110-- When using it we should keep in mind that a ringBuffer is a mutable data
111-- structure. We should not leak out references to it for immutable use.
114 { ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
115 , ringBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
118-------------------------------------------------------------------------------
120-------------------------------------------------------------------------------
122-- | Get the first address of the ring as a pointer.
123startOf :: Ring a -> Ptr a
124startOf = unsafeForeignPtrToPtr . ringStart
126-- | Create a new ringbuffer and return the ring buffer and the ringHead.
127-- Returns the ring and the ringHead, the ringHead is same as ringStart.
129new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
131 let size = count * max 1 (sizeOf (undefined :: a))
132 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
133 let p = unsafeForeignPtrToPtr fptr
136 , ringBound = p `plusPtr` size
139-- XXX Rename this to "new".
141-- | @newRing count@ allocates an empty array that can hold 'count' items. The
142-- memory of the array is uninitialized and the allocation is aligned as per
143-- the 'Storable' instance of the type.
146{-# INLINE newRing #-}
147newRing :: Int -> m (Ring a)
150-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
152{-# INLINE advance #-}
153advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
154advance Ring{..} ringHead =
155 let ptr = PTR_NEXT(ringHead,a)
156 in if ptr < ringBound
158 else unsafeForeignPtrToPtr ringStart
160-- | Move the ringHead by n items. The direction depends on the sign on whether
161-- n is positive or negative. Wrap around if we hit the beginning or end of the
164moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
165moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead
169 elemSize = STORABLE_SIZE_OF(a)
170 ringStartPtr = unsafeForeignPtrToPtr ringStart
171 lenInBytes = ringBound `minusPtr` ringStartPtr
172 offInBytes = ringHead `minusPtr` ringStartPtr
173 len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize
174 off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize
175 advanceFromHead = (off + by `mod` len) * elemSize
177-- XXX Move the writeLastN from array module here.
179-- | @writeN n@ is a rolling fold that keeps the last n elements of the stream
184writeN :: -- (Storable a, MonadIO m) =>
185 Int -> Fold m a (Ring a)
188-------------------------------------------------------------------------------
190-------------------------------------------------------------------------------
192-- | Cast a mutable array to a ring array.
193fromArray :: MutArray a -> Ring a
196-------------------------------------------------------------------------------
197-- Conversion to/from array
198-------------------------------------------------------------------------------
200-- | Modify a given index of a ring array using a modifier function.
203modifyIndex :: -- forall m a b. (MonadIO m, Storable a) =>
204 Ring a -> Int -> (a -> (a, b)) -> m b
205modifyIndex = undefined
207-- | /O(1)/ Write the given element at the given index in the ring array.
208-- Performs in-place mutation of the array.
210-- >>> putIndex arr ix val = Ring.modifyIndex arr ix (const (val, ()))
213{-# INLINE putIndex #-}
214putIndex :: -- (MonadIO m, Storable a) =>
215 Ring a -> Int -> a -> m ()
218-- | Insert an item at the head of the ring, when the ring is full this
219-- replaces the oldest item in the ring with the new item. This is unsafe
220-- beause ringHead supplied is not verified to be within the Ring. Also,
221-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
222{-# INLINE unsafeInsert #-}
223unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
224unsafeInsert rb ringHead newVal = do
226 -- touchForeignPtr (ringStart rb)
227 return $ advance rb ringHead
229-- | Insert an item at the head of the ring, when the ring is full this
230-- replaces the oldest item in the ring with the new item.
233slide :: -- forall m a. (MonadIO m, Storable a) =>
234 Ring a -> a -> m (Ring a)
237-------------------------------------------------------------------------------
239-------------------------------------------------------------------------------
241-- | Return the element at the specified index without checking the bounds.
243-- Unsafe because it does not check the bounds of the ring array.
244{-# INLINE_NORMAL getIndexUnsafe #-}
245getIndexUnsafe :: -- forall m a. (MonadIO m, Storable a) =>
247getIndexUnsafe = undefined
249-- | /O(1)/ Lookup the element at the given index. Index starts from 0.
251{-# INLINE getIndex #-}
252getIndex :: -- (MonadIO m, Storable a) =>
256-- | /O(1)/ Lookup the element at the given index from the end of the array.
257-- Index starts from 0.
259-- Slightly faster than computing the forward index and using getIndex.
261{-# INLINE getIndexRev #-}
262getIndexRev :: -- (MonadIO m, Storable a) =>
264getIndexRev = undefined
266-------------------------------------------------------------------------------
268-------------------------------------------------------------------------------
270-- | /O(1)/ Get the byte length of the array.
273{-# INLINE byteLength #-}
274byteLength :: Ring a -> Int
275byteLength = undefined
277-- | /O(1)/ Get the length of the array i.e. the number of elements in the
280-- Note that 'byteLength' is less expensive than this operation, as 'length'
281-- involves a costly division operation.
285length :: -- forall a. Storable a =>
289-- | Get the total capacity of an array. An array may have space reserved
290-- beyond the current used length of the array.
293{-# INLINE byteCapacity #-}
294byteCapacity :: Ring a -> Int
295byteCapacity = undefined
297-- | The remaining capacity in the array for appending more elements without
301{-# INLINE bytesFree #-}
302bytesFree :: Ring a -> Int
305-------------------------------------------------------------------------------
307-------------------------------------------------------------------------------
309-- XXX We can read the ring in a loop and use "take" to restrict the number of
310-- elements to be taken.
312-- | Read n elements from the ring starting at the supplied ring head. If n is
313-- more than the ring size it keeps reading the ring in a circular fashion.
315-- If the ring is not full the user must ensure than n is less than or equal to
316-- the number of valid elements in the ring.
319{-# INLINE_NORMAL read #-}
320read :: forall m a. (MonadIO m, Storable a) => Unfold m (Ring a, Ptr a, Int) a
321read = Unfold step return
325 step (rb, rh, n) = do
328 liftIO $ touchForeignPtr (ringStart rb)
331 x <- liftIO $ peek rh
332 let rh1 = advance rb rh
333 return $ Yield x (rb, rh1, n - 1)
335-- | Unfold a ring array into a stream in reverse order.
338{-# INLINE_NORMAL readRev #-}
339readRev :: -- forall m a. (MonadIO m, Storable a) =>
340 Unfold m (MutArray a) a
343-------------------------------------------------------------------------------
345-------------------------------------------------------------------------------
347-- XXX Move this module to a lower level Ring/Type module and move ringsOf to a
348-- higher level ring module where we can import "scan".
350-- | @ringsOf n stream@ groups the input stream into a stream of
351-- ring arrays of size n. Each ring is a sliding window of size n.
354{-# INLINE_NORMAL ringsOf #-}
355ringsOf :: -- forall m a. (MonadIO m, Storable a) =>
356 Int -> Stream m a -> Stream m (MutArray a)
357ringsOf = undefined -- Stream.scan (writeN n)
359-------------------------------------------------------------------------------
361-------------------------------------------------------------------------------
363-- | Cast an array having elements of type @a@ into an array having elements of
364-- type @b@. The array size must be a multiple of the size of type @b@.
368castUnsafe :: Ring a -> Ring b
369castUnsafe = undefined
371-- | Cast an @Array a@ into an @Array Word8@.
375asBytes :: Ring a -> Ring Word8
378-- | Cast an array having elements of type @a@ into an array having elements of
379-- type @b@. The length of the array should be a multiple of the size of the
380-- target element otherwise 'Nothing' is returned.
384cast :: forall a b. Storable b => Ring a -> Maybe (Ring b)
386 let len = byteLength arr
387 r = len `mod` STORABLE_SIZE_OF(b)
390 else Just $ castUnsafe arr
392-------------------------------------------------------------------------------
394-------------------------------------------------------------------------------
396-- XXX remove all usage of unsafeInlineIO
398-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
399-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
401{-# INLINE unsafeEqArrayN #-}
402unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
403unsafeEqArrayN Ring{..} rh A.Array{..} nBytes
404 | nBytes < 0 = error "unsafeEqArrayN: n should be >= 0"
406 | otherwise = unsafeInlineIO $ check (castPtr rh) 0
410 w8Contents = arrContents
413 (relem :: Word8) <- peek p
414 aelem <- peekByteIndex i w8Contents
416 then go (p `plusPtr` 1) (i + 1)
420 | i == nBytes = return True
421 | castPtr p == ringBound =
422 go (castPtr (unsafeForeignPtrToPtr ringStart)) i
423 | castPtr p == rh = touchForeignPtr ringStart >> return True
424 | otherwise = check p i
426-- XXX This is not modular. We should probably just convert the array and the
427-- ring buffer to streams and compare the two streams. Need to check perf
430-- | Byte compare the entire length of ringBuffer with the given array,
431-- starting at the supplied ringHead pointer. Returns true if the Array and
432-- the ringBuffer have identical contents.
434-- This is unsafe because the ringHead Ptr is not checked to be in range. The
435-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
437{-# INLINE unsafeEqArray #-}
438unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
439unsafeEqArray Ring{..} rh A.Array{..} =
440 unsafeInlineIO $ check (castPtr rh) 0
444 w8Contents = arrContents
447 (relem :: Word8) <- peek p
448 aelem <- peekByteIndex i w8Contents
450 then go (p `plusPtr` 1) (i + 1)
455 ringBound = go (castPtr (unsafeForeignPtrToPtr ringStart)) i
456 | castPtr p == rh = touchForeignPtr ringStart >> return True
457 | otherwise = check p i
459-------------------------------------------------------------------------------
461-------------------------------------------------------------------------------
463-- XXX We can unfold it into a stream and fold the stream instead.
466-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
467-- step function. This is useful to fold the items in the ring when the ring is
468-- not full. The supplied pointer is usually the end of the ring.
470-- Unsafe because the supplied Ptr is not checked to be in range.
471{-# INLINE unsafeFoldRing #-}
472unsafeFoldRing :: forall a b. Storable a
473 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
474unsafeFoldRing ptr f z Ring{..} =
475 let !res = unsafeInlineIO $ withForeignPtr ringStart $ \p ->
480 | p == q = return acc
483 go (f acc x) (PTR_NEXT(p,a)) q
485-- XXX Can we remove MonadIO here?
486withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
487withForeignPtrM fp fn = do
488 r <- fn $ unsafeForeignPtrToPtr fp
489 liftIO $ touchForeignPtr fp
492-- | Like unsafeFoldRing but with a monadic step function.
493{-# INLINE unsafeFoldRingM #-}
494unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
495 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
496unsafeFoldRingM ptr f z Ring {..} =
497 withForeignPtrM ringStart $ \x -> go z x ptr
500 | start == end = return acc
502 let !x = unsafeInlineIO $ peek start
504 go acc1 (PTR_NEXT(start,a)) end
506-- | Fold the entire length of a ring buffer starting at the supplied ringHead
507-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
508-- this would fold the ring starting from the oldest item to the newest item in
511-- Note, this will crash on ring of 0 size.
513{-# INLINE unsafeFoldRingFullM #-}
514unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
515 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
516unsafeFoldRingFullM rh f z rb@Ring {..} =
517 withForeignPtrM ringStart $ \_ -> go z rh
520 let !x = unsafeInlineIO $ peek start
522 let ptr = advance rb start
527-- | Fold @Int@ items in the ring starting at @Ptr a@. Won't fold more
528-- than the length of the ring.
530-- Note, this will crash on ring of 0 size.
532{-# INLINE unsafeFoldRingNM #-}
533unsafeFoldRingNM :: forall m a b. (MonadIO m, Storable a)
534 => Int -> Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
535unsafeFoldRingNM count rh f z rb@Ring {..} =
536 withForeignPtrM ringStart $ \_ -> go count z rh
540 go 0 acc _ = return acc
541 go !n !acc !start = do
542 let !x = unsafeInlineIO $ peek start
544 let ptr = advance rb start
545 if ptr == rh || n == 0
547 else go (n - 1) acc' ptr
549data Tuple4' a b c d = Tuple4' !a !b !c !d deriving Show
551-- | Like slidingWindow but also provides the entire ring contents as an Array.
552-- The array reflects the state of the ring after inserting the incoming
555-- IMPORTANT NOTE: The ring is mutable, therefore, the result of @(m (Array
556-- a))@ action depends on when it is executed. It does not capture the sanpshot
557-- of the ring at a particular time.
558{-# INLINE slidingWindowWith #-}
559slidingWindowWith :: forall m a b. (MonadIO m, Storable a, Unbox a)
560 => Int -> Fold m ((a, Maybe a), m (MutArray a)) b -> Fold m a b
561slidingWindowWith n (Fold step1 initial1 extract1) = Fold step initial extract
567 then error "Window size must be > 0"
570 (rb, rh) <- liftIO $ new n
573 Partial s -> Partial $ Tuple4' rb rh (0 :: Int) s
576 toArray foldRing rb rh = do
577 -- Using unpinned array here instead of pinned
578 arr <- liftIO $ MA.new n
579 let snoc' b a = liftIO $ MA.snocUnsafe b a
580 foldRing rh snoc' arr rb
582 step (Tuple4' rb rh i st) a
584 rh1 <- liftIO $ unsafeInsert rb rh a
585 liftIO $ touchForeignPtr (ringStart rb)
586 let action = toArray unsafeFoldRingM rb (PTR_NEXT(rh, a))
587 r <- step1 st ((a, Nothing), action)
590 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
593 old <- liftIO $ peek rh
594 rh1 <- liftIO $ unsafeInsert rb rh a
595 liftIO $ touchForeignPtr (ringStart rb)
596 r <- step1 st ((a, Just old), toArray unsafeFoldRingFullM rb rh1)
599 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
602 extract (Tuple4' _ _ _ st) = extract1 st
604-- | @slidingWindow collector@ is an incremental sliding window
605-- fold that does not require all the intermediate elements in a computation.
606-- This maintains @n@ elements in the window, when a new element comes it slides
607-- out the oldest element and the new element along with the old element are
608-- supplied to the collector fold.
610-- The 'Maybe' type is for the case when initially the window is filling and
611-- there is no old element.
613{-# INLINE slidingWindow #-}
614slidingWindow :: forall m a b. (MonadIO m, Storable a, Unbox a)
615 => Int -> Fold m (a, Maybe a) b -> Fold m a b
616slidingWindow n f = slidingWindowWith n (lmap fst f)
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