git.iko.soy/streamly
on fa75413 help sign in
streamly/src/Streamly/Memory/Ring.hs
194 lines 7.4 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)
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 qualified Streamly.Internal.Memory.Array.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 :: !(ForeignPtr a) -- first address
59 , ringBound :: !(Ptr a) -- first address beyond allocated memory
60 }
62-- | Create a new ringbuffer and return the ring buffer and the ringHead.
63-- Returns the ring and the ringHead, the ringHead is same as ringStart.
64{-# INLINE new #-}
65new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
66new count = do
67 let size = count * sizeOf (undefined :: a)
68 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
69 let p = unsafeForeignPtrToPtr fptr
70 return $ (Ring
71 { ringStart = fptr
72 , ringBound = p `plusPtr` size
73 }, p)
75-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
76-- array.
77{-# INLINE advance #-}
78advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
79advance Ring{..} ringHead =
80 let ptr = ringHead `plusPtr` sizeOf (undefined :: a)
81 in if ptr < ringBound
82 then ptr
83 else unsafeForeignPtrToPtr ringStart
85-- | Insert an item at the head of the ring, when the ring is full this
86-- replaces the oldest item in the ring with the new item. This is unsafe
87-- beause ringHead supplied is not verified to be within the Ring. Also,
88-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
89{-# INLINE unsafeInsert #-}
90unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
91unsafeInsert rb ringHead newVal = do
92 poke ringHead newVal
93 -- touchForeignPtr (ringStart rb)
94 return $ advance rb ringHead
96-- XXX remove all usage of A.unsafeInlineIO
97--
98-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
99-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
100-- be in range.
101{-# INLINE unsafeEqArrayN #-}
102unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
103unsafeEqArrayN Ring{..} rh A.Array{..} n =
104 let !res = A.unsafeInlineIO $ do
105 let rs = unsafeForeignPtrToPtr ringStart
106 let as = unsafeForeignPtrToPtr aStart
107 assert (aBound `minusPtr` as >= ringBound `minusPtr` rs) (return ())
108 let len = ringBound `minusPtr` rh
109 r1 <- A.memcmp (castPtr rh) (castPtr as) (min len n)
110 r2 <- if n > len
111 then A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
112 (min (rh `minusPtr` rs) (n - len))
113 else return True
114 -- XXX enable these, check perf impact
115 -- touchForeignPtr ringStart
116 -- touchForeignPtr aStart
117 return (r1 && r2)
118 in res
120-- | Byte compare the entire length of ringBuffer with the given array,
121-- starting at the supplied ringHead pointer. Returns true if the Array and
122-- the ringBuffer have identical contents.
123--
124-- This is unsafe because the ringHead Ptr is not checked to be in range. The
125-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
126-- ARE NOT CHECKED.
127{-# INLINE unsafeEqArray #-}
128unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
129unsafeEqArray Ring{..} rh A.Array{..} =
130 let !res = A.unsafeInlineIO $ do
131 let rs = unsafeForeignPtrToPtr ringStart
132 let as = unsafeForeignPtrToPtr aStart
133 assert (aBound `minusPtr` as >= ringBound `minusPtr` rs)
134 (return ())
135 let len = ringBound `minusPtr` rh
136 r1 <- A.memcmp (castPtr rh) (castPtr as) len
137 r2 <- A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
138 (rh `minusPtr` rs)
139 -- XXX enable these, check perf impact
140 -- touchForeignPtr ringStart
141 -- touchForeignPtr aStart
142 return (r1 && r2)
143 in res
145-- XXX use MonadIO
146--
147-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
148-- step function. This is useful to fold the items in the ring when the ring is
149-- not full. The supplied pointer is usually the end of the ring.
150--
151-- Unsafe because the supplied Ptr is not checked to be in range.
152{-# INLINE unsafeFoldRing #-}
153unsafeFoldRing :: forall a b. Storable a
154 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
155unsafeFoldRing ptr f z Ring{..} =
156 let !res = A.unsafeInlineIO $ withForeignPtr ringStart $ \p ->
157 go z p ptr
158 in res
159 where
160 go !acc !p !q
161 | p == q = return acc
162 | otherwise = do
163 x <- peek p
164 go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q
166-- | Like unsafeFoldRing but with a monadic step function.
167{-# INLINE unsafeFoldRingM #-}
168unsafeFoldRingM :: forall m a b. (Monad m, Storable a)
169 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
170unsafeFoldRingM ptr f z Ring{..} = go z (unsafeForeignPtrToPtr ringStart) ptr
171 where
172 go !acc !start !end
173 | start == end = return acc
174 | otherwise = do
175 let !x = A.unsafeInlineIO $ peek start
176 acc' <- f acc x
177 go acc' (start `plusPtr` sizeOf (undefined :: a)) end
179-- | Fold the entire length of a ring buffer starting at the supplied ringHead
180-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
181-- this would fold the ring starting from the oldest item to the newest item in
182-- the ring.
183{-# INLINE unsafeFoldRingFullM #-}
184unsafeFoldRingFullM :: forall m a b. (Monad m, Storable a)
185 => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
186unsafeFoldRingFullM rh f z rb@Ring{..} = go z rh
187 where
188 go !acc !start = do
189 let !x = A.unsafeInlineIO $ peek start
190 acc' <- f acc x
191 let ptr = advance rb start
192 if ptr == rh
193 then return acc'
194 else go acc' ptr