@@ -9,50 +9,156 @@
10module Streamly.Internal.Data.Ring
10module Streamly.Internal.Data.Ring
11 ( Ring(..)
11 ( Ring(..)
12 , createRing
14 , createRing
13− , unsafeInsertRing
15+ , writeLastN
19+ , unsafeInsertRingWith
16−import Control.Monad (when)
28+import Control.Monad.IO.Class (liftIO, MonadIO)
17−import Data.IORef (newIORef, readIORef, writeIORef, IORef)
29+import Streamly.Internal.Data.Stream.StreamD.Type (Stream)
30+import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
31+import Streamly.Internal.Data.Fold.Type (Fold(..))
18import Streamly.Internal.Data.Array.Generic.Mut.Type
32import Streamly.Internal.Data.Array.Generic.Mut.Type
19 ( MutArray(..)
33 ( MutArray(..)
22 , putIndexUnsafe
36 , putIndexUnsafe
39+-- import qualified Streamly.Internal.Data.Stream.StreamD.Type as Stream
40+import qualified Streamly.Internal.Data.Fold.Type as Fold
42+-- XXX Use MutableArray rather than keeping a MutArray here.
25data Ring a = Ring
43data Ring a = Ring
26 { ringArr :: MutArray a
44 { ringArr :: MutArray a
27− -- XXX This could be an unboxed IO ref
45+ -- XXX We can keep the current fill amount, Or we can keep a count of total
28− -- If we want to rely on an external Int count using mod ringMax as index
46+ -- elements inserted and compute ring head as well using mod on that,
29− -- into the ring then we do not need to maintain this, assuming the counter
47+ -- assuming it won't overflow. But mod could be expensive.
30− -- would not overflow.
31− , ringHead :: IORef Int -- current index to be over-written
48+ , ringHead :: !Int -- current index to be over-written
32− , ringMax :: !Int -- first index beyond allocated memory
49+ , ringMax :: !Int -- first index beyond allocated memory
52+-------------------------------------------------------------------------------
54+-------------------------------------------------------------------------------
35-- XXX If we align the ringMax to nearest power of two then computation of the
56-- XXX If we align the ringMax to nearest power of two then computation of the
36-- index to write could be cheaper.
57-- index to write could be cheaper.
37{-# INLINE createRing #-}
58{-# INLINE createRing #-}
38−createRing :: Int -> IO (Ring a)
59+createRing :: MonadIO m => Int -> m (Ring a)
39−createRing count = do
60+createRing count = liftIO $ do
40 arr <- new count
61 arr <- new count
41 arr1 <- uninit arr count
62 arr1 <- uninit arr count
43 return (Ring
63 return (Ring
44 { ringArr = arr1
64 { ringArr = arr1
45− , ringHead = ref
65+ , ringHead = 0
46 , ringMax = count
66 , ringMax = count
70+{-# INLINE writeLastN #-}
71+writeLastN :: MonadIO m => Int -> Fold m a (Ring a)
72+writeLastN n = Fold step initial extract
78+ then Fold.Done <$> createRing 0
81+ return $ Fold.Partial $ Tuple' rb (0 :: Int)
83+ step (Tuple' rb cnt) x = do
84+ rh1 <- liftIO $ unsafeInsertRingWith rb x
85+ return $ Fold.Partial $ Tuple' (rb {ringHead = rh1}) (cnt + 1)
87+ extract (Tuple' rb@Ring{..} cnt) =
90+ then Ring ringArr 0 ringHead
93+-------------------------------------------------------------------------------
95+-------------------------------------------------------------------------------
49-- XXX This is safe
97-- XXX This is safe
50−{-# INLINE unsafeInsertRing #-}
98+-- Take the ring head and return the new ring head.
51−unsafeInsertRing :: Ring a -> a -> IO ()
99+{-# INLINE unsafeInsertRingWith #-}
52−unsafeInsertRing Ring{..} x = do
100+unsafeInsertRingWith :: Ring a -> a -> IO Int
101+unsafeInsertRingWith Ring{..} x = do
102+ assertM(ringMax >= 1)
103+ assertM(ringHead < ringMax)
104+ putIndexUnsafe ringHead ringArr x
105+ let rh1 = ringHead + 1
106+ next = if rh1 == ringMax then 0 else rh1
109+-- | Move the ring head clockwise (+ve adj) or counter clockwise (-ve adj) by
112+seek :: MonadIO m => Int -> Ring a -> m (Ring a)
113+seek adj rng@Ring{..}
114+ | ringMax > 0 = liftIO $ do
115+ -- XXX try avoiding mod when in bounds
116+ let idx1 = ringHead + adj
117+ next = mod idx1 ringMax
118+ return $ Ring ringArr next ringMax
119+ | otherwise = pure rng
121+-------------------------------------------------------------------------------
123+-------------------------------------------------------------------------------
125+-- | @toMutArray rignHeadAdjustment lengthToRead ring@.
126+-- Convert the ring into a boxed mutable array. Note that the returned MutArray
127+-- may share the same underlying memory as the Ring.
128+{-# INLINE toMutArray #-}
129+toMutArray :: MonadIO m => Int -> Int -> Ring a -> m (MutArray a)
130+toMutArray adj n Ring{..} = do
53− when (ringMax > 0) $ do
131+ let len = min ringMax n
132+ let idx = mod (ringHead + adj) ringMax
136+ -- putSliceUnsafe ringArr idx arr1 0 len
137+ return $ ringArr { arrStart = idx, arrLen = len }
139+ -- XXX Just swap the elements in the existing ring and return the
140+ -- same array without reallocation.
141+ arr <- liftIO $ new len
142+ arr1 <- uninit arr len
143+ putSliceUnsafe ringArr idx arr1 0 (ringMax - idx)
144+ putSliceUnsafe ringArr 0 arr1 (ringMax - idx) (end - ringMax)
147+-- This would be theoretically slower than toMutArray because of a branch
148+-- introduced for each element in the second half of the ring.
150+-- | Seek by n and then read the entire ring. Use 'take' on the stream to
151+-- restrict the reads.
152+toStreamWith :: Int -> Ring a -> Stream m a
153+toStreamWith = undefined
155+toStreamWith n Ring{..}
156+ | ringMax > 0 = concatEffect $ liftIO $ do
54 idx <- readIORef ringHead
157 idx <- readIORef ringHead
55− putIndexUnsafe idx ringArr x
56− let idx1 = idx + 1
158+ let idx1 = idx + adj
57− next = if idx1 == ringMax then 0 else idx1
159+ next = mod idx1 ringMax
58− writeIORef ringHead next
160+ s1 = undefined -- stream initial slice
161+ s2 = undefined -- stream next slice
162+ return (s1 `Stream.append` s2)
163+ | otherwise = Stream.nil