git.iko.soy/streamly
on e49a37d help sign in
streamly/src/Streamly/Memory/Ring.hs
224 lines 8.5 kB history raw
1-- |
2-- Module : Streamly.Memory.Ring
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD3
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
10module Streamly.Memory.Ring
11 ( Ring(..)
13 -- * Construction
14 , new
15 , advance
16 , moveBy
17 , startOf
19 -- * Modification
20 , unsafeInsert
22 -- * Folds
23 , unsafeFoldRing
24 , unsafeFoldRingM
25 , unsafeFoldRingFullM
27 -- * Fast Byte Comparisons
28 , unsafeEqArray
29 , unsafeEqArrayN
30 ) where
32import Control.Exception (assert)
33import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
34import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
35import Foreign.Ptr (plusPtr, minusPtr, castPtr)
36import Foreign.Storable (Storable(..))
37import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
38import GHC.Ptr (Ptr(..))
39import Prelude hiding (length, concat)
41import Control.Monad.IO.Class (MonadIO(..))
43import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as A
45-- | A ring buffer is a mutable array of fixed size. Initially the array is
46-- empty, with ringStart pointing at the start of allocated memory. We call the
47-- next location to be written in the ring as ringHead. Initially ringHead ==
48-- ringStart. When the first item is added, ringHead points to ringStart +
49-- sizeof item. When the buffer becomes full ringHead would wrap around to
50-- ringStart. When the buffer is full, ringHead always points at the oldest
51-- item in the ring and the newest item added always overwrites the oldest
52-- item.
53--
54-- When using it we should keep in mind that a ringBuffer is a mutable data
55-- structure. We should not leak out references to it for immutable use.
56--
57data Ring a = Ring
58 { ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
59 , ringBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory
60 }
62-- | Get the first address of the ring as a pointer.
63startOf :: Ring a -> Ptr a
64startOf = unsafeForeignPtrToPtr . ringStart
66-- | Create a new ringbuffer and return the ring buffer and the ringHead.
67-- Returns the ring and the ringHead, the ringHead is same as ringStart.
68{-# INLINE new #-}
69new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
70new count = do
71 let size = count * sizeOf (undefined :: a)
72 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
73 let p = unsafeForeignPtrToPtr fptr
74 return (Ring
75 { ringStart = fptr
76 , ringBound = p `plusPtr` size
77 }, p)
79-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
80-- array.
81{-# INLINE advance #-}
82advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
83advance Ring{..} ringHead =
84 let ptr = ringHead `plusPtr` sizeOf (undefined :: a)
85 in if ptr < ringBound
86 then ptr
87 else unsafeForeignPtrToPtr ringStart
89-- | Move the ringHead by n items. The direction depends on the sign on whether
90-- n is positive or negative. Wrap around if we hit the beginning or end of the
91-- array.
92{-# INLINE moveBy #-}
93moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
94moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead
96 where
98 elemSize = sizeOf (undefined :: a)
99 ringStartPtr = unsafeForeignPtrToPtr ringStart
100 lenInBytes = ringBound `minusPtr` ringStartPtr
101 offInBytes = ringHead `minusPtr` ringStartPtr
102 len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize
103 off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize
104 advanceFromHead = (off + by `mod` len) * elemSize
106-- | Insert an item at the head of the ring, when the ring is full this
107-- replaces the oldest item in the ring with the new item. This is unsafe
108-- beause ringHead supplied is not verified to be within the Ring. Also,
109-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
110{-# INLINE unsafeInsert #-}
111unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
112unsafeInsert rb ringHead newVal = do
113 poke ringHead newVal
114 -- touchForeignPtr (ringStart rb)
115 return $ advance rb ringHead
117-- XXX remove all usage of A.unsafeInlineIO
118--
119-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
120-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
121-- be in range.
122{-# INLINE unsafeEqArrayN #-}
123unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
124unsafeEqArrayN Ring{..} rh A.Array{..} n =
125 let !res = A.unsafeInlineIO $ do
126 let rs = unsafeForeignPtrToPtr ringStart
127 as = unsafeForeignPtrToPtr aStart
128 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())
129 let len = ringBound `minusPtr` rh
130 r1 <- A.memcmp (castPtr rh) (castPtr as) (min len n)
131 r2 <- if n > len
132 then A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
133 (min (rh `minusPtr` rs) (n - len))
134 else return True
135 -- XXX enable these, check perf impact
136 -- touchForeignPtr ringStart
137 -- touchForeignPtr aStart
138 return (r1 && r2)
139 in res
141-- | Byte compare the entire length of ringBuffer with the given array,
142-- starting at the supplied ringHead pointer. Returns true if the Array and
143-- the ringBuffer have identical contents.
144--
145-- This is unsafe because the ringHead Ptr is not checked to be in range. The
146-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
147-- ARE NOT CHECKED.
148{-# INLINE unsafeEqArray #-}
149unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
150unsafeEqArray Ring{..} rh A.Array{..} =
151 let !res = A.unsafeInlineIO $ do
152 let rs = unsafeForeignPtrToPtr ringStart
153 let as = unsafeForeignPtrToPtr aStart
154 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)
155 (return ())
156 let len = ringBound `minusPtr` rh
157 r1 <- A.memcmp (castPtr rh) (castPtr as) len
158 r2 <- A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
159 (rh `minusPtr` rs)
160 -- XXX enable these, check perf impact
161 -- touchForeignPtr ringStart
162 -- touchForeignPtr aStart
163 return (r1 && r2)
164 in res
166-- XXX use MonadIO
167--
168-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
169-- step function. This is useful to fold the items in the ring when the ring is
170-- not full. The supplied pointer is usually the end of the ring.
171--
172-- Unsafe because the supplied Ptr is not checked to be in range.
173{-# INLINE unsafeFoldRing #-}
174unsafeFoldRing :: forall a b. Storable a
175 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
176unsafeFoldRing ptr f z Ring{..} =
177 let !res = A.unsafeInlineIO $ withForeignPtr ringStart $ \p ->
178 go z p ptr
179 in res
180 where
181 go !acc !p !q
182 | p == q = return acc
183 | otherwise = do
184 x <- peek p
185 go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q
187-- XXX Can we remove MonadIO here?
188withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
189withForeignPtrM fp fn = do
190 r <- fn $ unsafeForeignPtrToPtr fp
191 liftIO $ touchForeignPtr fp
192 return r
194-- | Like unsafeFoldRing but with a monadic step function.
195{-# INLINE unsafeFoldRingM #-}
196unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
197 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
198unsafeFoldRingM ptr f z Ring {..} =
199 withForeignPtrM ringStart $ \x -> go z x ptr
200 where
201 go !acc !start !end
202 | start == end = return acc
203 | otherwise = do
204 let !x = A.unsafeInlineIO $ peek start
205 acc' <- f acc x
206 go acc' (start `plusPtr` sizeOf (undefined :: a)) end
208-- | Fold the entire length of a ring buffer starting at the supplied ringHead
209-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
210-- this would fold the ring starting from the oldest item to the newest item in
211-- the ring.
212{-# INLINE unsafeFoldRingFullM #-}
213unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
214 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
215unsafeFoldRingFullM rh f z rb@Ring {..} =
216 withForeignPtrM ringStart $ \_ -> go z rh
217 where
218 go !acc !start = do
219 let !x = A.unsafeInlineIO $ peek start
220 acc' <- f acc x
221 let ptr = advance rb start
222 if ptr == rh
223 then return acc'
224 else go acc' ptr