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