git.iko.soy/streamly
on 745265e help sign in
streamly/src/Streamly/Memory/Ring.hs
200 lines 7.6 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
16 -- * Modification
17 , unsafeInsert
19 -- * Folds
20 , unsafeFoldRing
21 , unsafeFoldRingM
22 , unsafeFoldRingFullM
24 -- * Fast Byte Comparisons
25 , unsafeEqArray
26 , unsafeEqArrayN
27 ) where
29import Control.Exception (assert)
30import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
31import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
32import Foreign.Ptr (plusPtr, minusPtr, castPtr)
33import Foreign.Storable (Storable(..))
34import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
35import GHC.Ptr (Ptr(..))
36import Prelude hiding (length, concat)
38import Control.Monad.IO.Class (MonadIO(..))
40import qualified Streamly.Internal.Memory.Array.Types as A
42-- | A ring buffer is a mutable array of fixed size. Initially the array is
43-- empty, with ringStart pointing at the start of allocated memory. We call the
44-- next location to be written in the ring as ringHead. Initially ringHead ==
45-- ringStart. When the first item is added, ringHead points to ringStart +
46-- sizeof item. When the buffer becomes full ringHead would wrap around to
47-- ringStart. When the buffer is full, ringHead always points at the oldest
48-- item in the ring and the newest item added always overwrites the oldest
49-- item.
50--
51-- When using it we should keep in mind that a ringBuffer is a mutable data
52-- structure. We should not leak out references to it for immutable use.
53--
54data Ring a = Ring
55 { ringStart :: !(ForeignPtr a) -- first address
56 , ringBound :: !(Ptr a) -- first address beyond allocated memory
57 }
59-- | Create a new ringbuffer and return the ring buffer and the ringHead.
60-- Returns the ring and the ringHead, the ringHead is same as ringStart.
61{-# INLINE new #-}
62new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
63new count = do
64 let size = count * sizeOf (undefined :: a)
65 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
66 let p = unsafeForeignPtrToPtr fptr
67 return (Ring
68 { ringStart = fptr
69 , ringBound = p `plusPtr` size
70 }, p)
72-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
73-- array.
74{-# INLINE advance #-}
75advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
76advance Ring{..} ringHead =
77 let ptr = ringHead `plusPtr` sizeOf (undefined :: a)
78 in if ptr < ringBound
79 then ptr
80 else unsafeForeignPtrToPtr ringStart
82-- | Insert an item at the head of the ring, when the ring is full this
83-- replaces the oldest item in the ring with the new item. This is unsafe
84-- beause ringHead supplied is not verified to be within the Ring. Also,
85-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
86{-# INLINE unsafeInsert #-}
87unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
88unsafeInsert rb ringHead newVal = do
89 poke ringHead newVal
90 -- touchForeignPtr (ringStart rb)
91 return $ advance rb ringHead
93-- XXX remove all usage of A.unsafeInlineIO
94--
95-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
96-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
97-- be in range.
98{-# INLINE unsafeEqArrayN #-}
99unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
100unsafeEqArrayN Ring{..} rh A.Array{..} n =
101 let !res = A.unsafeInlineIO $ do
102 let rs = unsafeForeignPtrToPtr ringStart
103 as = unsafeForeignPtrToPtr aStart
104 assert (aBound `minusPtr` as >= ringBound `minusPtr` rs) (return ())
105 let len = ringBound `minusPtr` rh
106 r1 <- A.memcmp (castPtr rh) (castPtr as) (min len n)
107 r2 <- if n > len
108 then A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
109 (min (rh `minusPtr` rs) (n - len))
110 else return True
111 -- XXX enable these, check perf impact
112 -- touchForeignPtr ringStart
113 -- touchForeignPtr aStart
114 return (r1 && r2)
115 in res
117-- | Byte compare the entire length of ringBuffer with the given array,
118-- starting at the supplied ringHead pointer. Returns true if the Array and
119-- the ringBuffer have identical contents.
120--
121-- This is unsafe because the ringHead Ptr is not checked to be in range. The
122-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
123-- ARE NOT CHECKED.
124{-# INLINE unsafeEqArray #-}
125unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
126unsafeEqArray Ring{..} rh A.Array{..} =
127 let !res = A.unsafeInlineIO $ do
128 let rs = unsafeForeignPtrToPtr ringStart
129 let as = unsafeForeignPtrToPtr aStart
130 assert (aBound `minusPtr` as >= ringBound `minusPtr` rs)
131 (return ())
132 let len = ringBound `minusPtr` rh
133 r1 <- A.memcmp (castPtr rh) (castPtr as) len
134 r2 <- A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
135 (rh `minusPtr` rs)
136 -- XXX enable these, check perf impact
137 -- touchForeignPtr ringStart
138 -- touchForeignPtr aStart
139 return (r1 && r2)
140 in res
142-- XXX use MonadIO
143--
144-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
145-- step function. This is useful to fold the items in the ring when the ring is
146-- not full. The supplied pointer is usually the end of the ring.
147--
148-- Unsafe because the supplied Ptr is not checked to be in range.
149{-# INLINE unsafeFoldRing #-}
150unsafeFoldRing :: forall a b. Storable a
151 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
152unsafeFoldRing ptr f z Ring{..} =
153 let !res = A.unsafeInlineIO $ withForeignPtr ringStart $ \p ->
154 go z p ptr
155 in res
156 where
157 go !acc !p !q
158 | p == q = return acc
159 | otherwise = do
160 x <- peek p
161 go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q
163-- XXX Can we remove MonadIO here?
164withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
165withForeignPtrM fp fn = do
166 r <- fn $ unsafeForeignPtrToPtr fp
167 liftIO $ touchForeignPtr fp
168 return r
170-- | Like unsafeFoldRing but with a monadic step function.
171{-# INLINE unsafeFoldRingM #-}
172unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
173 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
174unsafeFoldRingM ptr f z Ring {..} =
175 withForeignPtrM ringStart $ \x -> go z x ptr
176 where
177 go !acc !start !end
178 | start == end = return acc
179 | otherwise = do
180 let !x = A.unsafeInlineIO $ peek start
181 acc' <- f acc x
182 go acc' (start `plusPtr` sizeOf (undefined :: a)) end
184-- | Fold the entire length of a ring buffer starting at the supplied ringHead
185-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
186-- this would fold the ring starting from the oldest item to the newest item in
187-- the ring.
188{-# INLINE unsafeFoldRingFullM #-}
189unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
190 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
191unsafeFoldRingFullM rh f z rb@Ring {..} =
192 withForeignPtrM ringStart $ \_ -> go z rh
193 where
194 go !acc !start = do
195 let !x = A.unsafeInlineIO $ peek start
196 acc' <- f acc x
197 let ptr = advance rb start
198 if ptr == rh
199 then return acc'
200 else go acc' ptr