git.iko.soy/streamly
on aaa775b help sign in
streamly/src/Streamly/Internal/Data/Stream/SVar/Generate.hs
291 lines 10 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.Generate
12-- Copyright : (c) 2017 Composewell Technologies
13-- License : BSD-3-Clause
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
16-- Portability : GHC
17--
18--
19module Streamly.Internal.Data.Stream.SVar.Generate
20 (
21 -- * Write to SVar
22 toSVar
24 -- * Read from SVar
25 -- $concurrentEval
26 , fromSVar
27 , fromSVarD
28 -- , fromStreamVar
29 )
30where
32#include "inline.hs"
34import Control.Exception (fromException)
35import Control.Monad (when)
36import Control.Monad.Catch (throwM)
37import Control.Monad.IO.Class (MonadIO(liftIO))
38import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
39import Data.Maybe (isNothing)
40import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
41import Streamly.Internal.Data.Stream.Serial (SerialT(..))
42import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
43import System.Mem (performMajorGC)
45import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
46import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
48import Streamly.Internal.Data.SVar
50#if __GLASGOW_HASKELL__ < 810
51#ifdef INSPECTION
52import Control.Exception (Exception)
53import Control.Monad.Catch (MonadThrow)
54import Control.Monad.Trans.Control (MonadBaseControl)
55import Data.Typeable (Typeable)
56import Test.Inspection (inspect, hasNoTypeClassesExcept)
57#endif
58#endif
60------------------------------------------------------------------------------
61-- Generating streams from SVar
62------------------------------------------------------------------------------
64-- $concurrentEval
65--
66-- Usually the SVar is used to concurrently evaluate multiple actions in a
67-- stream using many worker threads that push the results to the SVar and a
68-- single puller that pulls them from SVar generating the evaluated stream.
69--
70-- @
71-- input stream
72-- |
73-- <-----------------|<--------worker
74-- | exceptions |
75-- output stream <------SVar<------worker
76-- |
77-- |<--------worker
78--
79-- @
80--
81-- The puller itself schedules the worker threads based on demand.
82-- Exceptions are propagated from the worker threads to the puller.
84-------------------------------------------------------------------------------
85-- Write a stream to an SVar
86-------------------------------------------------------------------------------
88-- XXX this errors out for Parallel/Ahead SVars
89-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
90-- be read back from the SVar using 'fromSVar'.
91toSVar :: MonadAsync m => SVar SerialT m a -> SerialT m a -> m ()
92toSVar sv m = do
93 runIn <- askRunInIO
94 liftIO $ enqueue sv (runIn, m)
95 done <- allThreadsDone sv
96 -- XXX This is safe only when called from the consumer thread or when no
97 -- consumer is present. There may be a race if we are not running in the
98 -- consumer thread.
99 -- XXX do this only if the work queue is not empty. The work may have been
100 -- carried out by existing workers.
101 when done $
102 case yieldRateInfo sv of
103 Nothing -> pushWorker 0 sv
104 Just _ -> pushWorker 1 sv
106-------------------------------------------------------------------------------
107-- Read a stream from an SVar
108-------------------------------------------------------------------------------
110-- | Pull a stream from an SVar.
111{-# NOINLINE fromStreamVar #-}
112fromStreamVar :: MonadAsync m => SVar K.Stream m a -> K.Stream m a
113fromStreamVar sv = K.MkStream $ \st yld sng stp -> do
114 list <- readOutputQ sv
115 -- Reversing the output is important to guarantee that we process the
116 -- outputs in the same order as they were generated by the constituent
117 -- streams.
118 K.foldStream st yld sng stp $ processEvents $ reverse list
120 where
122 allDone stp = do
123 when (svarInspectMode sv) $ do
124 t <- liftIO $ getTime Monotonic
125 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
126 liftIO $ printSVar sv "SVar Done"
127 stp
129 {-# INLINE processEvents #-}
130 processEvents [] = K.MkStream $ \st yld sng stp -> do
131 done <- postProcess sv
132 if done
133 then allDone stp
134 else K.foldStream st yld sng stp $ fromStreamVar sv
136 processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
137 let rest = processEvents es
138 case ev of
139 ChildYield a -> yld a rest
140 ChildStop tid e -> do
141 accountThread sv tid
142 case e of
143 Nothing -> do
144 stop <- shouldStop tid
145 if stop
146 then liftIO (cleanupSVar sv) >> allDone stp
147 else K.foldStream st yld sng stp rest
148 Just ex ->
149 case fromException ex of
150 Just ThreadAbort ->
151 K.foldStream st yld sng stp rest
152 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
153 shouldStop tid =
154 case svarStopStyle sv of
155 StopNone -> return False
156 StopAny -> return True
157 StopBy -> do
158 sid <- liftIO $ readIORef (svarStopBy sv)
159 return $ tid == sid
161#if __GLASGOW_HASKELL__ < 810
162#ifdef INSPECTION
163-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
164-- space leak because the tuple gets allocated in every recursive call and each
165-- allocation holds on to the previous allocation. This test is to make sure
166-- that we do not use the constraint tuple type class.
167--
168inspect $ hasNoTypeClassesExcept 'fromStreamVar
169 [ ''Monad
170 , ''Applicative
171 , ''MonadThrow
172 , ''Exception
173 , ''MonadIO
174 , ''MonadBaseControl
175 , ''Typeable
176 , ''Functor
177 ]
178#endif
179#endif
181-- | Generate a stream from an SVar. An unevaluated stream can be pushed to an
182-- SVar using 'toSVar'. As we pull a stream from the SVar the input stream
183-- gets evaluated concurrently. The evaluation depends on the SVar style and
184-- the configuration parameters e.g. using the maxBuffer/maxThreads
185-- combinators.
186--
187{-# INLINE fromSVar #-}
188fromSVar :: MonadAsync m => SVar K.Stream m a -> SerialT m a
189fromSVar sv =
190 SerialT $ K.mkStream $ \st yld sng stp -> do
191 ref <- liftIO $ newIORef ()
192 _ <- liftIO $ mkWeakIORef ref hook
193 -- We pass a copy of sv to fromStreamVar, so that we know that it has
194 -- no other references, when that copy gets garbage collected "ref"
195 -- will get garbage collected and our hook will be called.
196 K.foldStreamShared st yld sng stp $
197 fromStreamVar sv{svarRef = Just ref}
198 where
200 hook = do
201 when (svarInspectMode sv) $ do
202 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
203 when (isNothing r) $
204 printSVar sv "SVar Garbage Collected"
205 cleanupSVar sv
206 -- If there are any SVars referenced by this SVar a GC will prompt
207 -- them to be cleaned up quickly.
208 when (svarInspectMode sv) performMajorGC
210data FromSVarState t m a =
211 FromSVarInit
212 | FromSVarRead (SVar t m a)
213 | FromSVarLoop (SVar t m a) [ChildEvent a]
214 | FromSVarDone (SVar t m a)
216-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
217--
218{-# INLINE_NORMAL fromSVarD #-}
219fromSVarD :: (MonadAsync m) => SVar t m a -> D.Stream m a
220fromSVarD svar = D.Stream step FromSVarInit
221 where
223 {-# INLINE_LATE step #-}
224 step _ FromSVarInit = do
225 ref <- liftIO $ newIORef ()
226 _ <- liftIO $ mkWeakIORef ref hook
227 -- when this copy of svar gets garbage collected "ref" will get
228 -- garbage collected and our GC hook will be called.
229 let sv = svar{svarRef = Just ref}
230 return $ D.Skip (FromSVarRead sv)
232 where
234 {-# NOINLINE hook #-}
235 hook = do
236 when (svarInspectMode svar) $ do
237 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
238 when (isNothing r) $
239 printSVar svar "SVar Garbage Collected"
240 cleanupSVar svar
241 -- If there are any SVars referenced by this SVar a GC will prompt
242 -- them to be cleaned up quickly.
243 when (svarInspectMode svar) performMajorGC
245 step _ (FromSVarRead sv) = do
246 list <- readOutputQ sv
247 -- Reversing the output is important to guarantee that we process the
248 -- outputs in the same order as they were generated by the constituent
249 -- streams.
250 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
252 step _ (FromSVarLoop sv []) = do
253 done <- postProcess sv
254 return $ D.Skip $ if done
255 then FromSVarDone sv
256 else FromSVarRead sv
258 step _ (FromSVarLoop sv (ev : es)) = do
259 case ev of
260 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
261 ChildStop tid e -> do
262 accountThread sv tid
263 case e of
264 Nothing -> do
265 stop <- shouldStop tid
266 if stop
267 then do
268 liftIO (cleanupSVar sv)
269 return $ D.Skip (FromSVarDone sv)
270 else return $ D.Skip (FromSVarLoop sv es)
271 Just ex ->
272 case fromException ex of
273 Just ThreadAbort ->
274 return $ D.Skip (FromSVarLoop sv es)
275 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
276 where
278 shouldStop tid =
279 case svarStopStyle sv of
280 StopNone -> return False
281 StopAny -> return True
282 StopBy -> do
283 sid <- liftIO $ readIORef (svarStopBy sv)
284 return $ tid == sid
286 step _ (FromSVarDone sv) = do
287 when (svarInspectMode sv) $ do
288 t <- liftIO $ getTime Monotonic
289 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
290 liftIO $ printSVar sv "SVar Done"
291 return D.Stop