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
72#include "ArrayMacros.h"
75import Control.Exception (assert)
76import Control.Monad.IO.Class (MonadIO(..))
77import Data.Word (Word8)
78import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
79import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
80import Foreign.Ptr (plusPtr, minusPtr, castPtr)
81import Foreign.Storable (Storable(..))
82import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
83import GHC.Ptr (Ptr(..))
84import Streamly.Internal.Data.Array.Foreign.Mut.Type (Array, memcmp)
85import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..), lmap)
86import Streamly.Internal.Data.Stream.Serial (SerialT(..))
87import Streamly.Internal.Data.Unfold.Type (Unfold(..))
88import Streamly.Internal.System.IO (unsafeInlineIO)
90import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
91import qualified Streamly.Internal.Data.Array.Foreign.Type as A
93import Prelude hiding (length, concat, read)
97-- >>> import qualified Streamly.Internal.Data.Ring.Foreign as Ring
99-- | A ring buffer is a mutable array of fixed size. Initially the array is
100-- empty, with ringStart pointing at the start of allocated memory. We call the
101-- next location to be written in the ring as ringHead. Initially ringHead ==
102-- ringStart. When the first item is added, ringHead points to ringStart +
103-- sizeof item. When the buffer becomes full ringHead would wrap around to
104-- ringStart. When the buffer is full, ringHead always points at the oldest
105-- item in the ring and the newest item added always overwrites the oldest
108-- When using it we should keep in mind that a ringBuffer is a mutable data
109-- structure. We should not leak out references to it for immutable use.
112 { ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
113 , ringBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
116-------------------------------------------------------------------------------
118-------------------------------------------------------------------------------
120-- | Get the first address of the ring as a pointer.
121startOf :: Ring a -> Ptr a
122startOf = unsafeForeignPtrToPtr . ringStart
124-- | Create a new ringbuffer and return the ring buffer and the ringHead.
125-- Returns the ring and the ringHead, the ringHead is same as ringStart.
127new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
129 let size = count * SIZE_OF(a)
130 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
131 let p = unsafeForeignPtrToPtr fptr
134 , ringBound = p `plusPtr` size
137-- XXX Rename this to "new".
139-- | @newRing count@ allocates an empty array that can hold 'count' items. The
140-- memory of the array is uninitialized and the allocation is aligned as per
141-- the 'Storable' instance of the type.
144{-# INLINE newRing #-}
145newRing :: Int -> m (Ring a)
148-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
150{-# INLINE advance #-}
151advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
152advance Ring{..} ringHead =
153 let ptr = PTR_NEXT(ringHead,a)
154 in if ptr < ringBound
156 else unsafeForeignPtrToPtr ringStart
158-- | Move the ringHead by n items. The direction depends on the sign on whether
159-- n is positive or negative. Wrap around if we hit the beginning or end of the
162moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
163moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead
167 elemSize = SIZE_OF(a)
168 ringStartPtr = unsafeForeignPtrToPtr ringStart
169 lenInBytes = ringBound `minusPtr` ringStartPtr
170 offInBytes = ringHead `minusPtr` ringStartPtr
171 len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize
172 off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize
173 advanceFromHead = (off + by `mod` len) * elemSize
175-- XXX Move the writeLastN from array module here.
177-- | @writeN n@ is a rolling fold that keeps the last n elements of the stream
182writeN :: -- (Storable a, MonadIO m) =>
183 Int -> Fold m a (Ring a)
186-------------------------------------------------------------------------------
188-------------------------------------------------------------------------------
190-- | Cast a mutable array to a ring array.
191fromArray :: Array a -> Ring a
194-------------------------------------------------------------------------------
195-- Conversion to/from array
196-------------------------------------------------------------------------------
198-- | Modify a given index of a ring array using a modifier function.
201modifyIndex :: -- forall m a b. (MonadIO m, Storable a) =>
202 Ring a -> Int -> (a -> (a, b)) -> m b
203modifyIndex = undefined
205-- | /O(1)/ Write the given element at the given index in the ring array.
206-- Performs in-place mutation of the array.
208-- >>> putIndex arr ix val = Ring.modifyIndex arr ix (const (val, ()))
211{-# INLINE putIndex #-}
212putIndex :: -- (MonadIO m, Storable a) =>
213 Ring a -> Int -> a -> m ()
216-- | Insert an item at the head of the ring, when the ring is full this
217-- replaces the oldest item in the ring with the new item. This is unsafe
218-- beause ringHead supplied is not verified to be within the Ring. Also,
219-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
220{-# INLINE unsafeInsert #-}
221unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
222unsafeInsert rb ringHead newVal = do
224 -- touchForeignPtr (ringStart rb)
225 return $ advance rb ringHead
227-- | Insert an item at the head of the ring, when the ring is full this
228-- replaces the oldest item in the ring with the new item.
231slide :: -- forall m a. (MonadIO m, Storable a) =>
232 Ring a -> a -> m (Ring a)
235-------------------------------------------------------------------------------
237-------------------------------------------------------------------------------
239-- | Return the element at the specified index without checking the bounds.
241-- Unsafe because it does not check the bounds of the ring array.
242{-# INLINE_NORMAL getIndexUnsafe #-}
243getIndexUnsafe :: -- forall m a. (MonadIO m, Storable a) =>
245getIndexUnsafe = undefined
247-- | /O(1)/ Lookup the element at the given index. Index starts from 0.
249{-# INLINE getIndex #-}
250getIndex :: -- (MonadIO m, Storable a) =>
254-- | /O(1)/ Lookup the element at the given index from the end of the array.
255-- Index starts from 0.
257-- Slightly faster than computing the forward index and using getIndex.
259{-# INLINE getIndexRev #-}
260getIndexRev :: -- (MonadIO m, Storable a) =>
262getIndexRev = undefined
264-------------------------------------------------------------------------------
266-------------------------------------------------------------------------------
268-- | /O(1)/ Get the byte length of the array.
271{-# INLINE byteLength #-}
272byteLength :: Ring a -> Int
273byteLength = undefined
275-- | /O(1)/ Get the length of the array i.e. the number of elements in the
278-- Note that 'byteLength' is less expensive than this operation, as 'length'
279-- involves a costly division operation.
283length :: -- forall a. Storable a =>
287-- | Get the total capacity of an array. An array may have space reserved
288-- beyond the current used length of the array.
291{-# INLINE byteCapacity #-}
292byteCapacity :: Ring a -> Int
293byteCapacity = undefined
295-- | The remaining capacity in the array for appending more elements without
299{-# INLINE bytesFree #-}
300bytesFree :: Ring a -> Int
303-------------------------------------------------------------------------------
305-------------------------------------------------------------------------------
307-- | Unfold a ring array into a stream.
310{-# INLINE_NORMAL read #-}
311read :: -- forall m a. (MonadIO m, Storable a) =>
315-- | Unfold a ring array into a stream in reverse order.
318{-# INLINE_NORMAL readRev #-}
319readRev :: -- forall m a. (MonadIO m, Storable a) =>
323-------------------------------------------------------------------------------
325-------------------------------------------------------------------------------
327-- XXX Move this module to a lower level Ring/Type module and move ringsOf to a
328-- higher level ring module where we can import "scan".
330-- | @ringsOf n stream@ groups the input stream into a stream of
331-- ring arrays of size n. Each ring is a sliding window of size n.
334{-# INLINE_NORMAL ringsOf #-}
335ringsOf :: -- forall m a. (MonadIO m, Storable a) =>
336 Int -> SerialT m a -> SerialT m (Array a)
337ringsOf = undefined -- Stream.scan (writeN n)
339-------------------------------------------------------------------------------
341-------------------------------------------------------------------------------
343-- | Cast an array having elements of type @a@ into an array having elements of
344-- type @b@. The array size must be a multiple of the size of type @b@.
348castUnsafe :: Ring a -> Ring b
349castUnsafe = undefined
351-- | Cast an @Array a@ into an @Array Word8@.
355asBytes :: Ring a -> Ring Word8
358-- | Cast an array having elements of type @a@ into an array having elements of
359-- type @b@. The length of the array should be a multiple of the size of the
360-- target element otherwise 'Nothing' is returned.
364cast :: forall a b. Storable b => Ring a -> Maybe (Ring b)
366 let len = byteLength arr
367 r = len `mod` SIZE_OF(b)
370 else Just $ castUnsafe arr
372-------------------------------------------------------------------------------
374-------------------------------------------------------------------------------
376-- XXX remove all usage of unsafeInlineIO
378-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
379-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
381{-# INLINE unsafeEqArrayN #-}
382unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
383unsafeEqArrayN Ring{..} rh A.Array{..} n =
384 let !res = unsafeInlineIO $ do
385 let rs = unsafeForeignPtrToPtr ringStart
387 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())
388 let len = ringBound `minusPtr` rh
389 r1 <- memcmp (castPtr rh) (castPtr as) (min len n)
391 then memcmp (castPtr rs) (castPtr (as `plusPtr` len))
392 (min (rh `minusPtr` rs) (n - len))
394 -- XXX enable these, check perf impact
395 -- touchForeignPtr ringStart
396 -- touchForeignPtr aStart
400-- | Byte compare the entire length of ringBuffer with the given array,
401-- starting at the supplied ringHead pointer. Returns true if the Array and
402-- the ringBuffer have identical contents.
404-- This is unsafe because the ringHead Ptr is not checked to be in range. The
405-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
407{-# INLINE unsafeEqArray #-}
408unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
409unsafeEqArray Ring{..} rh A.Array{..} =
410 let !res = unsafeInlineIO $ do
411 let rs = unsafeForeignPtrToPtr ringStart
413 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)
415 let len = ringBound `minusPtr` rh
416 r1 <- memcmp (castPtr rh) (castPtr as) len
417 r2 <- memcmp (castPtr rs) (castPtr (as `plusPtr` len))
419 -- XXX enable these, check perf impact
420 -- touchForeignPtr ringStart
421 -- touchForeignPtr aStart
425-------------------------------------------------------------------------------
427-------------------------------------------------------------------------------
429-- XXX We can unfold it into a stream and fold the stream instead.
432-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
433-- step function. This is useful to fold the items in the ring when the ring is
434-- not full. The supplied pointer is usually the end of the ring.
436-- Unsafe because the supplied Ptr is not checked to be in range.
437{-# INLINE unsafeFoldRing #-}
438unsafeFoldRing :: forall a b. Storable a
439 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
440unsafeFoldRing ptr f z Ring{..} =
441 let !res = unsafeInlineIO $ withForeignPtr ringStart $ \p ->
446 | p == q = return acc
449 go (f acc x) (PTR_NEXT(p,a)) q
451-- XXX Can we remove MonadIO here?
452withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
453withForeignPtrM fp fn = do
454 r <- fn $ unsafeForeignPtrToPtr fp
455 liftIO $ touchForeignPtr fp
458-- | Like unsafeFoldRing but with a monadic step function.
459{-# INLINE unsafeFoldRingM #-}
460unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
461 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
462unsafeFoldRingM ptr f z Ring {..} =
463 withForeignPtrM ringStart $ \x -> go z x ptr
466 | start == end = return acc
468 let !x = unsafeInlineIO $ peek start
470 go acc1 (PTR_NEXT(start,a)) end
472-- | Fold the entire length of a ring buffer starting at the supplied ringHead
473-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
474-- this would fold the ring starting from the oldest item to the newest item in
477-- Note, this will crash on ring of 0 size.
479{-# INLINE unsafeFoldRingFullM #-}
480unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
481 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
482unsafeFoldRingFullM rh f z rb@Ring {..} =
483 withForeignPtrM ringStart $ \_ -> go z rh
486 let !x = unsafeInlineIO $ peek start
488 let ptr = advance rb start
493-- | Fold @Int@ items in the ring starting at @Ptr a@. Won't fold more
494-- than the length of the ring.
496-- Note, this will crash on ring of 0 size.
498{-# INLINE unsafeFoldRingNM #-}
499unsafeFoldRingNM :: forall m a b. (MonadIO m, Storable a)
500 => Int -> Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
501unsafeFoldRingNM count rh f z rb@Ring {..} =
502 withForeignPtrM ringStart $ \_ -> go count z rh
506 go 0 acc _ = return acc
507 go !n !acc !start = do
508 let !x = unsafeInlineIO $ peek start
510 let ptr = advance rb start
511 if ptr == rh || n == 0
513 else go (n - 1) acc' ptr
515data Tuple4' a b c d = Tuple4' !a !b !c !d deriving Show
517-- | Like slidingWindow but also provides the entire ring contents as an Array.
518-- The array reflects the state of the ring after inserting the incoming
521-- IMPORTANT NOTE: The ring is mutable, therefore, the result of @(m (Array
522-- a))@ action depends on when it is executed. It does not capture the sanpshot
523-- of the ring at a particular time.
524{-# INLINE slidingWindowWith #-}
525slidingWindowWith :: forall m a b. (MonadIO m, Storable a)
526 => Int -> Fold m ((a, Maybe a), (m (Array a))) b -> Fold m a b
527slidingWindowWith n (Fold step1 initial1 extract1) = Fold step initial extract
533 then error "Window size must be > 0"
536 (rb, rh) <- liftIO $ new n
539 Partial s -> Partial $ Tuple4' rb rh (0 :: Int) s
542 toArray foldRing rb rh = do
543 arr <- liftIO $ MA.newArray n
544 let snoc' b a = liftIO $ MA.snocUnsafe b a
545 foldRing rh snoc' arr rb
547 step (Tuple4' rb rh i st) a
549 rh1 <- liftIO $ unsafeInsert rb rh a
550 liftIO $ touchForeignPtr (ringStart rb)
551 let action = toArray unsafeFoldRingM rb (PTR_NEXT(rh, a))
552 r <- step1 st ((a, Nothing), action)
555 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
558 old <- liftIO $ peek rh
559 rh1 <- liftIO $ unsafeInsert rb rh a
560 liftIO $ touchForeignPtr (ringStart rb)
561 r <- step1 st ((a, Just old), toArray unsafeFoldRingFullM rb rh1)
564 Partial s -> Partial $ Tuple4' rb rh1 (i + 1) s
567 extract (Tuple4' _ _ _ st) = extract1 st
569-- | @slidingWindow collector@ is an incremental sliding window
570-- fold that does not require all the intermediate elements in a computation.
571-- This maintains @n@ elements in the window, when a new element comes it slides
572-- out the oldest element and the new element along with the old element are
573-- supplied to the collector fold.
575-- The 'Maybe' type is for the case when initially the window is filling and
576-- there is no old element.
578{-# INLINE slidingWindow #-}
579slidingWindow :: forall m a b. (MonadIO m, Storable a)
580 => Int -> Fold m (a, Maybe a) b -> Fold m a b
581slidingWindow 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