git.iko.soy/streamly
on c3dcabc help sign in
streamly/src/Streamly/Internal/Data/Stream/SVar.hs
243 lines 8.6 kB history raw
1#ifdef __HADDOCK_VERSION__
2#undef INSPECTION
3#endif
5#ifdef INSPECTION
6{-# LANGUAGE TemplateHaskell #-}
7{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
8#endif
10-- |
11-- Module : Streamly.Internal.Data.Stream.SVar
12-- Copyright : (c) 2017 Harendra Kumar
13--
14-- License : BSD3
15-- Maintainer : streamly@composewell.com
16-- Stability : experimental
17-- Portability : GHC
18--
19--
20module Streamly.Internal.Data.Stream.SVar
21 ( fromSVar
22 , fromStreamVar
23 , fromProducer
24 , fromConsumer
25 , toSVar
26 , pushToFold
27 )
28where
30import Control.Exception (fromException)
31import Control.Monad (when, void)
32import Control.Monad.Catch (throwM)
33import Control.Monad.IO.Class (MonadIO(liftIO))
34import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
35import Data.Maybe (isNothing)
36import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
37import System.Mem (performMajorGC)
39import Streamly.Internal.Data.SVar
40import Streamly.Internal.Data.Stream.StreamK hiding (reverse)
42#if __GLASGOW_HASKELL__ < 810
43#ifdef INSPECTION
44import Control.Exception (Exception)
45import Control.Monad.Catch (MonadThrow)
46import Control.Monad.Trans.Control (MonadBaseControl)
47import Data.Typeable (Typeable)
48import Test.Inspection (inspect, hasNoTypeClassesExcept)
49#endif
50#endif
52-- | Pull a stream from an SVar.
53{-# NOINLINE fromStreamVar #-}
54fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
55fromStreamVar sv = MkStream $ \st yld sng stp -> do
56 list <- readOutputQ sv
57 -- Reversing the output is important to guarantee that we process the
58 -- outputs in the same order as they were generated by the constituent
59 -- streams.
60 foldStream st yld sng stp $ processEvents $ reverse list
62 where
64 allDone stp = do
65 when (svarInspectMode sv) $ do
66 t <- liftIO $ getTime Monotonic
67 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
68 liftIO $ printSVar sv "SVar Done"
69 stp
71 {-# INLINE processEvents #-}
72 processEvents [] = MkStream $ \st yld sng stp -> do
73 done <- postProcess sv
74 if done
75 then allDone stp
76 else foldStream st yld sng stp $ fromStreamVar sv
78 processEvents (ev : es) = MkStream $ \st yld sng stp -> do
79 let rest = processEvents es
80 case ev of
81 ChildYield a -> yld a rest
82 ChildStop tid e -> do
83 accountThread sv tid
84 case e of
85 Nothing -> do
86 stop <- shouldStop tid
87 if stop
88 then liftIO (cleanupSVar sv) >> allDone stp
89 else foldStream st yld sng stp rest
90 Just ex ->
91 case fromException ex of
92 Just ThreadAbort ->
93 foldStream st yld sng stp rest
94 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
95 shouldStop tid =
96 case svarStopStyle sv of
97 StopNone -> return False
98 StopAny -> return True
99 StopBy -> do
100 sid <- liftIO $ readIORef (svarStopBy sv)
101 return $ if tid == sid then True else False
103#if __GLASGOW_HASKELL__ < 810
104#ifdef INSPECTION
105-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
106-- space leak because the tuple gets allocated in every recursive call and each
107-- allocation holds on to the previous allocation. This test is to make sure
108-- that we do not use the constraint tuple type class.
109--
110inspect $ hasNoTypeClassesExcept 'fromStreamVar
111 [ ''Monad
112 , ''Applicative
113 , ''MonadThrow
114 , ''Exception
115 , ''MonadIO
116 , ''MonadBaseControl
117 , ''Typeable
118 , ''Functor
119 ]
120#endif
121#endif
123{-# INLINE fromSVar #-}
124fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
125fromSVar sv =
126 mkStream $ \st yld sng stp -> do
127 ref <- liftIO $ newIORef ()
128 _ <- liftIO $ mkWeakIORef ref hook
129 -- We pass a copy of sv to fromStreamVar, so that we know that it has
130 -- no other references, when that copy gets garbage collected "ref"
131 -- will get garbage collected and our hook will be called.
132 foldStreamShared st yld sng stp $
133 fromStream $ fromStreamVar sv{svarRef = Just ref}
134 where
136 hook = do
137 when (svarInspectMode sv) $ do
138 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
139 when (isNothing r) $
140 printSVar sv "SVar Garbage Collected"
141 cleanupSVar sv
142 -- If there are any SVars referenced by this SVar a GC will prompt
143 -- them to be cleaned up quickly.
144 when (svarInspectMode sv) performMajorGC
146-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
147-- be read back from the SVar using 'fromSVar'.
148toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
149toSVar sv m = toStreamVar sv $ toStream m
151-------------------------------------------------------------------------------
152-- Process events received by a fold consumer from a stream producer
153-------------------------------------------------------------------------------
155-- | Pull a stream from an SVar.
156{-# NOINLINE fromProducer #-}
157fromProducer :: forall m a . MonadAsync m => SVar Stream m a -> Stream m a
158fromProducer sv = mkStream $ \st yld sng stp -> do
159 list <- readOutputQ sv
160 -- Reversing the output is important to guarantee that we process the
161 -- outputs in the same order as they were generated by the constituent
162 -- streams.
163 foldStream st yld sng stp $ processEvents $ reverse list
165 where
167 allDone :: m r -> m r
168 allDone stp = do
169 when (svarInspectMode sv) $ do
170 t <- liftIO $ getTime Monotonic
171 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
172 liftIO $ printSVar sv "SVar Done"
173 sendStopToProducer sv
174 stp
176 {-# INLINE processEvents #-}
177 processEvents :: [ChildEvent a] -> Stream m a
178 processEvents [] = mkStream $ \st yld sng stp -> do
179 foldStream st yld sng stp $ fromProducer sv
181 processEvents (ev : es) = mkStream $ \_ yld _ stp -> do
182 let rest = processEvents es
183 case ev of
184 ChildYield a -> yld a rest
185 ChildStop tid e -> do
186 accountThread sv tid
187 case e of
188 Nothing -> allDone stp
189 Just _ -> error "Bug: fromProducer: received exception"
191-------------------------------------------------------------------------------
192-- Process events received by the producer thread from the consumer side
193-------------------------------------------------------------------------------
195-- XXX currently only one event is sent by a fold consumer to the stream
196-- producer. But we can potentially have multiple events e.g. the fold step can
197-- generate exception more than once and the producer can ignore those
198-- exceptions or handle them and still keep driving the fold.
199--
200{-# NOINLINE fromConsumer #-}
201fromConsumer :: MonadAsync m => SVar Stream m a -> m Bool
202fromConsumer sv = do
203 (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
204 -- Reversing the output is important to guarantee that we process the
205 -- outputs in the same order as they were generated by the constituent
206 -- streams.
207 processEvents $ reverse list
209 where
211 {-# INLINE processEvents #-}
212 processEvents [] = return False
213 processEvents (ev : _) = do
214 case ev of
215 ChildStop _ e -> do
216 case e of
217 Nothing -> return True
218 Just ex -> throwM ex
219 ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"
221-- push values to a fold worker via an SVar. Returns whether the fold is done.
222{-# INLINE pushToFold #-}
223pushToFold :: MonadAsync m => SVar Stream m a -> a -> m Bool
224pushToFold sv a = do
225 -- Check for exceptions before decrement so that we do not
226 -- block forever if the child already exited with an exception.
227 --
228 -- We avoid a race between the consumer fold sending an event and we
229 -- blocking on decrementBufferLimit by waking up the producer thread in
230 -- sendToProducer before any event is sent by the fold to the producer
231 -- stream.
232 let qref = outputQueueFromConsumer sv
233 done <- do
234 (_, n) <- liftIO $ readIORef qref
235 if (n > 0)
236 then fromConsumer sv
237 else return False
238 if done
239 then return True
240 else liftIO $ do
241 decrementBufferLimit sv
242 void $ send sv (ChildYield a)
243 return False