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